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

顺心 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）

# 核心函数签名

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:
    """按指定交接单号执行全批次比对（不依赖实到锚点）。"""

数据类型：

@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 端点

@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）

# 终端直接调
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）

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，确认与服务器时区一致