feat(db_compare): add PostgreSQL-based comparison engine with SF handling

Replace Excel-based undelivered comparison with DB queries for all four
sites. The engine anchors on actual scan_time, reverse-lookups handover
batches, and compares expected vs actual waybill-by-waybill.

Shunxin SF waybills: use COUNT(*) instead of COUNT(DISTINCT piece_no)
since SF piece numbers are random and not derivable from the waybill.

Changes:
- db_compare.py: new module with compare_site_date(), compare_site_batch(),
  write_result_excel(), and POST /compare API endpoint
- runtime.py: switch _site_undelivered_handler from compare.write_site_file
  (Excel) to db_compare (DB); downloads succeed independently of comparison
- server.py: add POST /compare endpoint with date validation
- docs: implementation plan for Shunxin DB comparison

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

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

@@ -36,6 +36,7 @@ from inbound_verify.runtime import (
     launch_and_prepare,
     run_heartbeat,
 )
+from inbound_verify import db_compare
 
 # 全部站点；百世固定下载当天，不可配置偏移
 ALL_SITES = ["顺心", "百世", "中通", "韵达", "安能"]
@@ -245,6 +246,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 就绪状态。"""

inbound_verify/db_compare.py

@@ -0,0 +1,534 @@
+# -*- 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
+
+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
+
+# ============================== 结果类型 ==============================
+
+
+@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"
+
+
+# ============================== 终端验证入口 ==============================
+
+
+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/runtime.py

@@ -476,24 +476,45 @@ 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)
             stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site))
             if os.path.exists(stale):
                 os.remove(stale)
             return False
 
+        # ── DB 比对（替代旧 Excel 比对）──
+        try:
+            from inbound_verify import db_compare  # 懒导入，避免成环
+
+            if date:
+                target_date = date
+            else:
+                offset = state_store.get_offset(site, "expected")
+                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