feat(deca_on_sale): 新增得卡DECA每日采集爬虫及统计报表功能

- 实现常驻无人值守的商家和商品增量采集爬虫，支持两级采集管道
- 商家发现模块遍历商家展位列表，upsert写入商家状态数据
- 商品采集模块免token批量拉取全量在售商品，upsert写入商品主表及每日快照
- 实现商品详情信息补全逻辑，回填规格、系列、模式及时间信息
- 设计Token持久化及续签机制，保证接口鉴权签名有效性
- 新增支持签名计算和请求重试的通用HTTP请求函数
- 采后对账机制自动下架不在最新全量在售集合中的商品
- 定时任务调度每日多次自动执行采集主流程
- 统计报表模块生成综合Excel和图片报告，支持多Sheet和图形展示
- 增加企业微信群机器人自动发送生成的报表文件功能
- 采用日志和异常处理保障稳定运行，配置灵活方便后续扩展

deca_spider/on_sale/README.md

@@ -0,0 +1,154 @@
+# 得卡 DECA App 逆向分析文档
+
+> 目标：采集得卡「商家展位列表 / 在售商品列表」，供业务侧统计 **新增 / 商家账号 / 商品售卖进度**。
+> 核心难点：所有接口带 `signature` 请求头签名校验，部分接口需 JWT 登录态。
+> 结论：signature 已 100% 还原（无加壳、纯 Java），登录+续签闭环打通，可长期无人值守。
+> 日期：2026/08/02 ｜ Python 3.12.10
+
+---
+
+## 1. 目标信息
+
+- **App**：得卡 DECA `得卡DECA_1.0.0.apk`（appVersion 1.0.0）
+- **接口域名**：`https://api.decalive.com`
+- **网络栈**：OkHttp 5.3.2（`User-Agent: okhttp/5.3.2`）
+- **鉴权/加密措施**：
+  - 每个请求都带 `signature`（32 位十六进制，双重 MD5）+ `currentTime`（13 位毫秒时间戳）。
+  - 部分接口需 `Authorization: Bearer <JWT>`，access token 有效期 **15 分钟**。
+- **加固情况**：**无加壳**、无字符串加密、签名不依赖 Native（so）；混淆为中等（类名字母混淆，salt 明文硬编码）。
+
+### 涉及接口
+
+| 用途 | 方法 & 路径 | 是否需登录 | 请求体 |
+|---|---|---|---|
+| 商家展位列表 | `POST /api/v1/app/merchant/exhibition/list` | 否 | `{"page":N}` |
+| 在售商品列表 | `POST /api/v1/app/groupbuy/merchant/on-sale-list` | **是** | `{"merchantUserId","page","pageSize":20}` |
+| 密码登录 | `POST /api/v1/app/auth/password/login` | 否 | `{"countryCode","password","phone"}` |
+| Token 续签 | `POST /api/v1/app/auth/token/refresh` | 否 | `{"refreshToken"}` |
+
+---
+
+## 2. 分析过程
+
+1. **抓包**：得到 3 个业务请求 + 1 个登录请求的 curl（含 4 组真实 `signature`/`currentTime`/`body` 样本，用于最终验证）。
+2. **反编译**：jadx 1.5.3 反编译 APK（`--no-res`），产物 16044 个 java 文件，`classes.dex`/`classes2.dex` 均为真实业务码，无壳。
+3. **定位签名**：从头部装配器 `mi/w.java`（组装全部公共头 + `signature`）顺藤摸瓜到签名计算核心 `mf/c0.java`。
+
+### 关键类
+
+| 类 / 方法 | 作用 |
+|---|---|
+| `mi.w.f(Context, url)` | 头部装配器，组装公共头 + signature |
+| `mf.c0.a(Map)` | **签名计算核心**：生成 currentTime + 双重 MD5 |
+| `mf.c0.b(String)` | MD5 → 小写 hex |
+| `mf.b0` | 每个参数 entry 渲染为 `key`+`value`（无分隔符）的 lambda |
+| `eg.p` | Retrofit 认证接口（login / sms / wechat / aliyun / token refresh） |
+
+---
+
+## 3. 逆向思路
+
+- **障碍**：所有接口非签名不可用；商品接口还要登录态，且 token 15 分钟即过期，无法直接长期跑。
+- **突破**：
+  1. 在 `mf/c0.a()` 读出签名逻辑——参数排序 + 双重 MD5 + 固定 salt，salt 是明文常量。
+  2. 用抓包的 4 个样本反推确认拼接细节（key 升序、key 直接拼 value、无 `=` 无分隔符、只取时间戳后 6 位）。
+  3. 登录接口样本（无 Authorization）也能仅凭 body 算出一致签名 → 证明签名不依赖 token，可脚本自动登录。
+  4. 在 `eg/*.java` 发现 `token/refresh` + refreshToken 存储 → 续签闭环成立。
+
+---
+
+## 4. 算法要点
+
+### 4.1 signature 生成规则（已用 4 样本验证一致）
+
+```
+plain      = 参数按 key 升序，每项 "key"+"value" 直接拼接（无分隔符），剔除 value 为 null 的项
+inner      = md5(plain)                         # 32 位小写 hex
+signature  = md5( inner + currentTime[-6:] + "biu_card_nbclass" )
+```
+
+- 哈希：**MD5**，UTF-8，输出**小写** hex，**双重** MD5。
+- **固定 salt**（硬编码于 `mf/c0.java`）：`biu_card_nbclass`
+- `currentTime`：13 位毫秒时间戳，**同一个值**同时放进 `currentTime` 头；签名只用其**后 6 位**。
+- 空 body：`plain=""` → `inner=md5("")`，无特殊分支。
+- **Authorization / token 不参与**签名。
+
+### 4.2 Python 复现
+
+```python
+import hashlib, time
+SALT = "biu_card_nbclass"
+def md5(s): return hashlib.md5(s.encode("utf-8")).hexdigest()
+def make_signature(params: dict, current_time: str) -> str:
+    clean = {k: v for k, v in params.items() if v is not None}
+    plain = "".join(f"{k}{clean[k]}" for k in sorted(clean))
+    return md5(md5(plain) + current_time[-6:] + SALT)
+# current_time = str(int(time.time()*1000))，同一值放进 currentTime 头
+```
+
+### 4.3 验证样本（4/4 全中）
+
+| plain | currentTime | signature |
+|---|---|---|
+| `page1` | 1785662750434 | c0670b1219e885fcca8f81631f9f6731 |
+| `page2` | 1785662751198 | 0dea97b71395f132a7f09a2fcea432f9 |
+| `merchantUserId593731690page1pageSize20` | 1785664667848 | 5416ca70361e645f75eaf3e960057b53 |
+| `countryCode86passwordpass2022phone19521500850` | 1785665137653 | 81464e9bf4285adf61e314228c3ae19f |
+
+### 4.4 登录 / Token
+
+- `password/login` 响应 `data`：`accessToken` / `refreshToken` / `expiresIn`(秒，900=15分钟) / `userId` 等。
+- 续签：`token/refresh` body `{"refreshToken":...}`，返回同结构新 token。
+- 脚本策略：access 剩余 <60s 先续签，续签失败回退重新登录（登录已验证，绝对兜底）。
+
+---
+
+## 5. 踩坑记录
+
+1. **body 必须紧凑 JSON**：抓包 `Content-Length:10` 对应 `{"page":1}`（无空格）。发送用 `json.dumps(..., separators=(",",":"))`，否则 Content-Length 与实际不符。签名本身只用参数串，与 body 序列化格式无关。
+2. **Windows 控制台 GBK**：脚本打印 emoji/部分字符会 `UnicodeEncodeError`。运行加 `PYTHONIOENCODING=utf-8 PYTHONUTF8=1`，日志文件本身写 UTF-8 无碍。
+3. **token 15 分钟过期**：一轮采集可能超 15 分钟，必须在请求前动态检查并续签，不能只登录一次。
+4. **数据库握手被拒（2013）**：内网库 `100.64.0.25` 需网络/白名单就绪，TCP 通但 MySQL 握手断时，先排查 VPN/授权白名单，非代码问题。
+5. **jadx 插件 NPE**：为旧版本编译的 `jadx-ai-mcp` 插件会导致 jadx 启动崩溃，反编译前需临时清空插件配置。
+6. **很多商家 `activeGroupbuyCount=0`**：即当前无在售，属正常；采商品时对每个商家都会发一次接口确认。
+7. **微信自动发送库 `wxauto` 已下架**：PyPI 上原 `wxauto` 已被作者移除（`pip install wxauto` 报 `from versions: none`），现改推 `wxauto4`（对应微信 4.0 客户端，免费）/ `wxautox4`（付费 plus）。安装 `pip install wxauto4`；import 名即 `wxauto4`，用 `WeChat().SendFiles(filepath, who, exact)` 发文件。若 PC 微信仍是 3.9 旧版，则改用 `wxautox`。发送依赖微信窗口可见的 UI 自动化，无人值守时需保证微信一直登录、窗口不最小化。
+
+---
+
+## 6. 举一反三
+
+- **同类 App 签名套路**：`sign = hash( 排序参数串 + 时间戳(部分位) + 固定 salt )` 是移动端最常见范式。定位思路统一：先找组装请求头的拦截器/装配器，再找里面调用的 hash 工具方法，salt 常为明文常量。
+- **是否 URL 编码**：`mf.b0.a` 声明抛 `UnsupportedEncodingException`，暗示内部可能对 value 做 `URLEncoder.encode(v,"UTF-8")`（空格→`+`）。当前样本均无特殊字符无法区分；若日后遇到搜索关键词等含空格/中文的参数，需对 value 先编码再拼。
+- **嵌套/数组参数**：`c0` 会把嵌套对象递归转 TreeMap、数组保序后按 `toString` 拼接。当前接口均为扁平 body，遇到嵌套 body 时需另取样本校准渲染格式。
+- **长期无人值守要点**：签名 + 自动登录 + 续签三者齐备才算闭环；只解签名不解 token 续签，跑不过 15 分钟。
+
+---
+
+## 附：项目文件
+
+| 文件 | 说明 |
+|---|---|
+| `deca_daily_spider.py` | 常驻每日爬虫（登录→商家→在售商品+每日快照；采完自动出图并发微信） |
+| `deca_report.py` | 统计报表出图（KPI + 今日新增商家 + 各商家在售/售卖进度），生成 `reports/` 下 PNG，并按开关自动发微信 |
+| `deca_wechat.py` | 通过 `wxauto4` 把图片发到 PC 微信好友/群 |
+| `schema.sql` | 三张表 DDL：`deca_shop_record` / `deca_product_record` / `deca_product_daily_record` |
+| `stats.sql` | 统计查询 SQL（新增 / 商家账号 / 售卖进度趋势） |
+| `init_db.py` | 一键建表 |
+| `requirements.txt` | 依赖清单（版本实测） |
+
+**部署**：
+1. 装依赖：`pip install -r requirements.txt`（`wxauto4` 若受 `requires-python` 限制加 `--ignore-requires-python`）；本地公共库 `charley-utils` 需 editable 安装（见 requirements 尾注）。
+2. 建表：`python init_db.py`
+3. 常驻运行：`python deca_daily_spider.py`（每天 09:00 / 15:00 及启动各采集一轮，采完自动生成报表并发微信）。
+
+**报表与微信推送**：
+- 报表图输出到 `reports/deca_report_YYYYMMDD_HHMMSS.png`，命名到秒，一天多张互不覆盖。
+- 微信发送依赖 **PC 微信 4.0 已登录且窗口可见**；目标好友备注名须精确等于 `deca_report.WECHAT_TO`（默认 `backup`）。
+- 手动补发：`python deca_wechat.py "reports/xxx.png" backup`。
+- 关闭自动发送：把 `deca_report.SEND_WECHAT` 置为 `False`。
+
+**登录与 token（避开验证码）**：
+- 密码登录 `password/login` 受阿里云验证码2.0 风控，脚本无法自动过验证码；改走 `token/refresh` 续期（**无验证码**）。
+- 首次：在 APP 正常登录后抓包，把 `refreshToken` 粘入项目根目录 `token.json` 的 `refresh` 字段：`{"access":null,"refresh":"<粘这里>","exp":0}`。
+- 之后爬虫用它自动续期并滚动写回 `token.json`；refreshToken 失效（过期/被踢）时日志会报警，需重新抓包更新。
+- 商品**上架时间/结束时间**来自详情接口 `POST api/v1/app/groupbuy/detail`（body `{"code":商品code}`，需 token），**仅对今日新增商品拉取**，存入 `deca_product_record` 的 `publish_at`/`sale_start_at`/`sale_end_at`。商家入驻时间接口无，新增按入库时间判定。

deca_spider/on_sale/deca_on_sale_daily_spider.py

