# -*- 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 顺序决定日志去向。 # # 日志目录优先取环境变量 CXX_LOG_DIR,缺省回落到项目根 logs/: # 服务器的日志采集只读固定目录(如容器内 /app/logs),而项目代码未必就放在该目录的上一级, # 故目录由部署环境指定、代码不写死;文件名固定 {time:YYYYMMDD}.log 以匹配采集端的格式要求。 # 部署示例:CXX_LOG_DIR=/app/logs python start_cxx_spider.py import schedule from loguru import logger LOG_DIR = os.environ.get("CXX_LOG_DIR") or os.path.join(BASE_DIR, "logs") os.makedirs(LOG_DIR, exist_ok=True) # 提前建目录:目录不存在或无权限时在此直接报错,而非等到首条日志才暴露 logger.remove() # backtrace=True 保留完整调用栈便于定位;diagnose=False 关掉 loguru 对 traceback 的变量值标注—— # 它会把各帧局部变量的值逐行打印出来,连库失败时会把数据库密码写进日志并被日志采集送往中心平台。 logger.add(os.path.join(LOG_DIR, "{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", backtrace=True, diagnose=False) 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()