fix(runtime): strip host-injected env vars when launching 安能

所有站点 with_retry 在最后一次重试失败、reset 之前自动截图，
保存到 logs/screenshots/。网页站点走 Playwright page.screenshot()，
安能走 CDP Page.captureScreenshot。截图失败绝不阻塞任务流程。

- paths.py: 新增 SCREENSHOT_DIR (BASE_DIR/logs/screenshots/)
- runtime.py: 新增 capture_error_screenshot() 工具函数
- .gitignore: 新增 logs/ 忽略规则

Co-Authored-By: Claude <noreply@anthropic.com>

.gitignore

@@ -42,6 +42,7 @@ desktop.ini
 downloads/
 output/
 state/
+logs/
 *.xlsx
 *.xls
 *.log

docs/2026-07-31-四站点差缺对比逻辑审查报告.md

@@ -0,0 +1,332 @@
+# 四站点差缺对比逻辑审查报告
+
+> 审查日期：2026-07-31
+> 审查范围：顺心、中通、韵达、安能 四个站点的应到 vs 实到差缺对比逻辑
+> 排除：百世（站点直供未到明细，不参与四站比对）
+
+---
+
+## 一、比对算法总览（四站共用）
+
+`compare.py:process()` 对四个站点执行**完全相同**的算法步骤。站点间的差异仅由 `domain.py:STATIONS` 配置注入——列名映射 + 实到单号解析器。
+
+```
+步骤1: 读应到Excel → 按运单号去重keep-first → 构建 {运单号 → (交接单号, 交接件数=n)}
+步骤2: 读实到Excel → 站点专用解析器 → 构建 {运单基号 → {已到单号集合}}
+步骤3: 逐运单比对
+        arrived_cnt >= n  → 足额到货，跳过
+        arrived_cnt == 0  → 完全未到
+        0 < arrived < n   → 部分未到
+步骤4: 产出未到明细（交接单号 | 运单号 | 总件数 | 已到单号1 | 已到单号2 | ...）
+```
+
+### 核心口径
+
+| 指标 | 口径 |
+|------|------|
+| 应到件数 | **交接件数**（非录单件数）；按运单号去重 keep-first |
+| 实到件数 | 单号去重计数（每扫描一件=一个单号） |
+| 未到件数 | max(0, 应到件数 − 实到件数) |
+| 未到率 | 未到件数 ÷ 应到件数 |
+
+### 未到明细输出约定
+
+- 仅列出**短少运单**（实到 < 应到）
+- 列出该运单**实际已到的单号**（已到单号1, 已到单号2, ...）
+- **不编造缺件子单号**——实到扫描顺序号乱序，无法反推缺了哪个顺序号
+
+### 统计指标
+
+| 指标 | 含义 |
+|------|------|
+| 运单数 | 应到运单去重数 |
+| 应到件 | Σ 交接件数 |
+| 已到件 | Σ 实到单号去重数 |
+| 未到件 | max(0, 应到件 − 已到件) |
+| 涉及运单 | 存在短少的运单数 |
+| 完全未到 | 整单零到货运单数 |
+| 部分未到 | 部分缺件运单数 |
+
+---
+
+## 二、四站配置对照
+
+`domain.py:STATIONS` — 所有差异集中于此配置表，比对核心代码不感知站点差异。
+
+| 维度 | 中通 | 顺心 | 韵达 | 安能 |
+|------|------|------|------|------|
+| 应到文件 | `中通-应到货物数据.xlsx` | `顺心-应到货物数据.xlsx` | `韵达-应到货物数据.xlsx` | `安能-应到货物数据.xlsx` |
+| 实到文件 | `中通-实到货物数据.xlsx` | `顺心-实到货物数据.xlsx` | `韵达-实到货物数据.xlsx` | `安能-实到货物数据.xlsx` |
+| 应到-运单号列 | `运单号` | `运单号` | `运单号` | `运单号` |
+| 应到-件数列 | `交接件数` | `交接件数` | `交接件数` | `交接件数` |
+| 应到-交接单号列 | `交接单号` | `交接单号` | `交接单号` | `交接单号` |
+| 实到-基号列 | —（从复合串推导） | `运单号` | **`主单号`** | **`所属单号`** |
+| 实到-单号列 | `运单号`（复合串） | `子单号` | `子单号` | `扫描单号` |
+| 解析器 | `arrived_pieces_zhongtong` | `arrived_pieces_by_cols` | `arrived_pieces_by_cols` | `arrived_pieces_by_cols` |
+
+---
+
+## 三、逐站点详细分析
+
+### 3.1 中通（ZTO）
+
+#### 业务逻辑
+
+实到货物数据中的「运单号」为复合串，由三部分构成：
+
+```
+┌──────────┬────────────┬──────────┐
+│ 运单号    │ 录单件数    │ 顺序号    │
+│ (12位)    │ (4位)      │ (4位)    │
+└──────────┴────────────┴──────────┘
+  总长 20 位
+
+示例: 330953527953 0001 0001
+      ├─ 运单号 ─┤├录单┤├顺序┤
+```
+
+- **运单号(12位)**: 与应到货物数据中的运单号对齐
+- **录单件数(4位)**: 该运单在系统中的录单总件数，0占位
+- **顺序号(4位)**: 0占位，如 `0001`, `0002`, `0003`, `0004`
+
+对比逻辑：
+1. 从应到数据取运单号 + 交接件数（**非录单件数**）
+2. 从实到数据取复合串，掐尾8位得运单基号，完整串为子运单号
+3. 按运单基号分组，子运单号去重得实到件数
+4. 实到件数 < 交接件数 → 差缺
+
+> **重要**: 录单件数仅作参考。举例：某运单录单件数=4、交接件数=2，实到最多出现2条数据。如果只出现了1条，我们只知道差缺了，但**无法判断具体差缺了哪一件**（顺序号乱序）。
+
+#### 代码实现
+
+`domain.py:17-26` — 实到解析器：
+
+```python
+def arrived_pieces_zhongtong(df):
+    res = defaultdict(set)
+    for v in df["运单号"]:
+        v = str(v).strip()
+        if len(v) > 8 and v[-4:].isdigit():
+            res[v[:-8]].add(v)    # 基号=前12位, 已到单号=完整20位复合串
+    return res
+```
+
+`domain.py:48-56` — 站点配置：
+
+```python
+{
+    "name": "中通",
+    "exp_qty": "交接件数",       # 应到件数口径：交接件数（非录单件数）
+    "exp_wb": "运单号",
+    "exp_jd": "交接单号",
+    "arrived_pieces": arrived_pieces_zhongtong,
+    "columns": ["交接单号", "运单号", "总件数"],
+}
+```
+
+#### 对齐情况：✅ 对齐
+
+代码实现与业务逻辑一致。`v[:-8]` 掐尾8位得12位运单基号，保留完整复合串作为已到单号——不解析、不推断录单件数和顺序号的具体含义。
+
+---
+
+### 3.2 安能（Anneng）
+
+#### 业务逻辑
+
+与中通相同的差缺对比逻辑。
+
+安能实到数据同样为复合串，结构：`运单号(12位) + 录单件数(4位) + 顺序号(4位)`（20位）。
+
+与中通的关键区别：安能实到表有**独立的「所属单号」列**（干净运单基号），无需像中通那样从复合串掐尾8位推导基号。
+
+#### 代码实现
+
+`domain.py:76-86`：
+
+```python
+{
+    "name": "安能",
+    "arrived_pieces": arrived_pieces_by_cols("所属单号", "扫描单号"),
+    ...
+}
+```
+
+安能使用 `arrived_pieces_by_cols` 而非 `arrived_pieces_zhongtong`——直接从「所属单号」列读基号、从「扫描单号」列读完整单号，效果等价。
+
+| 差异点 | 中通 | 安能 |
+|--------|------|------|
+| 实到基号来源 | 从复合串解析（`v[:-8]`） | 直接读「所属单号」列 |
+| 实到单号来源 | 复合串本身（「运单号」列） | 「扫描单号」列 |
+| 解析器 | `arrived_pieces_zhongtong` | `arrived_pieces_by_cols` |
+| 最终产出 | `{基号 → {完整单号集合}}` | 相同 |
+
+#### 数据库验证
+
+```
+piece_no=61003282264500140014 → waybill_no=610032822645 (12位), total=0014, seq=0014
+```
+
+#### 对齐情况：✅ 对齐
+
+---
+
+### 3.3 顺心（Shunxin）⚠️
+
+#### 业务逻辑
+
+顺心站点需区分两类运单：
+
+**A. 非SF开头运单（占 97%）:**
+
+实到「子单号」结构为两部分：
+
+```
+┌──────────┬──────────┐
+│ 运单号    │ 顺序号    │
+│ (不定长)  │ (3位)    │
+└──────────┴──────────┘
+
+示例: S71623721115 001
+      ├─ 运单号 ──┤├顺序┤
+
+注意：顺心子单号无录单件数部分（仅两部分）
+```
+
+对比时从实到取「子单号」列，按「运单号」分组，子单号去重得实到件数。
+
+**B. SF开头运单（占 3%）:**
+
+SF订单的「子单号」为**随机号码**（非由运单号衍生），不能用于差缺推导。
+
+对比逻辑：
+1. 在实到数据中按「运单号」字段查找，统计出现次数
+2. 出现次数 < 交接件数 → 差缺
+3. 将找到的子单号（虽随机但可以列出来）填入「已到单号」列
+
+SF订单的差缺判定：**只基于交接件数与实到运单号出现次数的比较**，不依赖子单号的结构解析。
+
+#### 代码实现
+
+`domain.py:57-65`：
+
+```python
+{
+    "name": "顺心",
+    "arrived_pieces": arrived_pieces_by_cols("运单号", "子单号"),
+}
+```
+
+**SF 与非 SF 没有任何区分处理。** 所有运单走同一条路径。
+
+#### 数据库验证
+
+**非SF（正常）:**
+```
+子单号=S71623721115001 → 运单号=S71623721115 + 后缀=001  ✅
+子单号=S71934073996002 → 运单号=S71934073996 + 后缀=002  ✅
+```
+
+**SF（异常）:**
+```
+运单号=SF1225002296515 的两条实到记录:
+  子单号=SF2025318183224  (随机SF号码)
+  子单号=SF1225002296515  (与运单号相同)
+```
+数据中有 10 个SF运单存在多条实到记录。
+
+#### 对齐情况：⚠️ 部分对齐，SF特殊逻辑缺失
+
+| 检查项 | 代码现状 | 业务要求 |
+|--------|----------|----------|
+| 非SF处理 | ✅ `arrived_pieces_by_cols("运单号", "子单号")` | 一致 |
+| 非SF子单号结构 | ✅ 运单号 + 顺序号（两部分） | 一致 |
+| SF处理 | ❌ 与非SF完全一致，使用子单号去重 | **不能**使用子单号，只按运单号行数计数 |
+| 功能影响 | 子单号虽随机但值唯一，按目前逻辑也能正确去重计数 | 但语义不正确——SF子单号不由运单号衍生 |
+
+---
+
+### 3.4 韵达（Yunda）❌
+
+#### 业务逻辑
+
+**去重规则:** 韵达实到数据存在重复行（同一子单号出现两次）。去重依据为「交接单号」字段：
+- **保留**交接单号为**空**的行
+- **丢弃**交接单号**非空**的行
+
+**子单号结构:** 两部分——单号 + 顺序号（无录单件数部分）。
+
+```
+┌──────────┬──────────┐
+│ 主单号    │ 顺序号    │
+│ (不定长)  │ (4位)    │
+└──────────┴──────────┘
+
+示例: 713326603 0003
+      ├─主单号─┤├顺序┤
+```
+
+**对比方式:** 与中通/安能同——按「主单号」分组，「子单号」去重得实到件数，与交接件数比对。
+
+#### 代码实现
+
+`store.py:316-320`（入库过滤）:
+
+```python
+if site == "韵达":
+    # 韵达业务清洗：抛弃「交接单号」为空的行（派件/签收等其他扫描无交接单号），
+    # 再按子单号去重（一件多扫只留一条；清洗后子单号已天然唯一，drop 为保险）。
+    df = df[df["交接单号"].astype(str).str.strip() != ""]    # ← 保留非空
+    df = df.drop_duplicates(subset=[cm["piece"]], keep="last")
+```
+
+`domain.py:67-75`（比对配置）:
+
+```python
+{
+    "name": "韵达",
+    "exp_wb": "运单号",
+    "arrived_pieces": arrived_pieces_by_cols("主单号", "子单号"),
+}
+```
+
+#### 对齐情况：❌ 交接单号过滤逻辑完全相反
+
+| 检查项 | 代码现状 | 业务要求 |
+|--------|----------|----------|
+| 交接单号过滤 | 保留 `!= ""`（**非空**） | 保留 `== ""`（**空**） |
+| 子单号结构 | ✅ `7133266030003` = wb`713326603` + seq`0003` | 一致 |
+| 实到解析 | ✅ `arrived_pieces_by_cols("主单号", "子单号")` | 一致 |
+| compare.py 过滤 | ❌ **无过滤**，所有行参与比对 | 需要过滤 |
+
+**影响分析:**
+
+1. `store.py` 过滤反了——入库时留下了错误的数据集
+2. `compare.py` 完全没有交接单号过滤——如果原始 Excel 中同时存在空和非空行，比对阶段会全部读入导致重复计数
+3. 当前数据库中韵达 3483 条记录全部为非空交接单号——说明当前 Excel 数据中空交接单号行偏少或不存在，但这不改变逻辑错误
+
+---
+
+## 四、差异汇总
+
+| # | 站点 | 问题 | 严重程度 | 影响范围 |
+|---|------|------|----------|----------|
+| 1 | **韵达** | 交接单号过滤反了：`!= ""` 应改为 `== ""` | ❌ 严重 | `store.py:319` + `compare.py` 需新增过滤 |
+| 2 | **顺心** | SF运单无特殊处理，与非SF混用子单号 | ⚠️ 中等 | `domain.py` 需新增SF判断分支 |
+| 3 | **中通** | 录单件数0占位描述与实际数据完全一致 | ✅ 无影响 | 代码不依赖此区分 |
+
+---
+
+## 五、代码位置索引
+
+| 逻辑 | 文件 | 行号 |
+|------|------|------|
+| 单站比对 `process()` | `compare.py` | 61-143 |
+| 站点配置 `STATIONS` | `domain.py` | 46-87 |
+| 中通实到解析器 | `domain.py` | 17-26 |
+| 通用实到解析器 | `domain.py` | 29-42 |
+| 单站未到文件写入 | `compare.py` | 258-273 |
+| 全量汇总报告 | `compare.py` | 293-324 |
+| 未到触发编排 | `runtime.py` | 478-497 |
+| 韵达入库过滤（需修） | `store.py` | 316-320 |
+| 顺心实到配置（需修） | `domain.py` | 57-65 |

