- store.get_existing_handover_nos: query PG expected_record.handover_no - inject dedup skip before submitting export in zto/yunda/anneng/shunxin - shunxin reads RTS handover_no from waybill-list view (method 1) - force-redownload switch threaded via task_spec -> dispatch -> impl - schema: add idx_expected_handover index Co-Authored-By: Claude <noreply@anthropic.com>
4.8 KiB
4.8 KiB
应到数据「提交导出任务前」去重 — 实现总结
日期:2026-07-29 范围:顺心 / 中通 / 韵达 / 安能 4 站**应到(expected)**数据;百世与实到不在本次范围。
一、运行机制
在周期 / 手动触发下载时,于提交导出任务之前按交接单号判断该批应到数据是否已落库,已落库则跳过,从源头消除重复下载与重复落库。
flowchart TD
TRIG[周期调度 / 手动触发<br/>task_spec: site, kind, force] --> DISP[dispatch_task → 站点 download_impl]
DISP --> LOAD{force 强制重下?}
LOAD -- 是 --> EMPTY[existing = 空集]
LOAD -- 否 --> QRY[查 PG expected_record.handover_no]
QRY -- cpolar 失败 --> EMPTY
QRY -- 成功 --> SET[existing = 已落库交接单号集合]
EMPTY --> LOOP[遍历本次查询到的班次/交接单号]
SET --> LOOP
LOOP --> JUDGE{交接单号 ∈ existing?}
JUDGE -- 是 → 已落库 --> SKIP[⏭️ 跳过:不提交导出<br/>不 append export_times]
JUDGE -- 否 → 新单 --> EXP[提交导出任务 → 轮询下载 → 入库 UPSERT]
SKIP --> DONE{全部处理完}
EXP --> DONE
DONE --> FINAL{本次提交了新任务?}
FINAL -- 无 → 全跳过 --> BAIL[空兜底 return:不进下载轮询]
FINAL -- 有 --> POLL[轮询导出任务管理页 → 下载 → 入库]
核心要点:
- 去重数据源:PostgreSQL
expected_record.handover_no(已落库的权威记录),新增store.get_existing_handover_nos(site)查询。 - 判断时机:提交导出任务之前(循环内逐单判断),而非下载之后。
- 安全降级:PG 不可用 /
force=true→existing=空集→ 当作未落库 → 继续提交(宁可重复、绝不漏,UPSERT 兜底)。 - 空兜底:全部跳过时
export_times为空 → 直接return,不进下载轮询(避免下载数校验失败 / 空转超时)。 - force 开关:前端 checkbox(默认关)→
POST /tasks.force→ 一路透传到 impl;周期调度恒不 force。
二、force 强制重下透传链路
flowchart LR
UI[前端 checkbox<br/>forceRedownload] --> POST["POST /api/tasks<br/>{site,kind,force}"]
POST --> BFF[Next BFF 透传]
BFF --> TS["task_spec<br/>{site,kind,force}"]
TS --> DISP[dispatch_task]
DISP --> HDR["handler(ctx, force)"]
HDR --> DL["download(pg, force)"]
DL --> IMPL["impl(pg, force)"]
IMPL --> DEC{force?}
DEC -- 是 --> EMPTY2["existing = 空集<br/>强制重下,跳过去重"]
DEC -- 否 --> LOAD2[查 PG 加载 existing]
周期调度(
_enqueue_fetch)投递任务时不带force→ 默认不强制。
三、4 站点标识获取
| 站点 | 提交前标识 | 来源 |
|---|---|---|
| 中通 / 韵达 / 安能 | 交接单号(原有代码已读取) | DOM 列 / CDP 复选框 |
| 顺心 | 交接单号 RTS\d{3}WJ\d+ |
点"运单列表"后从界面读取(方式1) |
顺心"班次号"业务上等同交接单号;4 站统一用交接单号(= DB
handover_no)作去重键。
四、改动概览
后端 InboundVerify:
| 文件 | 改动 |
|---|---|
schema.sql |
+idx_expected_handover 索引 |
inbound_verify/store.py |
+get_existing_handover_nos(site)(含 cpolar 降级) |
inbound_verify/cli/server.py |
TaskRequest.force + 透传到 task_spec |
inbound_verify/runtime.py |
dispatch_task + 所有 handler 透传 force 到 download(impl) |
inbound_verify/sites/{zto,yunda,anneng,shunxin}.py |
impl 加 force + 提交导出前注入去重 + 空兜底 |
inbound_verify/sites/baishi.py |
force 形参兼容 |
前端 dashboard:
| 文件 | 改动 |
|---|---|
app/page.tsx |
forceRedownload state + checkbox + trigger/triggerPrimary/onTrigger 透传 force |
BFF
app/api/tasks/route.ts是 generic 透传,无需改动。
五、验证结论
4 站去重 + force 开关均实测通过:
| 站点 | 二次触发行为 | 结果 |
|---|---|---|
| 中通 | ...801 已入库 → 跳过 → 空兜底 return |
20s(vs 首次 59s) ✅ |
| 顺心 | 双账号 4 班次全跳过(RTS023WJ375478 等) |
✅ |
| 韵达 | ...82001 已入库 → 跳过 → 原有空兜底 |
✅ |
| 安能 | 4008242619171180544 已入库 → 跳过 |
5s(vs 4 分钟) ✅ |
| force | [去重] 强制重下,跳过去重 + 已入库的重新导出 |
✅ |
实施过程中的两个问题均已解决:
- 顺心 RTS 正则:
RTS\d+遇字母 W 停(只抓RTS023)→ 改RTS[A-Z0-9]+抓完整RTS023WJ375320。 - 韵达 force 偶发失败:韵达站点自身 UI 不稳定(
section iframe匹配到 2 个 + 弹窗遮挡),与去重/force 无关;换安能验证 force 成功。
静态检查:Black(py310)+ compileall + tsc 全绿。