feat(spider): 统一启动入口加入订单与报告模块并优化调度机制

- 统一 cxx 爬虫与报表模块的定时启动入口，支持多线程独立运行任务
- 新增购买订单采集功能，包含增量与补抓两轮调度
- 纳入超新星已售每日报告，安排于订单采集后运行保证数据完整
- 实现任务锁机制避免同一任务多轮并行执行引发数据竞态
- 切换工作目录到项目根，保障在 Linux 环境使用 systemd/nohup 启动可靠
- 取消启动时每日商品采集自动执行，防止重复执行
- requirements.txt 新增 openpyxl 依赖支持报表生成
- 修正日志文件路径，日志统一输出到根目录 logs 中
- super_vault_his_report 统计口径调整，销售额改用实际收款 total_price
- 新增真实购买人数统计，依据订单数据区分匿名与实名用户
- 汇总与明细表格同步增加真实购买人数及相关计算
- super_vault_nova_daily_report 继承累计报告业务逻辑，改用 sold_time 作为成交时间锚点
- 新增用户明细 Sheet，统计真实买家按参与金额排序的明细列表
- 报告人数口径中卡人数改名，真实购买人数更精确表示客观购买行为
- 传入日志对象参数形式统一，支持异常捕获与线程内日志输出
- 代码风格及注释显著完善，提升维护可读性和稳定性

cxx_spider/requirements.txt

@@ -1,6 +1,7 @@
 -i https://mirrors.aliyun.com/pypi/simple/
 DBUtils==3.1.2
 loguru==0.7.3
+openpyxl==3.1.5
 PyMySQL==1.1.2
 PyYAML==6.0.1
 requests==2.32.5

cxx_spider/start_cxx_spider.py

@@ -1,38 +1,134 @@
 # -*- coding: utf-8 -*-
 # Author : Charley
-# Python : 3.10.8
+# 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 schedule
 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
-from super_vault_daily_spider import cxx_daily_main
-from super_vault_on_sale_spider import cxx_sale_main
 
 logger.remove()
-logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
+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")
 
-def run_threaded(job_func, *args, **kwargs):
+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()
+
 
-    :param job_func: 要运行的目标函数
-    :param args: 位置参数
-    :param kwargs: 关键字参数
+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: 无返回；线程以守护模式启动，主进程退出时一并结束。
     """
-    job_thread = threading.Thread(target=job_func, args=args, kwargs=kwargs)
+    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_daily_main, log=logger)
+    # time.sleep(5)
 
     run_threaded(cxx_sale_main, log=logger)
 
@@ -41,6 +137,15 @@ def schedule_task():
 
     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)

cxx_spider/stats/super_vault_his_report.py

@@ -16,24 +16,33 @@
     - 统计范围：默认 WIN_MODE="all"，剔除 sold_count=0（未开卖）与 status 0/1/2/3（未开始/预售/
       进行中等未成交态），只统计 status 4/6/7/8/9/10/12 等有效成交态的团。
     - 主播维度按 anchor_username 分组：同名不同 anchor_id（如两个「北京小周」）合并为同一位主播。
-    - 销售额 = SUM(sign_price * count)（单价 × 总份数 = 满仓总价；
-              注意随机卡种/未售罄团会高于实收 total_price）。
+    - 销售额 = SUM(total_price)（平台实收团总价，直接取库内 total_price）。
+              不再用 sign_price × count（单价×总份数=满仓价）计算——随机卡种/折扣/未售罄团会高估
+              （实测已售全量高约 17%：满仓 5104 万 vs 实收 4364 万），total_price 才是实收。
     - 成团数 = 该范围内的拼团商品数。
-    - 参与人数 = super_vault_report_record.user_name 去重（拆卡报告中卡用户，近似参团人头；
-              仅覆盖有拆卡报告的商品，非真实买家，偏低）。平台口径跨全部商品去重；各主播口径按
+    - 明细「总金额」= 该团 total_price（同上口径，实收）。
+    - 中卡人数（原「参与人数」）= super_vault_report_record.user_name 去重（拆卡报告中「抽中卡片的人」，
+              近似非真实参与者；仅覆盖有拆卡报告的商品，偏低）。平台口径跨全部商品去重；各主播口径按
               该主播商品单独去重，故各主播逐个相加(人次) ≥ 平台。
+    - 真实购买人数 = super_vault_order_record 订单表去重（非匿名按 uid/user_name 去重 + 匿名每笔计 1）。
+              数据由 super_vault_order_spider.py 采集；未抓订单的团为 0（建议报告前先跑订单爬虫）。
     - 均拼单价 = 销售额 / 成团数
-    - 人均消费 = 销售额 / 参与人数（按中卡近似，偏高，仅供参考）
-    - 明细「参与人数(本团)」= 该团拆卡报告 user_name 去重（中卡近似），口径与汇总一致、拆到单团。
-    - 本平台无 buy_record（真实买家）与 onsale 进度快照表，故无「真实参与人数」与「进度里程碑」列。
+    - 人均消费 = 销售额 / 真实购买人数（改用真实买家口径，比中卡近似更准；真实=0 的团留空）
+    - 明细「中卡人数(本团)」「真实购买人数(本团)」= 拆到单团的上述两口径，与汇总一致。
 
 从项目根目录运行：python stats/super_vault_his_report.py（cwd=根目录，读根目录 application.yml）
 """
 import os
 import sys
 import time
-# 把项目根目录加入 import 路径（与得卡报告一致：根目录留一份公共模块时可直接 import）
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+# 项目根目录（本文件在 stats/ 下，上一级即根目录）：公共模块 mysql_pool/YamlLoader、
+# 配置 application.yml、日志目录 logs 统一只在根目录留一份，stats 内不再放副本
+ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT_DIR)
+# 必须在 import mysql_pool 之前切到根目录：mysql_pool 在「导入期」就按 cwd 相对路径读
+# application.yml（读不到会直接 UnboundLocalError），而配置只在根目录留一份。
+# 入口脚本已切过，这里是幂等的；单独跑本文件时则靠这行兜住任意启动目录。
+os.chdir(ROOT_DIR)
 from datetime import date, datetime
 
 import schedule
@@ -42,10 +51,10 @@ from openpyxl import Workbook
 from openpyxl.styles import Font, Alignment, PatternFill, Border, Side
 from mysql_pool import MySQLConnectionPool
 
-# 日志：按天切分文件，保留 7 天（常驻定时运行）
-logger.remove()
-logger.add("./logs/super_vault_report_{time:YYYYMMDD}.log", encoding="utf-8", rotation="00:00",
-           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}", level="INFO", retention="7 day")
+# 日志统一写根目录 logs/{日期}.log，由入口脚本配置（见文件末尾 __main__）。
+# 模块级绝不配置 sink：被 start_cxx_spider 等入口 import 时若在此 logger.remove()/add()，
+# 会摘掉入口已装好的 sink，并凭空创建一个随即被摘、永不写入的 0 字节日志文件。
+LOG_PATH = os.path.join(ROOT_DIR, "logs", "{time:YYYYMMDD}.log")
 
 # 企微发送：本项目暂无 auto_send_wx_msg 模块，默认关闭；日后接入企微机器人时置 True 并放置发送模块即可
 SEND_WECHAT = False
@@ -64,19 +73,26 @@ STAT_END = "2026-09-01 00:00:00"     # WIN_MODE="range" 时生效
 # 开售时间(sell_time) 依次兜底。多数待发货团无 completion_time 但有 sold_time，故成交时间覆盖率大幅提升。
 TS_ANCHOR = "COALESCE({a}.completion_time, {a}.sold_time, {a}.sell_time)"
 