docs/2026-07-31-顺心DB差缺对比实施计划.md

@@ -0,0 +1,264 @@
+# 顺心 DB 差缺对比 — 实施计划
+
+> 日期：2026-07-31
+> 目标：将顺心站点差缺对比从 Excel 读取改为 PostgreSQL 查询，并修正 SF 运单特殊处理逻辑
+
+---
+
+## 一、背景
+
+### 当前状态（Excel 方式）
+
+```
+compare.py:process("顺心")
+  ├── 读 downloads/顺心-应到货物数据.xlsx
+  ├── 读 downloads/顺心-实到货物数据.xlsx
+  ├── arrived_pieces_by_cols("运单号", "子单号")  ← SF/non-SF 无区分
+  └── 产出 {站}-未到数据.xlsx + 统计 dict
+```
+
+### 需要解决的两个问题
+
+1. **从 Excel 切换到 DB**：数据已持久化到 PostgreSQL，比对应直接从 DB 查询
+2. **顺心 SF 运单特殊处理**：SF 运单的子单号为随机号码，不能用于去重计数，应使用行计数
+
+---
+
+## 二、数据结构
+
+### PostgreSQL 表
+
+**expected_record**（关键列）：
+
+| 列 | 类型 | 说明 |
+|----|------|------|
+| site | TEXT | 站点 |
+| waybill_no | TEXT | 运单号（唯一键之一，SF 以 "SF" 开头） |
+| handover_no | TEXT | 交接单号（批次标识） |
+| handover_pieces | INTEGER | 交接件数（应到口径） |
+| order_pieces | INTEGER | 录单件数（参考） |
+| business_date | DATE | 下载目标日期 |
+
+**actual_record**（关键列）：
+
+| 列 | 类型 | 说明 |
+|----|------|------|
+| site | TEXT | 站点 |
+| waybill_no | TEXT | 运单基号（关联 expected_record） |
+| piece_no | TEXT | 扫描单号（non-SF：运单号+顺序号；SF：随机号码） |
+| scan_time | TIMESTAMPTZ | 扫描时间（可靠，当天数据=当天扫描） |
+
+### SF 数据特征（已验证）
+
+- 顺心 actual_record 中 SF 运单：148 条
+- `piece_no == waybill_no`：86 条（58%）
+- `piece_no != waybill_no`：62 条（42%）← 随机 SF 号码
+- SF 运单 expected：99 条，分布在 31 个交接批次中
+
+---
+
+## 三、算法设计
+
+### 核心思路：以实到为锚，通过交接单号反推批次
+
+```
+输入: site="顺心", date="2026-07-25"
+
+Step 1 — 取实到锚点
+  SELECT DISTINCT waybill_no FROM actual_record
+  WHERE site='顺心' AND scan_time::date = '2026-07-25'
+
+Step 2 — 反推交接批次
+  SELECT DISTINCT handover_no FROM expected_record
+  WHERE site='顺心'
+    AND waybill_no IN (Step 1 的运单集合)
+
+Step 3 — 展开批次全量应到
+  SELECT waybill_no, handover_no, handover_pieces
+  FROM expected_record
+  WHERE site='顺心'
+    AND handover_no IN (Step 2 的交接单号集合)
+
+Step 4 — 取批次全量实到
+  SELECT waybill_no, piece_no FROM actual_record
+  WHERE site='顺心'
+    AND waybill_no IN (Step 3 的运单集合)
+
+Step 5 — 逐运单比对
+  for each waybill in Step 3:
+    if waybill_no LIKE 'SF%':
+        arrived_cnt = COUNT(*)         ← 行计数，不去重
+    else:
+        arrived_cnt = COUNT(DISTINCT piece_no)  ← 子单号去重
+    if arrived_cnt < handover_pieces → 差缺
+```
+
+### SF vs non-SF 处理差异
+
+| | non-SF | SF |
+|------|--------|-----|
+| piece_no 含义 | 运单号 + 顺序号（可推导） | 随机 SF 号码（无推导意义） |
+| 实到计数方式 | `COUNT(DISTINCT piece_no)` | `COUNT(*)`（行计数） |
+| 已到单号列表 | 列出去重后的子单号 | 列出所有 piece_no（含重复） |
+
+### 统计指标
+
+| 指标 | 公式 |
+|------|------|
+| 运单数 | Step 3 去重运单数 |
+| 应到件 | Σ handover_pieces |
+| 已到件 | Σ arrived_cnt |
+| 未到件 | max(0, 应到件 − 已到件) |
+| 涉及运单 | arrived_cnt < handover_pieces 的运单数 |
+| 完全未到 | arrived_cnt = 0 的运单数 |
+| 部分未到 | 0 < arrived_cnt < handover_pieces 的运单数 |
+| 未到率 | 未到件 ÷ 应到件 |
+
+### 边界情况覆盖
+
+| 情况 | 覆盖方式 |
+|------|----------|
+| 同日多批次 | Step 2 查出全部涉及的 handover_no |
+| 跨天到达（延迟） | Step 4 不限 scan_time，历史扫描全计入 |
+| 溢到（实到 > 应到） | arrived_cnt >= n 跳过，不进差缺表 |
+| 完全沉默批次 | 一件未扫 = 实到无锚点，该批次不会被触发——在首次有扫描那天被纳入 |
+| SF 子单号重复 | 用 COUNT(*) 而非 COUNT(DISTINCT)，不会漏计 |
+
+---
+
+## 四、模块设计
+
+### 新增文件
+
+**`inbound_verify/db_compare.py`** — DB 比对引擎（纯 PostgreSQL + Python）
+
+```python
+# 核心函数签名
+
+def compare_site_date(site: str, date: str) -> CompareResult | None:
+    """对指定站点和日期执行 DB 差缺比对。
+    
+    返回 CompareResult（stats + undelivered_rows），
+    当天无实到数据时返回 None。
+    """
+
+def compare_site_batch(site: str, handover_no: str) -> CompareResult | None:
+    """按指定交接单号执行全批次比对（不依赖实到锚点）。"""
+```
+
+**数据类型**：
+
+```python
+@dataclass
+class CompareResult:
+    stats: dict          # 统计指标
+    rows: list[dict]     # 差缺明细行
+    batches: list[str]   # 涉及的交接批次
+
+@dataclass  
+class UndeliveredRow:
+    handover_no: str     # 交接单号
+    waybill_no: str      # 运单号
+    total_pieces: int    # 总件数（=交接件数）
+    arrived_pieces: int  # 已到件数
+    arrived_list: list[str]  # 已到单号列表
+    is_sf: bool          # 是否 SF 运单
+```
+
+### 修改文件
+
+**`inbound_verify/cli/server.py`** — 新增 API 端点
+
+```python
+@app.post("/compare")
+def run_compare(req: CompareRequest):
+    """DB 比对：{site, date} → 返回差缺结果"""
+    
+@app.get("/compare/{site}/{date}")
+def get_compare(site: str, date: str):
+    """查询某站点某日的差缺结果（缓存）"""
+```
+
+### 现有文件保持不动
+
+- `compare.py` — 保留不动，Excel 比对继续可用
+- `domain.py` — 可能需要新增 DB 版站点配置（或复用现有）
+- `runtime.py` — 暂不改动，`_site_undelivered_handler` 仍走 Excel 路径
+
+---
+
+## 五、实施步骤
+
+### Phase 1 — `db_compare.py` 核心引擎
+
+- [ ] 新建 `inbound_verify/db_compare.py`
+- [ ] 实现 `compare_site_date("顺心", date)`
+- [ ] SF/non-SF 分支处理
+- [ ] 返回 `CompareResult`
+- [ ] 终端手动验证（直接调函数，打印结果）
+
+### Phase 2 — API 端点
+
+- [ ] 在 `server.py` 新增 `POST /compare`
+- [ ] `CompareRequest { site, date }`
+- [ ] 调用 `db_compare.compare_site_date()`
+- [ ] 返回 JSON：stats + undelivered rows
+- [ ] HTTP 验证：curl 调 `/compare` 对比不同日期结果
+
+### Phase 3 — Excel 输出（可选）
+
+- [ ] `db_compare` 生成 Excel 报告（复用现有 `compare.py` 的 openpyxl 样式）
+- [ ] 输出到 `output/顺心-{date}-未到数据.xlsx`
+- [ ] 或者只输出 JSON，前端自行渲染
+
+### Phase 4 — 替换 undelivered 任务流
+
+- [ ] `runtime.py` 新增 `_db_undelivered_handler`
+- [ ] 下载完成后不再调 Excel 比对，改调 DB 比对
+- [ ] 逐步替换 `TASK_HANDLERS` 中的顺心 undelivered handler
+
+### Phase 5 — 扩展到中通/韵达/安能
+
+- [ ] 各站适配（主要是 piece_no 去重方式差异）
+- [ ] 中通：`COUNT(DISTINCT piece_no)`，无 SF 问题
+- [ ] 韵达：同上
+- [ ] 安能：同上
+
+---
+
+## 六、测试策略
+
+### 手工验证（Phase 1）
+
+```python
+# 终端直接调
+from inbound_verify.db_compare import compare_site_date
+result = compare_site_date("顺心", "2026-07-25")
+print(result.stats)
+# 对比基于 Excel 版的 compare.process("顺心") 结果
+```
+
+### API 验证（Phase 2）
+
+```bash
+curl -X POST http://127.0.0.1:8000/compare \
+  -H "Content-Type: application/json" \
+  -d '{"site":"顺心","date":"2026-07-25"}'
+```
+
+### 回归验证
+
+- 新 DB 比对结果 vs 旧 Excel 比对结果（同一份数据）
+- SF 运单的 arrived_cnt 对比：DB 版（COUNT(*)）vs Excel 版（COUNT DISTINCT piece_no）
+- 确认 SF 运单不再被漏计
+
+---
+
+## 七、风险与注意事项
+
+| 风险 | 缓解 |
+|------|------|
+| DB 连接超时（cpolar 隧道） | 加 connect_timeout + try/except 降级 |
+| 全表扫描性能 | 依赖 (site, waybill_no) 和 (site, scan_time) 索引 |
+| SF 运单数据量小（~1%） | 测试覆盖可能不足——需找有 SF 差缺的日期验证 |
+| `scan_time` 时区 | 统一用 `::date` cast，确认与服务器时区一致 |

inbound_verify/cli/server.py