@@ -0,0 +1,744 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/02
+"""得卡 DECA 每日增量采集爬虫（常驻无人值守）。
+
+两级采集管道：
+    1. 商家发现：翻页遍历「商家展位列表」merchant/exhibition/list，
+       提取 userId/merchantName 等，upsert 写入 deca_onsale_shop_record（存最新状态）。
+    2. 商品采集：免 token 翻页拉「全量在售」home/search（空 query，saleStatus=0），
+       upsert 写入 deca_onsale_product_record（最新状态），并按天写入
+       deca_onsale_product_daily_record（每日售卖进度快照，用于统计推进趋势）。
+       变更（2026/08/07）：原为「逐商家遍历 on-sale-list[需 token]」，已改走免 token 的
+       home/search。对拍验证：home/search 与首页精选 recommend 返回的当前在售商品完全一致
+       （逐 code 交集相等），且商品项结构与 on-sale-list 同构，故 parse_product 直接复用。
+
+鉴权与签名（逆向自 APK mf/c0.java，已用 4 个抓包样本验证）：
+    - signature = md5( md5(plain) + currentTime[-6:] + "biu_card_nbclass" )
+      plain = 参数按 key 升序、key 直接拼 value、无分隔符（去掉 value 为 None 的项）。
+    - currentTime 为 13 位毫秒时间戳，同一值放进 currentTime 头与参与签名。
+    - 本任务全部业务接口（商家列表 / 全量在售 home/search / 商品详情）均无需 token，仅需 signature。
+    - token 机制（ensure_token/refresh_token 等）仍保留：供 km_spdier 的卡牌清单接口
+      groupbuy/card/list（需 Authorization: Bearer）复用，本模块自身不再调用。
+"""
+import os
+import sys
+import time
+import json
+import hashlib
+from datetime import date
+
+import requests
+import schedule
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+from mysql_pool import MySQLConnectionPool
+
+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")
+
+# ==================== 基础配置 ====================
+BASE = "https://api.decalive.com"  # 业务域名（来源：抓包）
+SALT = "biu_card_nbclass"          # 签名固定 salt（逆向自 mf/c0.java 硬编码常量）
+
+PAGE_SIZE = 20         # 在售商品每页条数（抓包实测 20，home/search 同）
+MAX_SHOP_PAGES = 50    # 商家列表翻页保护上限（接口1 无 total，靠 list 空停止 + 此上限兜底）
+MAX_PROD_PAGES = 100   # 单商家在售商品翻页保护上限（旧 on-sale-list 路径，get_product_list 仍保留）
+MAX_SEARCH_PAGES = 100 # 全量在售 home/search 翻页保护上限（当前约 143/20≈8 页，留足冗余）
+USE_PROXY = False      # 代理开关：默认直连，遇 IP 风控再置 True
+
+# 公共请求头（来源：抓包，四接口一致；currentTime/signature/Authorization 每次动态补）
+BASE_HEADERS = {
+    "appVersion": "1.0.1",
+    "buildType": "release",
+    "timeZone": "GMT+08:00",
+    "equipmentType": "Android",
+    "Content-Type": "application/json; charset=utf-8",
+    "Accept-Encoding": "gzip",
+    "User-Agent": "okhttp/5.3.2",
+}
+
+# Token 持久化文件：与上级 deca_sold_core.py 共用同一份（同账号，单一真相源），
+# 避免各存一份副本时 refreshToken 滚动更新后另一份变死副本、续期失效。
+# 上级(20:30~次日03:00)与本项目(09:00/15:00)运行时段不重叠，共用不会并发写冲突。
+# 变更日期：2026/08/06
+TOKEN_FILE = r"D:\work\2026-08-02(deca_spider)\token.json"
+
+# access token 缓存：access=当前令牌，refresh=续签令牌，exp=过期秒级时间戳
+_TOKEN = {"access": None, "refresh": None, "exp": 0}
+
+
+def after_log(retry_state):
+    """tenacity 重试回调，记录每次尝试的结果。
+
+    Args:
+        retry_state: tenacity 传入的 RetryCallState，含调用参数与结果。
+    """
+    # 约定挂了重试的业务函数首个位置参数为 log；取不到时回退全局 logger
+    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):
+    """获取隧道代理配置（默认不启用，见 USE_PROXY）。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        dict: requests 可用的 proxies 字典。
+
+    Raises:
+        Exception: 组装代理配置异常时向上抛出以触发重试。
+    """
+    tunnel = "x371.kdltps.com:15818"
+    user, pwd = "t13753103189895", "o0yefv6z"
+    try:
+        proxy = f"http://{user}:{pwd}@{tunnel}/"
+        return {"http": proxy, "https": proxy}
+    except Exception as e:
+        log.error(f"Error getting proxy: {e}")
+        raise e
+
+
+def _md5(s: str) -> str:
+    """对字符串做 UTF-8 MD5，返回小写十六进制。
+
+    Args:
+        s (str): 待哈希明文。
+
+    Returns:
+        str: 32 位小写十六进制 MD5。
+    """
+    return hashlib.md5(s.encode("utf-8")).hexdigest()
+
+
+def make_signature(params: dict, current_time: str) -> str:
+    """按逆向还原规则计算 signature 请求头。
+
+    规则(mf/c0.java)：plain = 参数按 key 升序、key 直接拼 value、无分隔符（剔除 None）；
+    signature = md5( md5(plain) + current_time[-6:] + SALT )。
+
+    Args:
+        params (dict): 参与签名的参数（POST 为 body 字段）。
+        current_time (str): 13 位毫秒时间戳字符串，同一值放进 currentTime 头。
+
+    Returns:
+        str: 32 位小写十六进制签名。
+    """
+    clean = {k: v for k, v in params.items() if v is not None}
+    plain = "".join(f"{k}{clean[k]}" for k in sorted(clean))
+    return _md5(_md5(plain) + current_time[-6:] + SALT)
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
+def do_request(log, path: str, body: dict, need_auth: bool = False) -> dict | None:
+    """通用带签名 POST 请求（带重试）。
+
+    自动补 currentTime + signature 头；need_auth 时补 Authorization。
+    body 以紧凑 JSON（无空格）发送，与抓包一致。
+
+    Args:
+        log: 日志对象。
+        path (str): 接口相对路径（以 / 开头）。
+        body (dict): JSON body，同时用于签名。
+        need_auth (bool, optional): 是否需要携带 Bearer token。Defaults to False。
+
+    Returns:
+        dict | None: 响应 JSON。
+
+    Raises:
+        RuntimeError: HTTP 状态码非 200 时抛出以触发重试。
+    """
+    ct = str(int(time.time() * 1000))
+    headers = BASE_HEADERS.copy()
+    headers["currentTime"] = ct
+    headers["signature"] = make_signature(body, ct)
+    if need_auth:
+        headers["Authorization"] = f"Bearer {ensure_token(log)}"
+
+    data = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+    proxies = get_proxys(log) if USE_PROXY else None
+    resp = requests.post(BASE + path, headers=headers, data=data, timeout=(5, 30), proxies=proxies)
+    if resp.status_code != 200:
+        log.error(f"请求失败 {resp.status_code}: {path}")
+        raise RuntimeError(f"HTTP {resp.status_code}")
+    return resp.json()
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
+def do_get(log, path: str, params: dict = None, need_auth: bool = False) -> dict | None:
+    """通用带签名 GET 请求（带重试）。
+
+    自动补 currentTime + signature 头；need_auth 时补 Authorization。签名参数取 query。
+
+    Args:
+        log: 日志对象。
+        path (str): 接口相对路径（以 / 开头）。
+        params (dict, optional): URL query 参数（同时用于签名）。Defaults to None。
+        need_auth (bool, optional): 是否需要携带 Bearer token。Defaults to False。
+
+    Returns:
+        dict | None: 响应 JSON。
+
+    Raises:
+        RuntimeError: HTTP 状态码非 200 时抛出以触发重试。
+    """
+    ct = str(int(time.time() * 1000))
+    headers = BASE_HEADERS.copy()
+    headers["currentTime"] = ct
+    headers["signature"] = make_signature(params or {}, ct)
+    if need_auth:
+        headers["Authorization"] = f"Bearer {ensure_token(log)}"
+
+    proxies = get_proxys(log) if USE_PROXY else None
+    resp = requests.get(BASE + path, headers=headers, params=params, timeout=(5, 30), proxies=proxies)
+    if resp.status_code != 200:
+        log.error(f"请求失败 {resp.status_code}: {path}")
+        raise RuntimeError(f"HTTP {resp.status_code}")
+    return resp.json()
+
+
+# ==================== 登录 / Token 管理 ====================
+def save_token():
+    """把当前 _TOKEN(access/refresh/exp) 持久化到 TOKEN_FILE，供进程重启后复用。"""
+    try:
+        with open(TOKEN_FILE, "w", encoding="utf-8") as f:
+            json.dump({"access": _TOKEN.get("access"), "refresh": _TOKEN.get("refresh"),
+                       "exp": _TOKEN.get("exp", 0)}, f)
+    except Exception as e:
+        logger.warning(f"token 持久化失败: {e}")
+
+
+def load_token():
+    """从 TOKEN_FILE 恢复 _TOKEN；文件不存在或损坏时静默跳过。"""
+    if not os.path.exists(TOKEN_FILE):
+        return
+    try:
+        with open(TOKEN_FILE, "r", encoding="utf-8") as f:
+            d = json.load(f)
+        _TOKEN["access"] = d.get("access")
+        _TOKEN["refresh"] = d.get("refresh")
+        _TOKEN["exp"] = d.get("exp", 0)
+    except Exception as e:
+        logger.warning(f"token 读取失败: {e}")
+
+
+def refresh_token(log) -> bool:
+    """用 refreshToken 续签 access token 并持久化（无验证码，是无人值守的主路径）。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        bool: 续签成功返回 True，否则 False。
+    """
+    if not _TOKEN.get("refresh"):
+        return False
+    try:
+        resp = do_request(log, "/api/v1/app/auth/token/refresh",
+                          {"refreshToken": _TOKEN["refresh"]}, need_auth=False)
+        data = (resp or {}).get("data") or {}
+        access = data.get("accessToken")
+        if not access:
+            log.warning(f"token 续签无返回 access: {resp.get('msg') if resp else None}")
+            return False
+        expires_in = int(data.get("expiresIn") or 900)
+        _TOKEN["access"] = access
+        if data.get("refreshToken"):
+            _TOKEN["refresh"] = data["refreshToken"]  # refreshToken 可能滚动更新
+        _TOKEN["exp"] = time.time() + expires_in
+        save_token()
+        log.info(f"token 续签成功，有效 {expires_in}s")
+        return True
+    except Exception as e:
+        log.warning(f"token 续签失败: {e}")
+        return False
+
+
+def ensure_token(log) -> str:
+    """返回一个有效的 access token：缓存有效→直接用；临期或缺失→用 refreshToken 续期。
+
+    仅走 token 续签，不再有账号密码登录兜底；refreshToken 失效时抛错，需更新 token.json。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        str: 有效的 access token。
+
+    Raises:
+        RuntimeError: refreshToken 缺失或续期失败时抛出。
+    """
+    # 1) 内存缓存仍有效(剩余>60s)
+    if _TOKEN.get("access") and time.time() < _TOKEN["exp"] - 60:
+        return _TOKEN["access"]
+    # 2) 内存没有 refreshToken 时，从持久化文件恢复
+    if not _TOKEN.get("refresh"):
+        load_token()
+    # 3) 有 refreshToken 就续期(无验证码，唯一路径)
+    if _TOKEN.get("refresh") and refresh_token(log):
+        return _TOKEN["access"]
+    raise RuntimeError("无法获取 access token：refreshToken 缺失或失效。"
+                       "请在 APP 重新登录后更新 token.json（可用 capture_token.py 自动抓取）。")
+
+
+# ==================== 一、商家发现 ====================
+def parse_shop(item: dict) -> dict | None:
+    """把商家展位列表项解析成 deca_onsale_shop_record 一行。
+
+    Args:
+        item (dict): merchant/exhibition/list 的 data.list 项。
+
+    Returns:
+        dict | None: 商家数据字典；无 userId 时返回 None。
+    """
+    uid = item.get("userId")
+    if not uid:
+        return None
+    return {
+        "merchant_user_id": str(uid),
+        "merchant_name": item.get("merchantName"),
+        "merchant_avatar_url": item.get("merchantAvatarUrl"),
+        "fans_count": item.get("fansCount"),
+        "active_groupbuy_count": item.get("activeGroupbuyCount"),
+    }
+
+
+def save_shops(log, rows: list, sql_pool) -> int:
+    """upsert 写入商家（存在则更新名称/头像/粉丝/在售数等动态字段）。
+
+    Args:
+        log: 日志对象。
+        rows (list[dict]): parse_shop 产出的商家字典列表。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本次写入/更新的商家数。
+    """
+    if not rows:
+        return 0
+    sql = ("INSERT INTO deca_onsale_shop_record "
+           "(merchant_user_id, merchant_name, merchant_avatar_url, fans_count, active_groupbuy_count) "
+           "VALUES (%s,%s,%s,%s,%s) "
+           "ON DUPLICATE KEY UPDATE "
+           "merchant_name=VALUES(merchant_name), merchant_avatar_url=VALUES(merchant_avatar_url), "
+           "fans_count=VALUES(fans_count), active_groupbuy_count=VALUES(active_groupbuy_count)")
+    args_list = [(r["merchant_user_id"], r["merchant_name"], r["merchant_avatar_url"],
+                  r["fans_count"], r["active_groupbuy_count"]) for r in rows]
+    sql_pool.insert_many(query=sql, args_list=args_list)
+    return len(args_list)
+
+
+def get_shop_list(log, sql_pool) -> int:
+    """翻页遍历商家展位列表，upsert 全部商家。
+
+    接口1 无 total 字段，靠 list 为空停止翻页，MAX_SHOP_PAGES 兜底。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 去重后发现的商家总数。
+    """
+    seen = set()
+    page = 1
+    while page <= MAX_SHOP_PAGES:
+        try:
+            resp = do_request(log, "/api/v1/app/merchant/exhibition/list", {"page": page})
+        except Exception as e:
+            log.error(f"商家列表第 {page} 页请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"商家列表返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        items = data.get("list") or []
+        if not items:
+            log.info(f"商家列表第 {page} 页无数据，停止翻页")
+            break
+
+        rows = []
+        for it in items:
+            row = parse_shop(it)
+            if row and row["merchant_user_id"] not in seen:
+                seen.add(row["merchant_user_id"])
+                rows.append(row)
+        save_shops(log, rows, sql_pool)
+        log.info(f"商家列表第 {page} 页完成，本页 {len(items)} 条，累计商家 {len(seen)}")
+        page += 1
+
+    return len(seen)
+
+
+# ==================== 二、商品采集 ====================
+def parse_product(item: dict) -> dict | None:
+    """把在售商品列表项解析成 deca_onsale_product_record 一行。
+
+    Args:
+        item (dict): on-sale-list 的 data.list 项（或商家内嵌 products 项，同构）。
+
+    Returns:
+        dict | None: 商品数据字典；无 code 时返回 None。
+    """
+    code = item.get("code")
+    if not code:
+        return None
+    merchant = item.get("merchant") or {}
+    return {
+        "product_code": code,
+        "merchant_user_id": str(merchant.get("merchantUserID")) if merchant.get("merchantUserID") else None,
+        "merchant_name": merchant.get("merchantName"),
+        "title": item.get("title"),
+        "card_product_title": item.get("cardProductTitle"),
+        "cover_image_url": item.get("coverImageUrl"),
+        "unit_price": item.get("unitPrice"),
+        "min_unit_price": item.get("minUnitPrice"),
+        "max_unit_price": item.get("maxUnitPrice"),
+        "card_count": item.get("cardCount"),
+        "sold_count": item.get("soldCount"),
+        "available_stock": item.get("availableStock"),
+        "groupbuy_status": item.get("groupbuyStatus"),
+        "groupbuy_status_name": item.get("groupbuyStatusName"),
+        "play_type": item.get("playType"),
+        "live_id": item.get("liveId")
+    }
+
+
+def save_products(log, rows: list, sql_pool) -> int:
+    """upsert 写入商品主表（更新售卖进度等动态字段）+ 当天售卖进度快照。
+
+    Args:
+        log: 日志对象。
+        rows (list[dict]): parse_product 产出的商品字典列表。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本次写入/更新的商品数。
+    """
+    if not rows:
+        return 0
+
+    # 1) 商品主表 upsert（最新状态）
+    cols = ["product_code", "merchant_user_id", "merchant_name", "title", "card_product_title",
+            "cover_image_url", "unit_price", "min_unit_price", "max_unit_price", "card_count",
+            "sold_count", "available_stock", "groupbuy_status", "groupbuy_status_name",
+            "play_type", "live_id"]
+    upd = [c for c in cols if c != "product_code"]  # 除唯一键外都更新为最新值
+    # is_on_sale=1：本函数写入的都是本轮 home/search 采到的当前在售商品；
+    # 重新上架的旧商品(原 is_on_sale=0)也会被 ON DUPLICATE 置回 1（自愈）。
+    prod_sql = (f"INSERT INTO deca_onsale_product_record ({', '.join(cols)}, is_on_sale) "
+                f"VALUES ({', '.join(['%s'] * len(cols))}, 1) "
+                f"ON DUPLICATE KEY UPDATE {', '.join(f'{c}=VALUES({c})' for c in upd)}, is_on_sale=1")
+    sql_pool.insert_many(query=prod_sql, args_list=[tuple(r[c] for c in cols) for r in rows])
+
+    # 2) 每日快照 upsert（product_code + snapshot_date 唯一，当天多次跑更新为最新值）
+    today = date.today().isoformat()
+    snap_cols = ["product_code", "merchant_user_id", "snapshot_date", "card_count",
+                 "sold_count", "available_stock", "groupbuy_status_name", "unit_price"]
+    snap_upd = ["card_count", "sold_count", "available_stock", "groupbuy_status_name", "unit_price"]
+    snap_sql = (f"INSERT INTO deca_onsale_product_daily_record ({', '.join(snap_cols)}) "
+                f"VALUES ({', '.join(['%s'] * len(snap_cols))}) "
+                f"ON DUPLICATE KEY UPDATE {', '.join(f'{c}=VALUES({c})' for c in snap_upd)}")
+    snap_args = [(r["product_code"], r["merchant_user_id"], today, r["card_count"],
+                  r["sold_count"], r["available_stock"], r["groupbuy_status_name"], r["unit_price"])
+                 for r in rows]
+    sql_pool.insert_many(query=snap_sql, args_list=snap_args)
+    return len(rows)
+
+
+def get_product_list(log, merchant_user_id: str, sql_pool) -> int:
+    """遍历某商家在售商品翻页，写入商品主表 + 每日快照。
+
+    接口2 返回 data.total，用于判断是否翻完；同时 list 为空也停止。
+
+    Args:
+        log: 日志对象。
+        merchant_user_id (str): 商家用户 ID。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本商家写入/更新的商品数。
+    """
+    page = 1
+    saved = 0
+    total = None
+    while page <= MAX_PROD_PAGES:
+        body = {"merchantUserId": merchant_user_id, "page": page, "pageSize": PAGE_SIZE}
+        try:
+            resp = do_request(log, "/api/v1/app/groupbuy/merchant/on-sale-list", body, need_auth=True)
+        except Exception as e:
+            log.error(f"商家 {merchant_user_id} 在售商品第 {page} 页请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"商家 {merchant_user_id} 在售商品返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        if total is None:
+            total = data.get("total")
+        items = data.get("list") or []
+        if not items:
+            break
+
+        rows = [r for r in (parse_product(it) for it in items) if r]
+        saved += save_products(log, rows, sql_pool)
+
+        # 翻页终止：已采集条数达到 total，或本页不足一页
+        if total is not None and page * PAGE_SIZE >= total:
+            break
+        if len(items) < PAGE_SIZE:
+            break
+        page += 1
+
+    if saved:
+        log.info(f"商家 {merchant_user_id} 在售商品采集完成，写入/更新 {saved} 个(total={total})")
+    return saved
+
+
+def fetch_all_onsale(log) -> tuple[list, bool]:
+    """免 token 翻页拉取「全量在售」商品（home/search 空关键词），只解析返回、不落库。
+
+    抽出纯拉取供多方复用（get_onsale_products 落库、onsale_alert_spider 自采筛商家），
+    避免重复实现翻页/签名。home/search 免登录即返回当前全量在售，商品项与 on-sale-list 同构，
+    直接用 parse_product 解析。翻到底：本页不足一页（<PAGE_SIZE）即最后一页。
+
+    Args:
+        log: 日志对象。
+
+    Returns:
+        tuple[list, bool]: (全站在售商品字典列表[parse_product 结果], 是否正常翻到底)。
+            bool 为 False 表示中途请求异常/未取全，调用方据此决定是否安全做下架对账。
+    """
+    page = 1
+    total = None
+    out = []
+    ok = False
+    while page <= MAX_SEARCH_PAGES:
+        # 空 query + saleStatus=0：实测返回当前全量在售；playType/sortType/specType=0 表示不额外过滤
+        body = {"categoryParam": 0, "page": page, "pageSize": PAGE_SIZE, "playType": 0,
+                "query": "", "saleStatus": 0, "sortType": 0, "specType": 0}
+        try:
+            resp = do_request(log, "/api/v1/app/home/search", body, need_auth=False)  # 免登录，仅需签名
+        except Exception as e:
+            log.error(f"全量在售第 {page} 页请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"全量在售返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        if total is None:
+            total = data.get("total")
+            log.info(f"全量在售共 {total} 个商品，开始翻页")
+        items = data.get("list") or []
+        if not items:
+            ok = True   # 翻到空页也算正常结束
+            break
+
+        out.extend(r for r in (parse_product(it) for it in items) if r)
+        if len(items) < PAGE_SIZE:  # 翻到底：本页不足一页即最后一页
+            ok = True
+            break
+        page += 1
+    return out, ok
+
+
+def get_onsale_products(log, sql_pool) -> int:
+    """免 token 拉全量在售（复用 fetch_all_onsale）并写主表 + 每日快照 + 采后下架对账。
+
+    取代旧的「逐商家遍历 on-sale-list（需 token）」路径。拉取交给 fetch_all_onsale，
+    本函数只负责落库与对账；商品真实状态以 item.groupbuyStatus 为准（saleStatus=0 取全量）。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本轮写入/更新的商品数。
+    """
+    rows, ok = fetch_all_onsale(log)
+    current_codes = {r["product_code"] for r in rows}   # 本轮全量在售 code 集合，用于采后对账下架
+    saved = save_products(log, rows, sql_pool) if rows else 0
+
+    # 采后对账下架：仅在本轮正常取到非空在售集合时执行，避免请求异常时误判全部下架
+    if ok and current_codes:
+        _sweep_offsale(log, sql_pool, current_codes)
+    else:
+        log.warning(f"本轮未正常取全在售集合(ok={ok}, 集合数={len(current_codes)})，跳过下架对账")
+
+    log.info(f"全量在售商品采集完成，写入/更新 {saved} 个")
+    return saved
+
+
+def _sweep_offsale(log, sql_pool, current_codes: set) -> int:
+    """对账下架：把 is_on_sale=1 但不在本轮全量在售集合内的商品置为 is_on_sale=0。
+
+    解决 home/search 只返回「当前在售」导致的陈旧问题：上午在售、下午卖完的商品，
+    下午这轮不在 current_codes 里，会被置为不在售，从而让 deca_report 统计准确。
+    商品重新上架时，save_products 的 upsert 会把 is_on_sale 置回 1（自愈）。
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+        current_codes (set): 本轮 home/search 全量在售的 product_code 集合（非空）。
+
+    Returns:
+        int: 被置为下架的商品数（rowcount）；取不到时返回 0。
+    """
+    placeholders = ",".join(["%s"] * len(current_codes))
+    sql = (f"UPDATE deca_onsale_product_record SET is_on_sale=0 "
+           f"WHERE is_on_sale=1 AND product_code NOT IN ({placeholders})")
+    cur = sql_pool.update_one(sql, tuple(current_codes))
+    try:
+        n = cur.rowcount if cur is not None else 0
+    except Exception:
+        n = 0
+    log.info(f"下架对账完成：{n} 个商品本轮不在全量在售集合，已置 is_on_sale=0")
+    return n
+
+
+def get_product_detail(log, code: str) -> dict:
+    """获取商品详情 groupbuy/detail，返回 data 节点（含 publishAt/saleStartAt/saleEndAt 等）。
+
+    Args:
+        log: 日志对象。
+        code (str): 商品编码（列表接口的 code）。
+
+    Returns:
+        dict: 详情 data 字典；无数据时返回空字典。
+    """
+    resp = do_request(log, "/api/v1/app/groupbuy/detail", {"code": code}, need_auth=False)  # 详情接口免登录，实测不带token也返回
+    return (resp or {}).get("data") or {}
+
+
+def fill_product_details(log, sql_pool) -> int:
+    """给未补齐详情的商品拉 groupbuy/detail，回填规格/系列/模式/时间。
+
+    选取范围：publish_at 或 spec_name 任一为空的商品（前者覆盖新商品初次补时间字段，
+    后者覆盖历史存量商品的规格/系列/模式回填）。详情字段入库后基本不变，无需重复拉取。
+
+    回填字段：
+        - publish_at / sale_start_at / sale_end_at ：上架、开售、结束时间
+        - spec_name       ：组队规格顶层(specTypeName)，如 原箱 / 单盒 / 单包 / LOT
+        - series_name     ：系列名(giftInfo.items[0].seriesName)，含 Hobby/Jumbo 版本词
+        - series_config   ：几箱几盒几包结构化(giftInfo.items[0].seriesConfig)
+        - play_type_name  ：玩法中文名(playTypeName)，如 自定义随机 / 随机球队
+
+    Args:
+        log: 日志对象。
+        sql_pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本轮成功补详情的商品数。
+    """
+    rows = sql_pool.select_all(
+        "SELECT product_code FROM deca_onsale_product_record "
+        "WHERE publish_at IS NULL OR spec_name IS NULL") or []
+    log.info(f"待补详情(时间/规格/模式)商品 {len(rows)} 个")
+    filled = 0
+    for (code,) in rows:
+        try:
+            d = get_product_detail(log, code)
+            if not d:
+                continue
+            # 系列信息在 giftInfo.items 首元素里；缺则给空字典兜底，字段 .get() 全部返回 None
+            first_item = ((d.get("giftInfo") or {}).get("items") or [{}])[0] or {}
+            sql_pool.update_one(
+                "UPDATE deca_onsale_product_record SET publish_at=%s, sale_start_at=%s, "
+                "  sale_end_at=%s, spec_name=%s, series_name=%s, series_config=%s, "
+                "  play_type_name=%s "
+                "WHERE product_code=%s",
+                (d.get("publishAt") or None, d.get("saleStartAt") or None,
+                 d.get("saleEndAt") or None,
+                 d.get("specTypeName") or None,
+                 first_item.get("seriesName") or None,
+                 first_item.get("seriesConfig") or None,
+                 d.get("playTypeName") or None,
+                 code))
+            filled += 1
+        except Exception as e:
+            log.error(f"商品 {code} 补详情失败: {e}")
+        time.sleep(0.3)  # 轻微限速
+    return filled
+
+
+# ==================== 主流程 ====================
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def main_task(log):
+    """得卡每日采集主函数：商家发现 → 免 token 全量在售商品采集(home/search, +每日快照)。
+
+    Args:
+        log: 日志对象。
+
+    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:
+        # 1) 商家发现（免 token）
+        try:
+            n = get_shop_list(log, sql_pool)
+            log.info(f"商家发现完成，去重商家 {n} 个")
+        except Exception as e:
+            log.error(f"get_shop_list error: {e}")
+
+        time.sleep(3)
+
+        # 2) 商品采集：免 token 翻页拉全量在售（home/search，取代旧的逐商家 on-sale-list[需 token]）
+        try:
+            n = get_onsale_products(log, sql_pool)
+            log.info(f"在售商品采集完成，写入/更新 {n} 个")
+        except Exception as e:
+            log.error(f"get_onsale_products error: {e}")
+
+        # 3) 补商品详情：只给今日新增商品拉 groupbuy/detail 的上架/开售/结束时间
+        try:
+            n = fill_product_details(log, sql_pool)
+            log.info(f"补详情完成，本轮 {n} 个")
+        except Exception as e:
+            log.error(f"fill_product_details error: {e}")
+
+        # 4) 采集完成：生成统计报表图片，并按 deca_report.SEND_WECHAT 开关自动发微信好友
+        try:
+            import deca_report
+            out = deca_report.main()
+            log.info(f"报表已生成并处理发送: {out}")
+        except Exception as e:
+            log.error(f"生成/发送报表失败: {e}")
+    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():
+    """定时任务入口：每天 09:00 与 15:00 各运行一次 main_task。"""
+    # main_task(log=logger)  # 立即跑一次（调试时取消注释）
+
+    schedule.every().day.at("09:00").do(main_task, log=logger)  # 上午场
+    schedule.every().day.at("15:00").do(main_task, log=logger)  # 下午场
+    schedule.every().day.at("20:00").do(main_task, log=logger)  # 晚上场
+    schedule.every().day.at("01:00").do(main_task, log=logger)  # 凌晨场
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    schedule_task()

deca_spider/on_sale/deca_on_sale_report.py

@@ -0,0 +1,727 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/02
+"""得卡 DECA 统计报表：从数据库生成一份 Excel(xlsx) + 一张综合图片(PNG)。
+
+产出：
+    - reports/得卡在售每日报告(新增商家／明细)YYYYMMDD_HH时.xlsx  多 Sheet 汇总（文件名带小时，每天 01/09/15/20 四档各留一份、互不覆盖）：
+        Sheet 概览 / 今日新增商家 / 其他商家 / 商品明细 / 在售趋势 / 上架时段分布
+        明细字段全量、不截断：商家 / 标题 / 系列 / 规格类型 / 规格详情 / 模式 /
+                                上架时间 / 预计结束时间 / 单价 / 已售数 / 总数 /
+                                已售总价 / 新品
+        2026/08/14 按监测清单新增两 Sheet：
+          · 在售趋势     ：今日+前3日 在售商家数/在售拼团数/新增商家数/新增拼团数（每日快照差分）
+          · 上架时段分布 ：近 7 日 24h 上架分布（迷你条形图，运营节奏）
+        进度分析不做「固定 10/22 点」双节点：报告每天发 4 次(09/15/20/01)，每次即一个时间
+        节点，商品明细「进度」列已是当次实时进度。
+    - reports/deca_report_YYYYMMDD_HHMMSS.png 综合图片（视觉版，一天多次生成互不覆盖）
+
+生成后可选自动发送到企业微信群机器人（见 SEND_WECHAT，依赖 auto_send_wx_msg.py）。
+
+用法：python deca_report.py
+"""
+import os
+import re
+import sys
+# 把项目根目录加入 import 路径：企微发送模块 auto_send_wx_msg.py 只在根目录留一份（WEBHOOK_URL 单点维护）
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from datetime import datetime, timedelta
+import time
+import matplotlib
+matplotlib.use("Agg")  # 无 GUI，直接保存文件
+import matplotlib.pyplot as plt
+from matplotlib import gridspec
+from loguru import logger
+from mysql_pool import MySQLConnectionPool
+from openpyxl import Workbook
+from openpyxl.styles import Alignment, Font, PatternFill, Border, Side
+from openpyxl.utils import get_column_letter
+
+logger.remove()
+logger.add("./logs/onsale_report_{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")
+
+# 中文字体 + 负号正常显示
+plt.rcParams["font.sans-serif"] = ["Microsoft YaHei", "SimHei"]
+plt.rcParams["axes.unicode_minus"] = False
+
+# 配色
+C_SHOP = "#2563eb"    # 主色-蓝
+C_PROD = "#f59e0b"    # 商品-橙
+C_PROG = "#16a34a"    # 进度-绿
+C_NEW = "#dc2626"     # 今日新增-红(强调)
+C_OTHER = "#0f766e"   # 其他商家-青绿(与新增红对比)
+C_CARD = "#f1f5f9"    # KPI 卡片底
+C_TEXT = "#0f172a"    # 主文字
+C_SUB = "#64748b"     # 次文字
+C_ROW = "#f8fafc"     # 表格隔行底色
+C_NEWROW = "#fef2f2"  # 新品行淡红底
+
+# 企微发送开关（生成 Excel 后自动发到企业微信群机器人；群由 auto_send_wx_msg.WEBHOOK_URL 决定）
+SEND_WECHAT = True
+PRODUCT_LIMIT = 80            # 在售商品明细最多显示的行数（防图过长）
+SHOP_LIMIT = 100              # 商家表(今日新增/其他)最多显示的行数
+# ---- 监测清单增强(2026/08/14)：在售趋势 / 上架时段 配置 ----
+TREND_DAYS = 4               # 平台在售趋势展示天数(今日 + 前3日)
+LISTING_HOUR_DAYS = 7        # 上架时段分布回看天数(平台 24h 上架节奏)
+
+
+def f(v, default=0.0):
+    """把可能为 None/Decimal 的值安全转 float。
+
+    Args:
+        v: 原始值（None / Decimal / 数字）。
+        default (float, optional): 空值时的默认。Defaults to 0.0。
+
+    Returns:
+        float: 转换后的浮点数。
+    """
+    return default if v is None else float(v)
+
+
+def _fmt_time(v) -> str:
+    """把时间字段渲染成短字符串；空值显示破折号。
+
+    Args:
+        v: 时间原始值（字符串/时间戳/None/空串）。
+
+    Returns:
+        str: 截断到 16 字符的时间文本，或 "—"。
+    """
+    if v in (None, "", 0):
+        return "—"
+    return str(v)[:16]
+
+
+def _fmt_money(v) -> str:
+    """把金额渲染成短字符串：>=1万显示"X.X万"，否则整数。
+
+    Args:
+        v: 金额原始值（None/Decimal/数字）。
+
+    Returns:
+        str: 格式化后的文本；空值返回 "—"。
+    """
+    x = f(v)
+    if x <= 0:
+        return "—"
+    if x >= 10000:
+        return f"{x / 10000:.1f}万"
+    return f"{x:.0f}" if x >= 100 else f"{x:.2f}"
+
+
+def _fmt_progress(sold, total) -> str:
+    """把已售/总数渲染成售出进度百分比字符串（用于 PNG 图）。
+
+    Args:
+        sold: 已售份数（sold_count）。
+        total: 总份数（card_count）。
+
+    Returns:
+        str: 形如 "68.5%" 的进度；总数为 0 或空时返回 "—"。
+    """
+    s = f(sold)
+    t = f(total)
+    if t <= 0:
+        return "—"
+    return f"{s / t * 100:.1f}%"
+
+
+def _fmt_spec(spec_name, series_config) -> str:
+    """把规格组合成一行：`spec_name · series_config`，缺失部分省略。
+
+    Args:
+        spec_name: 顶层规格(原箱/单盒/单包/LOT)。
+        series_config: 结构化规格(几张/包 几包/盒 几盒/箱 共X箱)。
+
+    Returns:
+        str: 组合后的字符串；两者都空返回 "—"。
+    """
+    parts = [p for p in (spec_name, series_config) if p]
+    return " · ".join(parts) if parts else "—"
+
+
+def _fmt_series(series_name, title) -> str:
+    """系列名优先取详情的 series_name（已含 Hobby/Jumbo）；缺失时回退到标题正则粗抽。
+
+    Args:
+        series_name: 详情接口 giftInfo.items[0].seriesName。
+        title: 商品完整标题，仅在 series_name 为空时用作兜底。
+
+    Returns:
+        str: 系列文本，截断到 22 字。
+    """
+    if series_name:
+        return series_name[:22]
+    if not title:
+        return "—"
+    # 兜底：粗略抽取 "品牌 系列 版本" 段，避免展示为空
+    import re
+    m = re.search(r"(topps|Panini)\s+([\w \-]+?(?:Hobby|Jumbo|Blaster|Retail))", title, flags=re.I)
+    return (m.group(0) if m else title)[:22]
+
+
+def fetch_data(pool) -> dict:
+    """从数据库汇总报表所需数据（KPI + 今日新增商家 + 其他商家 + 各商家在售商品明细）。
+
+    Args:
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        dict: 各区块所需数据集合。
+    """
+    d = {}
+    # ---- KPI ----
+    d["total_shops"] = pool.select_one("SELECT COUNT(*) FROM deca_onsale_shop_record")[0]
+    d["total_products"] = pool.select_one(
+        "SELECT COUNT(*) FROM deca_onsale_product_record WHERE is_on_sale=1")[0]
+    d["today_new_shops"] = pool.select_one(
+        "SELECT COUNT(*) FROM deca_onsale_shop_record WHERE DATE(gmt_create_time)=CURDATE()")[0]
+    # 今日新增商品口径：按商品「开售时间 publish_at」判定，而非入库时间 gmt_create_time
+    # 原因：入库时间受采集延迟/故障影响会与真实开售日错位；补抓的老商品也会被误标为今日新增
+    d["today_new_products"] = pool.select_one(
+        "SELECT COUNT(*) FROM deca_onsale_product_record "
+        "WHERE DATE(publish_at)=CURDATE() AND is_on_sale=1")[0]
+    row = pool.select_one(
+        "SELECT SUM(sold_count) FROM deca_onsale_product_daily_record WHERE snapshot_date=CURDATE()")
+    d["today_sold_total"] = int(f(row[0])) if row and row[0] is not None else 0
+
+    # ---- 一、今日新增商家（今天首次入库的商家 + 粉丝数 + 在售数 + 今日新增商品数）----
+    d["new_shop_rows"] = pool.select_all(
+        "SELECT s.merchant_name, s.fans_count, s.active_groupbuy_count, "
+        "  SUM(CASE WHEN DATE(p.publish_at)=CURDATE() AND p.is_on_sale=1 THEN 1 ELSE 0 END) AS today_new "
+        "FROM deca_onsale_shop_record s "
+        "LEFT JOIN deca_onsale_product_record p ON p.merchant_user_id = s.merchant_user_id "
+        "WHERE DATE(s.gmt_create_time)=CURDATE() "
+        "GROUP BY s.merchant_user_id, s.merchant_name, s.fans_count, s.active_groupbuy_count "
+        "ORDER BY s.active_groupbuy_count DESC, s.fans_count DESC "
+        "LIMIT %s", (SHOP_LIMIT,)) or []
+
+    # ---- 二、其他商家（非今日新增的存量商家，同格式含粉丝数）----
+    d["other_shop_rows"] = pool.select_all(
+        "SELECT s.merchant_name, s.fans_count, s.active_groupbuy_count, "
+        "  SUM(CASE WHEN DATE(p.publish_at)=CURDATE() AND p.is_on_sale=1 THEN 1 ELSE 0 END) AS today_new "
+        "FROM deca_onsale_shop_record s "
+        "LEFT JOIN deca_onsale_product_record p ON p.merchant_user_id = s.merchant_user_id "
+        "WHERE DATE(s.gmt_create_time) != CURDATE() "
+        "GROUP BY s.merchant_user_id, s.merchant_name, s.fans_count, s.active_groupbuy_count "
+        "ORDER BY s.active_groupbuy_count DESC, s.fans_count DESC, today_new DESC "
+        "LIMIT %s", (SHOP_LIMIT,)) or []
+    d["other_shop_total"] = pool.select_one(
+        "SELECT COUNT(*) FROM deca_onsale_shop_record WHERE DATE(gmt_create_time) != CURDATE()")[0]
+
+    # ---- 三、各商家在售商品明细（按商家分组；今日新增商品标记）----
+    # 新增：series_name(系列+版本Hobby/Jumbo)、spec_name(规格 原箱/单盒/LOT)、series_config(几箱几盒几包)、
+    #      play_type_name(模式 自定义随机/随机球队/...)、unit_price(单价)；
+    #      已售总价：随机团(选队随机/剩余随机) 用 team_total_amount(2026/08/11 新增)——
+    #      每支球队价格不同，用单一 unit_price×sold_count 会严重失真；具体口径见
+    #      docs/选队随机与剩余随机_总价口径与采集_20260811.md。
+    #      NULL(固定价团 或 数据缺失) 回落到 unit_price×sold_count。
+    # PNG 版受 PRODUCT_LIMIT 截断（防图过长）；Excel 版拉全量（product_rows_all，字段完整无截断）
+    prod_sql = (
+        "SELECT merchant_name, title, series_name, spec_name, series_config, play_type_name, "
+        "  unit_price, sold_count, card_count, publish_at, sale_end_at, "
+        "  CASE WHEN DATE(publish_at)=CURDATE() THEN 1 ELSE 0 END AS is_new, "  # 新品口径=开售日 publish_at
+        "  COALESCE(team_total_amount, unit_price * sold_count) AS total_amount "  # 随机团用 team_total_amount，其它回落
+        "FROM deca_onsale_product_record "
+        "WHERE is_on_sale=1 "                       # 只统计当前在售，排除已下架/售罄陈旧项
+        "ORDER BY merchant_name, is_new DESC, product_code")
+    d["product_rows"] = pool.select_all(prod_sql + " LIMIT %s", (PRODUCT_LIMIT,)) or []
+    d["product_rows_all"] = pool.select_all(prod_sql) or []
+    d["product_total"] = pool.select_one(
+        "SELECT COUNT(*) FROM deca_onsale_product_record WHERE is_on_sale=1")[0]
+
+    # ---- 四、监测清单增强：在售趋势(差分) / 上架时段(24h) ----
+    # 注：进度分析不做「固定 10/22 点」双节点——在售报告每天发 4 次(09/15/20/01)，每次发送
+    # 本身即一个时间节点，商品明细的「进度」列就是当次发送时刻的实时进度，无需再造人工节点。
+    d["trend"] = fetch_onsale_trend(pool, TREND_DAYS)
+    d["listing_hours"] = fetch_listing_hour_dist(pool, LISTING_HOUR_DAYS)
+    return d
+
+
+def _bar(value: int, max_value: int, width: int = 20) -> str:
+    """把数值渲染成等宽条形字符串（Excel 内迷你直方图）。
+
+    Args:
+        value (int): 当前值。
+        max_value (int): 该组最大值（归一化条长用）。
+        width (int, optional): 满值时的条长（字符数）。Defaults to 20。
+
+    Returns:
+        str: 由 █ 组成的条；max_value<=0 或 value<=0 时返回空串。
+    """
+    if max_value <= 0 or value <= 0:
+        return ""
+    return "█" * max(1, round(value / max_value * width))
+
+
+def fetch_onsale_trend(pool, days: int) -> list:
+    """按每日在售快照统计近 N 天的平台在售趋势（商家数/拼团数/新增商家/新增拼团）。
+
+    在售商家数/在售拼团数取当日快照去重计数；新增数为「当日快照相对前一日的差分」
+    （当日出现、前一日没有的 merchant_user_id / product_code 数），即真正的净新增。
+    源表 deca_onsale_product_daily_record（每日一份在售快照）。
+
+    Args:
+        pool (MySQLConnectionPool): MySQL 连接池。
+        days (int): 展示天数（含今天）。
+
+    Returns:
+        list[dict]: 按日期倒序，每项含 日期/在售商家数/在售拼团数/新增商家数/新增拼团数；
+            最早一天若缺前一日基线，新增列为 None。
+    """
+    # 多取一天做最早展示日的差分基线（WHERE 覆盖到 CURDATE()-days）
+    rows = pool.select_all(
+        "SELECT snapshot_date, merchant_user_id, product_code "
+        "FROM deca_onsale_product_daily_record "
+        "WHERE snapshot_date >= CURDATE() - INTERVAL %s DAY", (days,)) or []
+    day_shops, day_prods = {}, {}
+    for snap_date, mid, code in rows:
+        day_shops.setdefault(snap_date, set()).add(mid)
+        day_prods.setdefault(snap_date, set()).add(code)
+    dates = sorted(day_shops.keys(), reverse=True)   # 新 → 旧
+    result = []
+    for snap_date in dates[:days]:
+        prev = snap_date - timedelta(days=1)          # 前一日基线
+        shops, prods = day_shops[snap_date], day_prods[snap_date]
+        if prev in day_shops:
+            new_shops = len(shops - day_shops[prev])  # 净新增商家 = 当日有、前日无
+            new_prods = len(prods - day_prods[prev])  # 净新增拼团 = 当日有、前日无
+        else:
+            new_shops = new_prods = None              # 无基线，诚实留空
+        result.append({"日期": str(snap_date), "在售商家数": len(shops), "在售拼团数": len(prods),
+                       "新增商家数": new_shops, "新增拼团数": new_prods})
+    return result
+
+
+def fetch_listing_hour_dist(pool, days: int) -> list:
+    """统计近 N 天上架(publish_at)按小时的 24 桶分布，反映平台 24h 上架节奏。
+
+    Args:
+        pool (MySQLConnectionPool): MySQL 连接池。
+        days (int): 回看天数。
+
+    Returns:
+        list[int]: 长度 24，索引=小时(0~23)，值=该小时上架的商品数。
+    """
+    since = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
+    # publish_at 为 'YYYY-MM-DD HH:MM:SS' 字符串，字典序比较等价于时间比较；HOUR() 可直接解析
+    sql = ("SELECT HOUR(publish_at) h, COUNT(*) c FROM deca_onsale_product_record "
+           "WHERE publish_at IS NOT NULL AND publish_at <> '' AND publish_at >= %s "
+           "GROUP BY h")
+    dist = [0] * 24
+    for h, c in pool.select_all(sql, (since,)) or []:
+        if h is not None and 0 <= int(h) < 24:
+            dist[int(h)] = int(c)
+    return dist
+
+
+def _kpi_card(ax, title: str, value, color: str):
+    """在指定子图里画一个 KPI 卡片（大数字 + 标题）。
+
+    Args:
+        ax: matplotlib 子图。
+        title (str): 卡片标题。
+        value: 展示的数值。
+        color (str): 数字颜色。
+    """
+    ax.axis("off")
+    ax.add_patch(plt.Rectangle((0.03, 0.08), 0.94, 0.84, transform=ax.transAxes,
+                               facecolor=C_CARD, edgecolor="none", zorder=0))
+    ax.text(0.5, 0.60, str(value), transform=ax.transAxes, ha="center", va="center",
+            fontsize=28, fontweight="bold", color=color)
+    ax.text(0.5, 0.24, title, transform=ax.transAxes, ha="center", va="center",
+            fontsize=11, color=C_SUB)
+
+
+def _table(ax, title, title_color, col_labels, col_widths, cell_text,
+           left_cols=(0,), new_flags=None):
+    """在子图里画一张表（表头着色、隔行底色、指定列左对齐、新品行高亮）。
+
+    Args:
+        ax: matplotlib 子图。
+        title (str): 表标题。
+        title_color (str): 标题与表头底色。
+        col_labels (list[str]): 列名。
+        col_widths (list[float]): 各列宽（和为 1）。
+        cell_text (list[list[str]]): 单元格文本二维数组。
+        left_cols (tuple[int], optional): 需要左对齐的列下标。Defaults to (0,)。
+        new_flags (list[bool], optional): 每行是否为"今日新增"，True 则整行淡红底。Defaults to None。
+    """
+    ax.axis("off")
+    ax.set_title(title, fontsize=15, fontweight="bold", color=title_color, loc="left", pad=10)
+    if not cell_text:
+        ax.text(0.01, 0.5, "暂无数据", transform=ax.transAxes, fontsize=12, color=C_SUB)
+        return
+    tbl = ax.table(cellText=cell_text, colLabels=col_labels, colWidths=col_widths,
+                   cellLoc="center", loc="upper center")
+    tbl.auto_set_font_size(False)
+    tbl.set_fontsize(10.5)
+    tbl.scale(1, 1.5)
+    ncol = len(col_labels)
+    # 表头
+    for j in range(ncol):
+        c = tbl[0, j]
+        c.set_facecolor(title_color)
+        c.set_text_props(color="white", fontweight="bold")
+        c.set_height(c.get_height() * 1.1)
+    # 数据行
+    for i in range(1, len(cell_text) + 1):
+        is_new = bool(new_flags[i - 1]) if new_flags else False
+        for j in range(ncol):
+            cell = tbl[i, j]
+            if is_new:
+                cell.set_facecolor(C_NEWROW)
+            elif i % 2 == 0:
+                cell.set_facecolor(C_ROW)
+            if j in left_cols:
+                cell.set_text_props(ha="left")
+                cell.PAD = 0.03
+            # 新品列(最后一列)红字加粗
+            if new_flags and j == ncol - 1 and cell_text[i - 1][j]:
+                cell.set_text_props(color=C_NEW, fontweight="bold")
+
+
+def _shop_cells(rows: list) -> list:
+    """把商家汇总行渲染成表格单元格文本（列：商家 / 粉丝数 / 在售商品数 / 今日新增商品）。
+
+    Args:
+        rows (list): [(merchant_name, fans_count, active_groupbuy_count, today_new_products)] 列表。
+
+    Returns:
+        list[list[str]]: 单元格二维文本。
+    """
+    return [[(r[0] or "-")[:18], str(int(f(r[1]))), str(int(f(r[2]))), str(int(f(r[3])))]
+            for r in rows]
+
+
+def build_report(d: dict, out_path: str):
+    """根据数据渲染报表图并保存（KPI + 今日新增 + 其他商家 + 各商家在售明细）。
+
+    Args:
+        d (dict): fetch_data 返回的数据集合。
+        out_path (str): 输出 PNG 路径。
+    """
+    # 一、今日新增商家
+    new_cells = _shop_cells(d["new_shop_rows"])
+    # 二、其他商家
+    other_cells = _shop_cells(d["other_shop_rows"])
+    # 三、在售商品明细（按商家分组，同商家仅首行显示名）
+    # 每行 11 列：商家/系列/规格/模式/预计结束时间/单价/已售数/总数/进度/已售总价/新品
+    # 进度 = 已售数 ÷ 总数（售出百分比）
+    prod_cells, new_flags, last_name = [], [], None
+    for r in d["product_rows"]:
+        (name, title, series_name, spec_name, series_config, play_type_name,
+         unit_price, sold, card, pub, end, is_new, total_amount) = r
+        show_name = "" if name == last_name else (name or "-")[:12]
+        last_name = name
+        sold_i = int(f(sold))
+        card_i = int(f(card))
+        # 已售总价：随机团用 SQL 里 COALESCE 好的 team_total_amount，其它回落 unit_price×sold_count
+        total = f(total_amount)
+        prod_cells.append([
+            show_name,
+            _fmt_series(series_name, title),
+            _fmt_spec(spec_name, series_config)[:26],
+            (play_type_name or "—")[:12],
+            _fmt_time(end),
+            f"¥{f(unit_price):.2f}" if f(unit_price) > 0 else "—",
+            str(sold_i),
+            str(card_i) if card_i else "—",
+            _fmt_progress(sold_i, card_i),   # 进度 = 已售数 ÷ 总数
+            _fmt_money(total),
+            "新" if is_new else "",
+        ])
+        new_flags.append(bool(is_new))
+
+    n_new = max(len(new_cells), 1)
+    n_other = max(len(other_cells), 1)
+    n_prod = max(len(prod_cells), 1)
+    fig_h = 4.0 + 0.4 * n_new + 0.4 * n_other + 0.4 * n_prod
+    # 明细表列数从 6 加到 10，图宽加到 14 以保证列不挤
+    fig = plt.figure(figsize=(14, fig_h), facecolor="white")
+    gs = gridspec.GridSpec(5, 5, figure=fig,
+                           height_ratios=[0.8, 0.9,
+                                          0.4 * n_new + 0.5,
+                                          0.4 * n_other + 0.5,
+                                          0.4 * n_prod + 0.5],
+                           hspace=0.32, wspace=0.35)
+
+    # 标题
+    ax_title = fig.add_subplot(gs[0, :])
+    ax_title.axis("off")
+    ax_title.text(0.5, 0.75, "得卡 DECA 商家在售采集统计", ha="center", va="center",
+                  fontsize=22, fontweight="bold", color=C_TEXT)
+    ax_title.text(0.98, 0.18, f"生成时间：{datetime.now().strftime('%Y-%m-%d %H:%M')}",
+                  ha="right", va="center", fontsize=11, color=C_SUB)
+
+    # KPI 行（5 卡片，汇总保留）
+    kpis = [
+        ("商家总数", d["total_shops"], C_SHOP),
+        ("在售商品总数", d["total_products"], C_PROD),
+        ("今日新增商家", d["today_new_shops"], C_NEW),
+        ("今日新增商品", d["today_new_products"], C_NEW),
+        ("今日累计已售(份)", d["today_sold_total"], C_PROG),
+    ]
+    for i, (t, v, c) in enumerate(kpis):
+        _kpi_card(fig.add_subplot(gs[1, i]), t, v, c)
+
+    # 一、今日新增商家
+    _table(fig.add_subplot(gs[2, :]),
+           f"一、今日新增商家（{d['today_new_shops']} 家）", C_NEW,
+           ["商家", "粉丝数", "在售商品数", "今日新增商品"], [0.40, 0.20, 0.20, 0.20],
+           new_cells, left_cols=(0,))
+
+    # 二、其他商家（存量，非今日新增）
+    o_total, o_shown = d["other_shop_total"], len(other_cells)
+    o_suffix = f"（共 {o_total} 家，显示前 {o_shown}）" if o_total > o_shown else f"（共 {o_total} 家）"
+    _table(fig.add_subplot(gs[3, :]),
+           f"二、其他商家{o_suffix}", C_OTHER,
+           ["商家", "粉丝数", "在售商品数", "今日新增商品"], [0.40, 0.20, 0.20, 0.20],
+           other_cells, left_cols=(0,))
+
+    # 三、各商家在售商品明细（按商家分组，今日新上架标🆕）
+    # 列：商家 / 系列(含Hobby/Jumbo) / 规格(原箱·几箱几盒几包) / 模式(playTypeName) /
+    #     预计结束时间(sale_end_at) / 单价 / 已售数(sold_count) / 总数(card_count) /
+    #     进度(已售数÷总数) / 已售总价(单价×已售数) / 新品
+    p_total, p_shown = d["product_total"], len(prod_cells)
+    p_suffix = f"（在售商品共 {p_total}，显示前 {p_shown}）" if p_total > p_shown else f"（在售商品 {p_total}）"
+    _table(fig.add_subplot(gs[4, :]),
+           f"三、各商家在售商品明细{p_suffix}", C_SHOP,
+           ["商家", "系列", "规格", "模式", "预计结束时间", "单价", "已售数", "总数", "进度", "已售总价", "新"],
+           [0.09, 0.17, 0.15, 0.09, 0.11, 0.07, 0.07, 0.06, 0.06, 0.09, 0.04],
+           prod_cells, left_cols=(0, 1, 2, 3), new_flags=new_flags)
+
+    fig.savefig(out_path, dpi=150, bbox_inches="tight", facecolor="white")
+    plt.close(fig)
+
+
+# ==================== Excel 产出 ====================
+# 配色与已售报告(stats/daily_report.py)统一：浅蓝底 + 深蓝表头字、淡蓝灰斑马纹、浅灰细边框。
+# 2026/08/14 起各 Sheet 表头不再用蓝/红/青绿区分底色，统一改为已售同款(主公要求两报告风格一致)。
+XL_HEADER_FILL = PatternFill("solid", fgColor="D9E1F2")     # 表头：浅蓝底(同已售 FILL_HEADER)
+XL_HEADER_FONT = Font(name="Microsoft YaHei", bold=True, color="1F3864")  # 表头：深蓝字(同已售 FONT_HEADER)
+XL_CELL_FONT = Font(name="Microsoft YaHei", size=10)        # 正文
+XL_NEWROW_FILL = PatternFill("solid", fgColor="FEF2F2")     # 新品行淡红底(功能高亮，保留)
+XL_ZEBRA_FILL = PatternFill("solid", fgColor="F5F8FC")      # 隔行淡蓝灰(同已售 FILL_ZEBRA)
+XL_THIN_BORDER = Border(left=Side(style="thin", color="D6DCE5"),
+                        right=Side(style="thin", color="D6DCE5"),
+                        top=Side(style="thin", color="D6DCE5"),
+                        bottom=Side(style="thin", color="D6DCE5"))
+
+
+def _write_sheet(ws, headers, rows, new_flags=None,
+                 num_cols=None, money_cols=None, pct_cols=None):
+    """把一份表格数据（表头+多行）写入 Sheet，套统一样式（与已售报告一致）。
+
+    Args:
+        ws: openpyxl 的 worksheet 对象。
+        headers (list[str]): 表头列名。
+        rows (list[list]): 数据行，每行长度应与 headers 一致。
+        new_flags (list[bool], optional): 每行是否为"今日新增"，True 则整行淡红底。Defaults to None。
+        num_cols (set[int], optional): 需要按整数右对齐显示的列下标（0-based）。Defaults to None。
+        money_cols (set[int], optional): 需要按金额格式(¥#,##0.00) 显示的列下标（0-based）。Defaults to None。
+        pct_cols (set[int], optional): 需要按百分比格式(0.0%) 显示的列下标（0-based），值存占比小数如 0.685。Defaults to None。
+    """
+    num_cols = num_cols or set()
+    money_cols = money_cols or set()
+    pct_cols = pct_cols or set()
+
+    # 1) 写表头（浅蓝底 + 深蓝字，居中换行；与已售报告统一）
+    ws.append(headers)
+    for c in ws[1]:
+        c.font = XL_HEADER_FONT
+        c.fill = XL_HEADER_FILL
+        c.alignment = Alignment(horizontal="center", vertical="center", wrap_text=True)
+        c.border = XL_THIN_BORDER
+    ws.row_dimensions[1].height = 26
+
+    # 2) 写数据 + 隔行/新品行样式 + 数字/金额格式
+    for i, row in enumerate(rows, start=2):
+        ws.append(row)
+        is_new = bool(new_flags[i - 2]) if new_flags else False
+        fill = XL_NEWROW_FILL if is_new else (XL_ZEBRA_FILL if i % 2 == 0 else None)
+        for j, cell in enumerate(ws[i]):
+            cell.font = XL_CELL_FONT
+            cell.border = XL_THIN_BORDER
+            if fill:
+                cell.fill = fill
+            if j in money_cols:
+                cell.number_format = "¥#,##0.00"
+                cell.alignment = Alignment(horizontal="right", vertical="center")
+            elif j in pct_cols:
+                cell.number_format = "0.0%"     # 存占比小数(0.685)，显示为 68.5%
+                cell.alignment = Alignment(horizontal="right", vertical="center")
+            elif j in num_cols:
+                cell.number_format = "#,##0"
+                cell.alignment = Alignment(horizontal="right", vertical="center")
+            else:
+                cell.alignment = Alignment(horizontal="left", vertical="center", wrap_text=False)
+
+    # 3) 列宽：按内容估算，中文/表头都算宽度；上下限 [8, 60]
+    for j, header in enumerate(headers, start=1):
+        max_len = len(str(header)) * 2  # 表头基准（中文占 2）
+        for r in ws.iter_rows(min_row=2, min_col=j, max_col=j):
+            v = r[0].value
+            if v is None:
+                continue
+            s = str(v)
+            # 中文按 2 宽度，其他按 1
+            w = sum(2 if ord(ch) > 127 else 1 for ch in s)
+            if w > max_len:
+                max_len = w
+        ws.column_dimensions[get_column_letter(j)].width = max(8, min(60, max_len + 2))
+
+    # 4) 冻结表头
+    ws.freeze_panes = "A2"
+
+
+def build_excel(d: dict, out_path: str):
+    """把报表所有数据集合到一个 xlsx 文件（多 Sheet），字段完整不截断。
+
+    Sheet 结构：
+        1. 概览          ：KPI 5 项 + 生成时间
+        2. 今日新增商家  ：今天入库的商家（商家/粉丝数/在售商品数/今日新增商品）
+        3. 其他商家      ：存量商家（同上格式）
+        4. 商品明细      ：全量在售商品；14 列全字段（含标题、上架时间、规格详情、模式、进度、总价等）
+        5. 在售趋势      ：今日+前3日 在售商家数/在售拼团数/新增商家数/新增拼团数（每日快照差分）
+        6. 上架时段分布  ：近 7 日 24h 上架分布（含迷你条形图）
+
+    Args:
+        d (dict): fetch_data 返回的数据集合（含 product_rows_all 全量明细、trend/listing_hours
+            等增强区块数据）。
+        out_path (str): 输出 xlsx 文件路径。
+    """
+    wb = Workbook()
+
+    # Sheet 1：概览
+    ws1 = wb.active
+    ws1.title = "概览"
+    overview_rows = [
+        ["商家总数", int(f(d["total_shops"]))],
+        ["在售商品总数", int(f(d["total_products"]))],
+        ["今日新增商家", int(f(d["today_new_shops"]))],
+        ["今日新增商品", int(f(d["today_new_products"]))],
+        ["今日累计已售(份)", int(f(d["today_sold_total"]))],
+        ["生成时间", datetime.now().strftime("%Y-%m-%d %H:%M:%S")],
+    ]
+    _write_sheet(ws1, ["指标", "数值"], overview_rows, num_cols={1})
+
+    # Sheet 2：今日新增商家
+    ws2 = wb.create_sheet("今日新增商家")
+    new_rows = [[r[0] or "-", int(f(r[1])), int(f(r[2])), int(f(r[3]))] for r in d["new_shop_rows"]]
+    _write_sheet(ws2, ["商家", "粉丝数", "在售商品数", "今日新增商品"], new_rows,
+                 num_cols={1, 2, 3})
+
+    # Sheet 3：其他商家
+    ws3 = wb.create_sheet("其他商家")
+    other_rows = [[r[0] or "-", int(f(r[1])), int(f(r[2])), int(f(r[3]))] for r in d["other_shop_rows"]]
+    _write_sheet(ws3, ["商家", "粉丝数", "在售商品数", "今日新增商品"], other_rows,
+                 num_cols={1, 2, 3})
+
+    # Sheet 4：商品明细（全量、字段不截断）
+    ws4 = wb.create_sheet("商品明细")
+    prod_headers = ["商家", "标题", "系列", "规格类型", "规格详情", "模式",
+                    "上架时间", "预计结束时间", "单价", "已售数", "总数", "进度", "已售总价", "新品"]
+    prod_rows, new_flags = [], []
+    for r in d["product_rows_all"]:
+        (name, title, series_name, spec_name, series_config, play_type_name,
+         unit_price, sold, card, pub, end, is_new, total_amount) = r
+        sold_i = int(f(sold))
+        card_i = int(f(card))
+        unit = f(unit_price)
+        # 已售总价：随机团用 SQL 里 COALESCE 好的 team_total_amount，其它回落 unit_price×sold_count
+        total = f(total_amount)
+        progress = (sold_i / card_i) if card_i else None   # 进度 = 已售数 ÷ 总数（占比，Excel 用 0.0% 显示）
+        prod_rows.append([
+            name or "-", title or "-", series_name or "-",
+            spec_name or "-", series_config or "-", play_type_name or "-",
+            _fmt_time(pub), _fmt_time(end),
+            unit if unit > 0 else None,
+            sold_i, card_i,
+            progress,
+            total if total > 0 else None,
+            "新" if is_new else "",
+        ])
+        new_flags.append(bool(is_new))
+    _write_sheet(ws4, prod_headers, prod_rows,
+                 new_flags=new_flags,
+                 num_cols={9, 10},     # 已售数、总数
+                 pct_cols={11},        # 进度（已售数÷总数）
+                 money_cols={8, 12})   # 单价、已售总价
+
+    # Sheet 5：平台在售趋势（今日+前3日，按每日快照差分）—— 监测清单「平台层面」
+    ws5 = wb.create_sheet("在售趋势")
+    trend_rows = [[t["日期"], t["在售商家数"], t["在售拼团数"], t["新增商家数"], t["新增拼团数"]]
+                  for t in d["trend"]]
+    _write_sheet(ws5, ["日期", "在售商家数", "在售拼团数", "新增商家数(差分)", "新增拼团数(差分)"],
+                 trend_rows, num_cols={1, 2, 3, 4})
+
+    # Sheet 6：上架时段分布（近 N 日 24h 累计）—— 监测清单「运营节奏·上架时段热图」
+    ws6 = wb.create_sheet("上架时段分布")
+    lhours = d["listing_hours"]
+    lmax = max(lhours) if lhours else 0
+    lrows = [[f"{h:02d}时", lhours[h], _bar(lhours[h], lmax)] for h in range(24)]
+    _write_sheet(ws6, ["时段", "上架数", f"分布(近{LISTING_HOUR_DAYS}日累计)"], lrows,
+                 num_cols={1})
+
+    wb.save(out_path)
+
+
+def main() -> str:
+    """连库取数、生成 Excel 报表 + PNG 综合图，并按开关自动发送微信。
+
+    Returns:
+        str: 生成的 xlsx 路径；失败返回空串。
+    """
+    log = logger
+    pool = MySQLConnectionPool(log=log)
+    if not pool.check_pool_health():
+        log.error("数据库连接池异常")
+        return ""
+    data = fetch_data(pool)
+    os.makedirs("reports", exist_ok=True)
+
+    # 1) Excel 主产出：一份多 Sheet 汇总
+    # 文件名加"小时"以区分每天 01/09/15/20 四个时段，四份各自留档、互不覆盖
+    # 注：文件名括号内的分隔符用全角斜杠 U+FF0F，因为 Windows 不允许半角 `/` 出现在文件名里
+    xlsx_out = os.path.join(
+        "reports", f"得卡在售每日报告(新增商家／明细){datetime.now().strftime('%Y%m%d_%H时')}.xlsx")
+    build_excel(data, xlsx_out)
+    log.info(f"Excel 报表已生成: {xlsx_out}")
+
+    # 2) PNG 综合图（视觉版，一天多次生成互不覆盖，便于历史留档）
+    png_out = os.path.join("reports", f"deca_report_{datetime.now().strftime('%Y%m%d_%H%M%S')}.png")
+    build_report(data, png_out)
+    log.info(f"PNG 报表已生成: {png_out}")
+
+    out = xlsx_out
+
+    # 自动发企微（失败仅告警，不影响报表产出）：每次生成后把 Excel 明细发到企业微信群机器人（只发表格，不发图）
+    if SEND_WECHAT:
+        try:
+            from auto_send_wx_msg import send_wechat_group_file
+            send_wechat_group_file(log=log, file_path=out)   # out=xlsx，只发 Excel
+        except Exception as e:
+            log.warning(f"企微发送跳过: {e}")
+    return out
+
+
+if __name__ == "__main__":
+    # logger.remove()
+    # logger.add(sys.stderr, level="INFO")
+    # 用法：
+    #   python deca_on_sale_report.py       → 立即查库生成并发送一次（保留原行为）
+    #   python deca_on_sale_report.py loop  → 定时常驻：每天 09:00/15:00/20:00/01:00 各查库生成并发微信一次
+    if len(sys.argv) > 1 and sys.argv[1] == "loop":
+        import schedule
+        for _hhmm in ("09:00", "15:00", "20:00", "01:00"):
+            schedule.every().day.at(_hhmm).do(main)
+        logger.info("报告定时常驻启动：每天 09:00/15:00/20:00/01:00 查库生成并发微信（数据由 buy_record_spider 每分钟落库）")
+        while True:
+            schedule.run_pending()
+            time.sleep(1)
+    else:
+        print("报表已生成:", main())

deca_spider/on_sale/deca_sold_spider.py

@@ -0,0 +1,208 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/03
+"""得卡 DECA 商家历史成交采集爬虫（独立任务）。
+
+抓取每个商家的「历史成交」groupbuy/merchant/sold-list，INSERT IGNORE 写入 deca_sold_record。
+历史成交为终态、只增不改，用 product_code 唯一键去重增量采集。
+deca_sold_record 预留 report_state / replay_state，供后续通过 product_code / live_id
+采集「拆卡报告」与「视频回放」。
+
+鉴权：本接口需 Authorization Bearer（需 token），复用 deca_daily_spider 的登录/续期逻辑
+（优先 refreshToken 续期，避开阿里云验证码）。签名规则同样复用。
+"""
+import sys
+import time
+
+import schedule
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+from mysql_pool import MySQLConnectionPool
+import deca_daily_spider as deca  # 复用 do_request / ensure_token / make_signature / after_log
+
+# 覆盖 deca 导入时装的 logger handler，本任务日志独立到 sold_*.log
+logger.remove()
+logger.add("./logs/sold_{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")
+
+PAGE_SIZE = 20     # 历史成交每页条数（抓包实测 20）
+MAX_PAGES = 200    # 单商家历史成交翻页保护上限
+
+
+def parse_sold(item: dict) -> dict | None:
+    """把历史成交列表项解析成 deca_sold_record 一行。
+
+    Args:
+        item (dict): sold-list 的 data.list 项（结构同在售商品列表）。
+
+    Returns:
+        dict | None: 与 deca_sold_record 列对应的数据字典；无 code 时返回 None。
+    """
+    code = item.get("code")
+    if not code:
+        return None
+    m = item.get("merchant") or {}
+    return {
+        "product_code": code,
+        "merchant_user_id": str(m.get("merchantUserID")) if m.get("merchantUserID") else None,
+        "merchant_name": m.get("merchantName"),
+        "title": item.get("title"),
+        "card_product_title": item.get("cardProductTitle"),
+        "cover_image_url": item.get("coverImageUrl"),
+        "unit_price": item.get("unitPrice"),
+        "min_unit_price": item.get("minUnitPrice"),
+        "max_unit_price": item.get("maxUnitPrice"),
+        "card_count": item.get("cardCount"),
+        "sold_count": item.get("soldCount"),
+        "available_stock": item.get("availableStock"),
+        "groupbuy_status": item.get("groupbuyStatus"),
+        "groupbuy_status_name": item.get("groupbuyStatusName"),
+        "play_type": item.get("playType"),
+        "live_id": item.get("liveId"),
+        "completed_at": item.get("completedAt") or None,   # 成交完成时间
+        "publicity_at": item.get("publicityAt") or None,
+    }
+
+
+def get_sold_list(log, merchant_user_id: str, pool) -> int:
+    """翻页拉取某商家历史成交，INSERT IGNORE 写入 deca_sold_record。
+
+    历史成交为终态，用 product_code 唯一键去重，只增不改。
+
+    Args:
+        log: 日志对象。
+        merchant_user_id (str): 商家用户 ID。
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本商家写入的历史成交商品数（已去重）。
+    """
+    page = 1
+    saved = 0
+    total = None
+    while page <= MAX_PAGES:
+        body = {"merchantUserId": merchant_user_id, "page": page, "pageSize": PAGE_SIZE}
+        try:
+            resp = deca.do_request(log, "/api/v1/app/groupbuy/merchant/sold-list", body, need_auth=True)
+        except Exception as e:
+            log.error(f"商家 {merchant_user_id} 历史成交第 {page} 页请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"商家 {merchant_user_id} 历史成交返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        if total is None:
+            total = data.get("total")
+        items = data.get("list") or []
+        if not items:
+            break
+
+        rows = [r for r in (parse_sold(it) for it in items) if r]
+        if rows:
+            pool.insert_many(table="deca_sold_record", data_list=rows, ignore=True)
+            saved += len(rows)
+
+        # 翻页终止：已覆盖 total，或本页不足一页
+        if total is not None and page * PAGE_SIZE >= total:
+            break
+        if len(items) < PAGE_SIZE:
+            break
+        page += 1
+        time.sleep(0.2)
+
+    if saved:
+        log.info(f"商家 {merchant_user_id} 历史成交入库 {saved} 个(total={total})")
+    return saved
+
+
+def fill_sold_details(log, pool) -> int:
+    """给未补详情(publish_at 为空)的历史成交商品补上架/开售/结束时间。
+
+    调 groupbuy/detail 取 publishAt/saleStartAt/saleEndAt。历史成交为终态，
+    每个商品补一次即可（publish_at 非空后不再重复拉）。
+
+    Args:
+        log: 日志对象。
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本轮成功补详情的商品数。
+    """
+    rows = pool.select_all(
+        "SELECT product_code FROM deca_sold_record WHERE publish_at IS NULL") or []
+    log.info(f"待补详情历史成交商品 {len(rows)} 个")
+    filled = 0
+    for (code,) in rows:
+        try:
+            d = deca.get_product_detail(log, code)
+            if not d:
+                continue
+            pool.update_one(
+                "UPDATE deca_sold_record SET publish_at=%s, sale_start_at=%s, sale_end_at=%s "
+                "WHERE product_code=%s",
+                (d.get("publishAt") or None, d.get("saleStartAt") or None,
+                 d.get("saleEndAt") or None, code))
+            filled += 1
+        except Exception as e:
+            log.error(f"历史成交商品 {code} 补详情失败: {e}")
+        time.sleep(0.3)  # 轻微限速
+    return filled
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=deca.after_log)
+def main_task(log):
+    """遍历库内所有商家，采集历史成交写入 deca_sold_record。
+
+    商家名单取自 deca_shop_record（由 deca_daily_spider 维护）。
+
+    Args:
+        log: 日志对象。
+
+    Raises:
+        RuntimeError: 数据库连接池异常时抛出以触发重试。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name} 商家历史成交采集" + "." * 40)
+    pool = MySQLConnectionPool(log=log)
+    if not pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    try:
+        deca.ensure_token(log)  # 预取 token（后续请求自动续期）
+        rows = pool.select_all("SELECT merchant_user_id FROM deca_shop_record")
+        uids = [r[0] for r in rows] if rows else []
+        log.info(f"待采集商家 {len(uids)} 个")
+        for uid in uids:
+            try:
+                get_sold_list(log, uid, pool)
+            except Exception as e:
+                log.error(f"get_sold_list error（商家 {uid}）: {e}")
+            time.sleep(0.3)  # 轻微限速
+
+        # 补详情：给未补(publish_at 为空)的历史成交商品拉 groupbuy/detail 的上架/开售/结束时间
+        try:
+            n = fill_sold_details(log, pool)
+            log.info(f"历史成交补详情完成，本轮 {n} 个")
+        except Exception as e:
+            log.error(f"fill_sold_details error: {e}")
+    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():
+    """定时任务入口：每天 03:00 采集一次商家历史成交。"""
+    main_task(log=logger)  # 立即跑一次（调试时取消注释）
+
+    schedule.every().day.at("03:00").do(main_task, log=logger)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    schedule_task()

deca_spider/on_sale/deca_team_spider.py

@@ -0,0 +1,387 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/11
+"""得卡 DECA · 随机团 teams 明细高频采集（选队随机 + 剩余随机）。
+
+背景：
+  卡牌拼团分两种"随机团"玩法，同一团会经历两阶段：
+    · 选队随机(playTypeName 以「选队随机」开头)：每支球队一价，team-options 接口(带 token)
+      返回逐队 cardCount / availableStock / unitPrice；已售数 = cardCount − availableStock。
+    · 剩余随机(playTypeName == 剩余随机)：卖不动后转成的兜底阶段。团购价变成"剩余池加权
+      均价"、买家随机开一张剩余卡。team-options **对剩余随机永远返 29000**、详情 snapshot
+      冻结在转换那一刻（每队只有 unitPrice/availableStock，无原始 cardCount）。
+
+关键结论：**一旦转成剩余随机，原始 cardCount 就再也没接口能补回**。所以在选队阶段必须持续
+抓 team-options 存表(deca_groupbuy_team_record)，转剩余后我们才能用「存表的原始总价 − 当前
+剩余货值」算出实时已售额。转剩余前没被我们抓到的老团，team_total_amount 留 NULL、报告端
+标注"原始数据缺失"。
+
+一轮做什么：
+  1) 从 deca_onsale_product_record 拉当前在售·选队随机团 → team-options →
+     upsert 到 teams 表(data_source=team_options，同 (code, team_id) 覆盖成最新，captured_at
+     记本轮时刻) → 算 Σ 单价×(cardCount − availableStock) 写回 team_total_amount。
+  2) 从 deca_onsale_product_record 拉当前在售·剩余随机团 → 详情 snapshot 首次存表
+     (data_source=snapshot；冻结不变，重复见到直接跳过) → 若库里有该团选队阶段的原始
+     cardCount，则算 Σ 单价×cardCount − detail.unitPrice × detail.availableStock 写回
+     team_total_amount；否则留 NULL。
+
+登录态：
+  - team-options 强制带 token（不带返 10002）；
+  - 详情接口 groupbuy/detail 免登录。
+  - 与 alert / sold_daily 共用同一份根目录 token.json（HANDOFF 已说明）。
+
+运行：项目根目录 `python on_sale/deca_team_spider.py`
+"""
+import os
+import sys
+import time
+from datetime import datetime
+from decimal import Decimal, ROUND_HALF_UP
+
+import schedule
+from loguru import logger
+from tenacity import retry, stop_after_attempt, wait_fixed
+
+# 挂靠项目根：复用核心签名/token/请求 + application.yml
+_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+# 切到项目根：token.json / application.yml / ./logs 都在根目录，
+# 否则从 on_sale/ 目录里启动时 cwd=on_sale/ 找不到根目录的 token.json，team-options(需 token) 会一直失败
+os.chdir(_ROOT)
+
+import deca_sold_core as core            # noqa: E402
+from mysql_pool import MySQLConnectionPool  # noqa: E402
+
+# ==================== 配置 ====================
+INTERVAL_SEC = 300              # 采集间隔：5 分钟一轮（选队随机变化较快、又不至于压接口）
+BETWEEN_ITEM_SEC = 0.3          # 单商品之间的最小间隔，避免瞬时限流
+MAX_ITEM_PER_ROUND = 500        # 单轮采集商品数上限（保护；正常远小于此）
+
+ONSALE_TABLE = "deca_onsale_product_record"
+TEAM_TABLE   = "deca_groupbuy_team_record"
+
+# ==================== 日志 ====================
+logger.remove()
+logger.add("./logs/team_spider_{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")
+# logger.add(sys.stderr, level="INFO",
+#            format="[{time:HH:mm:ss}] {level} {message}")
+
+
+def after_log(retry_state):
+    """tenacity 重试回调，业务函数首参约定为 log。
+
+    Args:
+        retry_state: tenacity 传入的 RetryCallState，含调用参数与结果。
+    """
+    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")
+
+
+def _round2(x) -> Decimal | None:
+    """Decimal 化并四舍五入到 2 位小数。
+
+    Args:
+        x: 数值(int/float/str/Decimal)，None/异常直接返回 None。
+
+    Returns:
+        Decimal | None: 保留 2 位；None 表示不可算。
+    """
+    if x is None:
+        return None
+    try:
+        return Decimal(str(x)).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+    except Exception:
+        return None
+
+
+def fetch_onsale_random(log, pool) -> tuple[list, list]:
+    """从 deca_onsale_product_record 查当前在售·选队随机 / 剩余随机团 code 列表。
+
+    Args:
+        log: 日志对象。
+        pool: MySQL 连接池。
+
+    Returns:
+        tuple[list[str], list[str]]: (选队随机 codes, 剩余随机 codes)。
+    """
+    rows = pool.select_all(
+        f"SELECT product_code, play_type_name FROM {ONSALE_TABLE} "
+        f"WHERE is_on_sale=1 AND play_type_name IS NOT NULL "
+        f"LIMIT {MAX_ITEM_PER_ROUND}") or []
+    xd, sy = [], []
+    for code, ptn in rows:
+        if not ptn:
+            continue
+        if "选队随机" in ptn:
+            xd.append(code)
+        elif "剩余随机" in ptn:
+            sy.append(code)
+    log.info(f"[发现] 在售·选队随机 {len(xd)} / 剩余随机 {len(sy)}")
+    return xd, sy
+
+
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
+def fetch_team_options(log, code: str) -> list | None:
+    """打 team-options 接口拿一个团的逐队 teams(带 token)。
+
+    Args:
+        log: 日志对象。
+        code (str): 商品编码。
+
+    Returns:
+        list | None: teams 列表；接口报 29000(已转剩余随机)或失败时返回 None（None 是业务态，
+                     不视为需要重试的错误——只有网络级异常才会被 tenacity 拦下重试）。
+    """
+    r = core.do_request(log, "/api/v1/app/groupbuy/team-options", {"code": code}, need_auth=True)
+    if not r or r.get("code") != 0:
+        return None    # 29000 = 商家开启剩余随机中，正常业务态
+    return (r.get("data") or {}).get("list") or []
+
+
+@retry(stop=stop_after_attempt(3), wait=wait_fixed(2), after=after_log)
+def fetch_detail(log, code: str) -> dict | None:
+    """打 detail 接口拿详情(免 token)。
+
+    Args:
+        log: 日志对象。
+        code (str): 商品编码。
+
+    Returns:
+        dict | None: data 字典；失败返回 None。
+    """
+    d = core.do_request(log, "/api/v1/app/groupbuy/detail", {"code": code}, need_auth=False)
+    return (d or {}).get("data") or None
+
+
+def upsert_team_row(pool, code: str, ptn: str, source: str, team: dict,
+                    captured_at: str, snap_total: int | None = None) -> None:
+    """把一条 team 明细 upsert 进 deca_groupbuy_team_record。
+
+    唯一键 (product_code, team_id, data_source) 冲突时按最新覆盖——
+    这就把「在售团每轮覆盖成最新」自动做掉了。
+
+    Args:
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+        ptn (str): 采集时的 playTypeName。
+        source (str): 数据来源 'team_options' / 'snapshot'。
+        team (dict): 单条 team 数据(team-options 项 或 snapshot.teams 项)。
+        captured_at (str): 采集时刻 YYYY-MM-DD HH:MM:SS。
+        snap_total (int | None, optional): 剩余随机 snapshot.totalQuantity。Defaults to None。
+    """
+    tid = team.get("teamId")
+    if tid is None:
+        return
+    cc = team.get("cardCount")                 # 选队阶段有；剩余快照无 → NULL
+    av = team.get("availableStock")
+    up = _round2(team.get("unitPrice"))
+    sold = (cc - av) if (cc is not None and av is not None) else None
+    pool._execute(
+        f"INSERT INTO {TEAM_TABLE} "
+        f"(product_code, play_type_name, data_source, team_id, team_name_en, team_name_zh, "
+        f" team_logo_image_url, unit_price, card_count, available_stock, sold_count, "
+        f" snapshot_total_quantity, captured_at) "
+        f"VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s) "
+        f"ON DUPLICATE KEY UPDATE "
+        f"  play_type_name=VALUES(play_type_name), team_name_en=VALUES(team_name_en), "
+        f"  team_name_zh=VALUES(team_name_zh), team_logo_image_url=VALUES(team_logo_image_url), "
+        f"  unit_price=VALUES(unit_price), card_count=VALUES(card_count), "
+        f"  available_stock=VALUES(available_stock), sold_count=VALUES(sold_count), "
+        f"  snapshot_total_quantity=VALUES(snapshot_total_quantity), "
+        f"  captured_at=VALUES(captured_at)",
+        (code, ptn, source, tid, team.get("teamNameEn"), team.get("teamNameZh"),
+         team.get("teamLogoImageUrl"), up, cc, av, sold, snap_total, captured_at),
+        commit=True)
+
+
+def has_snapshot(pool, code: str) -> bool:
+    """判断某团是否已存过剩余随机 snapshot(冻结值，只需存一次)。
+
+    Args:
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+
+    Returns:
+        bool: True 已存过、可跳过；False 未存。
+    """
+    row = pool.select_one(
+        f"SELECT 1 FROM {TEAM_TABLE} WHERE product_code=%s AND data_source='snapshot' LIMIT 1",
+        (code,))
+    return bool(row)
+
+
+def get_original_total(pool, code: str) -> Decimal | None:
+    """从 teams 表查某团选队阶段抓过的「原始总价」= Σ 单价×cardCount。
+
+    Args:
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+
+    Returns:
+        Decimal | None: 原始总价；缺 team-options 记录时返回 None。
+    """
+    row = pool.select_one(
+        f"SELECT SUM(unit_price * card_count) FROM {TEAM_TABLE} "
+        f"WHERE product_code=%s AND data_source='team_options' AND card_count IS NOT NULL",
+        (code,))
+    if not row or row[0] is None:
+        return None
+    return _round2(row[0])
+
+
+def update_team_total(pool, code: str, amount: Decimal | None) -> None:
+    """把算好的团总价写回 deca_onsale_product_record.team_total_amount。
+
+    Args:
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+        amount (Decimal | None): 总价；None 会显式清空(表示"数据缺失")。
+    """
+    pool.update_one(
+        f"UPDATE {ONSALE_TABLE} SET team_total_amount=%s WHERE product_code=%s",
+        (amount, code))
+
+
+def process_xuandui(log, pool, code: str, captured_at: str) -> bool:
+    """处理一个在售·选队随机团：抓 team-options → 存表 → 算总价 → 回写商品。
+
+    Args:
+        log: 日志对象。
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+        captured_at (str): 本轮采集时刻。
+
+    Returns:
+        bool: True 成功；False 接口不可用(如刚转成剩余随机)。
+    """
+    teams = fetch_team_options(log, code)
+    if not teams:
+        return False   # 可能这一瞬间刚转成剩余随机，本轮跳过；下轮以剩余随机身份进另一分支
+    total = Decimal("0")
+    ptn = "选队随机"       # 精确 name 已在商品行 play_type_name；此处只作 teams 表内的存档
+    for t in teams:
+        upsert_team_row(pool, code, ptn, "team_options", t, captured_at)
+        cc = t.get("cardCount") or 0
+        av = t.get("availableStock") or 0
+        up = Decimal(str(t.get("unitPrice") or "0"))
+        total += up * (cc - av)
+    update_team_total(pool, code, _round2(total))
+    return True
+
+
+def process_shengyu(log, pool, code: str, captured_at: str) -> bool:
+    """处理一个在售·剩余随机团：首次存 snapshot；每轮从 detail 取实时剩余算总价回写。
+
+    公式：team_total_amount = 原始总价(存表 Σ 单价×cardCount) − detail.unitPrice × detail.availableStock。
+    若库里没有该团选队阶段的 cardCount 记录，team_total_amount 留 NULL 表示"数据缺失"。
+
+    Args:
+        log: 日志对象。
+        pool: MySQL 连接池。
+        code (str): 商品编码。
+        captured_at (str): 本轮采集时刻(仅首次存 snapshot 时用)。
+
+    Returns:
+        bool: True 处理成功；False 详情拉失败。
+    """
+    dd = fetch_detail(log, code)
+    if not dd:
+        return False
+    # 1) 首次存 snapshot(冻结不变，只存一次)
+    if not has_snapshot(pool, code):
+        snap = dd.get("remainingRandomTeamSnapshot") or {}
+        for t in (snap.get("teams") or []):
+            upsert_team_row(pool, code, "剩余随机", "snapshot", t, captured_at,
+                            snap_total=snap.get("totalQuantity"))
+    # 2) 每轮重算实时总价
+    original = get_original_total(pool, code)
+    if original is None:
+        # 转剩余随机前没被我们抓到过 team-options → 拿不到原始总价，显式留 NULL
+        update_team_total(pool, code, None)
+        log.warning(f"[数据缺失] {code} 无选队阶段 team-options 记录，team_total_amount=NULL")
+        return True
+    up_now = Decimal(str(dd.get("unitPrice") or "0"))
+    av_now = int(dd.get("availableStock") or 0)
+    remaining_value = up_now * av_now
+    total = original - remaining_value
+    if total < 0:
+        # 边界：原始总价是"名义价"，理论上不会 <0；出现即为数据异常，记日志、置 None
+        log.warning(f"[异常] {code} original={original} - remaining={remaining_value} <0，置 NULL")
+        update_team_total(pool, code, None)
+    else:
+        update_team_total(pool, code, _round2(total))
+    return True
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def main_task(log):
+    """采集主函数：一轮遍历在售随机团、更新 teams 表与 team_total_amount。
+
+    挂了每小时重试(无人值守)；单轮内各商品独立 try/except，单个失败不拖垮整轮。
+
+    Args:
+        log: 日志对象。
+
+    Raises:
+        RuntimeError: 数据库连接池异常时抛出以触发重试。
+    """
+    log.info(f"开始运行 {sys._getframe().f_code.co_name}" + "." * 40)
+    pool = MySQLConnectionPool(log=log)
+    if not pool.check_pool_health():
+        log.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+    try:
+        xd_codes, sy_codes = fetch_onsale_random(log, pool)
+        captured_at = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+
+        # 选队随机：每轮全量刷新 team-options
+        ok_xd = fail_xd = 0
+        for c in xd_codes:
+            try:
+                if process_xuandui(log, pool, c, captured_at):
+                    ok_xd += 1
+                else:
+                    fail_xd += 1
+            except Exception as e:
+                fail_xd += 1
+                log.error(f"[选队随机·失败] {c}: {e}")
+            time.sleep(BETWEEN_ITEM_SEC)
+        log.info(f"[选队随机] 处理完成: 成功 {ok_xd} 失败 {fail_xd}")
+
+        # 剩余随机：每轮从 detail 取实时剩余、更新总价
+        ok_sy = fail_sy = 0
+        for c in sy_codes:
+            try:
+                if process_shengyu(log, pool, c, captured_at):
+                    ok_sy += 1
+                else:
+                    fail_sy += 1
+            except Exception as e:
+                fail_sy += 1
+                log.error(f"[剩余随机·失败] {c}: {e}")
+            time.sleep(BETWEEN_ITEM_SEC)
+        log.info(f"[剩余随机] 处理完成: 成功 {ok_sy} 失败 {fail_sy}")
+    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():
+    """定时入口：每 INTERVAL_SEC 秒跑一次 main_task。启动时立即跑一次。"""
+    main_task(log=logger)   # 启动即刻跑一轮，避免等 5 分钟才开始
+    schedule.every(INTERVAL_SEC).seconds.do(main_task, log=logger)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == "__main__":
+    schedule_task()

deca_spider/on_sale/deca_track.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/8/3 15:36
+import sys
+from mysql_pool import MySQLConnectionPool
+import schedule
+import time
+from deca_daily_spider import do_request
+from deca_daily_spider import logger
+
+
+def main_task():
+    logger.info(f"开始运行 {sys._getframe().f_code.co_name} 得卡采集任务" + "." * 40)
+
+    sql_pool = MySQLConnectionPool(log=logger)
+    if not sql_pool.check_pool_health():
+        logger.error("数据库连接池异常")
+        raise RuntimeError("数据库连接池异常")
+
+    try:
+        url = "/api/v1/app/home/recommend/groupbuy/list"
+        data = {
+            "page": 1,
+            "pageSize": 20,
+            "categoryParam": 0
+        }
+        response = do_request(logger, url, body=data)
+        # print(response.json())
+        # print(response)
+        if response:
+
+            seen = set()
+            dict_list = []
+            data_list = response.get("data", {}).get("list", [])
+
+            for data in data_list:
+                merchant = data.get("merchant", {})
+                merchant_name = merchant.get("merchantName", "")
+                if merchant_name != '魔都兄弟球星卡':
+                    continue
+
+                product_code = data.get("code")
+                if product_code in seen:
+                    continue
+                seen.add(product_code)
+
+                dict_list.append({
+                    # "merchant_user_id": merchant.get("merchantUserID", ""),
+                    "merchant_name": merchant_name,
+                    "product_code": product_code,
+                    "title": data.get("title"),
+                    "unit_price": data.get("unitPrice", 0),
+                    "card_count": data.get("cardCount", 0),
+                    "sold_count": data.get("soldCount", 0),
+                })
+            sql_pool.insert_many(table="deca_track_record", data_list=dict_list)
+
+    except Exception as e:
+        logger.error(f"{sys._getframe().f_code.co_name} error: {e}")
+    finally:
+        logger.info(f"得卡采集 {sys._getframe().f_code.co_name} 运行结束，等待下一轮" + "." * 20)
+
+def schedule_task():
+    """定时任务入口：每天 09:00 与 15:00 各运行一次 main_task。"""
+    # main_task()  # 立即跑一次（调试时取消注释）
+
+    schedule.every(1).minutes.do(main_task)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == '__main__':
+    schedule_task()
+
+
+
+

deca_spider/on_sale/deca_wechat.py

@@ -0,0 +1,69 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/02
+"""通过 PC 版微信(wxauto4)把文件/图片发送给指定好友或群。
+
+前置条件：
+    1. 已装 wxauto4：pip install wxauto4
+       （原 wxauto 已从 PyPI 下架；wxauto4 对应微信 4.0 PC 客户端，免费，支持 Python 3.9~3.13）
+    2. Windows PC 版微信(4.0+) 已登录，窗口未最小化（wxauto4 靠 UI 自动化操作微信窗口，
+       运行期间尽量别抢占鼠标/键盘）。
+    3. 目标好友/群的备注名或昵称须与传入的 who 完全一致（测试用 "backup"）。
+
+用法：
+    - 作为模块：from deca_wechat import send_files; send_files("xx.png", who="backup")
+    - 命令行：  python deca_wechat.py <文件路径> [好友名(默认backup)]
+"""
+import sys
+
+from loguru import logger
+
+# 目标匹配是否精确：True=按 who 精确匹配备注/昵称（推荐，避免误发给名字相近的人）
+EXACT_MATCH = True
+
+
+def send_files(paths, who: str = "backup") -> bool:
+    """发送一个或多个文件给微信好友/群。
+
+    Args:
+        paths (str | list[str]): 单个文件路径或路径列表。
+        who (str, optional): 目标好友备注名/昵称或群名。Defaults to "backup"。
+
+    Returns:
+        bool: 全部发送成功返回 True；库缺失、连不上微信或发送异常返回 False。
+    """
+    try:
+        from wxauto4 import WeChat
+    except ImportError:
+        logger.error("未安装 wxauto4，请执行: pip install wxauto4")
+        return False
+
+    if isinstance(paths, str):
+        paths = [paths]
+
+    try:
+        wx = WeChat()  # 绑定当前已登录的 PC 微信（未登录/窗口不可见会失败）
+    except Exception as e:
+        logger.error(f"连接 PC 微信失败（确认微信4.0已登录且窗口可见）: {e}")
+        return False
+
+    ok = True
+    for p in paths:
+        try:
+            resp = wx.SendFiles(str(p), who=who, exact=EXACT_MATCH)
+            logger.info(f"微信已发送: {p} -> {who} | 返回: {resp}")
+        except Exception as e:
+            ok = False
+            logger.error(f"微信发送失败 {p} -> {who}: {e}")
+    return ok
+
+
+if __name__ == "__main__":
+    logger.remove()
+    logger.add(sys.stderr, level="INFO")
+    if len(sys.argv) < 2:
+        print("用法: python deca_wechat.py <文件路径> [好友名(默认backup)]")
+        sys.exit(1)
+    target = sys.argv[2] if len(sys.argv) > 2 else "backup"
+    send_files(sys.argv[1], target)

deca_spider/on_sale/km_spdier.py

@@ -0,0 +1,131 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.12.10
+# Date   : 2026/08/04
+"""得卡 DECA 商品卡牌清单(checklist)采集：拉某商品 groupbuy/card/list 的全部卡牌，入 deca_onsale_kami_record。
+
+翻页「翻到底」（靠 data.total 判断），uuid 唯一键 INSERT IGNORE 去重。
+本接口需 Authorization Bearer，复用 deca_on_sale_daily_spider 的签名与 token 续期逻辑。
+
+字段对应（列表项）：
+    sortOrder 序号 / aliasName 别名 / teamNameEn·teamNameZh 球队 / playerNameEn·playerNameZh 球员 /
+    cardType 卡种 / limitNumber 限编 / cardNumber 卡编 / uuid 卡牌唯一标识
+
+用法：
+    python km_spdier.py <商品code>      # 采集指定商品的全部卡牌清单，默认用示例 code
+    或作为模块：from km_spdier import get_card_list; get_card_list(log, code, pool)
+"""
+import sys
+import time
+
+from loguru import logger
+from mysql_pool import MySQLConnectionPool
+import deca_on_sale_daily_spider as deca  # 复用 do_request / ensure_token / make_signature
+
+# 覆盖 deca 导入时装的 logger handler，本任务日志独立到 kami_*.log
+logger.remove()
+logger.add("./logs/kami_{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")
+
+PAGE_SIZE = 20      # 每页条数（抓包实测 20）
+MAX_PAGES = 600    # 翻页保护上限（单商品清单可上万张，20/页约需数百页）
+
+
+def parse_kami(item: dict, product_code: str) -> dict | None:
+    """把卡牌清单项解析成 deca_onsale_kami_record 一行。
+
+    Args:
+        item (dict): card/list 的 data.list 项。
+        product_code (str): 所属商品编码（列表项不含，由外部带入）。
+
+    Returns:
+        dict | None: 与 deca_onsale_kami_record 列对应的数据字典；无 uuid 时返回 None。
+    """
+    uuid = item.get("uuid")
+    if not uuid:
+        return None
+    return {
+        "uuid": uuid,
+        "product_code": product_code,
+        "sort_order": item.get("sortOrder"),
+        "alias_name": item.get("aliasName"),
+        "team_name_en": item.get("teamNameEn"),
+        "team_name_zh": item.get("teamNameZh"),
+        "player_name_en": item.get("playerNameEn"),
+        "player_name_zh": item.get("playerNameZh"),
+        "card_type": item.get("cardType"),
+        "limit_number": item.get("limitNumber"),
+        "card_number": item.get("cardNumber"),
+    }
+
+
+def get_card_list(log, product_code: str, pool) -> int:
+    """翻页拉取某商品的全部卡牌清单（翻到底），INSERT IGNORE 写入 deca_onsale_kami_record。
+
+    Args:
+        log: 日志对象。
+        product_code (str): 商品编码。
+        pool (MySQLConnectionPool): MySQL 连接池。
+
+    Returns:
+        int: 本商品实际写入的卡牌数（已去重）。
+    """
+    page = 1
+    saved = 0
+    total = None
+    while page <= MAX_PAGES:
+        body = {"code": product_code, "page": page, "pageSize": PAGE_SIZE}
+        try:
+            log.info(f"商品 {product_code} 卡牌清单请求第 {page} 页")
+            resp = deca.do_request(log, "/api/v1/app/groupbuy/card/list", body, need_auth=True)
+        except Exception as e:
+            log.error(f"商品 {product_code} 卡牌清单第 {page} 页请求失败: {e}")
+            break
+        if not resp or resp.get("code") != 0:
+            log.info(f"商品 {product_code} 卡牌清单返回异常: {resp.get('msg') if resp else None}")
+            break
+
+        data = resp.get("data") or {}
+        if total is None:
+            total = data.get("total")
+            log.info(f"商品 {product_code} 卡牌清单共 {total} 张，开始翻页入库")
+        items = data.get("list") or []
+        if not items:
+            break
+
+        rows = [r for r in (parse_kami(it, product_code) for it in items) if r]
+        if rows:
+            pool.insert_many(table="deca_onsale_kami_record", data_list=rows, ignore=True)
+            saved += len(rows)
+
+        # 翻到底：本页不足一页(<PAGE_SIZE)即最后一页。
+        # total 是动态值(清单持续新增)，仅作日志参考，不用于终止判断。
+        if len(items) < PAGE_SIZE:
+            break
+        page += 1
+        if page % 50 == 0:
+            log.info(f"商品 {product_code} 已翻到第 {page} 页，累计入库 {saved}")
+        # time.sleep(1)  # 每页间隔 1s，降低请求频率
+
+    log.info(f"商品 {product_code} 卡牌清单入库完成 {saved} 张(total={total})")
+    return saved
+
+
+def main(product_code: str):
+    """采集单个商品的全部卡牌清单入库。
+
+    Args:
+        product_code (str): 商品编码。
+    """
+    pool = MySQLConnectionPool(log=logger)
+    if not pool.check_pool_health():
+        logger.error("数据库连接池异常")
+        return
+    deca.ensure_token(logger)  # 预取 token（后续请求自动续期）
+    get_card_list(logger, product_code, pool)
+
+
+if __name__ == "__main__":
+    code = sys.argv[1] if len(sys.argv) > 1 else "GB26080614002"
+    main(code)

deca_spider/on_sale/requirements.txt

@@ -0,0 +1,23 @@
+# 得卡 DECA 爬虫依赖（版本为当前环境实测，Python 3.12.10）
+# 安装：pip install -r requirements.txt
+
+# 采集 / 调度 / 日志 / 重试
+requests==2.33.1
+schedule==1.2.2
+loguru==0.7.3
+tenacity==9.1.4
+
+# 数据库（全局公共库 mysql_pool 的底层依赖）
+PyMySQL==1.1.2
+DBUtils==3.1.2
+PyYAML==6.0.3
+
+# 统计报表出图
+matplotlib==3.10.9
+
+# PC 微信(4.0) 自动发送图片；原 wxauto 已从 PyPI 下架，改用 wxauto4
+# 若因 requires-python 装不上可加：pip install wxauto4 --ignore-requires-python
+wxauto4==41.1.2
+
+# 本地公共库（含 mysql_pool / YamlLoader），非 PyPI，需 editable 安装：
+#   cd D:\work\common\charley-utils && pip install -e .