spider_work/courtyard_spider/courtyard_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/12/2 14:08
import shutil
import threading
import time
import inspect
import requests
import schedule
import user_agent
from loguru import logger
from parsel import Selector
from datetime import datetime
from mysql_pool import MySQLConnectionPool
from DrissionPage import ChromiumPage, ChromiumOptions
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")

headers = {
    "accept": "application/json",
    "referer": "https://courtyard.io/",
    "user-agent": user_agent.generate_user_agent()
}



# 全局变量标识首次运行是否完成
detail_first_run_completed = False

# 详情页浏览器需拦截的资源模式。目标: 该页只需 DOM 结构做 xpath 解析(Activity history),
# 故把「不影响 DOM 数据」的重资源全部拦掉——省请求、提速、降内存, 且不会 OOM。
# 注意: 拦 CSS/字体不影响 xpath(xpath 只认 DOM 不认样式), 图片已由 no_imgs 关闭。

# 视频/大媒体: 最占带宽与内存
VIDEO_PATTERNS = [
    '*.mp4',
    '*.webm',
    '*.avi',
    '*.mov',
    '*.flv',
    '*.m3u8',
    '*.ts',
    '*googlevideo.com*',
    '*videoplayback*',
    '*video.twimg.com*',
    '*videocdn*',
    '*.mpd',  # DASH视频流
]

# 字体: 纯展示资源, 拦掉可省请求且绝不影响 DOM/JS。
# 注意: 不要拦 *.css! courtyard.io 是 Next.js 应用, CSS 与 JS 同属路由 chunk 依赖图,
# 拦掉 CSS 会导致 chunk-loading Promise 被 reject → 触发 React 错误边界(client-side exception)白屏。
STYLE_FONT_PATTERNS = [
    '*.woff',
    '*.woff2',
    '*.ttf',
    '*.otf',
    '*.eot',
]

# 第三方埋点/统计/广告/监控: 与业务数据无关, 纯拖慢加载
TRACKER_PATTERNS = [
    '*google-analytics.com*',
    '*googletagmanager.com*',
    '*doubleclick.net*',
    '*facebook.net*',
    '*connect.facebook*',
    '*segment.com*',
    '*segment.io*',
    '*sentry.io*',
    '*mixpanel.com*',
    '*hotjar.com*',
    '*intercom.io*',
    '*fullstory.com*',
    '*amplitude.com*',
]

# 汇总: 实际下发给浏览器的拦截清单
BLOCKED_URL_PATTERNS = VIDEO_PATTERNS + STYLE_FONT_PATTERNS + TRACKER_PATTERNS

# 连续失败达到该阈值即判定浏览器卡死, 触发重启
MAX_CONSECUTIVE_FAILURES = 3

# 每连续处理该条数就主动重启一次浏览器, 释放 Chromium 累积内存(V8 堆/渲染进程/缓存随页面数线性增长)
BROWSER_RESTART_INTERVAL = 100
# 每处理该条数就清理一次缓存/cookies, 减缓单实例内存增长
BROWSER_CLEAR_CACHE_INTERVAL = 10