+# 「真实购买人数」统计表达式（用于订单表 super_vault_order_record，别名 o）：
+#   非匿名按 COALESCE(uid, user_name) 去重（登录用户用头像 uid、默认头像用户用稳定 handle），
+#   匿名无法区分故每笔按 1 人（上界近似）。数据由 super_vault_order_spider.py 采集；未抓的团为 0。
+REAL_BUYERS_EXPR = ("COUNT(DISTINCT CASE WHEN o.is_anonymous = 0 THEN COALESCE(o.uid, o.user_name) END)"
+                    " + COALESCE(SUM(o.is_anonymous), 0)")
+
 # 明细表列规格：(表头, 明细dict取值键, 是否千分位金额格式)；取值键为「序号」时按行号自增填充。
-# 「参与人数(本团)」= 该团拆卡报告 user_name 去重（中卡近似），放在总金额后、开售时间前。
+# 「中卡人数(本团)」= 该团拆卡报告 user_name 去重（抽中卡片的人，近似非真实参与者）；
+# 「真实购买人数(本团)」= 订单表真实买家去重（更贴近真实）。二者放在总金额后、开售时间前。
 DETAIL_COLS = [
     ("序号", "序号", False), ("团名(商品标题)", "团名", False), ("编号", "编号", False),
     ("卡种", "卡种", False), ("规格", "规格", False), ("单价", "单价", True),
     ("总份数", "总份数", False), ("进度%", "进度", False), ("总金额", "总金额", True),
-    ("参与人数(本团)", "参与人数", False),
+    ("中卡人数(本团)", "中卡人数", False), ("真实购买人数(本团)", "真实购买人数", False),
     ("开售时间", "开售时间", False), ("成交时间", "成交时间", False), ("售卖时长", "售卖时长", False),
 ]
 # 分区标题/大标题横向着色跨列数：取明细列数，保证蓝条覆盖整表宽
 SECTION_SPAN = len(DETAIL_COLS)
-# 汇总指标（平台/主播通用）
-SUMMARY_HEADERS = ["销售额", "成团数", "参与人数", "均拼单价", "人均消费"]
+# 汇总指标（平台/主播通用）：原「参与人数」更名为「中卡人数」，并新增「真实购买人数」
+SUMMARY_HEADERS = ["销售额", "成团数", "中卡人数", "真实购买人数", "均拼单价", "人均消费"]
 
 # ---- 样式 ----
 FONT_TITLE = Font(bold=True, size=14)
@@ -152,23 +168,27 @@ def fetch_platform_summary(pool) -> dict:
         pool (MySQLConnectionPool): MySQL 连接池。
 
     Returns:
-        dict: 含 主播数/销售额/成团数/参与人数/均拼单价/人均消费。
+        dict: 含 主播数/销售额/成团数/中卡人数/真实购买人数/均拼单价/人均消费。
     """
     sql = f"""
     SELECT
-        ROUND(SUM(p.sign_price * p.count), 2)                              AS amount,
+        ROUND(SUM(p.total_price), 2)                              AS amount,
         COUNT(DISTINCT p.anchor_username)                                  AS anchors,
         COUNT(*)                                                            AS grp,
         (SELECT COUNT(DISTINCT rr.user_name)
            FROM super_vault_report_record rr
            JOIN super_vault_product_record pp ON pp.pid = rr.pid
           WHERE {win_clause('pp')}
-            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people
+            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people,
+        (SELECT {REAL_BUYERS_EXPR}
+           FROM super_vault_order_record o
+           JOIN super_vault_product_record pp2 ON pp2.pid = o.pid
+          WHERE {win_clause('pp2')})                                        AS real_people
     FROM super_vault_product_record p
     WHERE {win_clause('p')}
     """
-    amount, anchors, groups, people = pool.select_all(sql)[0]
-    return _pack_summary(amount, groups, people, extra={"主播数": anchors})
+    amount, anchors, groups, people, real_people = pool.select_all(sql)[0]
+    return _pack_summary(amount, groups, people, real_people, extra={"主播数": anchors})
 
 
 def fetch_anchor_summaries(pool) -> list[dict]:
@@ -178,18 +198,22 @@ def fetch_anchor_summaries(pool) -> list[dict]:
         pool (MySQLConnectionPool): MySQL 连接池。
 
     Returns:
-        list[dict]: 每元素含 主播名/销售额/成团数/参与人数/均拼单价/人均消费。
+        list[dict]: 每元素含 主播名/销售额/成团数/中卡人数/真实购买人数/均拼单价/人均消费。
     """
     sql = f"""
     SELECT
         p.anchor_username                                                  AS aname,
-        ROUND(SUM(p.sign_price * p.count), 2)                              AS amount,
+        ROUND(SUM(p.total_price), 2)                              AS amount,
         COUNT(*)                                                            AS grp,
         (SELECT COUNT(DISTINCT rr.user_name)
            FROM super_vault_report_record rr
            JOIN super_vault_product_record pp ON pp.pid = rr.pid
           WHERE pp.anchor_username = p.anchor_username AND {win_clause('pp')}
-            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people
+            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people,
+        (SELECT {REAL_BUYERS_EXPR}
+           FROM super_vault_order_record o
+           JOIN super_vault_product_record pp2 ON pp2.pid = o.pid
+          WHERE pp2.anchor_username = p.anchor_username AND {win_clause('pp2')}) AS real_people
     FROM super_vault_product_record p
     WHERE {win_clause('p')}
     GROUP BY p.anchor_username
@@ -197,8 +221,8 @@ def fetch_anchor_summaries(pool) -> list[dict]:
     """
     rows = pool.select_all(sql) or []
     result = []
-    for aname, amount, groups, people in rows:
-        d = _pack_summary(amount, groups, people)
+    for aname, amount, groups, people, real_people in rows:
+        d = _pack_summary(amount, groups, people, real_people)
         d["主播名"] = aname or "(未知主播)"
         result.append(d)
     return result
@@ -214,15 +238,18 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
         aname (str): 主播名 anchor_username。
 
     Returns:
-        list[dict]: 每元素为一条明细（团名/编号/卡种/规格/单价/总份数/进度/总金额/参与人数/开售时间/成交时间/售卖时长）。
+        list[dict]: 每元素为一条明细（团名/编号/卡种/规格/单价/总份数/进度/总金额/中卡人数/真实购买人数/开售时间/成交时间/售卖时长）。
     """
     sql = f"""
     SELECT
         p.title, p.serial, p.type_name, p.standard_name, p.sign_price, p.sold_count, p.count,
-        ROUND(p.sign_price * p.count, 2)                                 AS amount,
-        -- 该团参与人数：拆卡报告 user_name 去重（中卡近似，仅覆盖有报告的团）
+        p.total_price                                                    AS amount,
+        -- 该团中卡人数：拆卡报告 user_name 去重（抽中卡片的人，近似，仅覆盖有报告的团）
         (SELECT COUNT(DISTINCT rr.user_name) FROM super_vault_report_record rr
           WHERE rr.pid = p.pid AND rr.user_name IS NOT NULL AND rr.user_name <> '') AS buyers,
+        -- 该团真实购买人数：订单表按 uid/user_name 去重 + 匿名每笔计 1（未抓订单的团为 0）
+        (SELECT {REAL_BUYERS_EXPR} FROM super_vault_order_record o
+          WHERE o.pid = p.pid)                                           AS real_buyers,
         p.sell_time,
         COALESCE(p.completion_time, p.sold_time)                         AS deal_time,
         TIMESTAMPDIFF(SECOND, p.sell_time, COALESCE(p.completion_time, p.sold_time)) AS duration_secs
@@ -232,13 +259,14 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
     """
     rows = pool.select_all(sql, (aname,)) or []
     result = []
-    for (title, serial, type_name, standard, price, sold, total, amount, buyers,
+    for (title, serial, type_name, standard, price, sold, total, amount, buyers, real_buyers,
          sell_time, deal_time, duration_secs) in rows:
         progress = round(sold / total * 100, 1) if total else None  # 售卖进度百分比
         result.append({
             "团名": title, "编号": serial, "卡种": type_name, "规格": standard,
             "单价": price, "总份数": total, "进度": progress, "总金额": amount,
-            "参与人数": buyers,   # 本团拆卡报告 user_name 去重（中卡近似）
+            "中卡人数": buyers,                    # 本团拆卡报告 user_name 去重（原「参与人数」）
+            "真实购买人数": int(real_buyers or 0),   # 本团订单表真实买家去重
             # 成交时间 = 完成时间(发货完成)优先，缺则用售罄成交时刻 sold_time；售卖时长=成交-开售
             "开售时间": _fmt_dt(sell_time), "成交时间": _fmt_dt(deal_time),
             "售卖时长": _fmt_duration(duration_secs),
@@ -246,25 +274,29 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
     return result
 
 
-def _pack_summary(amount, groups, people, extra: dict = None) -> dict:
-    """把销售额/成团数/参与人数打包成统一汇总 dict，并算均拼单价、人均消费。
+def _pack_summary(amount, groups, people, real_people, extra: dict = None) -> dict:
+    """把销售额/成团数/中卡人数/真实购买人数打包成统一汇总 dict，并算均拼单价、人均消费。
+
+    人均消费改按「真实购买人数」（订单表真实买家，比中卡近似更准）；real_people=0 时留空。
 
     Args:
         amount (Decimal | None): 销售额。
         groups (int | None): 成团数。
-        people (int | None): 参与人数。
+        people (int | None): 中卡人数（拆卡报告去重，原「参与人数」）。
+        real_people (int | None): 真实购买人数（订单表去重）；未抓订单的团为 0。
         extra (dict, optional): 额外要合并的字段（如主播数）。Defaults to None。
 
     Returns:
-        dict: 含 销售额/成团数/参与人数/均拼单价/人均消费(+extra)。
+        dict: 含 销售额/成团数/中卡人数/真实购买人数/均拼单价/人均消费(+extra)。
     """
     amount = float(amount) if amount is not None else 0.0
     groups = groups or 0
     people = people or 0
-    avg_group = round(amount / groups, 2) if groups else None   # 均拼单价=销售额/成团数
-    avg_person = round(amount / people, 2) if people else None  # 人均消费=销售额/参与人数
-    d = {"销售额": round(amount, 2), "成团数": groups, "参与人数": people,
-         "均拼单价": avg_group, "人均消费": avg_person}
+    real_people = int(real_people or 0)
+    avg_group = round(amount / groups, 2) if groups else None            # 均拼单价=销售额/成团数
+    avg_person = round(amount / real_people, 2) if real_people else None  # 人均消费=销售额/真实购买人数
+    d = {"销售额": round(amount, 2), "成团数": groups, "中卡人数": people,
+         "真实购买人数": real_people, "均拼单价": avg_group, "人均消费": avg_person}
     if extra:
         d.update(extra)
     return d
@@ -394,7 +426,8 @@ def _write_anchor_table(ws, r: int, anchors: list) -> int:
     _style_row(ws, r, len(headers), font=FONT_HEADER, fill=FILL_HEADER)
     r += 1
     for a in anchors:
-        vals = [a["主播名"], a["销售额"], a["成团数"], a["参与人数"], a["均拼单价"], a["人均消费"]]
+        # 按表头 key 通用取值，新增列（真实购买人数等）自动纳入，无需改硬编码
+        vals = [a["主播名"]] + [a.get(h) for h in SUMMARY_HEADERS]
         for i, v in enumerate(vals):
             cell = ws.cell(row=r, column=1 + i, value=v)
             if headers[i] in ("销售额", "均拼单价", "人均消费"):
@@ -474,14 +507,14 @@ def build_report(pool, out: str):
         details = fetch_anchor_details(pool, a["主播名"])
         ws.cell(row=r, column=1,
                 value=(f"　▼ {a['主播名']}　销售额 {a['销售额']:,.2f} / 成团 {a['成团数']} / "
-                       f"参与 {a['参与人数']}（共 {len(details)} 条明细）"))
+                       f"中卡 {a['中卡人数']} / 真实 {a['真实购买人数']}（共 {len(details)} 条明细）"))
         _style_row(ws, r, SECTION_SPAN, font=FONT_HEADER, border=False)
         r += 1
         r = _write_details(ws, r, details, DETAIL_COLS)
         r += 2
 
