spider_work/cxx_spider/start_cxx_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.13.17
# Date   : 2026/1/29 16:31
"""cxx 爬虫与报表模块的统一定时启动入口。

2026/11/04 变更：纳入购买订单采集（super_vault_order_spider）与超新星已售每日报告
（stats/super_vault_nova_daily_report），并改为「启动时固定工作目录」，使其可在
Linux 服务器上以 systemd / nohup 等任意方式拉起而不依赖调用方的 cwd。

单进程 + 线程模型：schedule 主循环只负责计时，每个任务在独立线程里跑，
互不阻塞；同名任务加锁，上一轮没跑完时本轮跳过，避免并发写同一批数据。
"""
import os
import sys
import time
import threading

# ---- 路径与工作目录（必须在导入本项目模块之前完成）----
# mysql_pool 在导入期就按 cwd 相对路径读 application.yml，而服务器上由 systemd/cron
# 拉起时 cwd 往往不是项目目录，故先切到本文件所在目录（项目根），让相对路径始终可用。
# 公共模块 mysql_pool/YamlLoader 与配置 application.yml 都只在根目录留一份，子目录不放副本。
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STATS_DIR = os.path.join(BASE_DIR, "stats")
os.chdir(BASE_DIR)
# 以「python start_cxx_spider.py」方式启动时，sys.path[0] 已是 BASE_DIR；
# stats 是普通子目录（非包），须显式加入才能直接导入其中的报表模块。
for _p in (BASE_DIR, STATS_DIR):
    if _p not in sys.path:
        sys.path.insert(0, _p)

# ---- 日志：全项目唯一的 sink，所有模块的输出都汇总到根目录 logs/{日期}.log ----
# 各业务模块只在自己作为 __main__ 独立运行时才配 sink，被 import 时不碰全局 logger，
# 故这里在 import 之前还是之后配置都一样，不再像从前那样靠 import 顺序决定日志去向。
import schedule
from loguru import logger

logger.remove()
logger.add(os.path.join(BASE_DIR, "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")

from super_vault_daily_spider import cxx_daily_main
from super_vault_on_sale_spider import cxx_sale_main
from super_vault_order_spider import cxx_order_main
from super_vault_nova_daily_report import run_once as cxx_nova_report_once

# 任务互斥锁：key=任务名，防同一任务上下两轮重叠执行
# （订单全量回填可能跑数小时，02:00 那轮没跑完时 08:00 不能再起一轮去「删旧重灌」同一批 pid）
_JOB_LOCKS = {}
_LOCKS_GUARD = threading.Lock()


def _get_job_lock(job_name: str) -> threading.Lock:
    """取（或懒创建）指定任务名的互斥锁。

    Args:
        job_name (str): 任务名，由 run_threaded 传入。

    Returns:
        threading.Lock: 该任务名对应的锁对象，同名任务复用同一把。
    """
    with _LOCKS_GUARD:
        if job_name not in _JOB_LOCKS:
            _JOB_LOCKS[job_name] = threading.Lock()
        return _JOB_LOCKS[job_name]


def _job_wrapper(job_name: str, job_func, args: tuple, kwargs: dict):
    """线程内执行任务：加锁防重入，并兜底捕获异常避免线程静默死掉。

    Args:
        job_name (str): 任务名，用于日志与锁。
        job_func: 要运行的目标函数。
        args (tuple): 传给目标函数的位置参数。
        kwargs (dict): 传给目标函数的关键字参数。

    Returns:
        None: 无返回；执行结果与异常均写入日志。
    """
    lock = _get_job_lock(job_name)
    # 非阻塞抢锁：抢不到说明上一轮还在跑，本轮直接跳过（而不是排队堆积）
    if not lock.acquire(blocking=False):
        logger.warning(f"任务 {job_name} 上一轮仍在运行，本轮跳过")
        return
    try:
        logger.info(f"任务 {job_name} 开始")
        job_func(*args, **kwargs)
        logger.info(f"任务 {job_name} 结束")
    except Exception as e:
        # 线程里抛出的异常不会被主循环捕获，必须在此落日志，否则无人值守时无从排查
        logger.exception(f"任务 {job_name} 异常退出: {e}")
    finally:
        lock.release()


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

    Args:
        job_func: 要运行的目标函数。
        *args: 传给目标函数的位置参数。
        job_name (str, optional): 任务名，用于日志与防重入锁。Defaults to None（取函数名）。
        **kwargs: 传给目标函数的关键字参数。

    Returns:
        None: 无返回；线程以守护模式启动，主进程退出时一并结束。
    """
    name = job_name or getattr(job_func, "__name__", "job")
    job_thread = threading.Thread(target=_job_wrapper, args=(name, job_func, args, kwargs),
                                  name=name, daemon=True)
    job_thread.start()


def schedule_task():
    """爬虫与报表模块的统一定时入口。

    排期（服务器时区须为 Asia/Shanghai，schedule 按本地时间触发）：
        启动即跑  在售商品采集（cxx_sale_main）
        每小时    在售商品采集（cxx_sale_main）
        07:01     每日商品采集（cxx_daily_main）
        02:00     购买订单采集（cxx_order_main），增量
        08:00     购买订单采集（cxx_order_main），补抓 02:00 失败/漏抓及新售罄的团
        09:30     超新星已售每日报告（统计前一天成交，生成 xlsx 并发企微群）

    Returns:
        None: 不返回，常驻循环直到进程被终止。
    """
    # 立即运行一次任务
    # run_threaded(cxx_daily_main, log=logger)
    # time.sleep(5)

    run_threaded(cxx_sale_main, log=logger)

    # 设置定时任务
    schedule.every().day.at("07:01").do(run_threaded, cxx_daily_main, log=logger)

    schedule.every().hour.do(run_threaded, cxx_sale_main, log=logger)

    # 购买订单采集：两轮均为增量，靠 order_state 只补未完成的团，重复跑安全
    # log 以位置参数传入——该函数挂了 @retry(after=after_log)，after_log 靠 args[0] 取日志对象
    schedule.every().day.at("02:00").do(run_threaded, cxx_order_main, logger)
    schedule.every().day.at("08:00").do(run_threaded, cxx_order_main, logger)

    # 已售每日报告：排在订单采集（08:00 那轮）之后，保证统计到的是当日补全后的数据
    schedule.every().day.at("09:30").do(run_threaded, cxx_nova_report_once, log=logger,
                                        job_name="cxx_nova_daily_report")

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


if __name__ == '__main__':
    schedule_task()