@@ -27,7 +27,7 @@ from fastapi import FastAPI, HTTPException
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
-from inbound_verify.paths import DOWNLOAD_DIR, OUTPUT_DIR
+from inbound_verify.paths import OUTPUT_DIR
 from inbound_verify import state_store
 from inbound_verify.runtime import (
     HEARTBEAT_INTERVAL,
@@ -36,6 +36,7 @@ from inbound_verify.runtime import (
     launch_and_prepare,
     run_heartbeat,
 )
+from inbound_verify import db_compare
 
 # 全部站点；百世固定下载当天，不可配置偏移
 ALL_SITES = ["顺心", "百世", "中通", "韵达", "安能"]
@@ -127,7 +128,10 @@ def _enqueue_fetch(site, kind):
     if not _in_active_window(cfg["active_start"], cfg["active_end"]):
         return  # 不在激活时段，跳过本次 fire
     try:
-        tid = state_store.create_task_if_idle(site, kind)
+        target_date = state_store.resolve_target_date(site, kind)
+        tid = state_store.create_task_if_idle(
+            site, kind, trigger="auto", target_date=target_date
+        )
         if tid is None:
             return  # 上一次同类任务还没跑完，跳过避免堆积
         task_queue.put((tid, {"site": site, "kind": kind}))
@@ -224,7 +228,13 @@ def create_task(req: TaskRequest):
             raise HTTPException(
                 status_code=400, detail="百世固定下载当天，不支持指定日期"
             )
-    task_id = state_store.create_task(req.site, req.kind)
+    task_id = state_store.create_task(
+        req.site,
+        req.kind,
+        trigger="manual",
+        target_date=state_store.resolve_target_date(req.site, req.kind, req.date),
+        force=req.force,
+    )
     spec = {"site": req.site, "kind": req.kind, "force": req.force}
     if req.date:
         spec["date"] = req.date
@@ -245,6 +255,71 @@ def list_tasks(limit: int = 20):
     return state_store.list_tasks(limit)
 
 
+# ── DB 比对（基于 PostgreSQL，不依赖 Excel 文件）──
+
+
+class CompareRequest(BaseModel):
+    site: str
+    date: str  # YYYY-MM-DD
+
+
+@app.post("/compare")
+def run_compare(req: CompareRequest):
+    """DB 差缺比对：以实到扫描日期为锚点，反推交接批次，展开全量比对。
+    返回统计指标 + 差缺明细。
+    """
+    # 合法性校验
+    if req.site not in db_compare.SITE_COMPARE_CONFIG:
+        raise HTTPException(
+            status_code=400,
+            detail=f"不支持的站点: {req.site}（支持: {list(db_compare.SITE_COMPARE_CONFIG.keys())}）",
+        )
+    try:
+        target_date = datetime.strptime(req.date, "%Y-%m-%d").date()
+    except ValueError:
+        raise HTTPException(
+            status_code=400, detail=f"date 格式非法，需 YYYY-MM-DD: {req.date}"
+        )
+    today = datetime.now().date()
+    if target_date > today:
+        raise HTTPException(status_code=400, detail=f"date 不可为未来日期: {req.date}")
+
+    result = db_compare.compare_site_date(req.site, req.date)
+    if result is None:
+        raise HTTPException(
+            status_code=404,
+            detail=f"{req.site} {req.date}: 当天无实到数据，无法比对",
+        )
+
+    return {
+        "site": result.site,
+        "date": result.date,
+        "batches": result.batches,
+        "stats": {
+            "waybill_count": result.stats.waybill_count,
+            "sf_wb_count": result.stats.sf_wb_count,
+            "expected_pieces": result.stats.expected_pieces,
+            "arrived_pieces": result.stats.arrived_pieces,
+            "undelivered_pieces": result.stats.undelivered_pieces,
+            "undelivered_wb": result.stats.undelivered_wb,
+            "full_miss": result.stats.full_miss,
+            "part_miss": result.stats.part_miss,
+            "sf_undelivered": result.stats.sf_undelivered,
+        },
+        "rows": [
+            {
+                "handover_no": r.handover_no,
+                "waybill_no": r.waybill_no,
+                "total_pieces": r.total_pieces,
+                "arrived_pieces": r.arrived_pieces,
+                "arrived_list": r.arrived_list,
+                "is_sf": r.is_sf,
+            }
+            for r in result.rows
+        ],
+    }
+
+
 @app.get("/status")
 def get_status():
     """各站登录态 + 数据态 + 入库态（前端状态盘用），另含 worker 就绪状态。"""
@@ -351,20 +426,6 @@ def download_report():
     return FileResponse(path, filename=REPORT_FILE)
 
 
-@app.get("/data/{filename}")
-def download_data(filename: str):
-    """下载 downloads/ 下的数据文件（防路径穿越）。"""
-    if not filename or "/" in filename or "\\" in filename or ".." in filename:
-        raise HTTPException(status_code=400, detail="非法文件名")
-    path = os.path.join(DOWNLOAD_DIR, filename)
-    # 双重校验：解析后绝对路径仍在 DOWNLOAD_DIR 内
-    if not os.path.abspath(path).startswith(os.path.abspath(DOWNLOAD_DIR) + os.sep):
-        raise HTTPException(status_code=400, detail="非法路径")
-    if not os.path.isfile(path):
-        raise HTTPException(status_code=404, detail="文件不存在")
-    return FileResponse(path, filename=filename)
-
-
 def main():
     """服务模式入口。传字符串导入路径（规范写法；不开 reload/workers 时进程内 import，行为等价）。"""
     uvicorn.run("inbound_verify.cli.server:app", host="0.0.0.0", port=8000)

inbound_verify/compare.py

@@ -75,6 +75,12 @@ def process(name):
     df_exp = pd.read_excel(exp_path, dtype=str).fillna("")
     df_act = pd.read_excel(act_path, dtype=str).fillna("")
 
+    if name == "韵达":
+        # 韵达实到数据有重复行（同子单号出现两次），保留交接单号为空的（到/接件扫描），
+        # 丢弃交接单号不为空的（派件/签收等），再按子单号去重。
+        df_act = df_act[df_act["交接单号"].astype(str).str.strip() == ""]
+        df_act = df_act.drop_duplicates(subset=["子单号"], keep="last")
+
     # 同一运单可能有多条交接记录，按运单号去重、保留首条
     dup = int(df_exp[cfg["exp_wb"]].duplicated().sum())
     df_exp = df_exp.drop_duplicates(subset=[cfg["exp_wb"]], keep="first")

inbound_verify/db_compare.py

@@ -0,0 +1,658 @@
+# -*- coding: utf-8 -*-
+"""
+db_compare.py — 基于 PostgreSQL 的应到未到差缺比对引擎。
+
+与 compare.py（Excel 版）并行：本模块直接从 DB 查询数据进行比对，
+不依赖 downloads/ 下的 Excel 文件。
+
+核心思路：以实到扫描日期为锚点 → 反推交接批次 → 展开批次全量比对。
+
+每个站点只需提供配置（waybill 列名 / piece 列名 / 是否有 SF 特殊处理），
+核心比对逻辑完全通用。
+
+顺心站点 SF 运单特殊处理：SF 运单的子单号（piece_no）为随机号码，不能用
+COUNT(DISTINCT piece_no) 去重计数，改为 COUNT(*) 行计数。
+
+用法：
+    from inbound_verify.db_compare import compare_site_date, SITE_COMPARE_CONFIG
+
+    result = compare_site_date("顺心", "2026-07-25")
+    if result:
+        print(result.stats)
+        for row in result.rows:
+            print(row)
+"""
+
+import os
+from dataclasses import dataclass, field
+from datetime import date, datetime, timedelta
+
+import psycopg
+import yaml
+from openpyxl import Workbook
+from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
+
+from inbound_verify.paths import CONFIG_PATH, OUTPUT_DIR, DOWNLOAD_DIR
+from inbound_verify.domain import _site_cfg, ALL_REPORT_SITES, BAISHI_COLUMNS
+
+# ============================== 结果类型 ==============================
+
+
+@dataclass
+class CompareStats:
+    """单站点/单批次比对统计。"""
+
+    waybill_count: int = 0  # 应到运单数
+    expected_pieces: int = 0  # 应到件数
+    arrived_pieces: int = 0  # 实到件数
+    undelivered_pieces: int = 0  # 未到件数
+    undelivered_wb: int = 0  # 差缺运单数
+    full_miss: int = 0  # 完全未到
+    part_miss: int = 0  # 部分未到
+    sf_wb_count: int = 0  # SF 运单数
+    sf_undelivered: int = 0  # SF 差缺数
+
+
+@dataclass
+class UndeliveredRow:
+    """单条差缺明细。"""
+
+    handover_no: str = ""  # 交接单号
+    waybill_no: str = ""  # 运单号
+    total_pieces: int = 0  # 总件数（交接件数）
+    arrived_pieces: int = 0  # 已到件数
+    arrived_list: list = field(default_factory=list)  # 已到单号列表
+    is_sf: bool = False  # 是否 SF 运单
+
+
+@dataclass
+class CompareResult:
+    """一次比对的完整结果。"""
+
+    site: str = ""
+    date: str = ""
+    batches: list = field(default_factory=list)  # 涉及的交接批次
+    stats: CompareStats = field(default_factory=CompareStats)
+    rows: list = field(default_factory=list)  # UndeliveredRow 列表
+
+
+# ============================== 站点比对配置 ==============================
+
+
+@dataclass
+class SiteCompareConfig:
+    """DB 比对的站点参数。"""
+
+    name: str  # 站点名
+    has_sf: bool = False  # 是否需要区分 SF 运单
+
+
+# 四站点 DB 比对配置（百世不参与 4 站比对）
+SITE_COMPARE_CONFIG: dict[str, SiteCompareConfig] = {
+    "顺心": SiteCompareConfig(name="顺心", has_sf=True),
+    "中通": SiteCompareConfig(name="中通", has_sf=False),
+    "韵达": SiteCompareConfig(name="韵达", has_sf=False),
+    "安能": SiteCompareConfig(name="安能", has_sf=False),
+}
+
+
+# ============================== DB 连接 ==============================
+
+
+def _load_pg_config():
+    """从 config.yaml 读 postgres 段。与 store.py 共用同一配置源。"""
+    if not os.path.exists(CONFIG_PATH):
+        raise FileNotFoundError(
+            f"未找到配置文件 {CONFIG_PATH}（请参考 config.example.yaml 创建 config.yaml）"
+        )
+    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
+        cfg = yaml.safe_load(f) or {}
+    pg = cfg.get("postgres") or {}
+    return {
+        "host": pg.get("host", "127.0.0.1"),
+        "port": int(pg.get("port", 5432)),
+        "user": pg.get("user", "postgres"),
+        "password": pg.get("password", ""),
+        "dbname": pg.get("dbname", "CQHXDB"),
+        "schema": pg.get("schema", "inbound_verify"),
+        "connect_timeout_seconds": int(pg.get("connect_timeout_seconds", 5)),
+    }
+
+
+def _connect():
+    c = _load_pg_config()
+    return psycopg.connect(
+        host=c["host"],
+        port=c["port"],
+        dbname=c["dbname"],
+        user=c["user"],
+        password=c["password"],
+        options=f"-c search_path={c['schema']} -c statement_timeout=30s",
+        connect_timeout=c["connect_timeout_seconds"],
+    )
+
+
+# ============================== 核心比对逻辑 ==============================
+
+
+def compare_site_date(site: str, target_date: str) -> CompareResult | None:
+    """对指定站点和日期执行 DB 差缺比对。
+
+    算法：
+      1. 取 scan_time::date = target_date 的实到运单（锚点）
+      2. 反推这些运单所属的交接批次（handover_no）
+      3. 展开批次全量应到运单
+      4. 查询批次全量实到扫描
+      5. 逐运单比对差缺（SF/non-SF 分支处理）
+
+    Args:
+        site: 站点名（"顺心"/"中通"/"韵达"/"安能"）
+        target_date: 日期 "YYYY-MM-DD"
+
+    Returns:
+        CompareResult 或 None（当天无实到数据时返回 None）
+    """
+    cfg = SITE_COMPARE_CONFIG.get(site)
+    if cfg is None:
+        print(f"[db_compare] 不支持的站点: {site}")
+        return None
+
+    try:
+        conn = _connect()
+        cur = conn.cursor()
+
+        # ── Step 1: 取实到锚点 ──
+        cur.execute(
+            """
+            SELECT DISTINCT waybill_no FROM actual_record
+            WHERE site = %s AND scan_time::date = %s
+            """,
+            (site, target_date),
+        )
+        anchor_wbs = [r[0] for r in cur.fetchall()]
+        if not anchor_wbs:
+            print(f"[db_compare] {site} {target_date}: 当天无实到数据")
+            conn.close()
+            return None
+
+        # ── Step 2: 反推交接批次 ──
+        cur.execute(
+            """
+            SELECT DISTINCT e.handover_no FROM expected_record e
+            WHERE e.site = %s AND e.waybill_no = ANY(%s)
+            """,
+            (site, anchor_wbs),
+        )
+        batches = [r[0] for r in cur.fetchall()]
+
+        # ── Step 3: 展开批次全量应到 ──
+        cur.execute(
+            """
+            SELECT waybill_no, handover_no, handover_pieces
+            FROM expected_record
+            WHERE site = %s AND handover_no = ANY(%s)
+            ORDER BY handover_no, waybill_no
+            """,
+            (site, batches),
+        )
+        exp_rows = cur.fetchall()  # [(waybill_no, handover_no, handover_pieces), ...]
+
+        if not exp_rows:
+            conn.close()
+            return None
+
+        all_wbs = [r[0] for r in exp_rows]
+
+        # ── Step 4: 取批次全量实到 ──
+        cur.execute(
+            """
+            SELECT waybill_no, piece_no FROM actual_record
+            WHERE site = %s AND waybill_no = ANY(%s)
+            ORDER BY waybill_no, piece_no
+            """,
+            (site, all_wbs),
+        )
+        act_rows = cur.fetchall()  # [(waybill_no, piece_no), ...]
+
+        conn.close()
+
+        # ── Step 5: 逐运单比对 ──
+        return _do_compare(site, target_date, batches, exp_rows, act_rows, cfg)
+
+    except Exception as e:
+        print(f"[db_compare] {site} {target_date} 比对异常: {e}")
+        return None
+
+
+def compare_site_batch(site: str, handover_no: str) -> CompareResult | None:
+    """按指定交接单号执行全批次比对（不依赖实到锚点）。
+
+    用于已知交接单号后精确比对某一批次。
+    """
+    cfg = SITE_COMPARE_CONFIG.get(site)
+    if cfg is None:
+        print(f"[db_compare] 不支持的站点: {site}")
+        return None
+
+    try:
+        conn = _connect()
+        cur = conn.cursor()
+
+        cur.execute(
+            """
+            SELECT waybill_no, handover_no, handover_pieces
+            FROM expected_record
+            WHERE site = %s AND handover_no = %s
+            ORDER BY waybill_no
+            """,
+            (site, handover_no),
+        )
+        exp_rows = cur.fetchall()
+        if not exp_rows:
+            conn.close()
+            return None
+
+        all_wbs = [r[0] for r in exp_rows]
+
+        cur.execute(
+            """
+            SELECT waybill_no, piece_no FROM actual_record
+            WHERE site = %s AND waybill_no = ANY(%s)
+            ORDER BY waybill_no, piece_no
+            """,
+            (site, all_wbs),
+        )
+        act_rows = cur.fetchall()
+
+        conn.close()
+
+        return _do_compare(
+            site,
+            f"batch:{handover_no}",
+            [handover_no],
+            exp_rows,
+            act_rows,
+            cfg,
+        )
+
+    except Exception as e:
+        print(f"[db_compare] {site} batch:{handover_no} 比对异常: {e}")
+        return None
+
+
+# ============================== 比对核心 ==============================
+
+
+def _do_compare(
+    site: str,
+    label: str,
+    batches: list[str],
+    exp_rows: list[tuple],  # [(waybill_no, handover_no, handover_pieces), ...]
+    act_rows: list[tuple],  # [(waybill_no, piece_no), ...]
+    cfg: SiteCompareConfig,
+) -> CompareResult:
+    """执行逐运单比对，产出统计 + 差缺明细。
+
+    与 compare.py:process() 口径一致：
+      - 应到件数 = handover_pieces（交接件数）
+      - 实到件数 = SF ? COUNT(*) : COUNT(DISTINCT piece_no)
+      - arrived_cnt >= handover_pieces → 足额到货，跳过
+    """
+    # 构建实到索引: waybill_no → [piece_no, ...]（保留所有行，不去重）
+    act_by_wb: dict[str, list[str]] = {}
+    for wb, piece in act_rows:
+        act_by_wb.setdefault(wb, []).append(piece)
+
+    stats = CompareStats()
+    rows: list[UndeliveredRow] = []
+    max_arrived = 0
+
+    for wb, handover_no, handover_pcs in exp_rows:
+        handover_pcs = handover_pcs or 0
+        if handover_pcs <= 0:
+            continue
+
+        stats.waybill_count += 1
+        stats.expected_pieces += handover_pcs
+
+        is_sf = cfg.has_sf and wb.startswith("SF")
+        if is_sf:
+            stats.sf_wb_count += 1
+
+        all_pieces = act_by_wb.get(wb, [])
+
+        if is_sf:
+            # SF: 行计数，不去重（piece_no 是随机号码）
+            arrived_cnt = len(all_pieces)
+            arrived_list = list(all_pieces)
+        else:
+            # non-SF: 子单号去重
+            unique_pieces = list(dict.fromkeys(all_pieces))  # 保序去重
+            arrived_cnt = len(unique_pieces)
+            arrived_list = unique_pieces
+
+        stats.arrived_pieces += arrived_cnt
+
+        if arrived_cnt >= handover_pcs:
+            continue  # 足额或溢到，不进差缺表
+
+        if arrived_cnt == 0:
+            stats.full_miss += 1
+        else:
+            stats.part_miss += 1
+
+        if is_sf:
+            stats.sf_undelivered += 1
+
+        max_arrived = max(max_arrived, arrived_cnt)
+        rows.append(
+            UndeliveredRow(
+                handover_no=handover_no,
+                waybill_no=wb,
+                total_pieces=handover_pcs,
+                arrived_pieces=arrived_cnt,
+                arrived_list=arrived_list,
+                is_sf=is_sf,
+            )
+        )
+
+    stats.undelivered_pieces = max(0, stats.expected_pieces - stats.arrived_pieces)
+    stats.undelivered_wb = stats.full_miss + stats.part_miss
+
+    result = CompareResult(
+        site=site,
+        date=label,
+        batches=batches,
+        stats=stats,
+        rows=rows,
+    )
+
+    # 打印摘要
+    print(
+        f"[db_compare] {site} {label}: "
+        f"batches={len(batches)}, "
+        f"wb={stats.waybill_count}(SF:{stats.sf_wb_count}), "
+        f"exp={stats.expected_pieces}, arr={stats.arrived_pieces}, "
+        f"miss={stats.undelivered_pieces}, "
+        f"miss_wb={stats.undelivered_wb}(full={stats.full_miss}, part={stats.part_miss})"
+    )
+    if stats.sf_undelivered:
+        print(f"  SF 差缺: {stats.sf_undelivered} 个运单")
+
+    return result
+
+
+# ============================== Excel 输出 ==============================
+
+
+# 样式常量（与 compare.py 对齐）
+_FONT = "微软雅黑"
+_BLUE = "305496"
+
+_HEADER_FILL = PatternFill("solid", fgColor=_BLUE)
+_HEADER_FONT = Font(name=_FONT, bold=True, color="FFFFFF", size=11)
+_BODY_FONT = Font(name=_FONT, size=10)
+_THIN = Side(style="thin", color="D9D9D9")
+_BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN)
+
+
+def write_result_excel(result: CompareResult, output_path: str | None = None) -> str:
+    """将比对结果写入 Excel 文件。
+
+    Args:
+        result: compare_site_date 或 compare_site_batch 的返回值
+        output_path: 输出路径，为 None 时自动生成：
+                     output/{站}-{日期}-未到数据.xlsx
+
+    Returns:
+        实际写入的文件路径
+    """
+    if output_path is None:
+        os.makedirs(OUTPUT_DIR, exist_ok=True)
+        date_tag = result.date.replace(":", "-").replace("batch:", "batch-")
+        output_path = os.path.join(
+            OUTPUT_DIR, f"{result.site}-{date_tag}-未到数据.xlsx"
+        )
+
+    wb = Workbook()
+    ws = wb.active
+    ws.title = result.site
+
+    _write_sheet(ws, result)
+    wb.save(output_path)
+    print(f"[db_compare] Excel 已输出: {output_path}")
+    return output_path
+
+
+def _write_sheet(ws, result: CompareResult):
+    """写单个站点的差缺明细 sheet。"""
+    s = result.stats
+    rows = result.rows
+
+    # 动态列: 交接单号 | 运单号 | 总件数 | 已到单号1 | 已到单号2 | ...
+    max_arrived = max((len(r.arrived_list) for r in rows), default=0)
+    columns = ["交接单号", "运单号", "总件数"] + [
+        f"已到单号{i + 1}" for i in range(max_arrived)
+    ]
+
+    ws.sheet_view.showGridLines = False
+
+    # 表头
+    ws.append(columns)
+    for c in range(1, len(columns) + 1):
+        cell = ws.cell(row=1, column=c)
+        cell.fill = _HEADER_FILL
+        cell.font = _HEADER_FONT
+        cell.alignment = Alignment(horizontal="center", vertical="center")
+        cell.border = _BORDER
+
+    # 数据行
+    for row in rows:
+        values = {
+            "交接单号": row.handover_no,
+            "运单号": row.waybill_no,
+            "总件数": row.total_pieces,
+        }
+        for i, piece in enumerate(row.arrived_list):
+            values[f"已到单号{i + 1}"] = piece
+        ws.append([values.get(c, "") for c in columns])
+
+    # 格式
+    for r in range(2, ws.max_row + 1):
+        for c, col in enumerate(columns, start=1):
+            cell = ws.cell(row=r, column=c)
+            cell.font = _BODY_FONT
+            cell.border = _BORDER
+            if col == "总件数":
+                cell.number_format = "#,##0"
+                cell.alignment = Alignment(horizontal="right", vertical="center")
+            else:
+                cell.number_format = "@"
+
+    # 列宽
+    for c, col in enumerate(columns, start=1):
+        body_lens = [
+            len(str(ws.cell(row=r, column=c).value or ""))
+            for r in range(2, ws.max_row + 1)
+        ]
+        width = min(max([len(str(col))] + body_lens) + 4, 36)
+        ws.column_dimensions[ws.cell(row=1, column=c).column_letter].width = max(
+            width, 12
+        )
+
+    ws.freeze_panes = "A2"
+
+
+# ============================== 全站汇总报表（DB 版）=============================
+
+
+def _stats_to_dict(s: CompareStats) -> dict:
+    """CompareStats -> build_summary 要的中文键 stats dict。"""
+    return {
+        "运单数": s.waybill_count,
+        "应到件": s.expected_pieces,
+        "已到件": s.arrived_pieces,
+        "未到件": s.undelivered_pieces,
+        "完全未到": s.full_miss,
+        "部分未到": s.part_miss,
+    }
+
+
+def _target_date_for(site: str) -> str:
+    """4 站比对锚点：today - actual_offset（以实到扫描日为锚，与 _site_undelivered_handler 一致）。"""
+    from inbound_verify import state_store  # 懒导入，避免成环
+
+    offset = state_store.get_offset(site, "actual")
+    return (date.today() - timedelta(days=offset)).strftime("%Y-%m-%d")
+
+
+def _baishi_from_pg(cur, target: str):
+    """查百世当日基数（baishi_daily_stats）+ 当天未到明细（undelivered_record 按 ingested_at 过滤）。
+    返回 (stats_dict_or_None, rows_or_None)；基数与明细均无 → (None, None)。
+
+    undelivered_record 是 UPSERT 累积表；按 ingested_at::date = target 取当天入库的未到快照
+    （= 当天下载的当前未到，站点已剔除已到），避免累积偏大。
+    """
+    cur.execute(
+        "SELECT expected_pieces, arrived_pieces, undelivered_pieces "
+        "FROM baishi_daily_stats WHERE site = %s AND business_date = %s",
+        ("百世", target),
+    )
+    basis = cur.fetchone()
+    cur.execute(
+        "SELECT waybill_no, piece_no, biz_type, last_scan FROM undelivered_record "
+        "WHERE site = %s AND ingested_at::date = %s",
+        ("百世", target),
+    )
+    detail = cur.fetchall()
+    if basis is None and not detail:
+        return (None, None)
+    exp = basis[0] if basis else None
+    arr = basis[1] if basis else None
+    # 未到件优先取基数差（baishi_daily_stats.undelivered_pieces，与应到/已到同源自洽）；
+    # 基数缺失时退回明细行数。
+    undel = basis[2] if (basis and basis[2] is not None) else len(detail)
+    wb_count = len({r[0] for r in detail if r[0]})  # 运单号去重
+    rows = [
+        {
+            "类型": r[2] or "",
+            "子单号": r[1] or "",
+            "运单号": r[0] or "",
+            "最新扫描记录": r[3] or "",
+        }
+        for r in detail
+    ]
+    stats = {
+        "运单数": wb_count,
+        "应到件": exp,
+        "已到件": arr,
+        "未到件": undel,
+        "完全未到": None,
+        "部分未到": None,
+    }
+    return (stats, rows)
+
+
+def build_full_report(date=None) -> str:
+    """DB 版全站汇总报表：4 站走 DB 比对、百世走 PG，复用 compare.build_summary 渲染。
+    产出 output/应到未到数据.xlsx（/report 下载）。date=None 时各站按 actual_offset 算锚点（以实到扫描日为锚）。
+    返回输出路径。"""
+    from inbound_verify import compare  # 复用 build_summary / write_station / OUTFILE
+
+    print("[db_compare] 开始生成全站汇总报表 ...")
+    wb = Workbook()
+    wb.remove(wb.active)
+    summary_ws = wb.create_sheet("汇总报表")
+
+    results = []  # [(name, stats_dict_or_None)]，顺序 ALL_REPORT_SITES
+    site_targets = {}  # name -> target_date（汇总表"数据日期"列）
+
+    conn = _connect()
+    cur = conn.cursor()
+    try:
+        for name in ALL_REPORT_SITES:
+            if name == "百世":
+                target = date or datetime.now().strftime("%Y-%m-%d")
+                site_targets[name] = target
+                stats, rows = _baishi_from_pg(cur, target)
+                results.append((name, stats))
+                if rows is not None:
+                    compare.write_station(wb.create_sheet(name), BAISHI_COLUMNS, rows)
+                continue
+            if name not in SITE_COMPARE_CONFIG:
+                results.append((name, None))
+                continue
+            target = date or _target_date_for(name)
+            site_targets[name] = target
+            result = compare_site_date(name, target)
+            if result is not None:
+                results.append((name, _stats_to_dict(result.stats)))
+                _write_sheet(wb.create_sheet(name), result)
+            else:
+                results.append((name, None))
+    finally:
+        conn.close()
+
+    compare.build_summary(
+        summary_ws,
+        results,
+        datetime.now().strftime("%Y-%m-%d %H:%M"),
+        dates=site_targets,
+    )
+    os.makedirs(OUTPUT_DIR, exist_ok=True)
+    wb.save(compare.OUTFILE)
+    print(f"[db_compare] 全站汇总已输出: {compare.OUTFILE}")
+    for name, s in results:
+        print(f"   {name}：未到 {s['未到件']} 件" if s else f"   {name}：无数据，跳过")
+    return compare.OUTFILE
+
+
+# ============================== 终端验证入口 ==============================
+
+
+def main():
+    """命令行验证入口：
+    python -m inbound_verify.db_compare 顺心 2026-07-25
+    """
+    import sys
+
+    site = sys.argv[1] if len(sys.argv) > 1 else "顺心"
+    target_date = sys.argv[2] if len(sys.argv) > 2 else "2026-07-25"
+
+    result = compare_site_date(site, target_date)
+    if result is None:
+        print(f"{site} {target_date}: 无结果")
+        return
+
+    print(f"\n=== {result.site} {result.date} 差缺明细 ===")
+    print(f"涉及批次: {result.batches}")
+    print(f"应到运单: {result.stats.waybill_count} (SF: {result.stats.sf_wb_count})")
+    print(f"应到件数: {result.stats.expected_pieces}")
+    print(f"实到件数: {result.stats.arrived_pieces}")
+    print(f"未到件数: {result.stats.undelivered_pieces}")
+    print(
+        f"差缺运单: {result.stats.undelivered_wb} (完全未到: {result.stats.full_miss}, 部分未到: {result.stats.part_miss})"
+    )
+    if result.stats.sf_undelivered:
+        print(f"SF 差缺: {result.stats.sf_undelivered}")
+
+    if result.rows:
+        print(f"\n--- 差缺明细 (共 {len(result.rows)} 条) ---")
+        for row in result.rows[:20]:
+            sf = "[SF]" if row.is_sf else ""
+            arrived_preview = row.arrived_list[:5]
+            print(
+                f"  {sf} {row.waybill_no}: "
+                f"应到{row.total_pieces}件, 实到{row.arrived_pieces}件"
+                f"  {f'已到: {arrived_preview}' if arrived_preview else ''}"
+            )
+        if len(result.rows) > 20:
+            print(f"  ... 还有 {len(result.rows) - 20} 条")
+
+    # 输出 Excel
+    path = write_result_excel(result)
+    print(f"\n结果文件: {path}")
+
+
+if __name__ == "__main__":
+    main()