-    # 列宽：A-M 兼顾明细表与主播汇总表；E 列取 14 以容纳主播汇总的「均拼单价」大额（否则显示 ######）
-    widths = [8, 50, 14, 18, 14, 11, 10, 8, 14, 12, 20, 20, 14]
+    # 列宽：A-N 兼顾明细表与主播汇总表；在「中卡人数」后插入「真实购买人数」列宽（14 列）
+    widths = [8, 50, 14, 18, 14, 11, 10, 8, 14, 12, 14, 20, 20, 14]
     for i, w in enumerate(widths):
         ws.column_dimensions[chr(65 + i)].width = w
     ws.freeze_panes = "A2"  # 冻结标题行
@@ -534,4 +567,9 @@ def schedule_task():
 
 
 if __name__ == "__main__":
+    # 独立运行才配置日志 sink：按天切分、保留 7 天，与入口脚本同格式同去向（根目录 logs/）
+    logger.remove()
+    logger.add(LOG_PATH, encoding="utf-8", rotation="00:00",
+               format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
+               level="DEBUG", retention="7 day")
     main()

cxx_spider/stats/super_vault_nova_daily_report.py

@@ -10,18 +10,42 @@
       有 sold_time 即代表该团已售罄成交，故不再叠加 status/sold_count 过滤。
     - 成交时间/售卖时长一律以 sold_time 为准（累计报告用的是 completion_time→sold_time 兜底锚点）。
 
-报告结构（自上而下单 Sheet 分区，与累计报告一致）：
+人数口径（本每日报告与累计报告的差异，仅本文件生效，base 累计报告不受影响）：
+    - 中卡人数：原「参与人数」更名。来源 super_vault_report_record 拆卡报告 user_name 去重，仅是
+      「抽中卡片的人」的近似，且待发货团常尚无报告故偏低，故更名为「中卡人数」以免误解。
+    - 真实购买人数：新增。来源 super_vault_order_record（super_vault_order_spider.py 抓的真实买家订单），
+      非匿名按 uid/user_name 去重 + 匿名每笔计 1（上界近似）。更贴近真实买家数。
+      依赖：需 super_vault_order_spider.py 先抓过窗口内商品（建议每日排在本报告之前跑），
+           否则未抓订单的团该值为 0。
+    - 人均消费：本每日报告改按「销售额 / 真实购买人数」（比 base 累计报告的中卡人数口径更准）；
+      未抓订单的团 real=0 时留空。
+
+报告结构（共两个 Sheet）：
+  Sheet1「已售每日报告」（自上而下单 Sheet 分区，与累计报告一致）：
     一、平台汇总（当日成交累计）
     二、各主播汇总（每主播一行，按销售额倒序）
     三、各主播每条组队明细（按主播销售额倒序分块，块内按总金额倒序）
