8 Commits

Author SHA1 Message Date
Misaka
f29a9c6b77 feat(runtime): mute chrome and anneng via --mute-audio (configurable) 2026-08-04 21:37:15 +08:00
Misaka_Company
5853cb34e7 docs: record english-only commit message convention in CLAUDE.md 2026-08-04 17:58:18 +08:00
Misaka_Company
8e656516ea docs: finalize actual-arrival design as full-day scans, zero processing 2026-08-04 17:57:21 +08:00
Misaka_Company
c708bff924 refactor(db_compare): actual arrival = full scans on target date, no extra processing 2026-08-04 17:57:21 +08:00
Misaka_Company
db06776ff7 docs: design for actual-arrival extraction by scan date 2026-08-04 17:04:03 +08:00
Misaka_Company
e99f311bca fix(yunda): locate visible business iframe by content probe, fixes multi-iframe strict-mode violation 2026-08-04 17:04:03 +08:00
Misaka_Company
2387593eda fix(anneng): real-type ZK dateboxes + commit-wait before query, skip out-of-range handovers 2026-08-04 14:45:57 +08:00
Misaka
36b204abe5 refactor(db_compare): 差缺统计改为应到驱动,以出库日为批次归属日
将未到统计从实到驱动(实到锚点反推批次)改为应到驱动(batch_out_date 归属日直接取应到批次),以吸收应到任务提前 1~2 天提交的扰动(韵达固定 +1、中通偶发 +1)。

- expected_record 新增 out_date / batch_out_date + 索引;入库解析出库时间并聚合批次归属日,支持历史回填

- 新增 compare_site_outdate 应到驱动入口,保留 compare_site_date 实到驱动作对照

- 未到任务 / POST /compare / 全站汇总切换到应到驱动

- 附现状梳理、设计、实现总结三篇文档
2026-08-03 22:52:28 +08:00
13 changed files with 1058 additions and 64 deletions

View File

@@ -141,4 +141,5 @@ playwright install chromium
切勿提交,也别把真实凭据写进 `config.example.yaml` 切勿提交,也别把真实凭据写进 `config.example.yaml`
- **不要自动提交 / 推送**:本仓库约定改动后等用户明确说"提交"再 commit/push - **不要自动提交 / 推送**:本仓库约定改动后等用户明确说"提交"再 commit/push
(覆盖全局 CLAUDE.md 的 auto-push 默认)。 (覆盖全局 CLAUDE.md 的 auto-push 默认)。
- **提交消息一律英文**:格式 `type(scope): 英文描述`,风格与历史提交保持一致,禁止中文提交消息。
- 改完 Python 文件**必须跑 Black**(全局规范)。 - 改完 Python 文件**必须跑 Black**(全局规范)。

View File

@@ -20,6 +20,16 @@ debug:
# 留空或不匹配时将回退为全量模式。 # 留空或不匹配时将回退为全量模式。
target_site: "" target_site: ""
# ----------------------------------------------------------------------------
# 自动化静音:打开站点/安能应用时是否静音。
# true默认= 启动时给 Chromium 和安能附加 --mute-audio 开关,整个自动化会话
# 全程静音(防止深夜任务执行时浏览器/应用发出提示音打扰);
# false = 不静音,站点声音正常播放。
# 运行期切换需重启服务生效。
# ----------------------------------------------------------------------------
mute:
enabled: true
# ---------------------------------------------------------------------------- # ----------------------------------------------------------------------------
# 顺心捷达 (https://sxne.sxjdfreight.com) # 顺心捷达 (https://sxne.sxjdfreight.com)
# ---------------------------------------------------------------------------- # ----------------------------------------------------------------------------

View File