inbound_verify/paths.py

@@ -18,3 +18,6 @@ CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")
 
 # 状态存储（SQLite，阶段0：心跳 / 登录态 / 数据态持久化，重启不丢）
 STATE_DB_PATH = os.path.join(BASE_DIR, "state", "state.db")
+
+# 错误截图目录（下载流程失败时自动截取，供问题排查）
+SCREENSHOT_DIR = os.path.join(BASE_DIR, "logs", "screenshots")

inbound_verify/runtime.py

@@ -16,13 +16,12 @@ import socket
 import subprocess
 import time
 import urllib.request
-from datetime import datetime, timedelta
+from datetime import date, datetime, timedelta
 
 import yaml
 from playwright.sync_api import sync_playwright
 
-from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
+from inbound_verify.paths import CONFIG_PATH, SCREENSHOT_DIR
-from inbound_verify.domain import SITE_UNDELIVERED_FILE
 
 from inbound_verify import state_store
 from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng
@@ -50,31 +49,6 @@ APP_SITES = {"安能"}
 # 心跳间隔（秒）
 HEARTBEAT_INTERVAL = 30
 
-# 各站最终数据文件名（探测"数据是否已跑出来"）；百世为单流程
-DATA_FILENAMES = {
-    "顺心": {
-        "expected": "顺心-应到货物数据.xlsx",
-        "actual": "顺心-实到货物数据.xlsx",
-        "undelivered": "顺心-未到数据.xlsx",
-    },
-    "中通": {
-        "expected": "中通-应到货物数据.xlsx",
-        "actual": "中通-实到货物数据.xlsx",
-        "undelivered": "中通-未到数据.xlsx",
-    },
-    "韵达": {
-        "expected": "韵达-应到货物数据.xlsx",
-        "actual": "韵达-实到货物数据.xlsx",
-        "undelivered": "韵达-未到数据.xlsx",
-    },
-    "安能": {
-        "expected": "安能-应到货物数据.xlsx",
-        "actual": "安能-实到货物数据.xlsx",
-        "undelivered": "安能-未到数据.xlsx",
-    },
-    "百世": {"expected": "", "actual": "", "undelivered": "百世-应到未到货物数据.xlsx"},
-}
-
 
 # ============================ 安能启动（CDP）============================
 