+  Sheet2「用户明细」（2026/10/09 新增，参考得卡报告「用户排行榜」）：
+    真实买家按参与金额倒序排行（排名/用户昵称/user_id/参与车数/参与金额/车均消费）。
+    身份口径同「真实购买人数」（uid 优先、默认头像按 user_name 去重）；匿名无稳定身份不进排行、单列汇总。
+    参与金额按团实收 total_price 分摊到份（金额合计 ≈ 平台销售额），依赖 super_vault_order_spider.py 先抓订单。
 
 从项目根目录运行：python stats/super_vault_nova_daily_report.py（cwd=根目录，读根目录 application.yml）
 """
 import os
+import re
 import sys
 import time
-# 把项目根目录加入 import 路径（读根目录 application.yml、复用根目录公共模块）
-sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import glob
+from datetime import datetime, timedelta
+# 项目根目录（本文件在 stats/ 下，上一级即根目录）：公共模块 mysql_pool/YamlLoader、
+# 配置 application.yml、日志目录 logs 统一只在根目录留一份，stats 内不再放副本
+ROOT_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT_DIR)
+# 必须在 import mysql_pool 之前切到根目录：mysql_pool 在「导入期」就按 cwd 相对路径读
+# application.yml（读不到会直接 UnboundLocalError），而配置只在根目录留一份。
+# 入口脚本已切过，这里是幂等的；单独跑本文件时则靠这行兜住任意启动目录。
+os.chdir(ROOT_DIR)
 
 import schedule
 from loguru import logger
@@ -31,21 +55,35 @@ from mysql_pool import MySQLConnectionPool
 # 复用累计报告脚本的渲染组件与样式，不重复造轮子；口径相关的 fetch_* 在本文件按 sold_time 重写
 import super_vault_his_report as base
 from super_vault_his_report import (
-    DETAIL_COLS, SUMMARY_HEADERS, SECTION_SPAN, FONT_TITLE,
+    DETAIL_COLS, SUMMARY_HEADERS, SECTION_SPAN, FONT_TITLE, REAL_BUYERS_EXPR,
     _fmt_dt, _fmt_duration, _pack_summary,
     _style_row, _write_summary_block, _write_section_title,
     _write_anchor_table, _write_details,
 )
 
-# 日志：按天切分，保留 7 天（覆盖 base 模块 import 时设置的 sink）
-logger.remove()
-logger.add("./logs/nova_daily_report_{time:YYYYMMDD}.log", encoding="utf-8", rotation="00:00",
-           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}", level="INFO", retention="7 day")
+# 日志统一写根目录 logs/{日期}.log，由入口脚本配置（见文件末尾 __main__）。
+# 模块级绝不配置 sink：被 start_cxx_spider 等入口 import 时若在此 logger.remove()/add()，
+# 会摘掉入口已装好的 sink，并凭空创建一个随即被摘、永不写入的 0 字节日志文件。
+LOG_PATH = os.path.join(ROOT_DIR, "logs", "{time:YYYYMMDD}.log")
 
 OUT_PREFIX = "超新星已售每日报告"    # 输出文件名前缀，后缀加统计日期
 
+# 历史报告保留天数：每日产出一份 xlsx，长期无人值守会堆满磁盘，只留最近 N 天（按文件名里的统计日期算）
+KEEP_DAYS = 30
+# 匹配本脚本产出的文件名：{前缀}_YYYYMMDD.xlsx。清理只认这种严格格式，手工改名/其他报表一律不碰
+OUT_FILE_PATTERN = re.compile(rf"^{re.escape(OUT_PREFIX)}_(\d{{8}})\.xlsx$")
+
 # 企微发送开关：默认关闭；接入企微机器人后置 True 即启用（发送模块 auto_send_wx_msg.py 在同目录）
-SEND_WECHAT = False
+SEND_WECHAT = True
+
+# 人数口径（中卡人数/真实购买人数）与列结构、渲染均复用 base（super_vault_his_report），
+# 本文件仅按 sold_time「昨天」窗口重写 fetch_*；REAL_BUYERS_EXPR/DETAIL_COLS 等直接 import。
+
+
+def amount_expr(alias: str) -> str:
+    """生成销售额 SQL 表达式：优先用实收总价 total_price，缺失时才回退满仓价。"""
+    return (f"CASE WHEN {alias}.total_price IS NOT NULL "
+            f"THEN {alias}.total_price ELSE {alias}.sign_price * {alias}.count END")
 
 
 def win_clause(alias: str) -> str:
@@ -85,23 +123,27 @@ def fetch_platform_summary(pool) -> dict:
         pool (MySQLConnectionPool): MySQL 连接池。
 
     Returns:
-        dict: 含 主播数/销售额/成团数/参与人数/均拼单价/人均消费。
+        dict: 含 主播数/销售额/成团数/中卡人数/真实购买人数/均拼单价/人均消费。
     """
     sql = f"""
     SELECT
-        ROUND(SUM(p.sign_price * p.count), 2)                              AS amount,
+        ROUND(SUM({amount_expr('p')}), 2)                                  AS amount,
         COUNT(DISTINCT p.anchor_username)                                  AS anchors,
         COUNT(*)                                                            AS grp,
         (SELECT COUNT(DISTINCT rr.user_name)
            FROM super_vault_report_record rr
            JOIN super_vault_product_record pp ON pp.pid = rr.pid
           WHERE {win_clause('pp')}
-            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people
+            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people,
+        (SELECT {REAL_BUYERS_EXPR}
+           FROM super_vault_order_record o
+           JOIN super_vault_product_record pp2 ON pp2.pid = o.pid
+          WHERE {win_clause('pp2')})                                        AS real_people
     FROM super_vault_product_record p
     WHERE {win_clause('p')}
     """
-    amount, anchors, groups, people = pool.select_all(sql)[0]
-    return _pack_summary(amount, groups, people, extra={"主播数": anchors})
+    amount, anchors, groups, people, real_people = pool.select_all(sql)[0]
+    return _pack_summary(amount, groups, people, real_people, extra={"主播数": anchors})
 
 
 def fetch_anchor_summaries(pool) -> list[dict]:
@@ -111,18 +153,22 @@ def fetch_anchor_summaries(pool) -> list[dict]:
         pool (MySQLConnectionPool): MySQL 连接池。
 
     Returns:
-        list[dict]: 每元素含 主播名/销售额/成团数/参与人数/均拼单价/人均消费。
+        list[dict]: 每元素含 主播名/销售额/成团数/中卡人数/真实购买人数/均拼单价/人均消费。
     """
     sql = f"""
     SELECT
         p.anchor_username                                                  AS aname,
-        ROUND(SUM(p.sign_price * p.count), 2)                              AS amount,
+        ROUND(SUM({amount_expr('p')}), 2)                                  AS amount,
         COUNT(*)                                                            AS grp,
         (SELECT COUNT(DISTINCT rr.user_name)
            FROM super_vault_report_record rr
            JOIN super_vault_product_record pp ON pp.pid = rr.pid
           WHERE pp.anchor_username = p.anchor_username AND {win_clause('pp')}
-            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people
+            AND rr.user_name IS NOT NULL AND rr.user_name <> '')            AS people,
+        (SELECT {REAL_BUYERS_EXPR}
+           FROM super_vault_order_record o
+           JOIN super_vault_product_record pp2 ON pp2.pid = o.pid
+          WHERE pp2.anchor_username = p.anchor_username AND {win_clause('pp2')}) AS real_people
     FROM super_vault_product_record p
     WHERE {win_clause('p')}
     GROUP BY p.anchor_username