@@ -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[定时任务<br/>fetch_schedule] --> D[Playwright/CDP 登录站点]
A2[手动任务<br/>POST /tasks] --> D
D --> E1[应到货物数据.xlsx]
D --> E2[实到货物数据.xlsx]
end
subgraph 持久化层["持久化层 store.py + PG"]
E1 --> F[ingest_task<br/>expected_record]
E2 --> G[ingest_task<br/>actual_record]
end
subgraph 统计层["统计层 db_compare.py"]
F --> H[compare_site_date<br/>实到锚点→反推批次→展开应到]
G --> H
H --> I[output/*-未到数据.xlsx]
H --> J[output/应到未到数据.xlsx<br/>全站汇总]
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 实到锚点<br/>SELECT DISTINCT waybill_no<br/>FROM actual_record<br/>WHERE site=? AND scan_time::date = target_date"]
S1 -->|当天无实到| X[返回 None<br/>“当天无实到数据,无法比对”]
S1 -->|有实到运单| S2
S2["Step2 反推交接批次<br/>SELECT DISTINCT handover_no<br/>FROM expected_record<br/>WHERE waybill_no = ANY(锚点运单)"]
S2 --> S3
S3["Step3 展开批次全量应到<br/>SELECT waybill_no, handover_no, handover_pieces<br/>FROM expected_record<br/>WHERE handover_no = ANY(批次)"]
S3 -->|无应到| X2[返回 None]
S3 --> S4
S4["Step4 取批次全量实到<br/>SELECT waybill_no, piece_no<br/>FROM actual_record<br/>WHERE waybill_no = ANY(展开的全部运单)"]
S4 --> S5
S5["Step5 逐运单比对 _do_compare"]
S5 --> R[CompareResult<br/>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` | 前端触发任务、状态盘、下载报表 |

View File

@@ -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/83100% | 0 |
| 中通 | 28/3093% | 2`071402``071902` |
| 韵达 | 0 | 29/29100%,固定提前) |
| 安能 | 31/31100% | 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 取应到批次<br/>expected_record<br/>WHERE site=? AND 出库日 = D"]
S1 -->|无应到| X[返回:明确当日无应到]
S1 -->|有批次| S2
S2["Step2 展开批次全量应到<br/>waybill_no, handover_no, handover_pieces"]
S2 --> S3
S3["Step3 查这些运单的全量实到<br/>actual_record WHERE waybill_no = ANY(应到)"]
S3 --> S4
S4["Step4 逐运单比对_do_compare<br/>应到=交接件数 实到=子单号去重/SF行计数"]
S4 --> R[CompareResult<br/>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 数据。
---
## 八、决策记录(已确认)
- ✅ 出库时间字段业务含义 = **货物实际发出时间**(按此处理)。
- ✅ 批次内出库日并列众数取法 = **取较早日期**
- ✅ "无应到"呈现 = 报表中**直接写**(如实呈现,无需特殊文案)。
-**保留实到驱动入口**,长期作为对照(不删除)。

View File

@@ -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/83100% | 0 |
| 中通 | 28/3093% | 2`071402``071902` |
| 韵达 | 0 | 29/29100%,固定提前) |
| 安能 | 31/31100% | 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 切到应到驱动 |

View File

@@ -0,0 +1,112 @@
# 实到按「归属日当天扫描」提取修正 · 设计方案
> 日期2026-08-04
> 分支:`refactor/undelivered-by-expected`
> 状态已实施2026-08-04 验证通过)
> 目标:把差缺对比中的「实到货物数据」提取口径修正为「目标归属日当天实际扫描」,与业务口径对齐
---
## 一、业务口径(与业务部门对齐后的最终陈述)
差缺对比在**归属日期业务日期D** 这一维度上进行:
1. **应到货物数据**:取**归属日 = D** 的应到批次(批次归属日 = 出库大头日,已实现为 `expected_record.batch_out_date`)。
- 站点"提前下载"(韵达固定提前 1 天、个别站点偶发提前 1~2 天)不影响取数——提前提交的批次按其**出库日**自动归属到 D 日。
2. **实到货物数据**:取 **D 日当天实际扫描**的记录(`actual_record.scan_time::date = D`**全量、零加工**
- "当天实际扫描"指扫描行为发生在 D 日,与运单属于哪个批次无关。
- 重复单号在入库时即被吸收(`actual_record` 唯一键 `(site, piece_no)` + upsert对比层不再做任何加工。
3. 对取到的双方数据按运单号进行差缺对比。
**关键原则**:说"同一天"指的是**归属日期**,不是数据产生/下载的日期。
---
## 二、现状问题
| 环节 | 现状 | 问题 |
|------|------|------|
| 应到提取 | `WHERE batch_out_date = D`(应到驱动入口) | ✅ 已对齐 |
| 实到提取 | `WHERE waybill_no = ANY(批次运单)`**无扫描日过滤** | ❌ 跨日扫描混入 |
**实证案例(安能 08-03**
- 08-03 实到下载文件 = 355 行,全部扫描于 08-03获取层正确
- 今天 15:03 另有一次 08-04 实到下载176 行),其中 **6 件扫描于 08-04、运单属于 08-03 批次**
- 对比 SQL 按运单号取实到 → 08-03 实到 = 350当天+ 6次日= **356**,与"当天实际扫描"定义不符。
根因:`db_compare.py` 三处取实到的 SQL 只按运单号匹配,未按扫描日过滤。
---
## 三、修正方案
### 3.1 实到提取统一为「目标日当天扫描全量」
| 函数 | 用途 | 改动 |
|------|------|------|
| `compare_site_outdate` | 汇总报表(应到驱动,主入口) | 实到 SQL 改为 `WHERE site=%s AND scan_time::date=%s`,取当天扫描**全量**(不再按批次运单过滤) |
| `compare_site_date` | 实到驱动对照模式 | 同上Step 1 锚点本已按当天扫描Step 4 同步为全量口径) |
| `compare_site_batch` | 单批次审计(无日期语境) | **保持全量**docstring 注明"全批次全量实到,不受日期过滤" |
修正后统计口径:
```
应到 = 归属日 D 的批次全量应到(交接件数,不变)
实到 = D 日当天实际扫描的全量记录(零加工)
未到 = max(0, 应到 实到)(件级汇总);差缺明细按运单匹配(足额/溢到跳过)
```
**去重说明**:单号去重已在入库时完成——`schema.sql``actual_record``UNIQUE(site, piece_no)``store.py` 以 ON CONFLICT DO UPDATE 写入(韵达另有 drop_duplicates。对比层拿到的实到本就是去重后的数据不再做任何额外加工。
### 3.2 边界语义确认
| 场景 | 修正后行为 |
|------|-----------|
| 批次 D 的件扫描于 D+1迟到件 | 计入 D+1 日实到(其扫描日在 D+1D 日差缺按运单匹配判定,未到件照常呈现 |
| 批次 D 的件提前扫描于 D1 | 计入 D1 日实到(扫描日为准),不入 D 日实到 |
| 当天扫描但运单不在 D 日应到批次(孤儿) | 仍计入 D 日实到件数(实到=当天全量);差缺明细按运单匹配,无对应应到运单则不出现 |
| 一件多扫(同日) | 入库唯一键 `(site, piece_no)` 已吸收重复,对比层零加工 |
### 3.3 不做的事(范围外)
- 下载层各站查询日期口径已验证正确08-03 四站文件 100% 当天扫描),**不改下载层**。
- 获取层"日期回读校验"(韵达/中通/顺心选完日期后未校验输入框值)另立改进项,不阻塞本次。
- 百世为站级日聚合(已扫/应扫,固定当天),不涉及。
---
## 四、涉及改动清单
| 文件 | 改动 |
|------|------|
| `inbound_verify/db_compare.py` | `compare_site_outdate``compare_site_date` 两处实到 SQL 改为 `scan_time::date = 目标日` 全量取数;`compare_site_batch` docstring 注明口径差异;`_do_compare` 实到件数 = 实到记录全量 |
仅此一处文件;无需 schema 改动(去重本已由入库唯一键保证)。
---
## 五、验证结果2026-08-04 已重跑)
按新口径重跑 08-03 全站汇总(`build_full_report("2026-08-03")`
| 站点 | 应到 | 实到 | 未到 |
|------|------|------|------|
| 顺心 | 147 | 128 | 19 |
| 中通 | 305 | 276 | 29 |
| 韵达 | 54 | 53 | 1 |
| 安能 | 351 | **355**= 下载文件 355 行,零加工) | 0 |
08-04同为全量口径顺心 134、中通 214、韵达 102、安能 176。
- 安能 08-03 实到 355 与 `安能-实到货物数据.xlsx` 完全一致,验证"实到=当天扫描全量、零加工"成立。
- 差缺明细各运单"已到单号"扫描日全部 = 08-03。
- 报表已输出:`InboundVerify/output/应到未到数据.xlsx`
---
## 六、决策记录
- ✅ 实到 = 归属日当天实际扫描的全量(`scan_time::date = D`),对比时零加工。
- ✅ 单号去重由入库唯一键保证(`UNIQUE(site, piece_no)` + upsert对比层不重复去重。
-`compare_site_batch` 保持全批次全量(审计工具语义,不按日过滤)。
- ✅ 差缺明细按运单匹配,足额/溢到跳过,未到件 = max(0, 应到 实到)。

View File

@@ -265,7 +265,7 @@ class CompareRequest(BaseModel):
@app.post("/compare") @app.post("/compare")
def run_compare(req: CompareRequest): def run_compare(req: CompareRequest):
"""DB 差缺比对:以实到扫描日期为锚点,反推交接批次,展开全量比对。 """DB 差缺比对应到驱动以批次归属日batch_out_date为锚,展开全量比对。
返回统计指标 + 差缺明细。 返回统计指标 + 差缺明细。
""" """
# 合法性校验 # 合法性校验
@@ -284,11 +284,11 @@ def run_compare(req: CompareRequest):
if target_date > today: if target_date > today:
raise HTTPException(status_code=400, detail=f"date 不可为未来日期: {req.date}") 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: if result is None:
raise HTTPException( raise HTTPException(
status_code=404, status_code=404,
detail=f"{req.site} {req.date}: 当天无实到数据,无法比对", detail=f"{req.site} {req.date}: 当日无应到批次,无法比对",
) )
return { return {

View File

@@ -142,7 +142,7 @@ def compare_site_date(site: str, target_date: str) -> CompareResult | None:
1. 取 scan_time::date = target_date 的实到运单(锚点) 1. 取 scan_time::date = target_date 的实到运单(锚点)
2. 反推这些运单所属的交接批次handover_no 2. 反推这些运单所属的交接批次handover_no
3. 展开批次全量应到运单 3. 展开批次全量应到运单
4. 查询批次全量实到扫描 4. 取目标日当天实际扫描的全部实到记录(不做批次加工)
5. 逐运单比对差缺SF/non-SF 分支处理) 5. 逐运单比对差缺SF/non-SF 分支处理)
Args: Args:
@@ -207,10 +207,10 @@ def compare_site_date(site: str, target_date: str) -> CompareResult | None:
cur.execute( cur.execute(
""" """
SELECT waybill_no, piece_no FROM actual_record SELECT waybill_no, piece_no FROM actual_record
WHERE site = %s AND waybill_no = ANY(%s) WHERE site = %s AND scan_time::date = %s
ORDER BY waybill_no, piece_no ORDER BY waybill_no, piece_no
""", """,
(site, all_wbs), (site, target_date),
) )
act_rows = cur.fetchall() # [(waybill_no, piece_no), ...] act_rows = cur.fetchall() # [(waybill_no, piece_no), ...]
@@ -224,10 +224,91 @@ def compare_site_date(site: str, target_date: str) -> CompareResult | None:
return 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. 提前提交的批次按其出库日归属,自动归入正确日期
4. 实到 = 目标日当天实际扫描的全部记录scan_time::date = 目标日,不做批次加工),
跨日扫描不计入;差缺按运单匹配
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 scan_time::date = %s
ORDER BY waybill_no, piece_no
""",
(site, target_date),
)
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: def compare_site_batch(site: str, handover_no: str) -> CompareResult | None:
"""按指定交接单号执行全批次比对(不依赖实到锚点)。 """按指定交接单号执行全批次比对(不依赖实到锚点)。
用于已知交接单号后精确比对某一批次。 用于已知交接单号后精确比对某一批次。
注意:本入口为批次审计工具、无日期语境,实到取该批次运单的**全量**扫描
(不按扫描日过滤),与按归属日统计的 compare_site_outdate/compare_site_date 口径不同。
""" """
cfg = SITE_COMPARE_CONFIG.get(site) cfg = SITE_COMPARE_CONFIG.get(site)
if cfg is None: if cfg is None:
@@ -293,10 +374,10 @@ def _do_compare(
) -> CompareResult: ) -> CompareResult:
"""执行逐运单比对,产出统计 + 差缺明细。 """执行逐运单比对,产出统计 + 差缺明细。
compare.py:process() 口径一致: 业务口径一致:
- 应到件数 = handover_pieces交接件数 - 应到件数 = handover_pieces交接件数
- 实到件数 = SF ? COUNT(*) : COUNT(DISTINCT piece_no) - 实到件数 = 实到记录全量(目标日当天扫描全量,不做批次加工)
- arrived_cnt >= handover_pieces → 足额到货,跳过 - 差缺判断按运单匹配:arrived_cnt >= handover_pieces → 足额到货,跳过
""" """
# 构建实到索引: waybill_no → [piece_no, ...](保留所有行,不去重) # 构建实到索引: waybill_no → [piece_no, ...](保留所有行,不去重)
act_by_wb: dict[str, list[str]] = {} act_by_wb: dict[str, list[str]] = {}
@@ -306,6 +387,9 @@ def _do_compare(
stats = CompareStats() stats = CompareStats()
rows: list[UndeliveredRow] = [] rows: list[UndeliveredRow] = []
max_arrived = 0 max_arrived = 0
# 实到件数 = 实到记录全量(目标日当天扫描全量 / 批次全量),不做任何加工;
# 差缺判断仍按运单逐一匹配act_by_wb 仅用于逐运单 arrived_cnt
stats.arrived_pieces = len(act_rows)
for wb, handover_no, handover_pcs in exp_rows: for wb, handover_no, handover_pcs in exp_rows:
handover_pcs = handover_pcs or 0 handover_pcs = handover_pcs or 0
@@ -331,8 +415,6 @@ def _do_compare(
arrived_cnt = len(unique_pieces) arrived_cnt = len(unique_pieces)
arrived_list = unique_pieces arrived_list = unique_pieces
stats.arrived_pieces += arrived_cnt
if arrived_cnt >= handover_pcs: if arrived_cnt >= handover_pcs:
continue # 足额或溢到,不进差缺表 continue # 足额或溢到,不进差缺表
@@ -499,11 +581,9 @@ def _stats_to_dict(s: CompareStats) -> dict:
def _target_date_for(site: str) -> str: def _target_date_for(site: str) -> str:
"""4 站比对锚点today - actual_offset以实到扫描日为锚与 _site_undelivered_handler 一致)。""" """4 站比对锚点(应到驱动):批次归属日默认取今天。
from inbound_verify import state_store # 懒导入,避免成环 各站统一以出库日batch_out_date为准不再依赖站点偏移配置。"""
return date.today().strftime("%Y-%m-%d")
offset = state_store.get_offset(site, "actual")
return (date.today() - timedelta(days=offset)).strftime("%Y-%m-%d")
def _baishi_from_pg(cur, target: str): def _baishi_from_pg(cur, target: str):
@@ -555,7 +635,7 @@ def _baishi_from_pg(cur, target: str):
def build_full_report(date=None) -> str: def build_full_report(date=None) -> str:
"""DB 版全站汇总报表4 站走 DB 比对、百世走 PG复用 compare.build_summary 渲染。 """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 from inbound_verify import compare # 复用 build_summary / write_station / OUTFILE
@@ -584,7 +664,7 @@ def build_full_report(date=None) -> str:
continue continue
target = date or _target_date_for(name) target = date or _target_date_for(name)
site_targets[name] = target site_targets[name] = target
result = compare_site_date(name, target) result = compare_site_outdate(name, target)
if result is not None: if result is not None:
results.append((name, _stats_to_dict(result.stats))) results.append((name, _stats_to_dict(result.stats)))
_write_sheet(wb.create_sheet(name), result) _write_sheet(wb.create_sheet(name), result)

View File

@@ -76,6 +76,16 @@ def _wait_cdp_up(port, timeout=60.0):
return False return False
def read_mute_enabled() -> bool:
"""读 config.yaml 的 mute.enabled缺失默认 True=静音;读取失败也按 True"""
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
cfg = yaml.safe_load(f) or {}
return bool((cfg.get("mute", {}) or {}).get("enabled", True))
except Exception:
return True
# 【环境兼容】宿主 shellCodex/VS Code 插件、WorkBuddy 等)会向子进程注入一批与业务 # 【环境兼容】宿主 shellCodex/VS Code 插件、WorkBuddy 等)会向子进程注入一批与业务
# 无关的变量,实测会让安能应用登录后反复弹出“获取试用网点接口报错”: # 无关的变量,实测会让安能应用登录后反复弹出“获取试用网点接口报错”:
# - HTTP(S)_PROXY=http://127.0.0.1:8800QuickQ 加速器代理):安能的 wnp.ane56.com # - HTTP(S)_PROXY=http://127.0.0.1:8800QuickQ 加速器代理):安能的 wnp.ane56.com
@@ -100,8 +110,11 @@ _ANNENG_STRIP_EXACT = {
} }
def launch_anneng(app_path): def launch_anneng(app_path, mute=True):
"""以调试模式启动安能 Electron 应用(自动选取空闲端口),返回子进程对象。""" """以调试模式启动安能 Electron 应用(自动选取空闲端口),返回子进程对象。
muteTrue 时附加 --mute-audio 开关(静音应用声音,由 config.yaml mute.enabled 决定)。
"""
anneng_env = { anneng_env = {
key: value key: value
for key, value in os.environ.items() for key, value in os.environ.items()
@@ -109,9 +122,10 @@ def launch_anneng(app_path):
} }
port = _find_free_port() port = _find_free_port()
print(f">> 以调试模式启动【安能】应用(端口 {port}{app_path}") print(f">> 以调试模式启动【安能】应用(端口 {port}{app_path}")
proc = subprocess.Popen( anneng_args = [app_path, f"--remote-debugging-port={port}"]
[app_path, f"--remote-debugging-port={port}"], env=anneng_env if mute:
) anneng_args.append("--mute-audio")
proc = subprocess.Popen(anneng_args, env=anneng_env)
anneng.set_cdp_port(port) anneng.set_cdp_port(port)
if not _wait_cdp_up(port): if not _wait_cdp_up(port):
raise RuntimeError( raise RuntimeError(
@@ -251,6 +265,10 @@ def launch_and_prepare(debug_mode=False, debug_target="", foreground=True):
print(f"⚠️ 读取 config.yaml(debug) 失败: {e}") print(f"⚠️ 读取 config.yaml(debug) 失败: {e}")
anneng_app_path = state_store.get_setting("安能", "app_path") anneng_app_path = state_store.get_setting("安能", "app_path")
# 1.5 静音开关config.yaml mute.enabled缺失默认静音
mute = read_mute_enabled()
print(f">> 静音已{'开启' if mute else '关闭'}config.yaml mute.enabled={mute}")
# 2. 确定要挂载的网页站点 + 安能标记 # 2. 确定要挂载的网页站点 + 安能标记
anneng_active = False anneng_active = False
if debug_mode: if debug_mode:
@@ -283,7 +301,9 @@ def launch_and_prepare(debug_mode=False, debug_target="", foreground=True):
pw = sync_playwright().start() pw = sync_playwright().start()
# 调试模式临时开启 CDP 端口,便于外部 Playwright如 Playwright CLI 技能) # 调试模式临时开启 CDP 端口,便于外部 Playwright如 Playwright CLI 技能)
# 通过 connectOverCDP 挂载到已登录的页面读取内容。仅调试模式生效,不影响服务模式。 # 通过 connectOverCDP 挂载到已登录的页面读取内容。仅调试模式生效,不影响服务模式。
_launch_args = ["--remote-debugging-port=9223"] if debug_mode else [] _launch_args = (["--mute-audio"] if mute else []) + (
["--remote-debugging-port=9223"] if debug_mode else []
)
browser = pw.chromium.launch(headless=False, args=_launch_args) browser = pw.chromium.launch(headless=False, args=_launch_args)
context = browser.new_context(viewport={"width": 1920, "height": 1080}) context = browser.new_context(viewport={"width": 1920, "height": 1080})
# 默认禁用麦克风/摄像头:在每个页面/iframe 加载前覆盖 getUserMedia 为“直接拒绝”, # 默认禁用麦克风/摄像头:在每个页面/iframe 加载前覆盖 getUserMedia 为“直接拒绝”,
@@ -338,7 +358,7 @@ def launch_and_prepare(debug_mode=False, debug_target="", foreground=True):
anneng_proc = None anneng_proc = None
if anneng_active: if anneng_active:
try: try:
anneng_proc = launch_anneng(anneng_app_path) anneng_proc = launch_anneng(anneng_app_path, mute=mute)
pages_map["安能"] = True # 哨兵:已启动(无 Playwright page pages_map["安能"] = True # 哨兵:已启动(无 Playwright page
except Exception as e: except Exception as e:
print(f"⚠️ 启动安能应用失败,已跳过安能:{e}") print(f"⚠️ 启动安能应用失败,已跳过安能:{e}")
@@ -492,23 +512,20 @@ def _site_undelivered_handler(site):
except Exception as e: except Exception as e:
print(f">> [入库] {site} 前置入库失败(不影响比对尝试): {e}") print(f">> [入库] {site} 前置入库失败(不影响比对尝试): {e}")
# ── DB 比对(替代旧 Excel 比对)── # ── DB 比对(应到驱动:以批次归属日 batch_out_date 为锚)──
try: try:
from inbound_verify import db_compare # 懒导入,避免成环 from inbound_verify import db_compare # 懒导入,避免成环
if date: if date:
target_date = date target_date = date
else: else:
offset = state_store.get_offset(site, "actual") target_date = datetime.now().date().strftime("%Y-%m-%d")
target_date = (datetime.now().date() - timedelta(days=offset)).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: if result is not None:
db_compare.write_result_excel(result) db_compare.write_result_excel(result)
else: else:
print(f">> [未到] {site} {target_date}: 当天无实到数据,跳过比对") print(f">> [未到] {site} {target_date}: 当日无应到批次,跳过比对")
except Exception as e: except Exception as e:
print(f">> [未到] {site} DB 比对异常(不影响下载结果): {e}") print(f">> [未到] {site} DB 比对异常(不影响下载结果): {e}")

View File

@@ -11,8 +11,8 @@
# #
# 应到 = 运单信息导出(每张交接单下应到的运单明细)。整体流程: # 应到 = 运单信息导出(每张交接单下应到的运单明细)。整体流程:
# 主页(重庆鱼洞镇) → 运营管理 → 进站管理 → 进站交接单查询(tab) # 主页(重庆鱼洞镇) → 运营管理 → 进站管理 → 进站交接单查询(tab)
# → 设日期 → 查询 → 等“带小数点的 0”出现数据加载完毕 # → 设日期 → 等 ZK 组件值提交确认 → 查询 → 等“带小数点的 0”出现数据加载完毕
# → 逐条交接单:记下交接单号 → 双击 → 等运单信息加载并核对单号 # → 逐条交接单(交接日期超出范围的不导出):记下交接单号 → 双击 → 等运单信息加载并核对单号
# → 导出 → 双击转移全部待选字段 → 导出数据 → 任务添加成功 → 确认 → 回交接单信息 # → 导出 → 双击转移全部待选字段 → 导出数据 → 任务添加成功 → 确认 → 回交接单信息
# → 关闭进站交接单查询 tab → 打开导出下载(tab) # → 关闭进站交接单查询 tab → 打开导出下载(tab)
# → 轮询直到本批“交接单明细”任务全部导出完成 → 逐个免对话框下载x-auth+TGC 直连 GET # → 轮询直到本批“交接单明细”任务全部导出完成 → 逐个免对话框下载x-auth+TGC 直连 GET
@@ -30,7 +30,7 @@ import sys
import time import time
import urllib.parse import urllib.parse
import urllib.request import urllib.request
from datetime import datetime, timedelta from datetime import date, datetime, timedelta
import pandas as pd import pandas as pd
import websocket import websocket
@@ -375,25 +375,57 @@ def wait_query_page_ready(tab_cdp):
def set_zk_datebox(tab_cdp, index, date_str): def set_zk_datebox(tab_cdp, index, date_str):
"""直接向第 index 个 z-datebox-inp 入日期YYYY-MM-DD并触发 change """向第 index 个 z-datebox-inp 真实键入日期YYYY-MM-DDJS blur 触发提交
安能日期可直接输入无需日历控件。ZK datebox 监听 inputchange 事件 安能日期框是 ZK 组件:直接用原生 setter 改 value + 派发 input/change 事件不会
触发 ZK 的 onChange AU服务端拿不到新值查询仍用旧范围必须用 CDP Input 域
真实键入insertText再 JS blur() 让 ZK 解析、校验并提交 onChange。
不用 Tab 按键触发 blurCDP 按键事件依赖窗口在前台,窗口在后台时会被丢弃。
超出页面允许的查询范围时 ZK 会拒绝(不发 onChange
wait_zk_datebox_committed 等待超时暴露出来。
""" """
js = ( tab_cdp.eval(
"(() => {" "(() => {"
f" const inp = document.querySelectorAll('input.z-datebox-inp')[{index}];" f" const inp = document.querySelectorAll('input.z-datebox-inp')[{index}];"
" if (!inp) return false;" " if (!inp) return false;"
" inp.focus();" " inp.focus();"
" const setter = Object.getOwnPropertyDescriptor(" " inp.select();"
" window.HTMLInputElement.prototype, 'value').set;" " return true;"
f" setter.call(inp, {json.dumps(date_str)});"
" inp.dispatchEvent(new Event('input', {bubbles:true}));"
" inp.dispatchEvent(new Event('change', {bubbles:true}));"
" inp.blur();"
" return inp.value;"
"})()" "})()"
) )
return tab_cdp.eval(js) time.sleep(0.15)
tab_cdp.call("Input.insertText", text=date_str)
time.sleep(0.2)
tab_cdp.eval(
"(() => {"
f" const inp = document.querySelectorAll('input.z-datebox-inp')[{index}];"
" if (inp) inp.blur();"
" return true;"
"})()"
)
def wait_zk_datebox_committed(tab_cdp, index, date_str, timeout=5.0):
"""轮询 ZK 组件值已提交到服务端(读 zk.Widget 的 getText()),确认后再返回。
真实键入 + blur 触发 onChange AU 后服务端接受才会把新值同步回组件setAttr
因此 getText() 等于目标日期即代表服务端已确认。超时说明日期被页面拒绝
(大概率超出该页允许的查询范围),会抛超时报错而不是用旧范围静默查询。
"""
js_date = json.dumps(date_str)
wait_until(
tab_cdp,
"(() => {"
f" const inp = document.querySelectorAll('input.z-datebox-inp')[{index}];"
" if (!inp || typeof zk === 'undefined') return false;"
" const w = zk.Widget.$(inp);"
" if (!w || typeof w.getText !== 'function') return false;"
" const t = String(w.getText() ?? '').trim();"
f" return t === {js_date} || t.split(' ')[0] === {js_date};"
"})()",
f"ZK 日期组件[{index}]已提交 {date_str}",
timeout=timeout,
)
def click_query_button(tab_cdp): def click_query_button(tab_cdp):
@@ -421,12 +453,54 @@ def wait_query_done(tab_cdp):
) )
def collect_jiaojie_dan_ids(tab_cdp): _JAVA_MONTHS = {
"""收集交接单信息表里所有交接单号(从行复选框的 ewbsListNo 解析,按出现顺序)。 m: i
for i, m in enumerate(
[
"Jan",
"Feb",
"Mar",
"Apr",
"May",
"Jun",
"Jul",
"Aug",
"Sep",
"Oct",
"Nov",
"Dec",
],
start=1,
)
}
交接单号是 19 位纯数字,必须按文本保留,避免精度丢失。
def _parse_java_date(text):
"""解析 Java Date.toString()(如 'Mon Aug 03 05:50:46 CST 2026')为 date。
固定 token 切分,避免 strptime 对 %Z/%a/%b 的 locale 依赖。
""" """
return ( parts = (text or "").split()
if len(parts) != 6:
return None
month = _JAVA_MONTHS.get(parts[1])
if not month:
return None
try:
return date(int(parts[5]), month, int(parts[2]))
except (TypeError, ValueError):
return None
def collect_jiaojie_dan_ids(tab_cdp, start=None, end=None):
"""收集交接单信息表里的交接单号(从行复选框的 ewbsListNo 解析,按出现顺序)。
同时解析每行 value 里的 createdTime交接时间。传 start/enddate
只保留交接日期在 [start, end] 内的交接单;超期或日期解析失败的只跳过、不中断,
数量计入返回的 skipped。交接单号是 19 位纯数字,必须按文本保留,避免精度丢失。
返回 (ids, skipped)。
"""
rows = (
tab_cdp.eval( tab_cdp.eval(
"(() => {" "(() => {"
" const boxes = [...document.querySelectorAll('.z-listbox')];" " const boxes = [...document.querySelectorAll('.z-listbox')];"
@@ -439,14 +513,25 @@ def collect_jiaojie_dan_ids(tab_cdp):
" const rows = [...box.querySelectorAll('tr.z-listitem')]" " const rows = [...box.querySelectorAll('tr.z-listitem')]"
" .filter(tr => tr.querySelector('input[value*=\"ewbsListNo=\"]'));" " .filter(tr => tr.querySelector('input[value*=\"ewbsListNo=\"]'));"
" return rows.map(tr => {" " return rows.map(tr => {"
" const inp = tr.querySelector('input[value*=\"ewbsListNo=\"]');" " const v = (tr.querySelector('input[value*=\"ewbsListNo=\"]').value) || '';"
" const m = /ewbsListNo=(\\d+)/.exec(inp.value || '');" " const m = /ewbsListNo=(\\d+)/.exec(v);"
" return m ? m[1] : null;" " if (!m) return null;"
" const c = /createdTime=([^,\\]]+)/.exec(v);"
" return {id: m[1], created: c ? c[1] : null};"
" }).filter(Boolean);" " }).filter(Boolean);"
"})()" "})()"
) )
or [] or []
) )
ids, skipped = [], 0
for rec in rows:
created = _parse_java_date(rec.get("created")) if rec.get("created") else None
if start is not None and end is not None:
if created is None or not (start <= created <= end):
skipped += 1
continue
ids.append(rec["id"])
return ids, skipped
def dblclick_jiaojie_dan_row(tab_cdp, ewbs_no): def dblclick_jiaojie_dan_row(tab_cdp, ewbs_no):
@@ -947,25 +1032,41 @@ def anneng_expected_download_impl(force=False, date=None):
wait_query_page_ready(tab_cdp) wait_query_page_ready(tab_cdp)
print("✅ 进站交接单查询页已加载") print("✅ 进站交接单查询页已加载")
# 2) 设日期 + 查询 + 等加载 # 2) 设日期 + 提交确认 + 查询 + 等加载
print(">> 正在设置查询时间范围(直接输入日期)...") print(">> 正在设置查询时间范围(直接输入日期)...")
set_zk_datebox(tab_cdp, 0, start_str) set_zk_datebox(tab_cdp, 0, start_str)
# ZK 组件值异步提交:逐个设置并等服务端确认后再设下一个,
# 避免前一个 datebox 的 AU 响应与后一个的键入互相干扰(曾导致偶发提交超时);
# 未确认前点查询会按旧的时间范围查询(安能页面特性)。
wait_zk_datebox_committed(tab_cdp, 0, start_str)
set_zk_datebox(tab_cdp, 1, today_str) set_zk_datebox(tab_cdp, 1, today_str)
time.sleep(0.3) wait_zk_datebox_committed(tab_cdp, 1, today_str)
print(">> 正在执行查询 ...") print(">> 时间范围已提交确认,正在执行查询 ...")
click_query_button(tab_cdp) click_query_button(tab_cdp)
wait_query_done(tab_cdp) wait_query_done(tab_cdp)
print("✅ 查询完成,数据已加载") print("✅ 查询完成,数据已加载")
# 3) 收集交接单号并逐条导出 # 3) 收集交接单号并逐条导出
# 行的渲染可能比统计的小数点晚一拍:轮询等行出现;持续为空才视为无数据。 # 空结果页面会出现「没有任何相关信息」空态文案;若旧行还没被清掉,
# 靠交接日期范围过滤兜底。持续为空才视为无数据。
target_ids = [] target_ids = []
skipped_ids = 0
collect_deadline = time.monotonic() + 15.0 collect_deadline = time.monotonic() + 15.0
while time.monotonic() < collect_deadline: while time.monotonic() < collect_deadline:
target_ids = collect_jiaojie_dan_ids(tab_cdp) if tab_cdp.eval("document.body.innerText.includes('没有任何相关信息')"):
print(">> 查询完成但无交接单数据(页面空态),结束。")
return
target_ids, skipped_ids = collect_jiaojie_dan_ids(
tab_cdp, start=target.date(), end=target.date()
)
if target_ids: if target_ids:
break break
time.sleep(0.5) time.sleep(0.5)
if skipped_ids:
print(
f">> 忽略 {skipped_ids} 条超期/日期解析失败的交接单"
f"(超出 [{start_str} ~ {today_str}] 范围)"
)
print(f">> 共捕获 {len(target_ids)} 条交接单记录") print(f">> 共捕获 {len(target_ids)} 条交接单记录")
if not target_ids: if not target_ids:
print(">> ⚠ 没有交接单数据,结束。") print(">> ⚠ 没有交接单数据,结束。")

View File

@@ -172,6 +172,37 @@ def _resolve_export_frame(ws_frame):
return ws_frame.frame_locator('iframe[name="myFrame"]') return ws_frame.frame_locator('iframe[name="myFrame"]')
def _resolve_business_frame(page, probe_selector, label):
"""定位当前「可见」的业务 iframe返回 Frame用法与 FrameLocator 兼容)。
韵达门户每次切换菜单都会保留历史隐藏 iframedisplay:none因此不能按
`section iframe` 全量匹配会命中多个strict mode 报错)。这里遍历
page.frames取「可见」且「包含指定控件」的 frameprobe_selector 需是
目标页面独有的控件:应到=#startTime实到=#startDate导出服务=.datagrid-view2。
"""
print(f">> 正在定位业务 iframe{label}...")
deadline = time.monotonic() + 20
last_err = None
while time.monotonic() < deadline:
for frame in page.frames:
if frame is page.main_frame:
continue
try:
if not frame.frame_element().is_visible():
continue
except Exception as e:
last_err = e
continue
try:
if frame.locator(probe_selector).count() > 0:
print(f" ✅ 已定位业务 iframe{label}: {frame.url[:100]}")
return frame
except Exception as e:
last_err = e
page.wait_for_timeout(500)
raise RuntimeError(f"未定位到可见业务 iframe{label}: {last_err}")
def yunda_login(page): def yunda_login(page):
"""韵达自动登录:未登录则填充表单并提交,已登录则跳过。""" """韵达自动登录:未登录则填充表单并提交,已登录则跳过。"""
print(">> 正在检查韵达登录状态...") print(">> 正在检查韵达登录状态...")
@@ -270,7 +301,7 @@ def yunda_expected_download_impl(page, force=False, date=None):
yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"]) yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"])
print(">> 正在定位【进站交接单查询】iframe...") print(">> 正在定位【进站交接单查询】iframe...")
ws_frame = page.frame_locator("section iframe") ws_frame = _resolve_business_frame(page, "#startTime", "进站交接单查询")
ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000) ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000)
print("✅ 进站交接单查询页面就绪") print("✅ 进站交接单查询页面就绪")
@@ -527,7 +558,7 @@ def yunda_actual_download_impl(page, date=None):
yunda_smart_menu_click(page, ["报表管理", "扫描记录查询"]) yunda_smart_menu_click(page, ["报表管理", "扫描记录查询"])
print(">> 正在定位【扫描记录查询】iframe...") print(">> 正在定位【扫描记录查询】iframe...")
ws_frame = page.frame_locator("section iframe") ws_frame = _resolve_business_frame(page, "#startDate", "扫描记录查询")
ws_frame.locator( ws_frame.locator(
".no-records-found", has_text="没有找到匹配的记录" ".no-records-found", has_text="没有找到匹配的记录"
@@ -714,7 +745,7 @@ def _yunda_poll_and_download_tasks(page, export_times, download_dir, final_filen
print("\n>> 正在前往【导出服务】界面...") print("\n>> 正在前往【导出服务】界面...")
yunda_smart_menu_click(page, ["基础数据", "导出服务"]) yunda_smart_menu_click(page, ["基础数据", "导出服务"])
export_ws_frame = page.frame_locator("section iframe") export_ws_frame = _resolve_business_frame(page, ".datagrid-view2", "导出服务")
export_ws_frame.get_by_role("cell", name="模块名称", exact=True).wait_for( export_ws_frame.get_by_role("cell", name="模块名称", exact=True).wait_for(
state="visible", timeout=15000 state="visible", timeout=15000

View File

@@ -173,6 +173,33 @@ def _to_int(v):
return None 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): def _parse_time(v):
"""尽力解析多种时间格式为 datetime失败返回 None原始值在 raw 里)。""" """尽力解析多种时间格式为 datetime失败返回 None原始值在 raw 里)。"""
if v is None: if v is None:
@@ -233,13 +260,16 @@ def _read_business_dates():
_SQL_EXPECTED = """ _SQL_EXPECTED = """
INSERT INTO expected_record INSERT INTO expected_record
(site, waybill_no, handover_no, handover_pieces, order_pieces, business_date, raw) (site, waybill_no, handover_no, handover_pieces, order_pieces,
VALUES (%s,%s,%s,%s,%s,%s,%s) 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 ON CONFLICT (site, waybill_no) DO UPDATE SET
handover_no = EXCLUDED.handover_no, handover_no = EXCLUDED.handover_no,
handover_pieces = EXCLUDED.handover_pieces, handover_pieces = EXCLUDED.handover_pieces,
order_pieces = EXCLUDED.order_pieces, order_pieces = EXCLUDED.order_pieces,
business_date = COALESCE(EXCLUDED.business_date, expected_record.business_date), 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, raw = EXCLUDED.raw,
ingested_at = now() ingested_at = now()
""" """
@@ -294,19 +324,31 @@ def _ingest_expected(cur, site, business_date):
df = pd.read_excel(path, dtype=str).fillna("") df = pd.read_excel(path, dtype=str).fillna("")
df = df.drop_duplicates(subset=[cfg["exp_wb"]], keep="first") df = df.drop_duplicates(subset=[cfg["exp_wb"]], keep="first")
biz = _parse_date(business_date) 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 = [] rows = []
for r in df.to_dict("records"): for r in df.to_dict("records"):
wb = str(r.get(cfg["exp_wb"], "")).strip() wb = str(r.get(cfg["exp_wb"], "")).strip()
if not wb: if not wb:
continue 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( rows.append(
( (
site, site,
wb, wb,
str(r.get(cfg["exp_jd"], "")).strip() or None, hn,
_to_int(r.get(cfg["exp_qty"])), _to_int(r.get(cfg["exp_qty"])),
_to_int(r.get("录单件数")), _to_int(r.get("录单件数")),
biz, biz,
out_date,
batch_out_date.get(hn),
Jsonb(_raw_row(r)), Jsonb(_raw_row(r)),
) )
) )
@@ -463,6 +505,69 @@ def ingest_task(site, kind):
return total 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): def get_existing_handover_nos(site):
"""查该站点已落库的交接单号集合expected_record.handover_no """查该站点已落库的交接单号集合expected_record.handover_no
"提交导出任务前"去重:已落库的交接单号不再重复提交导出任务。 "提交导出任务前"去重:已落库的交接单号不再重复提交导出任务。
@@ -553,6 +658,8 @@ def main():
sys.exit(1) sys.exit(1)
total = ingest_task(site, kind) total = ingest_task(site, kind)
print(f">> [ingest-one] {site}/{kind} 入库 {total}") print(f">> [ingest-one] {site}/{kind} 入库 {total}")
elif cmd == "backfill-out-date":
backfill_out_date(site)
else: else:
print(__doc__) print(__doc__)
sys.exit(1) sys.exit(1)

View File

@@ -19,12 +19,16 @@ CREATE TABLE IF NOT EXISTS expected_record (
handover_pieces INTEGER, -- 交接件数(应到件数口径) handover_pieces INTEGER, -- 交接件数(应到件数口径)
order_pieces INTEGER, -- 录单件数 order_pieces INTEGER, -- 录单件数
business_date DATE, -- 业务日期(属性,非唯一键;读不到则 NULL business_date DATE, -- 业务日期(属性,非唯一键;读不到则 NULL
out_date DATE, -- 出库日(批次归属日口径;从 raw.出库时间 解析)
batch_out_date DATE, -- 批次归属日(同交接单号出库日众数,并列取较早)
raw JSONB NOT NULL, -- 站点原始全列key=原列名) raw JSONB NOT NULL, -- 站点原始全列key=原列名)
ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(), ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
UNIQUE (site, waybill_no) 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_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_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 ( CREATE TABLE IF NOT EXISTS actual_record (