@@ -102,13 +76,37 @@ def _wait_cdp_up(port, timeout=60.0):
     return False
 
 
+# 【环境兼容】宿主 shell（Codex/VS Code 插件、WorkBuddy 等）会向子进程注入一批与业务
+# 无关的变量，实测会让安能应用登录后反复弹出“获取试用网点接口报错”：
+#   - HTTP(S)_PROXY=http://127.0.0.1:8800（QuickQ 加速器代理）：安能的 wnp.ane56.com
+#     接口请求被塞进第三方代理后返回 400/用户未登录；
+#   - VSCODE_* / CODEX_* / EFC_*：VS Code 扩展宿主注入（IPC、PID、NLS、ESM 等）；
+#   - NODE_TLS_REJECT_UNAUTHORIZED / DEBUG / RUST_LOG 等宿主调试变量。
+# 另：ELECTRON_RUN_AS_NODE=1 会把安能当作纯 Node 运行（拒绝 Chromium 参数、启动即
+# 退出 rc=9）；NODE_OPTIONS 同样会干扰。拉起前全部摘掉，尽量还原终端手动启动环境。
+_ANNENG_STRIP_PREFIXES = ("VSCODE_", "CODEX_", "EFC_")
+_ANNENG_STRIP_EXACT = {
+    "NODE_OPTIONS",
+    "ELECTRON_RUN_AS_NODE",
+    "HTTP_PROXY",
+    "HTTPS_PROXY",
+    "ALL_PROXY",
+    "NO_PROXY",
+    "NODE_TLS_REJECT_UNAUTHORIZED",
+    "NODEFAULTCURRENTDIRECTORYINEXEPATH",
+    "DEBUG",
+    "RUST_LOG",
+    "APPLICATION_INSIGHTS_NO_STATSBEAT",
+}
+
+
 def launch_anneng(app_path):
     """以调试模式启动安能 Electron 应用（自动选取空闲端口），返回子进程对象。"""
-    # 【环境兼容】WorkBuddy 等 shell 会注入 NODE_OPTIONS（含 --use-system-ca），
+    anneng_env = {
-    # Electron 内置 Node 拒绝该 flag 导致安能启动即退出（rc=9）。
+        key: value
-    # 拉起前从子进程环境里摘掉 NODE_OPTIONS。
+        for key, value in os.environ.items()
-    anneng_env = os.environ.copy()
+        if key not in _ANNENG_STRIP_EXACT and not key.startswith(_ANNENG_STRIP_PREFIXES)
-    anneng_env.pop("NODE_OPTIONS", None)
+    }
     port = _find_free_port()
     print(f">> 以调试模式启动【安能】应用（端口 {port}）：{app_path}")
     proc = subprocess.Popen(
@@ -150,19 +148,6 @@ def probe_site_login(site_name, pages_map):
         return False
 
 
-def probe_data_file(site_name, kind):
-    """探测单站应到/实到数据文件是否存在且为今天。返回 (is_today, mtime_str)。"""
-    fname = DATA_FILENAMES.get(site_name, {}).get(kind, "")
-    if not fname:
-        return (False, "")
-    path = os.path.join(DOWNLOAD_DIR, fname)
-    if not os.path.exists(path):
-        return (False, "")
-    dt = datetime.fromtimestamp(os.path.getmtime(path))
-    is_today = dt.date() == datetime.now().date()
-    return (is_today, dt.strftime("%Y-%m-%d %H:%M:%S"))
-
-
 # ============================ 运行上下文 ============================
 
 
@@ -476,28 +461,75 @@ def _web_handler(site, download_func):
 
 
 def _site_undelivered_handler(site):
-    """4 站未到：下应到+实到 → 比对写 downloads/<站>-未到数据.xlsx。
+    """4 站未到：下应到+实到 → DB 比对 → 写 output/<站>-<日期>-未到数据.xlsx。
-    任一下载失败 → 清掉旧未到文件、返回 False（前端不展示陈旧未到）。"""
+    应到全量去重（已落库则跳过导出），因此比对不依赖 Excel 文件，走数据库查询。
+    下载成功则返回 True（比对失败不影响任务判定，数据已入库）。"""
 
     def handler(ctx, force=False, date=None):
-        # 各站下载入口约定返回 True/False；顺心历史返回 None（视为成功，与 dispatch 一致）
         exp_ok = TASK_HANDLERS[(site, "expected")](ctx, force, date) is not False
         act_ok = (
             (TASK_HANDLERS[(site, "actual")](ctx, force, date) is not False)
             if exp_ok
             else False
         )
-        if exp_ok and act_ok:
+        if not exp_ok or not act_ok:
-            return compare.write_site_file(site)
+            return False
-        stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site))
+
-        if os.path.exists(stale):
+        # ── 先入库再比对（修复时序：比对须读到本次下载的数据，
-            os.remove(stale)
+        #     否则首次/force 时 PG 无当天数据，比对返回 None、不产出 Excel）──
-        return False
+        try:
+            _record_business_date(site, "undelivered", date)
+        except Exception:
+            pass
+        try:
+            from inbound_verify import store  # 懒导入，避免成环
+
+            if store.ingest_enabled():
+                store.ingest_task(
+                    site, "undelivered"
+                )  # 4 站 = ingest expected + actual
+                print(f">> [入库] {site} 前置入库完成")
+        except Exception as e:
+            print(f">> [入库] {site} 前置入库失败（不影响比对尝试）: {e}")
+
+        # ── DB 比对（替代旧 Excel 比对）──
+        try:
+            from inbound_verify import db_compare  # 懒导入，避免成环
+
+            if date:
+                target_date = date
+            else:
+                offset = state_store.get_offset(site, "actual")
+                target_date = (datetime.now().date() - timedelta(days=offset)).strftime(
+                    "%Y-%m-%d"
+                )
+
+            result = db_compare.compare_site_date(site, target_date)
+            if result is not None:
+                db_compare.write_result_excel(result)
+            else:
+                print(f">> [未到] {site} {target_date}: 当天无实到数据，跳过比对")
+        except Exception as e:
+            print(f">> [未到] {site} DB 比对异常（不影响下载结果）: {e}")
+
+        return True  # 下载成功即返回 True，比对失败不影响任务判定
 
     return handler
 
 