@@ -130,8 +176,8 @@ def fetch_anchor_summaries(pool) -> list[dict]:
     """
     rows = pool.select_all(sql) or []
     result = []
-    for aname, amount, groups, people in rows:
-        d = _pack_summary(amount, groups, people)
+    for aname, amount, groups, people, real_people in rows:
+        d = _pack_summary(amount, groups, people, real_people)
         d["主播名"] = aname or "(未知主播)"
         result.append(d)
     return result
@@ -147,15 +193,18 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
         aname (str): 主播名 anchor_username。
 
     Returns:
-        list[dict]: 每元素为一条明细（团名/编号/卡种/规格/单价/总份数/进度/总金额/参与人数/开售时间/成交时间/售卖时长）。
+        list[dict]: 每元素为一条明细（团名/编号/卡种/规格/单价/总份数/进度/总金额/中卡人数/真实购买人数/开售时间/成交时间/售卖时长）。
     """
     sql = f"""
     SELECT
         p.title, p.serial, p.type_name, p.standard_name, p.sign_price, p.sold_count, p.count,
-        ROUND(p.sign_price * p.count, 2)                                 AS amount,
-        -- 该团参与人数：拆卡报告 user_name 去重（中卡近似，仅覆盖有报告的团；待发货团常尚无报告，为 0）
+        {amount_expr('p')}                                                AS amount,
+        -- 该团中卡人数：拆卡报告 user_name 去重（抽中卡片的人，仅覆盖有报告的团；待发货团常尚无报告，为 0）
         (SELECT COUNT(DISTINCT rr.user_name) FROM super_vault_report_record rr
           WHERE rr.pid = p.pid AND rr.user_name IS NOT NULL AND rr.user_name <> '') AS buyers,
+        -- 该团真实购买人数：订单表按 uid/user_name 去重 + 匿名每笔计 1（未抓订单的团为 0）
+        (SELECT {REAL_BUYERS_EXPR} FROM super_vault_order_record o
+          WHERE o.pid = p.pid)                                           AS real_buyers,
         p.sell_time,
         p.sold_time                                                      AS deal_time,
         TIMESTAMPDIFF(SECOND, p.sell_time, p.sold_time)                  AS duration_secs
@@ -165,13 +214,14 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
     """
     rows = pool.select_all(sql, (aname,)) or []
     result = []
