spider_work/myslabs_spider/ms_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/5/12 15:43
import time
import inspect
import schedule
import requests
import user_agent
from loguru import logger
from parsel import Selector
from datetime import datetime
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

logger.remove()
logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="7 day")


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        log = retry_state.args[0]  # 获取传入的 logger
    else:
        log = logger  # 使用全局 logger

    if retry_state.outcome.failed:
        log.warning(
            f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_proxys(log):
    """
    获取代理
    :return: 代理
    """
    tunnel = "x371.kdltps.com:15818"
    kdl_username = "t13753103189895"
    kdl_password = "o0yefv6z"
    try:
        proxies = {
            "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel},
            "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}
        }
        return proxies
    except Exception as e:
        log.error(f"Error getting proxy: {e}")
        raise e


def conversion_date(log, date_str):
    """
    日期格式转换  将 "May 1, 2026" 格式转换为 "YYYY-MM-DD"
    :param log: logger对象
    :param date_str: 日期字符串，如 "May 1, 2026"
    :return: 格式化后的日期字符串 "YYYY-MM-DD"
    """
    if not date_str or not isinstance(date_str, str):
        return None

    try:
        date_str = date_str.strip()
        date_object = datetime.strptime(date_str, "%B %d, %Y")
        return date_object.strftime("%Y-%m-%d")
    except ValueError:
        try:
            date_object = datetime.strptime(date_str, "%b %d, %Y")
            return date_object.strftime("%Y-%m-%d")
        except ValueError:
            log.error(f"日期格式转换失败: {date_str}")
            return None


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_single_page(log, page, sql_pool):
    """
    获取单页数据
    :param log: logger对象
    :param page: 页码
    :param sql_pool: MySQL连接池
    :return: 该页数据条数
    """
    log.info(f"正在爬取第 {page} 页数据............")

    headers = {
        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
        "referer": "https://myslabs.com/browse/archive/",
        "user-agent": user_agent.generate_user_agent()
    }

    url = "https://myslabs.com/browse/archive/"
    params = {
        # "page": "2"
        "page": f"{page}"
    }
    # response = requests.get(url, headers=headers, cookies=cookies, params=params)
    response = requests.get(url, headers=headers, params=params, timeout=22)
    response.raise_for_status()

    selector = Selector(response.text)
    tag_div_list = selector.xpath('//div[@class="slabs_cont padded-primary-container"]/div')

    info_list = []
    for tag_div in tag_div_list:
        title = tag_div.xpath('./a/div/text()').get()
        price = tag_div.xpath('.//div[@class="item-price"]/text()').get()
        price = price.replace('$', '').strip() if price else None

        release_date = tag_div.xpath('.//small/text()').get()  # May 1, 2026
        release_date = conversion_date(log, release_date)

        front_img = tag_div.xpath(
            './/div[@class="slab_item_img"]//img/@src | .//div[@class="slab_item_img"]//img/@data-src').get()
        # print(front_img)
        front_img = front_img.split('?')[0] if front_img and '?' in front_img else front_img
        back_img = front_img.replace("1.png", "2.png") if front_img else None
        # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_1.png"
        # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_2.png"
        # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_1.png?width=360&height=610&quality=85"
        detail_url = tag_div.xpath('./a/@href').get()
        # detail_url = "https://myslabs.com" + detail_url if detail_url else None
        detail_id = detail_url.split("/")[-2] if detail_url else 1

        data_dict = {
            "title": title.strip() if title else None,
            "price": price.replace(',', '') if price else None,
            "release_date": release_date,
            "front_img": front_img,
            "back_img": back_img,
            "detail_id": int(detail_id) if detail_id else 1
        }
        # print(data_dict)
        info_list.append(data_dict)

    # 保存数据
    sql_pool.insert_many(table="slabs_record", data_list=info_list, ignore=True)

    return len(tag_div_list)


def get_sold_list(log, sql_pool):
    page = 1
    max_page = 10
    while page <= max_page:
        try:
            single_count = get_single_page(log, page, sql_pool)

            if single_count < 72:
                log.warning(f"第 {page} 页数据不足 72 条，结束爬取")
                break

        except Exception as e:
            log.error(f"爬取第 {page} 页数据时出错: {e}")

        page += 1


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def ms_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool:
        log.error("MySQL数据库连接失败")
        raise Exception("MySQL数据库连接失败")

    try:
        get_sold_list(log, sql_pool)

    except Exception as e:
        log.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


def schedule_task():
    """
    设置定时任务
    """
    # ms_main(log=logger)

    schedule.every().day.at("05:00").do(ms_main, log=logger)
    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    # ms_main(log=logger)
    schedule_task()