-# 「跑比对」= 纯离线比对（用 downloads/ 现有文件生成全站汇总；下载交由各站定时/手动）。
+# 「跑比对」= DB 版全站汇总报表（替代旧 compare.main Excel 路径；下载交由各站定时/手动）。
+
+
+def _run_db_full_report(date=None):
+    """生成 DB 版全站汇总报表（output/应到未到数据.xlsx）。
+    懒导入 db_compare，best-effort：失败只告警，返回 True（与旧 lambda 契约一致）。"""
+    try:
+        from inbound_verify import db_compare
+
+        db_compare.build_full_report(date)
+    except Exception as e:
+        print(f">> [跑比对] DB 汇总报表生成失败: {e}")
+    return True
 
 
 TASK_HANDLERS = {
@@ -526,8 +558,8 @@ TASK_HANDLERS = {
         force=force, date=date
     ),
     ("安能", "undelivered"): _site_undelivered_handler("安能"),
-    ("__compare__", "compare"): lambda ctx, force=False, date=None: (
+    ("__compare__", "compare"): lambda ctx, force=False, date=None: _run_db_full_report(
-        compare.main() or True
+        date
     ),
 }
 
@@ -541,11 +573,10 @@ def _record_business_date(site, kind, date=None):
       undelivered：百世直供（恒当天）写 undelivered；4 站未到由 _site_undelivered_handler
         内部连带下了 expected+actual（不经 dispatch，无业务日期写入），故此处一并补写
         expected/actual/undelivered 三列——actual 用 actual 偏移、未到跟随 expected 偏移。
-    顺带置 ready=True，让前端不必等心跳即可反映下载成功；写入失败仅告警、不影响任务判定。"""
+    只写业务日期；ready 语义已移交「入库成功」（_persist_to_db 置位），此处不再碰 ready。"""
     if site == "__compare__":
         return
     today = datetime.now().date()
-    now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
     def _write(k, biz_or_off):
         # biz_or_off: int=偏移（today−off）；str=已确定业务日期（date）
@@ -555,9 +586,7 @@ def _record_business_date(site, kind, date=None):
             else biz_or_off
         )
         try:
-            state_store.set_data_state(
+            state_store.set_business_date(site, k, biz)
-                site, k, ready=True, generated_at=now, business_date=biz
-            )
         except Exception as e:
             print(f">> [状态] 写业务日期失败 {site}/{k}: {e}")
 
@@ -576,6 +605,77 @@ def _record_business_date(site, kind, date=None):
         _write("undelivered", date if date else off("expected"))
 
 
+def _ready_flags(site):
+    """从 PG 业务表派生单站三就绪态（ready = DB 数据真相）。
+
+    expected/actual = PG 中存在对应 target_date（today − offset）的数据；
+    百世 undelivered = baishi_daily_stats 中存在 target_date 的数据；
+    4 站 undelivered = expected_ready ∧ actual_ready（派生）。
+    PG 不可达时返回全 False（降级安全，不阻塞心跳）。
+
+    返回 (flags: {kind: bool}, dates: {kind: target_date_str})。
+    dates 与 flags 同源——ready=True 时 business_date 即该 target_date，
+    彻底消除 ready 与 business_date 不同源导致的日期标签漂移。"""
+    from inbound_verify import store  # 懒导入：避免模块级循环
+
+    today = date.today()
+    today_str = today.isoformat()
+
+    if site == "百世":
+        has_und, _ = store.has_data(site, "undelivered", today_str)
+        return (
+            {"expected": False, "actual": False, "undelivered": has_und},
+            {"undelivered": today_str},
+        )
+
+    exp_off = state_store.get_offset(site, "expected")
+    act_off = state_store.get_offset(site, "actual")
+    exp_date = (today - timedelta(days=exp_off)).isoformat()
+    act_date = (today - timedelta(days=act_off)).isoformat()
+
+    has_exp, _ = store.has_data(site, "expected", exp_date)
+    has_act, _ = store.has_data(site, "actual", act_date)
+    return (
+        {"expected": has_exp, "actual": has_act, "undelivered": has_exp and has_act},
+        {"expected": exp_date, "actual": act_date, "undelivered": exp_date},
+    )
+
+
+def _apply_ready(site, flags, dates=None):
+    """写入单站就绪态 + 业务日期（同源：ready 与 business_date 均据 PG + offset 派生）。
+    ready=True 时同步写入 target_date 作为 business_date，消除不同源导致的日期标签漂移。
+    失败仅告警。"""
+    for k, rdy in flags.items():
+        try:
+            state_store.set_ready(site, k, rdy)
+            if rdy and dates and dates.get(k):
+                state_store.set_business_date(site, k, dates[k])
+        except Exception as e:
+            print(f">> [状态] 置就绪态失败 {site}/{k}: {e}")
+
+
+def capture_error_screenshot(page, site, kind, attempt, error):
+    """流程失败时截取当前页面，保存到 logs/screenshots/。
+    page: Playwright Page 对象（安能传 None 走 CDP 分支，调用方自行处理）。
+    截图失败绝不外抛——只打告警，不干扰任务重试/清场流程。"""
+    try:
+        os.makedirs(SCREENSHOT_DIR, exist_ok=True)
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        err_short = (error or "unknown")[:40].replace("/", "_").replace("\\", "_")
+        fname = f"{site}_{kind}_{ts}_attempt{attempt}_{err_short}.png"
+        path = os.path.join(SCREENSHOT_DIR, fname)
+        page.screenshot(path=path, full_page=False)
+        print(f"📸 【{site}-{kind}】错误截图已保存: {path}")
+    except Exception as se:
+        print(f"📸 【{site}-{kind}】截图失败（不影响任务）: {se}")
+
+
+def _refresh_ready(site):
+    """入库后立即据 PG 派生并写入该站就绪态（省 30s 心跳等待，与心跳同源）。"""
+    flags, dates = _ready_flags(site)
+    _apply_ready(site, flags, dates)
+
+
 def _persist_to_db(site, kind):
     """下载成功后把本次数据入库 PostgreSQL（尽力而为，绝不外抛，不影响任务判定）。
     - __compare__ 无源数据，跳过。
@@ -595,7 +695,16 @@ def _persist_to_db(site, kind):
             print(">> [入库] 已关闭 (auto_ingest=false)，跳过")
             return
         count = store.ingest_task(site, kind)
-        state_store.set_ingest_state(site, kind, ok=True, count=count)
+        # 4 站 undelivered 连带入了 expected+actual：按实际入库的类补记 ingest_state，
+        # 否则心跳派生 ready（expected ∧ actual → undelivered）会读到陈旧值。
+        logged = (
+            ["expected", "actual", "undelivered"]
+            if kind == "undelivered" and site != "百世"
+            else [kind]
+        )
+        for k in logged:
+            state_store.set_ingest_state(site, k, ok=True, count=count)
+        _refresh_ready(site)  # 入库成功 → 立即据 ingest_state 派生就绪态（与心跳同源）
         print(f">> [入库] {site}/{kind} 成功，{count} 条")
     except Exception as e:
         print(f">> [warn] 入库失败 {site}/{kind}: {e}")
@@ -639,8 +748,11 @@ def dispatch_task(ctx, task_spec):
 
 
 def run_heartbeat(ctx):
-    """一轮心跳：探测各站登录态 + 数据文件，写状态库；登录态变化时提示。
+    """一轮心跳：探测各站登录态 + 据 PG 业务表派生数据就绪态；登录态变化时提示。
 
+    ready 直接查询 PG 业务表（expected_record / actual_record / baishi_daily_stats），
+    以「目标业务日期是否有数据」为唯一依据，彻底消除 ingest_state 日期比对带来的每日零点重置。
+    _refresh_ready 在入库瞬间即据 PG 派生（省 30s 等待），心跳同源复核。
     只在 Playwright 所属线程调用。
     """
     prev = state_store.get_all_status()
@@ -651,6 +763,5 @@ def run_heartbeat(ctx):
         now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT
         if prev_login and prev_login not in (now_login, state_store.LOGIN_UNKNOWN):
             print(f"\n   ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}")
-        for kind in ("expected", "actual", "undelivered"):
+        flags, dates = _ready_flags(site_name)
-            ready, gen_at = probe_data_file(site_name, kind)
+        _apply_ready(site_name, flags, dates)
-            state_store.set_data_state(site_name, kind, ready, gen_at)

inbound_verify/sites/anneng.py

@@ -43,10 +43,37 @@ from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
 from inbound_verify import state_store
 
 
-def with_retry(site_name, label, flow, reset, max_attempts=3):
+def _capture_error_screenshot(site, kind, attempt, error):
+    """安能 CDP 错误截图（best-effort；失败仅告警，绝不外抛）。"""
+    try:
+        import base64, os
+        from datetime import datetime
+        from inbound_verify.paths import SCREENSHOT_DIR
+
+        os.makedirs(SCREENSHOT_DIR, exist_ok=True)
+        pages = list_pages()
+        if not pages:
+            return
+        cdp = CDP(pages[0]["webSocketDebuggerUrl"])
+        ts = datetime.now().strftime("%Y%m%d_%H%M%S")
+        err_short = (error or "unknown")[:40].replace("/", "_").replace("\\", "_")
+        fname = f"{site}_{kind}_{ts}_attempt{attempt}_{err_short}.png"
+        path = os.path.join(SCREENSHOT_DIR, fname)
+        result = cdp.call("Page.captureScreenshot", format="png")
+        with open(path, "wb") as f:
+            f.write(base64.b64decode(result["data"]))
+        cdp.close()
+        print(f"📸 【{site}-{kind}】错误截图已保存: {path}")
+    except Exception as se:
+        print(f"📸 【{site}-{kind}】截图失败（不影响任务）: {se}")
+
+
+def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
     """异常兜底：flow 失败 → 重置回初始态 → 重试，最多 max_attempts 次（含首次）。
 
     每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。
+    最后一次失败时（重置前）自动截图保存到 logs/screenshots/，供问题排查。
+    （安能通过 CDP 截图，page 参数忽略；保留为统一签名兼容。）
     flow 为零参可调用；返回 False 视为失败，其余视为成功。
     返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。
     """
@@ -60,6 +87,11 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
             return True
         except Exception as e:
             print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
+            if attempt == max_attempts:
+                try:
+                    _capture_error_screenshot(site_name, label, attempt, str(e))
+                except Exception:
+                    pass
             print(f"   → 重置【{site_name}】到初始态，清理环境 ...")
             try:
                 reset()

inbound_verify/sites/baishi.py

@@ -8,10 +8,11 @@ from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
 from inbound_verify import state_store
 
 
-def with_retry(site_name, label, flow, reset, max_attempts=3):
+def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
     """异常兜底：flow 失败 → 重置回初始态 → 重试，最多 max_attempts 次（含首次）。
 
     每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。
+    最后一次失败时（重置前）自动截图保存到 logs/screenshots/，供问题排查。
     flow 为零参可调用；返回 False 视为失败，其余视为成功。
     返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。
     """
@@ -25,6 +26,13 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
             return True
         except Exception as e:
             print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
+            if attempt == max_attempts and page is not None:
+                try:
+                    from inbound_verify.runtime import capture_error_screenshot
+
+                    capture_error_screenshot(page, site_name, label, attempt, str(e))
+                except Exception:
+                    pass
             print(f"   → 重置【{site_name}】到初始态，清理环境 ...")
             try:
                 reset()
@@ -147,6 +155,7 @@ def baishi_download_undelivered_data(page, force=False, date=None):
         "应到未到",
         lambda: baishi_download_undelivered_data_impl(page),
         lambda: baishi_reset(page),
+        page=page,
     )
 
 
@@ -200,6 +209,11 @@ def baishi_download_undelivered_data_impl(page):
             if _exp_n > 0:
                 state_store.set_setting("百世", "scan_expected_pieces", str(_exp_n))
                 state_store.set_setting("百世", "scan_arrived_pieces", str(_arr_n))
+                from inbound_verify import (
+                    store,
+                )  # 直接落库 PG（一步，不绕 state_store→store）
+
+                store.upsert_baishi_daily_stats(_exp_n, _arr_n)
                 print(f"   ℹ️ 已记录百世应到/实到基数：应扫 {_exp_n} / 已扫 {_arr_n}")
         except Exception as _e:
             # 抓取失败绝不影响未到明细下载主流程

inbound_verify/sites/shunxin.py

@@ -11,10 +11,11 @@ from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
 from inbound_verify import state_store
 
 
-def with_retry(site_name, label, flow, reset, max_attempts=3):
+def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
     """异常兜底：flow 失败 → 重置回初始态 → 重试，最多 max_attempts 次（含首次）。
 
     每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。
+    最后一次失败时（重置前）自动截图保存到 logs/screenshots/，供问题排查。
     flow 为零参可调用；返回 False 视为失败，其余视为成功。
     返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。
     """
@@ -28,6 +29,13 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
             return True
         except Exception as e:
             print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
+            if attempt == max_attempts and page is not None:
+                try:
+                    from inbound_verify.runtime import capture_error_screenshot
+
+                    capture_error_screenshot(page, site_name, label, attempt, str(e))
+                except Exception:
+                    pass
             print(f"   → 重置【{site_name}】到初始态，清理环境 ...")
             try:
                 reset()
@@ -237,6 +245,7 @@ def shunxin_expected_download(pages, foreground=True, force=False, date=None):
                 p, out_tag=t, force=f, date=d
             ),
             lambda p=pg: shunxin_reset(p),
+            page=pg,
         )
         if not ok:
             return False  # 某账号重试耗尽 → 整体失败，不融合（避免部分数据）
