diff --git a/docs/2026-08-02-未到差缺统计逻辑现状梳理.md b/docs/2026-08-02-未到差缺统计逻辑现状梳理.md new file mode 100644 index 0000000..d7da464 --- /dev/null +++ b/docs/2026-08-02-未到差缺统计逻辑现状梳理.md @@ -0,0 +1,232 @@ +# 未到（差缺）统计逻辑现状梳理 + +> 梳理日期：2026-08-02 +> 分支：`refactor/undelivered-by-expected` +> 范围：顺心、中通、韵达、安能四站的"应到 / 实到 / 差缺"统计链路（百世为站点直供未到，单独处理） +> 目的：固定当前真实逻辑，为"改为基于应到数据统计差缺"的重构提供基线 + +--- + +## 一、系统总览与数据流 + +系统分三层： + +1. **下载层**（`inbound_verify/sites/*.py`）：用 Playwright / CDP 登录各物流站点，按目标日期导出「应到货物数据」「实到货物数据」Excel 到 `downloads/`。 +2. **持久化层**（`store.py` + PostgreSQL）：把 Excel 幂等 UPSERT 进 `expected_record`（应到，运单级）/ `actual_record`（实到，扫描件级）/ `undelivered_record`（百世直供未到，子单级）/ `baishi_daily_stats`（百世日聚合基数）。 +3. **统计层**（`db_compare.py`）：直接查 PostgreSQL，以**实到扫描日期为锚点**反推交接批次，展开批次全量应到后逐运单比对，产出差缺统计与明细，落盘 `output/` 下 Excel，供前端下载。 + +```mermaid +flowchart LR + subgraph 下载层["下载层 sites/*.py"] + A1[定时任务
fetch_schedule] --> D[Playwright/CDP 登录站点] + A2[手动任务
POST /tasks] --> D + D --> E1[应到货物数据.xlsx] + D --> E2[实到货物数据.xlsx] + end + + subgraph 持久化层["持久化层 store.py + PG"] + E1 --> F[ingest_task
expected_record] + E2 --> G[ingest_task
actual_record] + end + + subgraph 统计层["统计层 db_compare.py"] + F --> H[compare_site_date
实到锚点→反推批次→展开应到] + G --> H + H --> I[output/*-未到数据.xlsx] + H --> J[output/应到未到数据.xlsx
全站汇总] + end + + I --> K[前端下载] + J --> K +``` + +--- + +## 二、数据模型 + +### 2.1 PostgreSQL 业务表（schema.sql） + +| 表 | 粒度 | 业务唯一键 | 关键列 | +|----|------|-----------|--------| +| `expected_record` | 运单级，一运单一行 | `(site, waybill_no)` | `handover_no`、`handover_pieces`（交接件数=应到口径）、`order_pieces`（录单件数）、`business_date`（属性，非唯一键）、`raw` JSONB | +| `actual_record` | 扫描件级，一扫描一行 | `(site, piece_no)` | `waybill_no`（运单基号）、`piece_no`（扫描/子单号）、`scan_time`、`scan_site` | +| `undelivered_record` | 百世直供未到明细，子单级 | `(site, piece_no)` | `biz_type`、`last_scan` | +| `baishi_daily_stats` | 百世站级日聚合 | `(site, business_date)` | `expected_pieces`（应扫）、`arrived_pieces`（已扫）、`undelivered_pieces`（未扫） | + +> 索引：`expected_record` 有 `(site, business_date)` 与 `(site, handover_no)` 索引；`actual_record` 有 `(site, waybill_no)` 与 `(scan_time)` 索引——差缺比对的主查询路径均命中。 + +### 2.2 SQLite 状态库（state.db） + +- `site_config`：各站 `expected_offset` / `actual_offset`（0=今天，最大回溯 30 天）。 +- `fetch_schedule`：周期抓取开关、激活时段、间隔（分钟）。 +- `task_history`：任务记录（`trigger`、`target_date`、`force`）。 +- `ingest_state`：最近一次入库健康状态（`ok` / `count` / `error`）。 +- `site_status`：登录态 + 各 kind 的 `*_ready` 与 `*_business_date`（由 PG 派生，见 §4.4）。 + +--- + +## 三、当前差缺统计算法（核心） + +### 3.1 一句话概括 + +> **以目标日期的实到扫描为锚点 → 反推这些运单所属的交接批次 → 展开批次全量应到 → 逐运单比对差缺。** + +即：**先有实到，再找应到**。这是本次重构要推翻的核心假设。 + +### 3.2 算法步骤（`db_compare.compare_site_date`） + +```mermaid +flowchart TD + S[compare_site_date site, target_date] --> S1 + S1["Step1 实到锚点
SELECT DISTINCT waybill_no
FROM actual_record
WHERE site=? AND scan_time::date = target_date"] + S1 -->|当天无实到| X[返回 None
“当天无实到数据，无法比对”] + S1 -->|有实到运单| S2 + S2["Step2 反推交接批次
SELECT DISTINCT handover_no
FROM expected_record
WHERE waybill_no = ANY(锚点运单)"] + S2 --> S3 + S3["Step3 展开批次全量应到
SELECT waybill_no, handover_no, handover_pieces
FROM expected_record
WHERE handover_no = ANY(批次)"] + S3 -->|无应到| X2[返回 None] + S3 --> S4 + S4["Step4 取批次全量实到
SELECT waybill_no, piece_no
FROM actual_record
WHERE waybill_no = ANY(展开的全部运单)"] + S4 --> S5 + S5["Step5 逐运单比对 _do_compare"] + S5 --> R[CompareResult
stats + 差缺明细 rows] +``` + +### 3.3 逐运单比对口径（`_do_compare`） + +对展开出的每一条应到运单： + +| 判断 | 结论 | +|------|------| +| `handover_pieces <= 0` | 跳过，不计入应到 | +| 实到件数 `>=` 应到件数（`handover_pieces`） | **足额到货**（含溢到），不进差缺 | +| 实到件数 `== 0` | **完全未到**（`full_miss++`） | +| `0 <` 实到件数 `<` 应到件数 | **部分未到**（`part_miss++`） | + +实到件数口径（SF / 非 SF 分支）： + +- **非 SF（中通/韵达/安能）**：`COUNT(DISTINCT piece_no)`，子单号去重。 +- **顺心 SF 运单**（`waybill_no` 以 `SF` 开头）：`COUNT(*)` 行计数，不去重（SF 子单号为随机号码，不能去重计数）。 + +### 3.4 统计指标定义（`CompareStats`） + +| 指标 | 定义 | +|------|------| +| `waybill_count` | 应到运单数（`handover_pieces>0` 的展开运单） | +| `expected_pieces` | Σ `handover_pieces`（交接件数，非录单件数） | +| `arrived_pieces` | Σ 各运单实到件数（按上节口径） | +| `undelivered_pieces` | `max(0, expected_pieces − arrived_pieces)` | +| `undelivered_wb` | `full_miss + part_miss`（差缺运单数） | +| `full_miss` / `part_miss` | 完全未到 / 部分未到运单数 | +| `sf_wb_count` / `sf_undelivered` | 顺心 SF 运单总数 / 其中差缺数 | + +### 3.5 差缺明细（`UndeliveredRow`） + +仅含短少运单：`交接单号 | 运单号 | 总件数(交接件数) | 已到单号1 | 已到单号2 | ...`。已到单号按实到记录顺序列出，**不编造缺件子单号**（扫描顺序号乱序，无法反推缺了哪个）。 + +### 3.6 另一个入口：按批次比对（`compare_site_batch`） + +已知交接单号时可直接按 `handover_no` 精确比对，**不依赖实到锚点**。展开该批次全量应到 → 取全量实到 → 走同一 `_do_compare`。此入口当前未接入任务链路，主要用于调试/复核。 + +--- + +## 四、任务触发与执行链路 + +### 4.1 任务种类 + +| kind | 含义 | 四站行为 | 百世行为 | +|------|------|---------|---------| +| `expected` | 应到下载 | 下载应到 Excel → 入库 | 不支持 | +| `actual` | 实到下载 | 下载实到 Excel → 入库 | 不支持 | +| `undelivered` | 未到（差缺） | 应到+实到 → 入库 → DB 比对 → 写单站未到 Excel | 站点直供未到 → 入库 | +| `compare`（`__compare__`） | 全站跑比对 | 4 站 DB 比对 + 百世 PG → 全站汇总 Excel | 同上 | + +### 4.2 触发方式 + +- **定时**：APScheduler `IntervalTrigger` 按 `fetch_schedule` 配置周期投递，激活时段内才投；任务空闲才投（`create_task_if_idle`）。 +- **手动**：`POST /tasks {site, kind, force?, date?}`，前端「获取未到数据」主按钮触发各站 primary kind，前端「跑比对」按钮触发 `__compare__/compare`。 + +任务执行统一走 `dispatch_task`（`runtime.py`）：登录态校验 → 调 handler → 成功后写业务日期（`_record_business_date`）→ 入库（`_persist_to_db`）。 + +### 4.3 目标日期（target_date）怎么定 + +`state_store.resolve_target_date(site, kind, date)`： + +- 显式传 `date` → 用之（前端重试按钮会回放原 `target_date`）。 +- 未传 → `today − offset`：`expected` 用 `expected_offset`、`actual` 用 `actual_offset`、四站 `undelivered` 跟随 `expected_offset`、百世恒当天。 + +### 4.4 就绪态与业务日期（ready / business_date） + +- `_ready_flags`：直接查 PG，`expected` / `actual` 看对应 `today − offset` 日期是否有数据；四站 `undelivered = expected ∧ actual`；百世看 `baishi_daily_stats` 当天。 +- `_record_business_date`：任务成功后在 SQLite 快照本次数据日期，供前端状态盘与报表「数据日期」列展示。 +- **注意**：`expected_business_date` / `actual_business_date` 是**按偏移派生的目标日期**，不是下载文件内的真实业务日期——`expected_record.business_date` 入库时取自状态库的 `expected_business_date`（`store._read_business_dates`），可能与交接单实际生成日不同（详见 §6 疑点）。 + +--- + +## 五、站点差异与配置 + +### 5.1 比对配置（`db_compare.SITE_COMPARE_CONFIG`） + +| 站点 | has_sf | 备注 | +|------|--------|------| +| 顺心 | `True` | SF 运单走行计数 | +| 中通 / 韵达 / 安能 | `False` | 子单号去重计数 | + +### 5.2 入库列映射（`store.ACTUAL_COLMAP` 与 `domain.STATIONS`） + +| 站点 | 应到运单号列 | 实到运单基号列 | 实到单件列 | 扫描时间列 | +|------|-------------|---------------|-----------|-----------| +| 顺心 | 运单号 | 运单号 | 子单号 | 操作时间 | +| 中通 | 运单号 | 由子单号复合串 `v[:-8]` 推导 | 运单号（复合串） | 扫描时间 | +| 韵达 | 运单号 | 主单号 | 子单号 | 扫描时间 | +| 安能 | 运单号 | 所属单号 | 扫描单号 | 扫描时间 | + +### 5.3 入库清洗特例 + +- **韵达实到**：只保留「交接单号为空」的行（到/接件扫描），丢弃非空行（派件/签收等重复数据）；再按子单号去重 keep-last。 +- **中通实到**：`piece_no` 为复合串 `H+运单号(12)+总数(4)+顺序(4)`，基号 = `v[:-8]`，每串计一件。 + +--- + +## 六、当前逻辑的特征与疑点（重构输入） + +### 6.1 特征 + +1. **锚点=实到扫描日**：某天实到为空 → 比对直接返回 None，不产出任何差缺（"先有实到才有结论"）。 +2. **批次是反推出来的**：只要当天有 1 个运单扫到，其所属整个交接批次都会被展开，把该批次的历史未到也一起统计进来（跨日旧账混入当天报表）。 +3. **应到=批次全量**：`expected_pieces` 是"被命中批次的全部应到"，不是"当天的应到交接单"；`business_date` 标签也可能滞后/漂移。 +4. **溢到不抵消**：`undelivered_pieces = max(0, Σ应到 − Σ实到)` 是全局差；单运单溢到（实到>应到）只会让该运单不进差缺，不会抵消他单未到。 + +### 6.2 疑点（来自 08-02 韵达实测） + +- 08-02 实到 130 条/58 运单全部命中应到且件数一致，但差缺 9 单里有 7 单是 07-21 批次的历史未到——因为当天有 1 个运单（`295468511`）属于该批次被扫到，整批被展开。 +- `task_history` 无记录却有 18:28 入库 130 条实到：入库链路与常规任务链路不一致（`ingest_state` 也未刷新），说明存在绕过任务系统的直入路径，需在重构时统一入口。 + +--- + +## 七、重构目标对照（待细化） + +| 维度 | 现状（实到驱动） | 重构方向（应到驱动） | +|------|----------------|--------------------| +| 统计起点 | 目标日期实到扫描运单 | 目标日期/批次的应到数据（`expected_record`） | +| 批次来源 | 实到锚点反推 | 应到自身携带的 `handover_no` / `business_date` | +| 无数据表现 | 实到空 → 不产出 | 应到空 → 明确"无应到"；应到有、实到空 → 全部记差缺 | +| 历史批次混入 | 会（1 单命中即整批展开） | 应到锚定，天然按应到口径隔离 | +| 报表口径 | 批次全量 | 按应到日期/批次统计 | + +> 重构后仍需保持：应到=交接件数、实到=子单号去重（SF 行计数）、未到=`max(0, 应到−实到)`、差缺明细含已到单号。 + +--- + +## 八、涉及文件清单 + +| 文件 | 职责 | +|------|------| +| `inbound_verify/db_compare.py` | DB 差缺比对引擎 + 全站汇总（重构主战场） | +| `inbound_verify/runtime.py` | 任务派发、未到 handler、入库钩子、就绪派生 | +| `inbound_verify/store.py` | Excel → PG 入库（expected/actual/百世） | +| `inbound_verify/state_store.py` | SQLite 状态/配置/任务/偏移/目标日期 | +| `inbound_verify/domain.py` | 站点/文件/列映射单一配置源 | +| `inbound_verify/schema.sql` | PG 表结构 | +| `inbound_verify/cli/server.py` | FastAPI 端点（/tasks /compare /status /config /report） | +| `dashboard/app/page.tsx` | 前端触发任务、状态盘、下载报表 | diff --git a/docs/2026-08-03-应到驱动差缺统计重构-design.md b/docs/2026-08-03-应到驱动差缺统计重构-design.md new file mode 100644 index 0000000..e88c8af --- /dev/null +++ b/docs/2026-08-03-应到驱动差缺统计重构-design.md @@ -0,0 +1,179 @@ +# 应到驱动差缺统计重构 · 设计方案 + +> 日期：2026-08-03 +> 分支：`refactor/undelivered-by-expected` +> 状态：待审核 +> 目标：差缺统计从"实到驱动（反推应到）"改为"应到驱动（以应到为统计起点）"，并用「出库日」作为批次归属日，吸收"提前提交"扰动 + +--- + +## 一、背景与问题 + +### 1.1 业务诉求 + +- 业务部门日常看的是**应到货物数据**：要"应到了哪些、实到了哪些、差缺是什么"。 +- 当前实现是**实到驱动**：以目标日实到扫描为锚点 → 反推交接批次 → 展开批次全量应到 → 逐单比对。 +- 方向与业务诉求相反，需重构为**应到驱动**。 + +### 1.2 核心扰动：应到任务可能被提前提交 + +- 应到数据理论当天提交（韵达固定提前 1 天），但实际可能提前 1~2 天。 +- 若完全按"下载日"统计应到，会把"提前提交、货次日才到"的批次计入当天，产生假差缺。 + +### 1.3 数据实证（7 月全量） + +对 `downloads/archive/` 7 月应到文件 × PG 实到数据交叉验证： + +| 站点 | 下载日=出库大头日 | 出库大头日=下载日+1 | +|------|-----------------|-------------------| +| 顺心 | 83/83（100%） | 0 | +| 中通 | 28/30（93%） | 2（`071402`、`071902`） | +| 韵达 | 0 | 29/29（100%，固定提前） | +| 安能 | 31/31（100%） | 0 | + +**关键结论**： + +1. **出库大头日 ≈ 实到峰值日**（154/158 一致，97.5%）——"出库日"基本等于"这批货实际到的那天"。 +2. "提前提交"的批次（中通 `071402` 出库 7/15、`071902` 出库 7/20）在**出库时间上如实体现**了真实归属日。 +3. 四站应到文件的「出库时间」字段非空率 100%，且已完整保留在 PG `expected_record.raw` JSONB 中（四站 100% 有值）。 + +--- + +## 二、核心口径 + +### 2.1 批次归属日 = 出库大头日 + +一个交接批次内，取运单「出库时间」的**日期众数（大头日）**作为该批次归属日： + +``` +批次归属日 = mode(运单.出库时间::date) +``` + +- 顺心/安能：归属日 = 下载日（无扰动） +- 中通：偶发提前批次自动归属次日（`071402` → 7/15） +- 韵达：所有批次归属日 = 下载日 + 1（与实到对齐，不再依赖 `expected_offset=1`） + +### 2.2 统计 D 日差缺 = 取所有「出库日 = D」的应到批次 + +无论批次在 D / D-1 / D-2 哪天下载（`business_date` 为何），只要**出库日 = D** 即纳入 D 日统计： + +``` +目标批次 = expected_record WHERE site=? AND 出库日 = D +``` + +这样： + +- 提前提交的批次（下载于 D-1/D-2、出库于 D）会被**自然归入 D 日**，不再遗漏也不提前计入； +- 不再需要"实到为 0 → 抛弃/标记留存"的状态机； +- 不再需要为韵达单独配置 `expected_offset`。 + +--- + +## 三、统计流程（compare_site_date 重构后） + +```mermaid +flowchart TD + S[查询 D 日差缺] --> S1 + S1["Step1 取应到批次
expected_record
WHERE site=? AND 出库日 = D"] + S1 -->|无应到| X[返回：明确当日无应到] + S1 -->|有批次| S2 + S2["Step2 展开批次全量应到
waybill_no, handover_no, handover_pieces"] + S2 --> S3 + S3["Step3 查这些运单的全量实到
actual_record WHERE waybill_no = ANY(应到)"] + S3 --> S4 + S4["Step4 逐运单比对（_do_compare）
应到=交接件数 实到=子单号去重/SF行计数"] + S4 --> R[CompareResult
stats + 差缺明细] +``` + +### 3.1 与现实现的差异 + +| 环节 | 现状 | 重构后 | +|------|------|--------| +| 应到来源 | 实到锚点反推批次 | 按出库日直接取应到批次 | +| 无实到表现 | 返回 None（不产出） | 应到空 → 明确"无应到"；应到有实到空 → 记差缺 | +| 历史批次混入 | 1 单命中即整批展开 | 按出库日隔离，天然干净 | +| 提前提交 | 无感知（靠实到锚定） | 出库日归属，自动吸收 | + +### 3.2 保留的能力 + +- `compare_site_batch`（按交接单号精确比对）保留，供复核。 +- 实到驱动入口 `compare_site_date` 旧逻辑保留为对照模式（或通过配置切换），便于回溯验证差异。 +- 统计口径不变：应到=交接件数、实到=子单号去重（顺心 SF 行计数）、未到=`max(0,应到−实到)`、明细含已到单号。 + +--- + +## 四、数据层改造 + +### 4.1 新增列：`out_date` + +`expected_record` 新增 `out_date DATE`（出库日，批次归属日的持久化依据）： + +```sql +ALTER TABLE expected_record ADD COLUMN IF NOT EXISTS out_date DATE; +CREATE INDEX IF NOT EXISTS idx_expected_out_date ON expected_record (site, out_date); +``` + +- 入库时（`store._ingest_expected`）：从 raw 的「出库时间」解析出日期写入 `out_date`。 +- 历史数据回填：一次性 UPDATE，从 `raw->>'出库时间'` 提取日期。 +- 出库时间缺失/解析失败 → `out_date` 置 NULL，统计时回退 `business_date`（下载日），保证不丢数据。 + +### 4.2 `business_date` 语义保持不变 + +- `business_date` 继续表示"下载目标日快照"（兼容就绪态派生 / 报表数据日期列 / 现有 API）。 +- 差缺统计改用 `out_date`，两者解耦，避免连锁改动。 + +### 4.3 出库时间字段来源（已验证） + +| 站点 | 字段 | 覆盖率 | +|------|------|--------| +| 顺心 | `出库时间` | raw 100% | +| 中通 | `出库时间` | raw 100% | +| 韵达 | `出库时间` | raw 100% | +| 安能 | `出库时间` | raw 100% | + +--- + +## 五、边界与特殊处理 + +| 场景 | 处理 | +|------|------| +| 批次内出库日跨多天 | 取**大头日（众数）**；众数并列时取较早日期 | +| 出库时间缺失/解析失败 | `out_date` 置 NULL，回退 `business_date` | +| 应到有、实到空 | 全部计入差缺（不再因"无实到"而返回 None） | +| 实到有、应到无（孤儿） | 保持现状，报表/明细可另行提示，不混入应到统计 | +| 韵达 `expected_offset` | 保留配置但重构后不再参与归属日计算（由 `out_date` 取代） | +| 异常小批次 | 规模很小（1~4 单）按常规逻辑走；如出现系统性偏差再单独讨论 | + +--- + +## 六、涉及改动清单 + +| 文件 | 改动 | +|------|------| +| `schema.sql` | `expected_record` 增 `out_date` 列 + 索引 | +| `store.py` | `_ingest_expected` 写 `out_date`；新增历史回填逻辑（CLI） | +| `db_compare.py` | `compare_site_date` 改为按 `out_date` 取应到；新增"无应到"返回语义；保留批次入口与实到驱动对照 | +| `runtime.py` | `_site_undelivered_handler` 锚点日期逻辑随新口径调整 | +| `cli/server.py` | `/compare` 响应补充 `out_date` 语义说明；行为兼容 | +| `state_store.py` | 视需要暴露 `out_date` 相关查询 | +| 前端 `dashboard` | 报表说明文案（批次归属=出库日）；无结构变更预期 | + +--- + +## 七、验证计划 + +1. **单元验证**：`out_date` 回填后，抽样核对与归档 Excel 出库日一致。 +2. **回溯对照**：用 7 月归档应到 + PG 实到，分别跑"旧实到驱动"与"新应到驱动"，对比差缺差异，重点： + - 韵达 7 月各日（应到归属日整体 +1 是否对齐实到） + - 中通 7/14、7/19（`071402`/`071902` 是否归入次日） + - 顺心/安能（应无差异） +3. **报表烟测**：跑一次 `__compare__/compare` 全站汇总，人工核对韵达 08-02 数据。 + +--- + +## 八、决策记录（已确认） + +- ✅ 出库时间字段业务含义 = **货物实际发出时间**（按此处理）。 +- ✅ 批次内出库日并列众数取法 = **取较早日期**。 +- ✅ "无应到"呈现 = 报表中**直接写**（如实呈现，无需特殊文案）。 +- ✅ **保留实到驱动入口**，长期作为对照（不删除）。 diff --git a/docs/2026-08-03-应到驱动差缺统计重构-实现总结.md b/docs/2026-08-03-应到驱动差缺统计重构-实现总结.md new file mode 100644 index 0000000..10fb6cc --- /dev/null +++ b/docs/2026-08-03-应到驱动差缺统计重构-实现总结.md @@ -0,0 +1,120 @@ +# 应到驱动差缺统计重构 · 实现总结（备忘录） + +> 完成日期：2026-08-03 +> 分支：`refactor/undelivered-by-expected` +> 关联文档：`docs/2026-08-02-未到差缺统计逻辑现状梳理.md`、`docs/2026-08-03-应到驱动差缺统计重构-design.md` +> 状态：已完成并验证，待提交 + +--- + +## 一、背景 + +原差缺统计为**实到驱动**：以目标日实到扫描为锚点 → 反推交接批次 → 展开批次全量应到 → 逐单比对。 + +业务部门日常以**应到数据**为依据，且应到任务可能被**提前提交 1~2 天**（韵达固定提前 1 天），导致： + +- 完全按应到统计会出现假差缺； +- 实到驱动会把历史批次混入当天报表（1 单命中即整批展开）。 + +重构目标：改为**应到驱动**，并用「出库日」作为批次归属日，自动吸收提前提交。 + +## 二、核心口径 + +- **批次归属日 = 批次内运单「出库时间」日期众数，并列取较早**（字段：`expected_record.batch_out_date`）。 +- **统计 D 日差缺 = 取所有 `batch_out_date = D` 的应到批次**，展开全量应到 → 查全量实到 → 逐运单比对。 +- 无论批次在 D / D-1 / D-2 哪天下载（`business_date` 为何），只要出库日 = D 即纳入 D 日统计。 + +## 三、数据实证（7 月全量） + +| 站点 | 下载日=出库大头日 | 出库大头日=下载日+1 | +|------|-----------------|-------------------| +| 顺心 | 83/83（100%） | 0 | +| 中通 | 28/30（93%） | 2（`071402`、`071902`） | +| 韵达 | 0 | 29/29（100%，固定提前） | +| 安能 | 31/31（100%） | 0 | + +关键结论： + +- **出库大头日 ≈ 实到峰值日**（154/158 一致，97.5%）——"出库日"基本等于"这批货实际到的那天"。 +- "提前提交"的批次（中通 `071402` 出库 7/15、`071902` 出库 7/20）在出库时间上如实体现真实归属日。 +- 四站应到文件的「出库时间」字段非空率 100%，且已完整保留在 PG `expected_record.raw` JSONB 中。 + +## 四、代码改动 + +### 4.1 数据层 + +- `schema.sql`：`expected_record` 新增 `out_date DATE`（运单出库日）、`batch_out_date DATE`（批次归属日）+ `idx_expected_out_date` / `idx_expected_batch_out_date` 索引。 +- `store.py`： + - `_ingest_expected`：解析「出库时间」写 `out_date`；按交接单号聚合出库日众数（并列取较早）写 `batch_out_date`。 + - 新增 `_batch_out_date_map()` 辅助函数。 + - 新增 `backfill_out_date()` + CLI 子命令 `backfill-out-date`，历史数据一次性回填。 + +### 4.2 比对层 + +- `db_compare.py`： + - 新增 `compare_site_outdate(site, target_date)`：应到驱动入口，`WHERE batch_out_date = target_date` 取批次 → 展开 → 比对。 + - 保留 `compare_site_date()`（实到驱动）作对照，不删除。 + - `_target_date_for()` 改为默认取今天（不再依赖 actual_offset）。 + - `build_full_report()` 改用 `compare_site_outdate`。 +- `runtime.py`：`_site_undelivered_handler` 切到应到驱动，锚点日期默认今天。 +- `cli/server.py`：`POST /compare` 切到应到驱动。 + +## 五、实施与验证 + +### 5.1 数据迁移 + +``` +python -m inbound_verify.store init # 建表/补列（幂等） +python -m inbound_verify.store backfill-out-date # 历史回填 +``` + +回填结果： + +- `out_date`：顺心 3290 / 中通 5419 / 韵达 1640 / 安能 3459 条，共 13808 条。 +- `batch_out_date`：顺心 93 / 中通 35 / 韵达 32 / 安能 35 个批次。 +- 抽样核对 PG `out_date` vs 归档 Excel 出库日一致率 96~100%。 + +### 5.2 批次归属验证 + +| 批次 | 下载日 | batch_out_date | 预期 | +|------|--------|----------------|------| +| 中通 `...071401` | 7/14 | 7/14 | 正常 | +| 中通 `...071402` | 7/14 | **7/15** | 提前提交归位 | +| 中通 `...071901` | 7/19 | 7/19 | 正常 | +| 中通 `...071902` | 7/19 | **7/20** | 提前提交归位 | +| 韵达 `...07312001` | 7/31 | 8/1 | 固定 +1 | +| 顺心/安能 | — | = 下载日 | 无扰动 | + +### 5.3 回溯对照（7/02~7/31） + +新应到驱动 vs 旧实到驱动，差异方向符合设计： + +- 旧驱动混入历史批次（如中通 7/12 旧 236 件 vs 新 11 件；顺心 7/13 旧 59 vs 新 1）。 +- 新驱动只统计出库日=当天批次，数字更聚焦。 +- 个别日期新驱动未到偏大（如韵达 7/03、安能 7/29），属"当天出库、次日扫描"的真实差缺口径。 + +### 5.4 接口联调（真实后端） + +| 用例 | 结果 | +|------|------| +| `POST /compare` 韵达 2026-08-02 | 批次 1 个（`...08012001`），差缺 2 件/2 单（`988350756`、`988415586`），历史批次不再混入 | +| `POST /compare` 中通 2026-07-15 | 提前提交批次 `...071402` 正确归位到 7/15 | +| `__compare__/compare` 全站汇总 2026-08-01 | 顺心 17 件 / 中通 23 件 / 韵达 0 件 / 安能 0 件，合计 40 件，报表正常生成 | + +## 六、待确认 / 遗留事项 + +- `out_date` / `batch_out_date` 依赖站点「出库时间」字段语义（当前按"货物实际发出时间"处理，已与业务确认）。 +- 批次内出库日并列众数取较早（已确认）。 +- "无应到"时报表直接写（如实呈现，无特殊文案，已确认）。 +- 实到驱动入口保留作对照（已确认）。 +- `docs/2026-08-02-未到统计重构讨论纪要与下一步.md` 中记录的 18:28 直入入库等链路疑点，本重构未处理，留待后续。 + +## 七、附：涉及文件 + +| 文件 | 说明 | +|------|------| +| `schema.sql` | 表结构：新增 `out_date` / `batch_out_date` | +| `inbound_verify/store.py` | 入库解析 + 历史回填 | +| `inbound_verify/db_compare.py` | 应到驱动比对入口（保留实到驱动对照） | +| `inbound_verify/runtime.py` | 未到任务切到应到驱动 | +| `inbound_verify/cli/server.py` | `/compare` API 切到应到驱动 | diff --git a/inbound_verify/cli/server.py b/inbound_verify/cli/server.py index 182f1e5..88e63dc 100644 --- a/inbound_verify/cli/server.py +++ b/inbound_verify/cli/server.py @@ -265,7 +265,7 @@ class CompareRequest(BaseModel): @app.post("/compare") def run_compare(req: CompareRequest): - """DB 差缺比对：以实到扫描日期为锚点，反推交接批次，展开全量比对。 + """DB 差缺比对（应到驱动）：以批次归属日（batch_out_date）为锚，展开全量比对。 返回统计指标 + 差缺明细。 """ # 合法性校验 @@ -284,11 +284,11 @@ def run_compare(req: CompareRequest): if target_date > today: raise HTTPException(status_code=400, detail=f"date 不可为未来日期: {req.date}") - result = db_compare.compare_site_date(req.site, req.date) + result = db_compare.compare_site_outdate(req.site, req.date) if result is None: raise HTTPException( status_code=404, - detail=f"{req.site} {req.date}: 当天无实到数据，无法比对", + detail=f"{req.site} {req.date}: 当日无应到批次，无法比对", ) return { diff --git a/inbound_verify/db_compare.py b/inbound_verify/db_compare.py index b9deca9..d674842 100644 --- a/inbound_verify/db_compare.py +++ b/inbound_verify/db_compare.py @@ -224,6 +224,83 @@ def compare_site_date(site: str, target_date: str) -> CompareResult | None: return None +def compare_site_outdate(site: str, target_date: str) -> CompareResult | None: + """应到驱动差缺比对：以「批次归属日（batch_out_date）」为准取应到。 + + 与 compare_site_date（实到驱动）区别： + 1. 应到来源 = expected_record WHERE batch_out_date = target_date + 2. 不再依赖实到锚点反推；应到空时返回 None（明确"当日无应到"） + 3. 提前提交的批次按其出库日归属，自动归入正确日期 + + Args: + site: 站点名（"顺心"/"中通"/"韵达"/"安能"） + target_date: 目标业务日期 "YYYY-MM-DD" + + Returns: + 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() + + # ── Step 1: 取目标日应到批次（按批次归属日）── + cur.execute( + """ + SELECT DISTINCT handover_no FROM expected_record + WHERE site = %s AND batch_out_date = %s + ORDER BY handover_no + """, + (site, target_date), + ) + batches = [r[0] for r in cur.fetchall()] + if not batches: + print(f"[db_compare] {site} {target_date}: 当日无应到批次") + conn.close() + return None + + # ── Step 2: 展开批次全量应到 ── + 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() + if not exp_rows: + conn.close() + return None + + all_wbs = [r[0] for r in exp_rows] + + # ── Step 3: 取批次全量实到 ── + 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() + + # ── Step 4: 逐运单比对 ── + 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: """按指定交接单号执行全批次比对（不依赖实到锚点）。 @@ -499,11 +576,9 @@ def _stats_to_dict(s: CompareStats) -> dict: 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") + """4 站比对锚点（应到驱动）：批次归属日默认取今天。 + 各站统一以出库日（batch_out_date）为准，不再依赖站点偏移配置。""" + return date.today().strftime("%Y-%m-%d") def _baishi_from_pg(cur, target: str): @@ -555,7 +630,7 @@ def _baishi_from_pg(cur, target: str): def build_full_report(date=None) -> str: """DB 版全站汇总报表：4 站走 DB 比对、百世走 PG，复用 compare.build_summary 渲染。 - 产出 output/应到未到数据.xlsx（/report 下载）。date=None 时各站按 actual_offset 算锚点（以实到扫描日为锚）。 + 产出 output/应到未到数据.xlsx（/report 下载）。date=None 时各站取今天为批次归属锚点（应到驱动）。 返回输出路径。""" from inbound_verify import compare # 复用 build_summary / write_station / OUTFILE @@ -584,7 +659,7 @@ def build_full_report(date=None) -> str: continue target = date or _target_date_for(name) site_targets[name] = target - result = compare_site_date(name, target) + result = compare_site_outdate(name, target) if result is not None: results.append((name, _stats_to_dict(result.stats))) _write_sheet(wb.create_sheet(name), result) diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index fde67ff..39c1687 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -492,23 +492,20 @@ def _site_undelivered_handler(site): except Exception as e: print(f">> [入库] {site} 前置入库失败（不影响比对尝试）: {e}") - # ── DB 比对（替代旧 Excel 比对）── + # ── DB 比对（应到驱动：以批次归属日 batch_out_date 为锚）── 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" - ) + target_date = datetime.now().date().strftime("%Y-%m-%d") - result = db_compare.compare_site_date(site, target_date) + result = db_compare.compare_site_outdate(site, target_date) if result is not None: db_compare.write_result_excel(result) else: - print(f">> [未到] {site} {target_date}: 当天无实到数据，跳过比对") + print(f">> [未到] {site} {target_date}: 当日无应到批次，跳过比对") except Exception as e: print(f">> [未到] {site} DB 比对异常（不影响下载结果）: {e}") diff --git a/inbound_verify/store.py b/inbound_verify/store.py index 1f71fa1..14a22e4 100644 --- a/inbound_verify/store.py +++ b/inbound_verify/store.py @@ -173,6 +173,33 @@ def _to_int(v): return None +def _batch_out_date_map(df, cfg, out_col="出库时间"): + """按交接单号分组，计算批次归属日：出库日期众数，并列取较早日期。 + 返回 {handover_no: date}；无出库时间/无交接单号的行不参与。""" + from collections import Counter + + jd_col = cfg.get("exp_jd", "交接单号") + if jd_col not in df.columns or out_col not in df.columns: + return {} + counter: dict[str, Counter] = {} + for _, r in df.iterrows(): + hn = str(r.get(jd_col, "")).strip() + if not hn or hn == "nan": + continue + d = _parse_time(r.get(out_col)) + if d is None: + continue + counter.setdefault(hn, Counter())[d.date()] += 1 + out = {} + for hn, cnt in counter.items(): + if not cnt: + continue + max_n = max(cnt.values()) + earliest = min(d for d, n in cnt.items() if n == max_n) + out[hn] = earliest + return out + + def _parse_time(v): """尽力解析多种时间格式为 datetime；失败返回 None（原始值在 raw 里）。""" if v is None: @@ -233,13 +260,16 @@ def _read_business_dates(): _SQL_EXPECTED = """ INSERT INTO expected_record - (site, waybill_no, handover_no, handover_pieces, order_pieces, business_date, raw) - VALUES (%s,%s,%s,%s,%s,%s,%s) + (site, waybill_no, handover_no, handover_pieces, order_pieces, + business_date, out_date, batch_out_date, raw) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (site, waybill_no) DO UPDATE SET handover_no = EXCLUDED.handover_no, handover_pieces = EXCLUDED.handover_pieces, order_pieces = EXCLUDED.order_pieces, business_date = COALESCE(EXCLUDED.business_date, expected_record.business_date), + out_date = COALESCE(EXCLUDED.out_date, expected_record.out_date), + batch_out_date = COALESCE(EXCLUDED.batch_out_date, expected_record.batch_out_date), raw = EXCLUDED.raw, ingested_at = now() """ @@ -294,19 +324,31 @@ def _ingest_expected(cur, site, business_date): df = pd.read_excel(path, dtype=str).fillna("") df = df.drop_duplicates(subset=[cfg["exp_wb"]], keep="first") biz = _parse_date(business_date) + out_col = "出库时间" + has_out_col = out_col in df.columns + # 批次归属日：同交接单号出库日众数，并列取较早 + batch_out_date = _batch_out_date_map(df, cfg, out_col) rows = [] for r in df.to_dict("records"): wb = str(r.get(cfg["exp_wb"], "")).strip() if not wb: continue + out_date = None + if has_out_col: + out_dt = _parse_time(r.get(out_col)) + if out_dt is not None: + out_date = out_dt.date() + hn = str(r.get(cfg["exp_jd"], "")).strip() or None rows.append( ( site, wb, - str(r.get(cfg["exp_jd"], "")).strip() or None, + hn, _to_int(r.get(cfg["exp_qty"])), _to_int(r.get("录单件数")), biz, + out_date, + batch_out_date.get(hn), Jsonb(_raw_row(r)), ) ) @@ -463,6 +505,69 @@ def ingest_task(site, kind): return total +def backfill_out_date(site=None): + """历史数据回填：从 raw->>'出库时间' 解析出库日，写入 out_date； + 再按交接单号聚合出库日众数，回填 batch_out_date。 + site 为空时处理全部站点。返回回填 out_date 条数。""" + sites = [site] if site else ALL_SITES + total = 0 + with _connect(_load_pg_config()["dbname"]) as conn: + with conn.cursor() as cur: + for s in sites: + # ── Step 1: 回填 out_date（仅 NULL 行）── + cur.execute( + "SELECT id, raw FROM expected_record " + "WHERE site=%s AND out_date IS NULL", + (s,), + ) + rows = cur.fetchall() + updates = [] + for rid, raw in rows: + if not isinstance(raw, dict): + continue + out_dt = _parse_time(raw.get("出库时间")) + if out_dt is None: + continue + updates.append((out_dt.date(), rid)) + if updates: + cur.executemany( + "UPDATE expected_record SET out_date=%s WHERE id=%s", + updates, + ) + total += len(updates) + print(f" [回填] {s}：{len(updates)}/{len(rows)} 条") + + # ── Step 2: 回填 batch_out_date（仅 NULL 行）── + cur.execute( + "SELECT id, handover_no, out_date FROM expected_record " + "WHERE site=%s AND batch_out_date IS NULL", + (s,), + ) + rows = cur.fetchall() + if rows: + from collections import Counter + + cnt: dict[str, Counter] = {} + for _, hn, od in rows: + if not hn or od is None: + continue + cnt.setdefault(hn, Counter())[od] += 1 + batch_map = {} + for hn, c in cnt.items(): + max_n = max(c.values()) + batch_map[hn] = min(d for d, n in c.items() if n == max_n) + if batch_map: + cur.executemany( + "UPDATE expected_record SET batch_out_date=%s " + "WHERE site=%s AND handover_no=%s", + [(d, s, hn) for hn, d in batch_map.items()], + ) + print(f" [回填] {s} batch_out_date：{len(batch_map)} 个批次") + conn.commit() + print(f">> [回填] out_date 完成，共 {total} 条") + return total + + def get_existing_handover_nos(site): """查该站点已落库的交接单号集合（expected_record.handover_no）。 供"提交导出任务前"去重：已落库的交接单号不再重复提交导出任务。 @@ -553,6 +658,8 @@ def main(): sys.exit(1) total = ingest_task(site, kind) print(f">> [ingest-one] {site}/{kind} 入库 {total} 条") + elif cmd == "backfill-out-date": + backfill_out_date(site) else: print(__doc__) sys.exit(1) diff --git a/schema.sql b/schema.sql index cef79ff..fba8f53 100644 --- a/schema.sql +++ b/schema.sql @@ -19,12 +19,16 @@ CREATE TABLE IF NOT EXISTS expected_record ( handover_pieces INTEGER, -- 交接件数（应到件数口径） order_pieces INTEGER, -- 录单件数 business_date DATE, -- 业务日期（属性，非唯一键；读不到则 NULL） + out_date DATE, -- 出库日（批次归属日口径；从 raw.出库时间 解析） + batch_out_date DATE, -- 批次归属日（同交接单号出库日众数，并列取较早） raw JSONB NOT NULL, -- 站点原始全列（key=原列名） ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(), UNIQUE (site, waybill_no) ); CREATE INDEX IF NOT EXISTS idx_expected_site_date ON expected_record (site, business_date); CREATE INDEX IF NOT EXISTS idx_expected_handover ON expected_record (site, handover_no); +CREATE INDEX IF NOT EXISTS idx_expected_out_date ON expected_record (site, out_date); +CREATE INDEX IF NOT EXISTS idx_expected_batch_out_date ON expected_record (site, batch_out_date); -- 实到货物（扫描件级：一扫描一行；每扫描一件系统生成一个单号） CREATE TABLE IF NOT EXISTS actual_record (