-    for (title, serial, type_name, standard, price, sold, total, amount, buyers,
+    for (title, serial, type_name, standard, price, sold, total, amount, buyers, real_buyers,
          sell_time, deal_time, duration_secs) in rows:
         progress = round(sold / total * 100, 1) if total else None  # 售卖进度百分比
         result.append({
             "团名": title, "编号": serial, "卡种": type_name, "规格": standard,
             "单价": price, "总份数": total, "进度": progress, "总金额": amount,
-            "参与人数": buyers,
+            "中卡人数": buyers,                    # 拆卡报告中卡去重（原「参与人数」，非真实参与者）
+            "真实购买人数": int(real_buyers or 0),   # 订单表真实买家去重（Decimal 转 int）
             # 成交时间 = 售罄成交时刻 sold_time；售卖时长 = sold_time − 开售时间
             "开售时间": _fmt_dt(sell_time), "成交时间": _fmt_dt(deal_time),
             "售卖时长": _fmt_duration(duration_secs),
@@ -179,6 +229,121 @@ def fetch_anchor_details(pool, aname: str) -> list[dict]:
     return result
 
 
+# 「用户明细」Sheet 列规格：(表头, 取值键, 是否金额格式)；「排名」复用 _write_details 的行号自增
+USER_DETAIL_COLS = [
+    ("排名", "序号", False), ("用户昵称", "用户昵称", False), ("user_id", "user_id", False),
+    ("参与车数", "参与车数", False), ("参与金额", "参与金额", True), ("车均消费", "车均消费", True),
+]
+
+
+def _share_expr() -> str:
+    """生成「按团实收分摊到份」的单用户金额 SQL 表达式。
+
+    用户在某团的花费 = 团实收（优先 total_price，缺失回退满仓价）× 该用户份数 ÷ 团总份数，
+    与平台/主播汇总的 total_price 实收口径一致，避免用户明细金额合计与平台销售额对不上。
+
+    Returns:
+        str: 可嵌入 SUM() 的金额分摊表达式。
+    """
+    # NULLIF 防团总份数为 0 时除零（售罄团 count>0，仅兜底）
+    return f"({amount_expr('p')}) * o.count / NULLIF(p.count, 0)"
+
+
+def fetch_user_details(pool) -> list[dict]:
+    """统计昨天成交团的真实买家明细（非匿名），按参与金额倒序、全部返回。
+
+    身份口径与报告「真实购买人数」一致：登录用户按 uid、默认头像用户按 user_name 去重
+    （GROUP BY COALESCE(uid, user_name)）。匿名订单无稳定身份，不在此列（见 fetch_anon_summary）。
+
+    Args:
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        list[dict]: 每元素为 {用户昵称, user_id, 参与车数, 参与金额, 车均消费}；无数据时为空列表。
+    """
+    sql = f"""
+    SELECT
+        MAX(o.user_name)                      AS user_name,
+        MAX(o.uid)                            AS uid,
+        COUNT(DISTINCT o.pid)                 AS cars,
+        ROUND(SUM({_share_expr()}), 2)        AS amount
+    FROM super_vault_order_record o
+    JOIN super_vault_product_record p ON p.pid = o.pid
+    WHERE {win_clause('p')} AND o.is_anonymous = 0
+    GROUP BY COALESCE(o.uid, o.user_name)
+    ORDER BY amount DESC
+    """
+    rows = pool.select_all(sql) or []
+    result = []
+    for user_name, uid, cars, amount in rows:
+        amount = float(amount or 0)
+        result.append({
+            "用户昵称": user_name,
+            "user_id": uid,                                      # 默认头像用户无 uid，留空
+            "参与车数": cars,
+            "参与金额": amount,
+            "车均消费": round(amount / cars, 2) if cars else None,  # 车均 = 参与金额 ÷ 参与车数
+        })
+    return result
+
+
+def fetch_anon_summary(pool) -> tuple[int, float]:
+    """统计昨天成交团的匿名订单笔数与合计金额（无稳定身份、无法去重，仅汇总展示）。
+
+    Args:
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        tuple[int, float]: (匿名订单笔数, 匿名合计金额)。
+    """
+    sql = f"""
+    SELECT COUNT(*) AS cnt, ROUND(SUM({_share_expr()}), 2) AS amount
+    FROM super_vault_order_record o
+    JOIN super_vault_product_record p ON p.pid = o.pid
+    WHERE {win_clause('p')} AND o.is_anonymous = 1
+    """
+    cnt, amount = pool.select_all(sql)[0]
+    return int(cnt or 0), float(amount or 0)
+
+
+def build_user_sheet(wb, pool):
+    """在工作簿追加「用户明细」Sheet：真实买家按参与金额倒序排行，匿名单独汇总。
+
+    金额口径与主报告一致（团实收 total_price 分摊到份），故本表金额合计 ≈ 平台销售额。
+
+    Args:
+        wb (Workbook): openpyxl 工作簿（已含主报告 Sheet）。
+        pool (MySQLConnectionPool): MySQL 连接池。
+    """
+    users = fetch_user_details(pool)
+    anon_cnt, anon_amount = fetch_anon_summary(pool)
+
+    ws = wb.create_sheet("用户明细")
+    span = len(USER_DETAIL_COLS)
+    r = 1
+
+    # 标题行
+    ws.cell(row=r, column=1,
+            value=(f"用户明细 · 超新星（共 {len(users)} 人参与，全部展示，按参与金额倒序；"
+                   f"金额按团实收 total_price 分摊到份）"))
+    _style_row(ws, r, span, font=FONT_TITLE, border=False)
+    r += 1
+
+    # 副标题：匿名汇总（匿名无稳定身份、不进上方排行，单独说明）
+    ws.cell(row=r, column=1,
+            value=f"另有匿名订单 {anon_cnt} 笔 / 合计 {anon_amount:,.2f}（无稳定身份，未计入上方排行）")
+    _style_row(ws, r, span, font=base.FONT_HEADER, border=False)
+    r += 2  # 留一空行
+
+    # 排行表（复用明细渲染组件，「排名」列由 _write_details 按行号自增）
+    _write_details(ws, r, users, USER_DETAIL_COLS)
+
+    # 列宽 + 冻结标题/表头（标题1 / 副标题2 / 空行3 / 表头4 / 数据5+）
+    for i, w in enumerate([8, 22, 16, 12, 16, 14]):
+        ws.column_dimensions[chr(65 + i)].width = w
+    ws.freeze_panes = "A5"
+
+
 def build_report(pool, out: str):
     """汇总各段数据并生成单 Sheet 分区 Excel（超新星·已售每日报告）。
 
@@ -218,24 +383,72 @@ def build_report(pool, out: str):
         details = fetch_anchor_details(pool, a["主播名"])
         ws.cell(row=r, column=1,
                 value=(f"　▼ {a['主播名']}　销售额 {a['销售额']:,.2f} / 成团 {a['成团数']} / "
-                       f"参与 {a['参与人数']}（共 {len(details)} 条明细）"))
+                       f"中卡 {a['中卡人数']} / 真实 {a['真实购买人数']}（共 {len(details)} 条明细）"))
         _style_row(ws, r, SECTION_SPAN, font=base.FONT_HEADER, border=False)
         r += 1
         r = _write_details(ws, r, details, DETAIL_COLS)
         r += 2
 
-    # 列宽：A-M 兼顾明细表与主播汇总表；E 列取 14 以容纳主播汇总的「均拼单价」大额（否则显示 ######）
-    widths = [8, 50, 14, 18, 14, 11, 10, 8, 14, 12, 20, 20, 14]
+    # 列宽：在原明细列基础上，「中卡人数」后插入「真实购买人数」列宽（14 列）
+    widths = [8, 50, 14, 18, 14, 11, 10, 8, 14, 12, 14, 20, 20, 14]
     for i, w in enumerate(widths):
         ws.column_dimensions[chr(65 + i)].width = w
     ws.freeze_panes = "A2"  # 冻结标题行
 
+    # 追加「用户明细」Sheet（真实买家按参与金额排行 + 匿名汇总），口径同主报告
+    build_user_sheet(wb, pool)
+
     wb.save(out)
 
 
+def cleanup_old_reports(log, out_dir: str, keep_days: int = KEEP_DAYS) -> int:
+    """清理超出保留期的历史报告 xlsx，只留最近 keep_days 天。
+
+    保留期按**文件名里的统计日期**判断（而非文件 mtime）——报告可能被重跑、被拷贝，
+    mtime 会变而统计日期不会，用它做基准才和「只留最近 30 天的报告」语义一致。
+    文件名不符合 {前缀}_YYYYMMDD.xlsx 的一律跳过，避免误删手工文件或其他报表。
+
+    Args:
+        log: 日志对象。
+        out_dir (str): 报告所在目录。
+        keep_days (int, optional): 保留天数，按统计日期算。Defaults to KEEP_DAYS。
+
+    Returns:
+        int: 实际删除的文件数；无可删文件时返回 0。
+    """
+    # 以「今天」为基准倒推保留边界：统计日期早于该日的报告删掉
+    cutoff = (datetime.now() - timedelta(days=keep_days)).date()
+    removed = 0
+    # glob 先粗筛同前缀文件，再用正则严格校验日期段，双重过滤防误删
+    for path in glob.glob(os.path.join(out_dir, f"{OUT_PREFIX}_*.xlsx")):
+        m = OUT_FILE_PATTERN.match(os.path.basename(path))
+        if not m:
+            continue
+        try:
+            file_date = datetime.strptime(m.group(1), "%Y%m%d").date()
+        except ValueError:
+            # 8 位数字但不是合法日期（如 20261340），当异常文件留着，不猜不删
+            log.warning(f"报告文件名日期非法，跳过清理: {os.path.basename(path)}")
+            continue
+        if file_date >= cutoff:
+            continue
+        try:
+            os.remove(path)
+            removed += 1
+            log.info(f"清理过期报告: {os.path.basename(path)}")
+        except OSError as e:
+            # 占用中（Excel 打开着）或权限不足：仅告警，下轮再清，不影响报告产出
+            log.warning(f"清理失败 {os.path.basename(path)}: {e}")
+    if removed:
+        log.info(f"过期报告清理完成，共删除 {removed} 个（保留最近 {keep_days} 天）")
+    return removed
+
+
 def run_once(log) -> str:
     """连库生成超新星·已售每日报告（统计昨天成交）。
 
+    生成并发送后顺带清理超出保留期的历史 xlsx（见 cleanup_old_reports）。
+
     Args:
         log: 日志对象。
 
@@ -253,12 +466,18 @@ def run_once(log) -> str:
     log.info(f"报告已生成 -> {out_file}")
 
     # 发企微群（只发 Excel；失败仅告警，不影响报告产出）
-    # if SEND_WECHAT:
-    #     try:
-    #         from auto_send_wx_msg import send_wechat_group_file
-    #         send_wechat_group_file(log=log, file_path=out_file)
-    #     except Exception as e:
-    #         log.warning(f"企微发送跳过: {e}")
+    if SEND_WECHAT:
+        try:
+            from auto_send_wx_msg import send_wechat_group_file
+            send_wechat_group_file(log=log, file_path=out_file)
+        except Exception as e:
+            log.warning(f"企微发送跳过: {e}")
+
+    # 清理过期报告放在发送之后：确保本次产物已送出，清理异常也不会影响报告与推送
+    try:
+        cleanup_old_reports(log, out_dir)
+    except Exception as e:
+        log.warning(f"过期报告清理跳过: {e}")
     return out_file
 
 
@@ -277,4 +496,10 @@ def schedule_task():
 
 
 if __name__ == "__main__":
-    main()
+    # 独立运行才配置日志 sink：按天切分、保留 7 天，与入口脚本同格式同去向（根目录 logs/）
+    logger.remove()
+    logger.add(LOG_PATH, encoding="utf-8", rotation="00:00",
+               format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
+               level="DEBUG", retention="7 day")
+    # main()
+    schedule_task()

cxx_spider/super_vault_order_spider.py

@@ -0,0 +1,366 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/21
+"""超级仓库 SuperVault · 购买订单（真实买家）爬虫。
+
+背景：
+    已售统计报告 stats/super_vault_his_report.py 的「参与人数」原本用拆卡报告
+    super_vault_report_record.user_name 去重近似——那只是「中卡用户」，偏低、也不是真实买家。
+    order/simple/list 接口按「每笔购买订单」返回买家(username/avatar) + 下单时间 + 份数，
+    用它可统计每个拼团商品的「真实购买人数」，补上报告缺的真实买家维度。
+
+接口探测结论（2026/08/21 实测）：
+    - URL：POST https://cxx.cardsvault.net/app/order/simple/list，body {"tid","pageNum","pageSize"}。
+    - 免登录：不带 Authorization 也返回 200，故本脚本不带任何 token（登录态最小化）。
+    - pageSize 被服务端锁死为 9：请求 100/200 均只回 9 条，无法加大。
+    - 深翻页可用：可一路翻到末页（超尾页会被钳到最后一页），故能事后抓已售商品的全部订单。
+    - tid == pid（与拆卡报告口径一致，即 super_vault_product_record.pid）。
+
+去重/幂等策略（重要）：
+    源接口无订单号，且 (user_name, order_time, count) 在源头就不唯一（热门团存在「多个匿名同秒各买
+    1 份」等真实但同键的不同订单，实测大团约 0.1%）。故不建唯一键、不靠该组合去重——那会误删真实订单。
+    改为按 pid「删旧 + 整体重灌」：每次抓完一个商品的全部订单后，先删该 pid 旧行再批量写入，
+    幂等且不丢任何一笔。表上仅留 idx_pid 普通索引。
+
+真实购买人数口径（留给报告层，本爬虫只如实存每一笔）：
+    登录用户头像路径内嵌 uid（/avatar/{uid}/avatar_pic/...），可精确去重且识别复购；
+    默认头像用户无 uid 但 user_name 为系统稳定 handle，也可按 user_name 去重；
+    唯独「匿名」天生无法区分（多个匿名同名同默认头像）。参考统计 SQL：
+        SELECT
+          COUNT(*)                                         AS 订单笔数,
+          SUM(`count`)                                     AS 总份数,
+          COUNT(DISTINCT COALESCE(uid, user_name))         AS 去重身份数,
+          SUM(is_anonymous)                                AS 匿名笔数,
+          -- 真实购买人数 ≈ 非匿名去重身份 + 匿名笔数(每笔按1人,上界近似)
+          COUNT(DISTINCT CASE WHEN is_anonymous=0 THEN COALESCE(uid, user_name) END)
+            + SUM(is_anonymous)                            AS 真实购买人数近似
+        FROM super_vault_order_record WHERE pid = %s;
+
+运行方式：
+    从项目根目录运行 python super_vault_order_spider.py（读根目录 application.yml）。
+    首次为全量回填(984 个已售商品，数小时)，可断点续跑：每个商品抓完置 order_state=1，
+    重跑只补 order_state 未完成(NULL/0/3)的商品；之后每天增量抓新增已售商品，成本很低。
+"""
+import os
+import re
+import sys
+import time
+import math
+# 必须在 import mysql_pool 之前切到本文件所在目录（即项目根目录）：mysql_pool 在「导入期」
+# 就按 cwd 相对路径读 application.yml（读不到会直接 UnboundLocalError）。
+# 入口脚本已切过，这里是幂等的；单独跑本文件时则靠这行兜住任意启动目录。
+ROOT_DIR = os.path.dirname(os.path.abspath(__file__))
+os.chdir(ROOT_DIR)
+
+import requests
+import schedule
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+from mysql_pool import MySQLConnectionPool
+
+# 日志统一写根目录 logs/{日期}.log，由入口脚本配置（见文件末尾 __main__）。
+# 模块级绝不配置 sink：被 start_cxx_spider 等入口 import 时若在此 logger.remove()/add()，
+# 会摘掉入口已装好的 sink（或反被后续 import 的模块摘掉），日志去向变成由 import 顺序决定。
+LOG_PATH = os.path.join(ROOT_DIR, "logs", "{time:YYYYMMDD}.log")
+
+# 接口地址
+ORDER_LIST_URL = "https://cxx.cardsvault.net/app/order/simple/list"
+
+# 请求头：免登录，故不带 Authorization（登录态最小化）
+HEADERS = {
+    "User-Agent": "okhttp/4.9.0",
+    "CXX-APP-API-VERSION": "V2",  # 必须添加
+    "Content-Type": "application/json; charset=UTF-8",
+}
+
+PAGE_SIZE = 9          # 服务端锁死为 9，请求更大值无效，不要改
+USE_PROXY = False      # 代理开关：默认直连，遇 IP 风控再置 True
+PAGE_SLEEP = 0.2       # 每页请求间隔(秒)，限速避免触发风控
+PRODUCT_SLEEP = 0.5    # 每个商品之间间隔(秒)
+MAX_PRODUCTS_PER_RUN = 0  # 单轮最多处理的商品数，0=不限（断点续跑靠 order_state，全量可放心跑满）
+
+# order_state 状态含义（写入 super_vault_product_record.order_state）
+#   NULL       = 从未抓取
+#   0(ACTIVE)  = 已抓但商品仍在售(未售罄)，订单还会增长，下轮继续重灌刷新
+#   1(DONE)    = 已售罄/冻结，抓完不再重抓
+#   2(EMPTY)   = 无购买记录
+#   3(FAIL)    = 抓取失败/中断，下轮重试
+STATE_ACTIVE = 0   # 在售未售罄，已抓但待刷新
+STATE_DONE = 1     # 售罄冻结，已抓完
+STATE_EMPTY = 2    # 无购买记录(total=0)
+STATE_FAIL = 3     # 抓取失败/中断
+
+# 从头像 url 提取用户 uid：形如 .../avatar/1444/avatar_pic/xxx.jpg → 1444；默认头像无此段
+UID_PATTERN = re.compile(r"/avatar/(\d+)/")
+
+
+def after_log(retry_state):
+    """tenacity 重试回调，记录每次尝试的结果。
+
+    Args:
+        retry_state: tenacity 传入的 RetryCallState，其 args[0] 约定为日志对象。
+    """
+    log = retry_state.args[0] if retry_state.args else 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):
+    """获取隧道代理配置。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        dict: requests 可用的 proxies 字典。
+
+    Raises:
+        Exception: 组装代理配置异常时向上抛出以触发重试。
+    """
+    tunnel = "x371.kdltps.com:15818"
+    kdl_username = "t13753103189895"
+    kdl_password = "o0yefv6z"
+    try:
+        proxy = f"http://{kdl_username}:{kdl_password}@{tunnel}/"
+        return {"http": proxy, "https": proxy}
+    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_order_single_page(log, pid, page_num):
+    """获取某拼团商品单页的购买订单数据（带重试）。
+
+    Args:
+        log: 日志对象。
+        pid (int): 拼团商品 id（接口里的 tid）。
+        page_num (int): 页码，从 1 开始。
+
+    Returns:
+        dict | None: {"total": int, "current_page": int, "items": list}；业务码非 200 时返回 None。
+
+    Raises:
+        requests.HTTPError: HTTP 状态码非 2xx 时抛出以触发重试。
+    """
+    data = {"tid": str(pid), "pageNum": page_num, "pageSize": PAGE_SIZE}
+    proxies = get_proxys(log) if USE_PROXY else None
+    response = requests.post(ORDER_LIST_URL, headers=HEADERS, json=data, timeout=(5, 30), proxies=proxies)
+    response.raise_for_status()
+
+    result = response.json()
+    if result.get("status") == 200:
+        d = result.get("data", {}) or {}
+        total = d.get("total", 0)
+        current_page = d.get("pageNum", page_num)
+        items = d.get("data", []) or []
+        log.debug(f"pid={pid} 第 {current_page} 页 / 共 {total} 条，本页 {len(items)} 条")
+        return {"total": total, "current_page": current_page, "items": items}
+
+    log.error(f"pid={pid} 第 {page_num} 页 API 返回错误: {result.get('msg', '未知错误')}")
+    return None
+
+
+def parse_order_items(pid, items):
+    """把接口返回的订单列表解析成待入库的 dict 列表。
+
+    Args:
+        pid (int): 拼团商品 id，写入每行。
+        items (list[dict]): 接口 data.data，每元素含 userInfo/createTime/count。
+
+    Returns:
+        list[dict]: 每元素为 {pid, user_name, uid, is_anonymous, avatar, count, order_time}。
+    """
+    parsed = []
+    for item in items:
+        user_info = item.get("userInfo", {}) or {}
+        user_name = user_info.get("username")
+        avatar = user_info.get("avatar")
+
+        # 从头像路径提取 uid（登录用户才有；默认头像/匿名为空）
+        m = UID_PATTERN.search(avatar) if avatar else None
+        uid = int(m.group(1)) if m else None
+
+        parsed.append({
+            "pid": pid,
+            "user_name": user_name,
+            "uid": uid,
+            "is_anonymous": 1 if user_name == "匿名" else 0,  # 匿名天生无法区分，单独标记供报告层处理
+            "avatar": avatar,
+            "count": item.get("count"),          # 该笔购买份数
+            "order_time": item.get("createTime"),  # 下单时间
+        })
+    return parsed
+
+
+def crawl_product_orders(log, sql_pool, pid, frozen=True):
+    """抓取单个拼团商品的全部购买订单并入库，返回落库条数。
+
+    策略「攒齐全部页 → 删旧 → 整体重灌」：先逐页翻完 ceil(total/9) 页、把所有订单攒进内存，
+    全部成功后再 DELETE 该 pid 旧行 + 批量插入。这样：
+      - 天然幂等：重跑=整体替换，不会累加重复；
+      - 不误删真实订单：不靠 (user,time,count) 唯一键去重，匿名同秒/同份数的不同订单都完整保留；
+      - 失败安全：任一页多次重试仍失败即抛出，此时既未删也未插，旧数据/状态原样保留，交上层记 STATE_FAIL。
+    源接口按 createTime 倒序，对已售罄(冻结)商品列表稳定，故一次翻页即为精确全量快照。
+
+    终态标记：售罄(frozen=True)商品订单已冻结，标 STATE_DONE 不再重抓；仍在售(frozen=False)的商品
+    订单会继续增长，标 STATE_ACTIVE 让下轮重灌刷新，直到售罄后被标 DONE。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): 数据库连接池。
+        pid (int): 拼团商品 id。
+        frozen (bool, optional): 该商品是否已售罄冻结。True 标 DONE，False 标 ACTIVE 待刷新。Defaults to True。
+
+    Returns:
+        int: 落库的订单条数；无购买记录返回 0。
+
+    Raises:
+        RuntimeError: 某页多次重试后仍返回空时抛出，触发上层记 STATE_FAIL。
+    """
+    # 第 1 页：拿 total 与本页数据
+    first = get_order_single_page(log, pid, 1)
+    if not first:
+        raise RuntimeError(f"pid={pid} 第 1 页返回空")
+
+    total = first["total"]
+    if total == 0:
+        # 售罄却无记录→标 EMPTY 不再抓；在售无记录→标 ACTIVE 待其产生订单
+        empty_state = STATE_EMPTY if frozen else STATE_ACTIVE
+        log.info(f"pid={pid} 无购买记录，置 order_state={empty_state}")
+        sql_pool.update_one_or_dict(
+            table="super_vault_product_record",
+            data={"order_state": empty_state},
+            condition={"pid": pid})
+        return 0
+
+    total_pages = math.ceil(total / PAGE_SIZE)
+    log.info(f"pid={pid} 共 {total} 笔订单，{total_pages} 页，frozen={frozen}，开始抓取")
+
+    # 攒齐所有页（先全部成功，再动数据库）
+    all_items = parse_order_items(pid, first["items"])
+    for page_num in range(2, total_pages + 1):
+        time.sleep(PAGE_SLEEP)
+        page = get_order_single_page(log, pid, page_num)
+        if not page:
+            raise RuntimeError(f"pid={pid} 第 {page_num} 页返回空")
+        all_items.extend(parse_order_items(pid, page["items"]))
+
+    # 全部页成功 → 删旧 + 整体重灌 + 标记状态（售罄=DONE 不再抓；在售=ACTIVE 待刷新）
+    final_state = STATE_DONE if frozen else STATE_ACTIVE
+    sql_pool.update_one_or_dict(
+        query="DELETE FROM super_vault_order_record WHERE pid = %s",
+        args=(pid,))
+    if all_items:
+        sql_pool.insert_many(table="super_vault_order_record", data_list=all_items)
+    sql_pool.update_one_or_dict(
+        table="super_vault_product_record",
+        data={"order_state": final_state},
+        condition={"pid": pid})
+    log.info(f"pid={pid} 抓取完成，落库 {len(all_items)} 笔，order_state={final_state}")
+    return len(all_items)
+
+
+def get_pending_products(log, sql_pool, limit=0):
+    """取待抓订单的已售商品列表（口径与已售报告一致），含「是否售罄冻结」标记。
+
+    已售 = sold_count>0 且 status NOT IN(0,1,2,3)；只挑 order_state 未完成的
+    （NULL/0=未抓或在售待刷新、3=失败重试），跳过 1=已抓完(售罄冻结)与 2=无记录；
+    按成交时间倒序，新团优先补。frozen = sold_count>=count（售罄），决定抓完后标 DONE 还是 ACTIVE。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): 数据库连接池。
+        limit (int, optional): 最多返回的商品数，0 表示不限。Defaults to 0。
+
+    Returns:
+        list[tuple[int, bool]]: 每元素为 (pid, frozen)，frozen 表示是否已售罄冻结。
+    """
+    sql = (
+        "SELECT pid, (count > 0 AND sold_count >= count) AS frozen FROM super_vault_product_record "
+        "WHERE sold_count > 0 AND status NOT IN (0, 1, 2, 3) "
+        "AND (order_state IS NULL OR order_state NOT IN (1, 2)) "
+        "ORDER BY COALESCE(completion_time, sold_time, sell_time) DESC")
+    if limit and limit > 0:
+        sql += f" LIMIT {int(limit)}"
+    rows = sql_pool.select_all(sql) or []
+    products = [(r[0], bool(r[1])) for r in rows]
+    log.info(f"待抓订单的已售商品 {len(products)} 个")
+    return products
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def cxx_order_main(log, limit=0):
+    """购买订单采集主函数（可断点续跑）。
+
+    Args:
+        log: 日志对象。
+        limit (int, optional): 单轮最多处理的商品数，0=不限。Defaults to 0。
+
+    Raises:
+        RuntimeError: 数据库连接池异常时抛出以触发重试。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 购买订单采集" + "." * 40)
+
+    sql_pool = MySQLConnectionPool(log=log)
+    if not sql_pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    try:
+        run_limit = limit or MAX_PRODUCTS_PER_RUN
+        products = get_pending_products(log, sql_pool, run_limit)
+        done, empty, fail = 0, 0, 0
+        for idx, (pid, frozen) in enumerate(products, 1):
+            log.info(f"[{idx}/{len(products)}] 处理 pid={pid} (frozen={frozen})")
+            try:
+                fetched = crawl_product_orders(log, sql_pool, pid, frozen)
+                if fetched == 0:
+                    empty += 1
+                else:
+                    done += 1
+            except Exception as e:
+                # 单个商品失败不拖垮整轮：记 order_state=3，下轮重试
+                log.error(f"pid={pid} 抓取失败: {e}")
+                fail += 1
+                sql_pool.update_one_or_dict(
+                    table="super_vault_product_record",
+                    data={"order_state": STATE_FAIL},
+                    condition={"pid": pid})
+            time.sleep(PRODUCT_SLEEP)
+        log.info(f"本轮结束：成功 {done}，无记录 {empty}，失败 {fail}，共处理 {len(products)}")
+    except Exception as e:
+        log.error(f"{sys._getframe().f_code.co_name} error: {e}")
+    finally:
+        log.info(f"{sys._getframe().f_code.co_name} 运行结束，等待下一轮" + "." * 20)
+
+
+def schedule_task():
+    """定时入口：启动即增量抓一次，并每天 02:00、08:00 各跑一次（08:00 为保险补跑）。
+
+    三次触发均为增量：靠 order_state 只补未完成(NULL/0/3)的团，已抓完(1)/无记录(2)跳过；
+    抓取按 pid「删旧重灌」天然幂等，schedule 单线程顺序执行不会并发重入，故重复跑安全、开销小。
+    08:00 一轮用于补 02:00 失败(3)/漏抓的团、重刷在售(0)团，以及 02:00~08:00 间新售罄的团。
+    """
+    # cxx_order_main(logger)  # 启动即先增量跑一次，不必等到 02:00
+    schedule.every().day.at("02:00").do(cxx_order_main, logger)
+    schedule.every().day.at("08:00").do(cxx_order_main, logger)  # 保险：再补一轮
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    # 独立运行才配置日志 sink：按天切分、保留 7 天，与入口脚本同格式同去向（根目录 logs/）
+    logger.remove()
+    logger.add(LOG_PATH, encoding='utf-8', rotation="00:00",
+               format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
+               level="DEBUG", retention="7 day")
+    # 全量回填（数小时，断点续跑）：直接 cxx_order_main(logger)
+    # 小样本验证：cxx_order_main(logger, limit=5)
+    # cxx_order_main(logger)
+    schedule_task()