@@ -602,6 +611,7 @@ def shunxin_actual_download(pages, foreground=True, force=False, date=None):
                 p, out_tag=t, date=d
             ),
             lambda p=pg: shunxin_reset(p),
+            page=pg,
         )
         if not ok:
             return False

inbound_verify/sites/yunda.py

@@ -11,10 +11,11 @@ from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
 from inbound_verify import state_store
 
 
-def with_retry(site_name, label, flow, reset, max_attempts=3):
+def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
     """异常兜底：flow 失败 → 重置回初始态 → 重试，最多 max_attempts 次（含首次）。
 
     每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。
+    最后一次失败时（重置前）自动截图保存到 logs/screenshots/，供问题排查。
     flow 为零参可调用；返回 False 视为失败，其余视为成功。
     返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。
     """
@@ -28,6 +29,13 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
             return True
         except Exception as e:
             print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
+            if attempt == max_attempts and page is not None:
+                try:
+                    from inbound_verify.runtime import capture_error_screenshot
+
+                    capture_error_screenshot(page, site_name, label, attempt, str(e))
+                except Exception:
+                    pass
             print(f"   → 重置【{site_name}】到初始态，清理环境 ...")
             try:
                 reset()
@@ -235,6 +243,7 @@ def yunda_expected_download(page, force=False, date=None):
         "应到",
         lambda: yunda_expected_download_impl(page, force=force, date=date),
         lambda: yunda_reset(page),
+        page=page,
     )
 
 
@@ -492,6 +501,7 @@ def yunda_actual_download(page, force=False, date=None):
         "实到",
         lambda: yunda_actual_download_impl(page, date=date),
         lambda: yunda_reset(page),
+        page=page,
     )
 
 

inbound_verify/sites/zto.py

@@ -4,17 +4,18 @@ import os
 import re
 import time
 import yaml
-from datetime import datetime
+from datetime import datetime, timedelta
 import pandas as pd
 
 from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH
 from inbound_verify import state_store
 
 
-def with_retry(site_name, label, flow, reset, max_attempts=3):
+def with_retry(site_name, label, flow, reset, max_attempts=3, page=None):
     """异常兜底：flow 失败 → 重置回初始态 → 重试，最多 max_attempts 次（含首次）。
 
     每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。
+    最后一次失败时（重置前）自动截图保存到 logs/screenshots/，供问题排查。
     flow 为零参可调用；返回 False 视为失败，其余视为成功。
     返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。
     """
@@ -28,6 +29,13 @@ def with_retry(site_name, label, flow, reset, max_attempts=3):
             return True
         except Exception as e:
             print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}")
+            if attempt == max_attempts and page is not None:
+                try:
+                    from inbound_verify.runtime import capture_error_screenshot
+
+                    capture_error_screenshot(page, site_name, label, attempt, str(e))
+                except Exception:
+                    pass
             print(f"   → 重置【{site_name}】到初始态，清理环境 ...")
             try:
                 reset()
@@ -93,20 +101,50 @@ def _dom_click(locator):
     )
 
 
+def _zto_compute_target_time(offset):
+    """直接从 Python datetime 计算目标日期的毫秒级时间戳（本地时区零点）。
+
+    不再从 DOM 的 real-today 元素读取 time 属性，避免双月视图下 real-today
+    同时出现在 month1（隐藏 ghost cell）和 month2（可见）导致 .first 取到隐藏元素。
+    """
+    target_date = datetime.now().date() - timedelta(days=offset)
+    target_dt = datetime(target_date.year, target_date.month, target_date.day)
+    return int(target_dt.timestamp() * 1000)
+
+
+def _zto_find_visible_day(frame_locator, target_time):
+    """在双月日期控件中查找可见的日期格子。
+
+    jQuery Date Range Picker 双月视图下，同一天可能出现在两个面板中：
+    - month1（左面板）的溢出 ghost cell：display:none，不可见
+    - month2（右面板）的正常 cell：可见
+
+    同一日期在 DOM 中可能有毫秒级差异（零点 vs 23:59:59），遍历匹配并返回
+    第一个 visible 的；无可见匹配返回 None。
+    """
+    # 尝试两个时间变体：零点 和 23:59:59（部分 checked/selected 格用后者）
+    for time_variant in (target_time, target_time + 86399000):
+        sel = f"td div.day[time='{time_variant}']"
+        cells = frame_locator.locator(sel)
+        count = cells.count()
+        for i in range(count):
+            if cells.nth(i).is_visible():
+                return cells.nth(i)
+    return None
+
+
 def _zto_flip_to_target_month(frame_locator, page, target_time, max_flips=12):
     """中通日历(jQuery-Date-Range-Picker 双月视图)跨月导航：目标日期不在当前视窗时，
-    循环点 .prev 把目标月翻进 month1 视窗。offset 恒指向过去，故只往前翻；步长 1 月/次。
+    循环点 .prev 把目标月翻进视窗。用 _zto_find_visible_day 判可见（跳过隐藏 ghost cell）。
-    用 JS 派发点击(.evaluate("el=>el.click()"))避开 .date-range-length-tip 等 hover 遮挡。
     返回 True 若目标格子最终可见。"""
-    sel = f"td div.day[time='{target_time}']"
     for _ in range(max_flips):
-        if frame_locator.locator(sel).first.is_visible():
+        if _zto_find_visible_day(frame_locator, target_time) is not None:
             return True
         frame_locator.locator(".date-picker-wrapper .prev").first.evaluate(
             "el => el.click()"
         )
         page.wait_for_timeout(450)
-    return frame_locator.locator(sel).first.is_visible()
+    return _zto_find_visible_day(frame_locator, target_time) is not None
 
 
 def zto_smart_menu_click(page, menu_path):
@@ -133,6 +171,7 @@ def zto_expected_download(page, force=False, date=None):
         "应到",
         lambda: zto_expected_download_impl(page, force=force, date=date),
         lambda: zto_reset(page),
+        page=page,
     )
 
 
@@ -170,32 +209,25 @@ def zto_expected_download_impl(page, force=False, date=None):
         ewb_frame.locator("#beginDate").click()
         page.wait_for_timeout(500)
 
-        today_cell = ewb_frame.locator("td div.day.real-today").first
+        # 直接从 Python datetime 计算目标时间戳，不再依赖 DOM real-today（双月视图
-        today_cell.wait_for(state="visible")
+        # 下 real-today 可能同时出现在 month1 隐藏 ghost cell 和 month2 可见 cell，
+        # .first 会取到隐藏的那个导致 wait_for(visible) 超时）。
+        target_time = _zto_compute_target_time(offset)
+        target_cell = _zto_find_visible_day(ewb_frame, target_time)
 
-        today_time_str = today_cell.get_attribute("time")
+        if target_cell is None:
-        if today_time_str:
+            print("   ℹ️ 目标日期不在当前视窗，正在翻月导航 ...")
-            today_time = int(today_time_str)
+            if not _zto_flip_to_target_month(ewb_frame, page, target_time):
-            target_time = today_time - offset * 86400000
+                raise RuntimeError(f"翻月后仍无法定位目标日期格子(time={target_time})")
-            target_cell = ewb_frame.locator(f"td div.day[time='{target_time}']").first
+            target_cell = _zto_find_visible_day(ewb_frame, target_time)
+            if target_cell is None:
+                raise RuntimeError(f"翻月后仍无法定位目标日期格子(time={target_time})")
 
-            # 偏移日期跨月时目标格子不在当前双月视窗 → 向前翻月把它带进视窗，不再降级为当天
+        # 日期格子用 _dom_click 直接派发事件：.click() 会先 hover 格子，触发
-            if not target_cell.is_visible():
+        # "范围长度"提示气泡(.date-range-length-tip)盖住格子导致点击被遮挡超时。
-                print("   ℹ️ 偏移日期跨月，正在向前翻月导航到目标 ...")
+        _dom_click(target_cell)
-                if not _zto_flip_to_target_month(ewb_frame, page, target_time):
+        page.wait_for_timeout(300)
-                    raise RuntimeError(
+        _dom_click(target_cell)
-                        f"翻月后仍无法定位目标日期格子(time={target_time})"
-                    )
-            # 日期格子用 _dom_click 直接派发事件（同实到）：.click() 会先 hover 格子，触发
-            # “范围长度”提示气泡(.date-range-length-tip)盖住格子导致点击被遮挡超时，
-            # 跨月选中非今日格子时尤为明显。
-            _dom_click(target_cell)
-            page.wait_for_timeout(300)
-            _dom_click(target_cell)
-        else:
-            today_cell.click()
-            page.wait_for_timeout(300)
-            today_cell.click()
 
         page.wait_for_timeout(500)
 
@@ -386,6 +418,7 @@ def zto_actual_download(page, force=False, date=None):
         "实到",
         lambda: zto_actual_download_impl(page, date=date),
         lambda: zto_reset(page),
+        page=page,
     )
 
 
@@ -422,31 +455,25 @@ def zto_actual_download_impl(page, date=None):
         arr_frame.locator("#daterange").click()
         page.wait_for_timeout(500)
 
-        today_cell = arr_frame.locator("td div.day.real-today").first
+        # 直接从 Python datetime 计算目标时间戳，不再依赖 DOM real-today（双月视图
-        today_cell.wait_for(state="visible")
+        # 下 real-today 可能同时出现在 month1 隐藏 ghost cell 和 month2 可见 cell，
+        # .first 会取到隐藏的那个导致 wait_for(visible) 超时）。
+        target_time = _zto_compute_target_time(offset)
+        target_cell = _zto_find_visible_day(arr_frame, target_time)
 
-        today_time_str = today_cell.get_attribute("time")
+        if target_cell is None:
-        if today_time_str:
+            print("   ℹ️ 目标日期不在当前视窗，正在翻月导航 ...")
-            today_time = int(today_time_str)
+            if not _zto_flip_to_target_month(arr_frame, page, target_time):
-            target_time = today_time - offset * 86400000
+                raise RuntimeError(f"翻月后仍无法定位目标日期格子(time={target_time})")
-            target_cell = arr_frame.locator(f"td div.day[time='{target_time}']").first
+            target_cell = _zto_find_visible_day(arr_frame, target_time)
+            if target_cell is None:
+                raise RuntimeError(f"翻月后仍无法定位目标日期格子(time={target_time})")
 
-            # 日期格子用 _dom_click 直接派发事件：Playwright 的 .click() 会先 hover 格子，
+        # 日期格子用 _dom_click 直接派发事件：Playwright 的 .click() 会先 hover 格子，
-            # 触发“范围长度”提示气泡(.date-range-length-tip)盖住格子，导致点击被判遮挡而超时。
+        # 触发"范围长度"提示气泡(.date-range-length-tip)盖住格子，导致点击被判遮挡而超时。
-            # 偏移日期跨月时目标格子不在当前双月视窗 → 向前翻月把它带进视窗，不再降级为当天。
+        _dom_click(target_cell)
-            if not target_cell.is_visible():
+        page.wait_for_timeout(300)
-                print("   ℹ️ 偏移日期跨月，正在向前翻月导航到目标 ...")
+        _dom_click(target_cell)
-                if not _zto_flip_to_target_month(arr_frame, page, target_time):
-                    raise RuntimeError(
-                        f"翻月后仍无法定位目标日期格子(time={target_time})"
-                    )
-            _dom_click(target_cell)
-            page.wait_for_timeout(300)
-            _dom_click(target_cell)
-        else:
-            _dom_click(today_cell)
-            page.wait_for_timeout(300)
-            _dom_click(today_cell)
 
         page.wait_for_timeout(500)
 
@@ -695,7 +722,7 @@ def _zto_poll_and_download_tasks(page, export_times, download_dir, final_filenam
         )
 
     # ====================================================================
-    # 所有目标文件下载完成后，关闭“导出任务管理”标签页
+    # 所有目标文件下载完成后，关闭"导出任务管理"标签页
     # ====================================================================
     print(">> 【导出任务管理】下载完成，正在关闭标签页...")
     try:

inbound_verify/state_store.py

@@ -7,7 +7,7 @@
 
 import os
 import sqlite3
-from datetime import datetime
+from datetime import datetime, timedelta
 
 from inbound_verify.paths import STATE_DB_PATH
 
@@ -70,9 +70,22 @@ def init_db():
                 status      TEXT,
                 started_at  TEXT,
                 finished_at TEXT,