# auto_port 临时用户目录的基路径(放 D 盘专属子目录)。auto_port 每次用随机端口, 会在此
# 目录下不断新建 userData/{port} 且退出不自动删除, 故每次新建浏览器前会清空整个此目录
BROWSER_TMP_PATH = r'D:\Drissionpage_temp\courtyard'

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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_goods_list(log, sql_pool):
    """
    获取商品列表
    :param log: logger对象
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 开始获取商品列表 ==========================")
    url = "https://api.courtyard.io/vending-machines"
    response = requests.get(url, headers=headers, timeout=22)
    # print(response.text)
    response.raise_for_status()

    # 同样用 `or []` 兜底: 防止 vendingMachines 为 null 时下方 for 迭代抛 TypeError
    vendingMachines = response.json().get("vendingMachines") or []

    for item in vendingMachines:
        bag_id = item.get("id")
        bag_title = item.get("title")
        # sealed_pack_animation = item.get("sealedPackAnimation")
        # sealed_pack_image = item.get("sealedPackImage")
        category_title = item.get("category", {}).get("title")
        price = item.get("saleDetails", {}).get("salePriceUsd")

        data_dict = {
            "bag_id": bag_id,
            "bag_title": bag_title,
            "category": category_title,
            "price":price
        }
        # log.info(f'get_goods_list: {data_dict}')

        try:
            get_goods_detail(log, data_dict, sql_pool)
        except Exception as e:
            log.error(f"Error processing item: {e}")

    # 保存数据
    # if info_list:
    #     log.info(f"获取商品列表成功, 共 {len(info_list)} 条数据")
    #     sql_pool.insert_many(table="courtyard_vending_machines_record", data_list=info_list, ignore= True)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_goods_detail(log, query_dict: dict, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    :param query_dict: query_dict
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品详情 ==========================")
    url = "https://api.courtyard.io/index/query/recent-pulls"
    params = {
        "limit": "250",
        # "vendingMachineIds": "pkmn-basic-pack"
        "vendingMachineIds": query_dict["bag_id"]
    }
    response = requests.get(url, headers=headers, params=params, timeout=22)
    # print(response.text)
    response.raise_for_status()

    resp_json = response.json()
    # 接口偶发返回 {"error": ...}（无 assets 键，多为限流/暂无数据），显式跳过并告警，避免静默吞掉
    if isinstance(resp_json, dict) and "assets" not in resp_json:
        log.warning(f"recent-pulls 返回无 assets, bag_id: {query_dict['bag_id']}, resp: {str(resp_json)[:150]}")
        return

    # 注意: dict.get(key, default) 的 default 仅在 key 缺失时生效;
    # 若 key 存在但值为 null(None) 仍返回 None, 故统一用 `or []` 兜底, 防止后续 len()/迭代抛 TypeError
    pulls = resp_json.get("assets") or []
    info_list = []
    for item in pulls:
        detail_title = item.get("title")
        detail_id = item.get("proof_of_integrity")

        if not detail_id:
            log.error(f"信息异常, detail_id: {detail_id}")
            continue

        # asset_pictures 实测存在为 null 的情况, 原写法 len(None) 是本次 TypeError 崩溃主因, 用 `or []` 兜底
        asset_pictures = item.get("asset_pictures") or []
        img_front = asset_pictures[0] if len(asset_pictures) > 0 else None
        img_back = asset_pictures[1] if len(asset_pictures) > 1 else None

        # crawl_date = time.strftime("%Y-%m-%d", time.localtime())

        data_dict = {
            "bag_id": query_dict["bag_id"],
            "bag_title": query_dict["bag_title"],
            "category": query_dict["category"],
            "price": query_dict["price"],
            "detail_id": detail_id,
            "detail_title": detail_title,
            "img_front": img_front,
            "img_back": img_back,
            # "crawler_date": crawl_date
        }
        # log.info(f'data_dict:{data_dict}')
        info_list.append(data_dict)

    # 保存数据
    if info_list:
        log.info(f"获取商品详情成功, 共 {len(info_list)} 条数据")
        sql_pool.insert_many(table="courtyard_list_record", data_list=info_list, ignore=True)


def convert_time_format(time_str):
    """
    将时间字符串转换为标准格式
    :param time_str: 原始时间字符串，如 "December 3, 2025 at 4:29 PM"
    :return: 标准时间格式字符串，如 "2025-12-03 16:29:00"
    """
    if not time_str:
        return None

    try:
        dt_obj = datetime.strptime(time_str, "%B %d, %Y at %I:%M %p")
        return dt_obj.strftime("%Y-%m-%d %H:%M:%S")
    except ValueError as e:
        logger.warning(f"时间转换失败: {time_str}, 错误: {e}")
        return None


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    :param page: page对象
    :param sql_id: 数据库id
    :param detail_id: 商品详情id
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品 <sale> 详情, sql_id: {sql_id} ==========================")
    # page_url = "https://courtyard.io/asset/a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4"
    page_url = f"https://courtyard.io/asset/{detail_id}"
    page.get(page_url)
    # 加载策略为 none, get() 立即返回; 这里只等目标节点 "Activity history" 渲染出现即可开始解析,
    # 不等整页 load 完成——目标节点在, 就代表要抓的数据已就绪, 大幅缩短单条耗时
    target = page.ele('xpath://h6[text()="Activity history"]', timeout=25)
    if not target:
        log.error(f'{inspect.currentframe().f_code.co_name} -> 目标节点未出现(可能加载失败/被限流), 重试..........')
        raise Exception('目标节点未出现, 重新加载........')  # 抛出异常以便重试
    log.debug(f'{inspect.currentframe().f_code.co_name} -> 目标节点已就绪, url: {page_url}')

    html = page.html
    if not html:
        log.error(f'{inspect.currentframe().f_code.co_name} -> 页面加载失败...........')
        raise Exception('页面加载失败, 重新加载........')  # 抛出异常以便重试

    selector = Selector(text=html)

    # 方法一：通过文本内容匹配（优先）
    correlation_spans = selector.xpath('//span[contains(text(), ":")]/text()')
    correlation_id = None

    for text_selector in correlation_spans:
        correlation_id = text_selector.get()  # ✅ 获取字符串

        # match = re.search(r'[\w\s]+:\s*(\d+)', text)
        # if match:
        #     correlation_id = match.group(1)
        #     break  # 获取第一个有效 ID
    # correlation_spans = selector.xpath('//span[contains(text(), ":")]')
    # correlation_text = None
    #
    # for span in correlation_spans:
    #     text = span.get()
    #     if text and ":" in text:
    #         correlation_text = text
    #         break

    # 如果方法一失败，使用方法二：通过结构定位（备用）
    if not correlation_id:
        correlation_span = selector.xpath('//a[contains(@href, "cgccards.com")]/preceding-sibling::span[1]/text()')
        correlation_id = correlation_span.get()

    # 初始化所有可能的字段为None
    data_dict = {"detail_id": detail_id, "correlation_id": correlation_id, "burn_from": None,
                 "burn_to": None, "burn_time": None, "sale_price": None, "sale_from": None, "sale_to": None,
                 "sale_time": None, "mint_price": None, "mint_from": None, "mint_to": None, "mint_time": None}

    # 获取 "Activity history" 后面的 div
    activity_div = selector.xpath('//h6[text()="Activity history"]/following-sibling::div[1]/div')
    for tag_div in activity_div:
        tag_name = tag_div.xpath('./div[1]/div/span/text()').get()
        if not tag_name:
            continue

        if tag_name == "Burn":
            data_dict["burn_from"] = tag_div.xpath('./div[2]/div[1]//h6/text()').get()
            data_dict["burn_to"] = tag_div.xpath('./div[2]/div[2]//h6/text()').get()
            data_dict["burn_time"] = tag_div.xpath('./div[2]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["burn_time"] = convert_time_format(data_dict["burn_time"])
        elif tag_name == "Sale":
            sale_price = tag_div.xpath('./div[2]/span/text()').get()
            if sale_price:
                sale_price = sale_price.replace("$", "").replace(",", "")
            data_dict["sale_price"] = sale_price

            data_dict["sale_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get()
            data_dict["sale_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get()
            data_dict["sale_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["sale_time"] = convert_time_format(data_dict["sale_time"])
        elif tag_name == "Mint":
            mint_price = tag_div.xpath('./div[2]/span/text()').get()
            if mint_price:
                mint_price = mint_price.replace("$", "").replace(",", "")
            data_dict["mint_price"] = mint_price

            data_dict["mint_from"] = tag_div.xpath('./div[3]/div[1]//h6/text()').get()
            data_dict["mint_to"] = tag_div.xpath('./div[3]/div[2]//h6/text()').get()
            data_dict["mint_time"] = tag_div.xpath('./div[3]/div[3]/span/@aria-label').get()
            # December 3, 2025 at 4:29 PM 转换时间格式
            data_dict["mint_time"] = convert_time_format(data_dict["mint_time"])

    # log.info(f'Sale detail data: {data_dict}')

    # 保存数据
    sql_pool.insert_one_or_dict(table="courtyard_detail_record", data=data_dict, ignore=True)
    sql_pool.update_one("UPDATE courtyard_list_record SET state = 1 WHERE id = %s", (sql_id,))


def _create_detail_browser(log):
    """创建并配置用于详情采集的浏览器实例。

    集中管理浏览器启动参数、视频拦截规则与超时, 供首次启动和卡死重启复用。

    Args:
        log: loguru logger 对象, 从调用方透传。

    Returns:
        ChromiumPage: 已配置拦截规则与超时的浏览器页面对象。
    """
    # auto_port 的临时用户目录退出不会自动清理, 每次随机端口都会在 BROWSER_TMP_PATH 下留一份;
    # 故新建实例前先清空整个目录, 保证不残留缓存、不累积占磁盘(重启时旧实例已 quit, 无占用)
    shutil.rmtree(BROWSER_TMP_PATH, ignore_errors=True)

    options = ChromiumOptions()
    # 无需登录态, 不做任何持久化: auto_port 自动分配空闲端口(每次全新进程, 确保内存被彻底释放),
    # 相比固定端口: 分批重启不会因端口被残留进程占用而失败/误连到旧浏览器
    options.auto_port(True)
    # 临时用户目录基路径放到 D 盘专属子目录(默认在 C 盘 %TEMP%/DrissionPage)
    options.set_tmp_path(BROWSER_TMP_PATH)
    # options.set_proxy("http://" + tunnel)

    options.no_imgs(True)

    # 加载策略设为 none: page.get() 不等整页 load 完成立即返回, 之后由业务代码只等
    # 目标元素(Activity history)出现即开始解析——这是提速最关键的一招(SPA 整页 load
    # 往往还挂着一堆无关请求/长连接, 傻等会白白拖慢每一条)
    options.set_load_mode('none')

    # 禁止媒体相关设置
    options.set_argument('--autoplay-policy=user-gesture-required')
    options.set_argument('--disable-features=PreloadMediaEngagementData')

    # ---- 省资源(真正有效且无副作用的项): 减少下载与缓存占用, 不掐内存上限 ----
    # 说明: 不再使用 --max-old-space-size / --renderer-process-limit=1 / --process-per-site。
    # 前者会在重 SPA 需要更多 V8 堆时直接触发渲染进程 OOM 崩溃(即"喔唷崩溃啦 Out of Memory");
    # 后两者强制单渲染进程复用, 反而关闭了浏览器靠"进程销毁"回收内存的机制, 让内存只增不减。
    # 真正的内存回收依赖每 BROWSER_RESTART_INTERVAL 条整浏览器重启(见 get_sale_detail_list)。
    options.set_argument('--disk-cache-size=1')                   # 几乎禁用磁盘缓存
    options.set_argument('--media-cache-size=1')                  # 几乎禁用媒体缓存
    options.set_argument('--disable-application-cache')           # 禁用应用缓存
    options.set_argument('--disable-dev-shm-usage')               # 不占用共享内存, 避免共享内存耗尽
    options.set_argument('--disable-extensions')                  # 禁用扩展
    options.set_argument('--disable-background-networking')       # 关闭后台网络活动

    # 最大化
    options.set_argument("--start-maximized")
    options.set_argument("--disable-gpu")
    options.set_argument("-accept-lang=en-US")
    page = ChromiumPage(options)

    # 使用 set.blocked_urls() 拦截视频/样式/字体/埋点等无关资源(不影响 DOM 取数)
    page.set.blocked_urls(BLOCKED_URL_PATTERNS)
    # 设置超时: 页面加载最多等 30s, 基础操作 20s, 避免浏览器卡死时无限等待
    page.set.timeouts(base=20, page_load=30)
    return page


def _restart_detail_browser(log, page, reason='卡死'):
    """关闭旧浏览器并重建一个新实例。

    先尽力 quit 旧实例(失败忽略), 等待端口释放后重新创建。既用于连续采集失败时的
    自愈重启, 也用于每处理 BROWSER_RESTART_INTERVAL 条后的主动内存释放重启。

    Args:
        log: loguru logger 对象, 从调用方透传。
        page (ChromiumPage): 待关闭的旧浏览器实例。
        reason (str, optional): 重启原因, 仅用于日志区分。Defaults to '卡死'。

    Returns:
        ChromiumPage: 全新的浏览器页面对象。
    """
    try:
        page.quit()
    except Exception as e:
        # 卡死的浏览器 quit 本身也可能超时/报错, 忽略即可
        log.warning(f'关闭旧浏览器失败(忽略): {e}')
    # 等待端口(9138)与用户数据目录释放, 再重建, 避免端口占用导致启动失败
    time.sleep(3)
    log.warning(f'浏览器{reason}, 正在重启新实例..........')
    return _create_detail_browser(log)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_sale_detail_list(log, sql_pool=None):
    """
    获取商品详情
    :param log: logger对象
    # :param detail_id_list: 详情id列表
    :param sql_pool: MySQL连接池对象
    :return:
    """
    log.info(f"========================== 获取商品 <sale> 详情  LIST ==========================")
    page = _create_detail_browser(log)
    # 连续失败计数: 用于区分"单条数据问题"与"浏览器整体卡死", 后者需重启浏览器
    consecutive_failures = 0

    try:
        sql_detail_id_list = sql_pool.select_all("SELECT id, detail_id FROM courtyard_list_record WHERE state = 0")

        for idx, sql_detail_id in enumerate(sql_detail_id_list):
            sql_id = sql_detail_id[0]
            detail_id = sql_detail_id[1]

            # 每处理 BROWSER_RESTART_INTERVAL 条主动重启浏览器, 彻底释放 Chromium 累积内存;
            # 单实例连续跑几百页时 V8 堆/渲染进程/缓存会线性增长到数 GB, 这是内存/CPU 高的主因
            if idx > 0 and idx % BROWSER_RESTART_INTERVAL == 0:
                log.info(f'已连续处理 {idx} 条, 达到批次阈值, 主动重启浏览器释放内存..........')
                page = _restart_detail_browser(log, page, reason='达到批次重启阈值')
                consecutive_failures = 0

            try:
                get_sale_detail_single_page(log, page, sql_id, detail_id, sql_pool)
                consecutive_failures = 0  # 成功即清零, 只累计"连续"失败
                # 定期清理缓存/cookies, 减缓单实例在两次批次重启之间的内存增长
                if idx > 0 and idx % BROWSER_CLEAR_CACHE_INTERVAL == 0:
                    try:
                        page.clear_cache()
                    except Exception as ce:
                        log.warning(f'清理浏览器缓存失败(忽略): {ce}')
            except Exception as e:
                consecutive_failures += 1
                log.error(f'get_sale_detail_single_page error '
                          f'({consecutive_failures}/{MAX_CONSECUTIVE_FAILURES}), sql_id: {sql_id}: {e}')

                if consecutive_failures >= MAX_CONSECUTIVE_FAILURES:
                    # 连续多条失败, 判定浏览器已卡死: 不标记 state=2(避免误伤),
                    # 保留 state=0 待下轮重试, 并重启浏览器后继续处理后续条目
                    page = _restart_detail_browser(log, page)
                    consecutive_failures = 0
                else:
                    # 未达阈值, 视为该条数据自身问题, 标记 state=2 跳过
                    sql_pool.update_one("UPDATE courtyard_list_record SET state = 2 WHERE id = %s", (sql_id,))

    except Exception as e:
        log.error(f'get_response error: {e}')
        raise  # 直接透传原始异常(原写法 raise 字符串会抛 TypeError 掩盖真因)
    finally:
        try:
            page.quit()
        except Exception as e:
            log.warning(f'最终关闭浏览器失败(忽略): {e}')


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

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        try:
            log.debug('------------------- 开始获取商品列表 -------------------')
            get_goods_list(log, sql_pool)
        except Exception as e:
            log.error(f'get_goods_list error: {e}')

    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} 运行结束,等待下一轮的采集任务............')

    end = time.time()
    elapsed_time = end - start
    log.info(f'============================== 本次爬虫运行时间:{elapsed_time:.2f} 秒 ===============================')

    return elapsed_time


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

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    global detail_first_run_completed

    try:
        # 获取详情页信息
        try:
            log.debug('------------------- 获取商品 detail 数据 -------------------')
            get_sale_detail_list(log, sql_pool)
        except Exception as e:
            log.error(f'get_sale_detail_list error: {e}')

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


def control_list_mask(log):
    """
    控制列表爬虫任务  每10分钟运行
    :param log: logger对象
    """
    while True:
        log.info(
            f'--------------------- 开始运行 {inspect.currentframe().f_code.co_name} 新一轮的爬虫任务 ---------------------')
        elapsed_time = list_main(log)
        # 计算剩余时间
        wait_time = max(0, 300 - int(elapsed_time))
        if wait_time > 0:
            log.info(f"程序运行时间{elapsed_time:.2f}秒, 小于 5 分钟，等待 {wait_time:.2f} 秒后再开始下一轮任务")
            time.sleep(wait_time)
        else:
            log.info("程序运行时间大于等于5分钟，直接开始下一轮任务")


def scheduled_detail_main(log):
    """定时任务调用的包装函数"""
    global detail_first_run_completed
    if detail_first_run_completed:
        detail_main(log)
    else:
        log.info("Skipping scheduled task as first run is not completed yet")


def run_threaded(job_func, *args, **kwargs):
    """
    在新线程中运行给定的函数，并传递参数。

    :param job_func: 要运行的目标函数
    :param args: 位置参数
    :param kwargs: 关键字参数
    """
    job_thread = threading.Thread(target=job_func, args=args, kwargs=kwargs)
    job_thread.start()



def schedule_task():
    """
    设置定时任务
    """
    # 启动 control_list_mask 任务线程
    list_thread = threading.Thread(target=control_list_mask, args=(logger,))
    list_thread.daemon = True  # 设置为守护线程，主程序退出时自动结束
    list_thread.start()

    # 启动 detail_main 任务线程（首次运行）
    detail_thread = threading.Thread(target=detail_main, args=(logger,))
    detail_thread.daemon = True
    detail_thread.start()

    # 设置定时任务  每天
    # schedule.every().day.at("00:01").do(run_threaded, detail_main, logger)
    schedule.every().day.at("00:01").do(run_threaded, scheduled_detail_main, logger)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    schedule_task()
    # detail_main(log=logger)
    # get_sale_detail_list(log, ((1, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4'),
    #                            (2, 'a4f0bbebd858370567f1779fddf0f55630810116d80965e33940fc8ff5ac94b4')), sql_pool)