-                error       TEXT
+                error       TEXT,
+                trigger     TEXT NOT NULL DEFAULT '',
+                target_date TEXT NOT NULL DEFAULT '',
+                force       INTEGER NOT NULL DEFAULT 0
             )
             """)
+        # 旧库迁移：补触发方式/目标日期/强制重下三列（新库已含；重复添加抛 OperationalError，忽略）
+        for _col, _typedef in [
+            ("trigger", "TEXT NOT NULL DEFAULT ''"),
+            ("target_date", "TEXT NOT NULL DEFAULT ''"),
+            ("force", "INTEGER NOT NULL DEFAULT 0"),
+        ]:
+            try:
+                conn.execute(f"ALTER TABLE task_history ADD COLUMN {_col} {_typedef}")
+            except sqlite3.OperationalError:
+                pass
         conn.execute("""
             CREATE TABLE IF NOT EXISTS site_config (
                 site             TEXT PRIMARY KEY,
@@ -242,6 +255,25 @@ def set_data_state(site, kind, ready, generated_at, business_date=None):
         _upsert(conn, site, **fields)
 
 
+def set_business_date(site, kind, business_date):
+    """仅写业务日期快照（不碰 ready/generated_at）。
+
+    下载成功钩子用：ready 语义已移交「入库成功」（见 reset_data_ready / _persist_to_db），
+    下载阶段只记业务日期，供前端状态盘显示「是哪天的数据」。
+    """
+    with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
+        _upsert(conn, site, **{f"{kind}_business_date": business_date or ""})
+
+
+def set_ready(site, kind, ready):
+    """仅写就绪态（不碰 business_date/generated_at）。
+
+    供心跳从 ingest_state 派生 ready 用——ready 现为 DB 入库真相的派生视图，
+    非启动重置、不读 Excel。"""
+    with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
+        _upsert(conn, site, **{f"{kind}_ready": 1 if ready else 0})
+
+
 def get_all_status():
     """返回 {site: {各字段}}；库不存在则返回 {}。"""
     if not os.path.exists(STATE_DB_PATH):
@@ -338,6 +370,27 @@ def set_offset(site, kind, offset):
     return offset
 
 
+def resolve_target_date(site, kind, date=None):
+    """计算一条任务的目标下载日期（YYYY-MM-DD，供任务日志展示 / 重试回放）。
+    有 date 用 date；否则按站点偏移推算（与 runtime._record_business_date 同源）：
+      expected → 应到偏移；actual → 实到偏移；百世 undelivered → 当天；
+      4 站 undelivered → 跟随应到偏移。__compare__ 无数据概念，返回 ''。"""
+    if site == "__compare__":
+        return ""
+    if date:
+        return date
+    today = datetime.now().date()
+    if kind == "expected":
+        return (today - timedelta(days=get_offset(site, "expected"))).strftime(
+            "%Y-%m-%d"
+        )
+    if kind == "actual":
+        return (today - timedelta(days=get_offset(site, "actual"))).strftime("%Y-%m-%d")
+    if site == "百世":
+        return today.strftime("%Y-%m-%d")
+    return (today - timedelta(days=get_offset(site, "expected"))).strftime("%Y-%m-%d")
+
+
 def set_schedule(site, enabled, time_str):
     """【DEPRECATED】旧"每日单时点定时"——已被 fetch_schedule 的周期+激活时段模式取代。
     保留死代码以防外部残留调用；新代码请用 set_fetch_schedule。"""
@@ -520,26 +573,28 @@ def get_site_settings(site):
 # ============================ 任务历史 ============================
 
 
-def create_task(site, kind):
+def create_task(site, kind, trigger="manual", target_date="", force=False):
-    """新建一条 pending 任务，返回其 id。"""
+    """新建一条 pending 任务（手动触发），返回其 id。trigger='manual'/'auto'；
+    target_date 为该任务的目标下载日期（YYYY-MM-DD，可为 ''）；force 是否强制重下。"""
     now = _now()
     with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
         cur = conn.execute(
-            "INSERT INTO task_history (site, kind, status, started_at, finished_at, error) "
+            "INSERT INTO task_history "
-            "VALUES (?, ?, ?, ?, '', '')",
+            "(site, kind, status, started_at, finished_at, error, trigger, target_date, force) "
-            (site, kind, TASK_PENDING, now),
+            "VALUES (?, ?, ?, ?, '', '', ?, ?, ?)",
+            (site, kind, TASK_PENDING, now, trigger, target_date, 1 if force else 0),
         )
         conn.commit()
         return cur.lastrowid
 
 
-def create_task_if_idle(site, kind):
+def create_task_if_idle(site, kind, trigger="auto", target_date=""):
     """周期调度专用：若该 (site,kind) 已有 pending/running 任务则返回 None（跳过本次周期），
     否则建一条 pending 任务返回其 id。单连接内 check-then-insert，靠 SQLite 写锁把竞态压到忽略不计。
 
     与 create_task 的区别：手动触发(POST /tasks)用 create_task（用户点的必建）；周期 job 用本函数
     ——上一次还没跑完时跳过，避免同 (site,kind) 任务堆积。手动建的任务会让紧随其后的周期 fire
-    判到 inflight 而跳过，天然互斥。"""
+    判到 inflight 而跳过，天然互斥。trigger='auto'；target_date 为目标下载日期（YYYY-MM-DD）。"""
     now = _now()
     with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
         row = conn.execute(
@@ -550,9 +605,10 @@ def create_task_if_idle(site, kind):
         if row:
             return None
         cur = conn.execute(
-            "INSERT INTO task_history (site, kind, status, started_at, finished_at, error) "
+            "INSERT INTO task_history "
-            "VALUES (?, ?, ?, ?, '', '')",
+            "(site, kind, status, started_at, finished_at, error, trigger, target_date, force) "
-            (site, kind, TASK_PENDING, now),
+            "VALUES (?, ?, ?, ?, '', '', ?, ?, 0)",
+            (site, kind, TASK_PENDING, now, trigger, target_date),
         )
         conn.commit()
         return cur.lastrowid
@@ -579,7 +635,8 @@ def get_task(task_id):
     """返回单条任务 dict，不存在返回 None。"""
     with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
         row = conn.execute(
-            "SELECT id, site, kind, status, started_at, finished_at, error "
+            "SELECT id, site, kind, status, started_at, finished_at, error, "
+            "trigger, target_date, force "
             "FROM task_history WHERE id=?",
             (task_id,),
         ).fetchone()
@@ -593,6 +650,9 @@ def get_task(task_id):
         "started_at": row[4],
         "finished_at": row[5],
         "error": row[6],
+        "trigger": row[7],
+        "target_date": row[8],
+        "force": bool(row[9]),
     }
 
 
@@ -600,7 +660,8 @@ def list_tasks(limit=20):
     """返回最近 limit 条任务（按 id 倒序）。"""
     with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn:
         rows = conn.execute(
-            "SELECT id, site, kind, status, started_at, finished_at, error "
+            "SELECT id, site, kind, status, started_at, finished_at, error, "
+            "trigger, target_date, force "
             "FROM task_history ORDER BY id DESC LIMIT ?",
             (limit,),
         ).fetchall()
@@ -613,6 +674,9 @@ def list_tasks(limit=20):
             "started_at": r[4],
             "finished_at": r[5],
             "error": r[6],
+            "trigger": r[7],
+            "target_date": r[8],
+            "force": bool(r[9]),
         }
         for r in rows
     ]

inbound_verify/store.py

@@ -268,6 +268,18 @@ _SQL_UNDELIVERED = """
         ingested_at = now()
 """
 
+_SQL_BAISHI_DAILY_STATS = """
+    INSERT INTO baishi_daily_stats
+        (site, business_date, expected_pieces, arrived_pieces, undelivered_pieces, raw)
+    VALUES (%s,%s,%s,%s,%s,%s)
+    ON CONFLICT (site, business_date) DO UPDATE SET
+        expected_pieces    = COALESCE(EXCLUDED.expected_pieces, baishi_daily_stats.expected_pieces),
+        arrived_pieces     = COALESCE(EXCLUDED.arrived_pieces, baishi_daily_stats.arrived_pieces),
+        undelivered_pieces = COALESCE(EXCLUDED.undelivered_pieces, baishi_daily_stats.undelivered_pieces),
+        raw                = EXCLUDED.raw,
+        ingested_at        = now()
+"""
+
 
 # ============================== 入库 ==============================
 
@@ -314,9 +326,10 @@ def _ingest_actual(cur, site):
     cm = ACTUAL_COLMAP[site]
     df = pd.read_excel(path, dtype=str).fillna("")
     if site == "韵达":
-        # 韵达业务清洗：抛弃「交接单号」为空的行（派件/签收等其他扫描无交接单号），
+        # 韵达业务清洗：保留「交接单号」为空的行（到/接件扫描），
+        # 抛弃「交接单号」不为空的行（派件/签收等，属重复数据）。
         # 再按子单号去重（一件多扫只留一条；清洗后子单号已天然唯一，drop 为保险）。
-        df = df[df["交接单号"].astype(str).str.strip() != ""]
+        df = df[df["交接单号"].astype(str).str.strip() == ""]
         df = df.drop_duplicates(subset=[cm["piece"]], keep="last")
     rows = []
     for r in df.to_dict("records"):
@@ -368,6 +381,34 @@ def _ingest_undelivered_baishi(cur):
     return len(rows)
 
 
+def upsert_baishi_daily_stats(exp, arr, business_date=None):
+    """直接落库百世当日应到/实到基数（应扫/已扫，站级日聚合）。
+    供 baishi 下载时抓到基数后直接调用（一步落库，不绕 state_store→store）。
+    business_date 默认今天（百世固定当天）。best-effort：失败只告警，不影响下载流程。"""
+    biz = business_date or date.today()
+    if exp is None and arr is None:
+        return
+    undel = (exp - arr) if (exp is not None and arr is not None) else None
+    try:
+        with _connect(_load_pg_config()["dbname"]) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    _SQL_BAISHI_DAILY_STATS,
+                    (
+                        "百世",
+                        biz,
+                        exp,
+                        arr,
+                        undel,
+                        Jsonb({"expected": exp, "arrived": arr, "undelivered": undel}),
+                    ),
+                )
+            conn.commit()
+        print(f"   [基数] 百世 {biz}: 应扫 {exp} / 已扫 {arr} / 未扫 {undel}")
+    except Exception as e:
+        print(f"   [基数] 百世 {biz} 入库失败（不影响下载）: {e}")
+
+
 def ingest(site=None):
     """入库：指定 site 则单站（百世只入未到），否则全站。返回总条数。"""
     dates = _read_business_dates()
@@ -441,6 +482,52 @@ def get_existing_handover_nos(site):
         return set()
 
 
+# ============================== PG 数据存在性查询 ==============================
+
+
+def has_data(site, kind, target_date):
+    """查询 PG：指定站点在 target_date 是否有业务数据。
+    target_date: str 'YYYY-MM-DD' 或 date 对象。
+    返回 (has_rows: bool, count: int)。
+    PG 不可达时返回 (False, 0)，不抛异常——调用方按「未确认存在」处理。
+
+    kind 路由：
+      expected     → expected_record (business_date)
+      actual       → actual_record (scan_time::date)
+      undelivered  → 百世: baishi_daily_stats；4 站: 不单独查（由调用方 expected∧actual 派生）
+    """
+    if site == "百世" and kind == "undelivered":
+        sql = (
+            "SELECT COUNT(*) FROM baishi_daily_stats"
+            " WHERE site = %s AND business_date = %s"
+        )
+        params = (site, target_date)
+    elif kind == "expected":
+        sql = (
+            "SELECT COUNT(*) FROM expected_record"
+            " WHERE site = %s AND business_date = %s"
+        )
+        params = (site, target_date)
+    elif kind == "actual":
+        sql = (
+            "SELECT COUNT(*) FROM actual_record"
+            " WHERE site = %s AND scan_time::date = %s"
+        )
+        params = (site, target_date)
+    else:
+        return (False, 0)
+    try:
+        with _connect(_load_pg_config()["dbname"]) as conn:
+            with conn.cursor() as cur:
+                cur.execute(sql, params)
+                row = cur.fetchone()
+                cnt = int(row[0]) if row else 0
+                return (cnt > 0, cnt)
+    except Exception as e:
+        print(f">> [状态] PG 查询 {site}/{kind}/{target_date} 失败: {e}")
+        return (False, 0)
+
+
 # ============================== 命令行 ==============================
 
 

schema.sql

@@ -53,3 +53,16 @@ CREATE TABLE IF NOT EXISTS undelivered_record (
     ingested_at  TIMESTAMPTZ  NOT NULL DEFAULT now(),
     UNIQUE (site, piece_no)
 );
+
+-- 百世日聚合（应扫/已扫基数：站级日聚合，区别于运单级/件级/子单级表）
+CREATE TABLE IF NOT EXISTS baishi_daily_stats (
+    id                 BIGSERIAL    PRIMARY KEY,
+    site               TEXT         NOT NULL,   -- 百世
+    business_date      DATE         NOT NULL,   -- 业务日期（百世固定当天）
+    expected_pieces    INTEGER,                  -- 应扫（应到基数）
+    arrived_pieces     INTEGER,                  -- 已扫（实到基数）
+    undelivered_pieces INTEGER,                  -- 未扫（=应扫-已扫，任一缺失则 NULL）
+    raw                JSONB        NOT NULL,    -- 原始抓取值
+    ingested_at        TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    UNIQUE (site, business_date)
+);