From f710622a3d0d74d7e7c4644108749facae513446 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Wed, 29 Jul 2026 10:57:07 +0800 Subject: [PATCH] feat: dedup expected data by handover_no before export submit - 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 --- docs/2026-07-29-应到提交导出去重-实现总结.md | 102 +++++ docs/2026-07-29-应到提交导出去重实施计划.md | 399 +++++++++++++++++++ inbound_verify/cli/server.py | 5 +- inbound_verify/runtime.py | 26 +- inbound_verify/sites/anneng.py | 33 +- inbound_verify/sites/baishi.py | 2 +- inbound_verify/sites/shunxin.py | 47 ++- inbound_verify/sites/yunda.py | 26 +- inbound_verify/sites/zto.py | 31 +- inbound_verify/store.py | 19 + schema.sql | 1 + 11 files changed, 663 insertions(+), 28 deletions(-) create mode 100644 docs/2026-07-29-应到提交导出去重-实现总结.md create mode 100644 docs/2026-07-29-应到提交导出去重实施计划.md diff --git a/docs/2026-07-29-应到提交导出去重-实现总结.md b/docs/2026-07-29-应到提交导出去重-实现总结.md new file mode 100644 index 0000000..188ef1a --- /dev/null +++ b/docs/2026-07-29-应到提交导出去重-实现总结.md @@ -0,0 +1,102 @@ +# 应到数据「提交导出任务前」去重 — 实现总结 + +> 日期：2026-07-29 +> 范围：顺心 / 中通 / 韵达 / 安能 4 站**应到（expected）**数据；百世与实到不在本次范围。 + +## 一、运行机制 + +在周期 / 手动触发下载时，于**提交导出任务之前**按交接单号判断该批应到数据是否已落库，已落库则跳过，从源头消除重复下载与重复落库。 + +```mermaid +flowchart TD + TRIG[周期调度 / 手动触发
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[⏭️ 跳过：不提交导出
不 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 强制重下透传链路 + +```mermaid +flowchart LR + UI[前端 checkbox
forceRedownload] --> POST["POST /api/tasks
{site,kind,force}"] + POST --> BFF[Next BFF 透传] + BFF --> TS["task_spec
{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 = 空集
强制重下，跳过去重"] + 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 | `[去重] 强制重下，跳过去重` + 已入库的重新导出 | ✅ | + +实施过程中的两个问题均已解决： +1. **顺心 RTS 正则**：`RTS\d+` 遇字母 W 停（只抓 `RTS023`）→ 改 `RTS[A-Z0-9]+` 抓完整 `RTS023WJ375320`。 +2. **韵达 force 偶发失败**：韵达站点自身 UI 不稳定（`section iframe` 匹配到 2 个 + 弹窗遮挡），与去重/force 无关；换安能验证 force 成功。 + +静态检查：Black（py310）+ compileall + tsc 全绿。 diff --git a/docs/2026-07-29-应到提交导出去重实施计划.md b/docs/2026-07-29-应到提交导出去重实施计划.md new file mode 100644 index 0000000..3c2dbae --- /dev/null +++ b/docs/2026-07-29-应到提交导出去重实施计划.md @@ -0,0 +1,399 @@ +# 应到数据「提交导出任务前」去重 实施计划 v2 + +> **执行约定：** 本仓库无 pytest，验证靠实跑站点流程 + DB 核对（见第 7 节）。 +> 本仓库约定 **不自动提交**；所有改动落地后等用户明确说"提交"再 commit/push。 +> 本次会话额外约定：未获用户明确指示前不动代码、不提交。 + +**Goal：** 在周期性自动落库场景下，于「提交导出任务」之前，按**交接单号**（顺心=运单列表界面里的交接单号）判断该批应到数据是否已落库，已落库则跳过提交导出任务，从源头消除重复下载与重复落库；并提供一个"强制重下"开关（默认关）兜底。 + +**Architecture：** +- 去重数据源 = PostgreSQL `expected_record.handover_no`（已存在字段，权威）；`store.py` 新增 `get_existing_handover_nos(site)`（含 cpolar 降级）。 +- 在 4 站**应到**下载循环内、提交导出动作之前注入"命中已落库则 `continue`"。 +- `force` 开关经 `task_spec` → `dispatch_task` → handler → 各站 `download(page, force)` → `impl(page, force)` 透传；`force=True` 时跳过去重。周期 job 默认不 force。 + +**Tech Stack：** Python 3.10+ / Playwright（网页 3 站）/ 裸 CDP（安能）/ psycopg / SQLite；前端 Next.js 16 + React 19。 + +## Global Constraints + +- Python 一律 `.venv`；改完任何 `.py` 必须跑 `.venv/Scripts/python.exe -m black inbound_verify`。 +- 改完跑 `.venv/Scripts/python.exe -m py_compile inbound_verify` 自检。 +- **不自动提交/推送**（覆盖全局 auto-push 默认）。 +- **不动 `export_times` 时间容差（≤40s）匹配机制**（CLAUDE.md 约定）。 +- 安能 CDP 驱动，**绝不用 `Page.reload`**。 +- `config.yaml` 已 gitignore，不提交真实凭据。 +- 前端是 **Next.js 16（有 breaking changes）**，写前端代码前先查 `node_modules/next/dist/docs/`。 +- 行号基于 2026-07-29 快照，实现时以当前代码为准、就近定位。 + +## 1. 已定决策（v1 审核反馈） + +| 决策 | 结论 | +|---|---| +| A. 去重数据源 | **查 PostgreSQL `expected_record.handover_no`** | +| B. 范围 | **本次只做应到（expected）**；实到/百世不动 | +| C. 强制开关 | **加 force 开关，默认不强制重下** | +| 顺心标识 | **方式1：点"运单列表"后、点导出前，从运单列表界面读交接单号**（与其他 3 站统一用交接单号去重） | + +## 2. 背景与问题根源 + +周期链路：`fetch_schedule(IntervalTrigger) → task_queue → worker → dispatch_task → handler → 提交导出+下载 → _persist_to_db(UPSERT)`。 +- DB 已幂等（`expected_record` 按 `(site, waybill_no)` UPSERT）。 +- 但 `dispatch_task` 调 handler 前**无"是否需要下载"判断**，周期触发重复"提交导出→下载→解析"。 +- 本方案在「提交导出任务」前按交接单号去重，从源头省掉重复下载。 + +## 3. 各站探索结论（注入点） + +| 站点 | 文件 | 提交导出位置 | 提交前标识 | 来源 | +|---|---|---|---|---| +| 中通 | `sites/zto.py` | `zto_expected_download_impl` L264 | ✅ 已有 `handover_no`（L251） | 主表行 `td.nth(3)` 正则 18 位 | +| 韵达 | `sites/yunda.py` | `yunda_expected_download_impl` L342 | ✅ 已有 `raw_no`（L291） | 列表行 `td.nth(1)` | +| 安能 | `sites/anneng.py` | 主循环 L927（逐条） | ✅ 已有 `ewbs_no`（L907-913） | CDP 复选框 `ewbsListNo=` 正则 19 位 | +| 顺心 | `sites/shunxin.py` | L319 点导出 | 🆕 方式1：L316 后读运单列表界面交接单号 | DOM 待实勘 | + +**顺心方式1关键事实**（已探明）：raw 里「班次号」「交接单号」都有，且一个交接单 = 一个班次 = 多条运单；交接单号即入库 `handover_no`，与另 3 站同键。 + +## 4. 文件结构 + +| 文件 | 改动 | +|---|---| +| `schema.sql` | 加 `expected_record(site, handover_no)` 索引 | +| `inbound_verify/store.py` | 新增 `get_existing_handover_nos(site)` | +| `inbound_verify/cli/server.py` | `TaskRequest` 加 `force`；`create_task` 透传 force（周期不 force） | +| `inbound_verify/runtime.py` | `dispatch_task` 读 force 传 handler；所有 handler 加 `force` 形参并透传到 `download_func` | +| `inbound_verify/sites/zto.py` | `download/impl` 加 `force`；应到循环注入去重 + 空兜底 | +| `inbound_verify/sites/yunda.py` | 同上（空兜底已存在） | +| `inbound_verify/sites/anneng.py` | 同上 | +| `inbound_verify/sites/shunxin.py` | `download/impl` 加 `force`；方式1：点运单列表后读交接单号去重 + 退回 + 空兜底 | +| `dashboard/app/page.tsx` | 加 `forceRedownload` state + checkbox；`trigger/triggerPrimary` 透传 force | + +> 注：4 站的 actual（实到）`download` 入口也统一加 `force=False` 形参（接收但不用，仅让 `_web_handler` 的统一调用成立），actual impl 不改。 + +## 5. 任务分解 + +### Task 1：schema.sql 加索引 + +**Files:** Modify `schema.sql`（`idx_expected_site_date` 之后） + +```sql +CREATE INDEX IF NOT EXISTS idx_expected_handover ON expected_record (site, handover_no); +``` + +**验证：** `.venv/Scripts/python.exe -m inbound_verify.store init`（幂等）。 + +--- + +### Task 2：store.py 新增查已落库交接单号集合 + +**Files:** Modify `inbound_verify/store.py`（`ingest_task` 之后） + +**Produces:** `get_existing_handover_nos(site: str) -> set[str]` + +```python +def get_existing_handover_nos(site): + """查该站点已落库的交接单号集合（expected_record.handover_no）。 + 供"提交导出任务前"去重：已落库的不再重复提交导出。 + PG 不可用（cpolar 抖动等）时返回空集 + 告警，调用方按"未确认存在"处理 + （继续提交导出，UPSERT 兜底，绝不因去重查询失败而漏数据）。""" + try: + with _connect(_load_pg_config()["dbname"]) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT handover_no FROM expected_record " + "WHERE site=%s AND handover_no IS NOT NULL AND handover_no <> ''", + (site,), + ) + return {str(r[0]).strip() for r in cur.fetchall()} + except Exception as e: + print(f">> [去重] 查询已落库交接单号失败（{site}），本次不去重: {e}") + return set() +``` + +**验证：** `.venv/Scripts/python.exe -c "from inbound_verify import store; print(len(store.get_existing_handover_nos('中通')))"` 不抛异常即过。 + +--- + +### Task 3：force 开关后端骨架（server + runtime） + +让 `force` 从 `task_spec` 一路透传到各站 `download(page, force)`。`with_retry` 的 `flow` 是零参 lambda，force 经闭包捕获，**with_retry 不动**。 + +**(a) `server.py` TaskRequest + create_task：** + +```python +class TaskRequest(BaseModel): + site: str + kind: str + force: bool = False # 新增：强制重下（忽略已落库去重），默认关 +``` + +```python +# create_task 内 +task_queue.put((task_id, {"site": req.site, "kind": req.kind, "force": req.force})) +``` + +> `_enqueue_fetch`（L133 周期投递）**保持不变**（不带 force → 默认 False）✅。 + +**(b) `runtime.py` dispatch_task 透传 force：** + +```python +def dispatch_task(ctx, task_spec): + site = task_spec.get("site") + kind = task_spec.get("kind") + force = bool(task_spec.get("force", False)) # 新增 + ... + try: + ret = handler(ctx, force) # 改：原 handler(ctx) +``` + +**(c) `runtime.py` 所有 handler 加 force 形参：** + +`_web_handler`： +```python + def handler(ctx, force=False): + pg = ctx.pages_map[site] + if isinstance(pg, list): # 顺心双账号 + return download_func(pg, foreground=ctx.foreground, force=force) + if ctx.foreground: + pg.bring_to_front() + return download_func(pg, force=force) +``` + +`_site_undelivered_handler`： +```python + def handler(ctx, force=False): + exp_ok = TASK_HANDLERS[(site, "expected")](ctx, force) is not False + act_ok = (TASK_HANDLERS[(site, "actual")](ctx, force) is not False) if exp_ok else False + ... +``` + +安能 expected/actual 与 compare（签名兼容即可）： +```python +("安能", "expected"): lambda ctx, force=False: anneng.anneng_expected_download(force=force), +("安能", "actual"): lambda ctx, force=False: anneng.anneng_actual_download(force=force), +("__compare__", "compare"): lambda ctx, force=False: (compare.main() or True), +``` + +**验证：** `py_compile` 通过；服务重启后 `POST /tasks {site,kind,force:true}` 不报 TypeError（此时各站 `download` 的 force 形参由 Task 4-7 补齐，连续实施）。 + +--- + +### Task 4：中通 zto.py（download/impl 加 force + 去重） + +**Files:** Modify `inbound_verify/sites/zto.py` + +**(a) 入口与 impl 加 force（闭包透传，with_retry 不动）：** +```python +def zto_expected_download(page, force=False): + return with_retry( + "中通", "应到", + lambda: zto_expected_download_impl(page, force=force), + lambda: zto_reset(page), + ) + +def zto_expected_download_impl(page, force=False): + ... +# zto_actual_download / zto_actual_download_impl 同样加 force=False 形参（actual 不用 force，仅兼容） +``` + +**(b) 循环前加载已落库集合**（L241 print 之后、L243 `for` 之前）： +```python + # 【去重】加载本站已落库交接单号；force=True 或查询失败时 existing=空集（不去重） + if force: + existing = set() + print(">> [去重] 强制重下，跳过去重。") + else: + try: + from inbound_verify import store + existing = store.get_existing_handover_nos("中通") + except Exception as _e: + existing = set() + print(f">> [去重] 加载失败，本次不去重: {_e}") +``` + +**(c) 循环内命中跳过**（L252 print 之后、L254 `row.dblclick()` 之前）： +```python + print(f" -> 当前交接单号：{handover_no}") + if handover_no in existing: + print(f" ⏭️ 交接单号 {handover_no} 已落库，跳过提交导出。") + continue + row.dblclick() +``` + +**(d) 空列表兜底**（L305 循环后、进入 `_zto_poll_and_download_tasks` 之前）： +```python + if not export_times: + print(">> 本次无新交接单需导出（全部已落库或无数据），结束。") + return +``` + +--- + +### Task 5：韵达 yunda.py（同构） + +**Files:** Modify `inbound_verify/sites/yunda.py` + +(a) `yunda_expected_download(page, force=False)` + impl 加 force（actual 同理加形参）；(b) 循环前加载 existing（同 Task 4b，站点"韵达"）； + +**(c) 循环内命中跳过**（L291 `raw_no = ...` 之后、L293 `# 跳过已绑定的交接单` 之前）： +```python + raw_no = current_row.locator("td").nth(1).inner_text().strip() + if raw_no in existing: + print(f" ⏭️ 交接单号 {raw_no} 已落库，跳过提交导出。") + continue + # 跳过已绑定的交接单 + bind_status = current_row.locator("td").nth(2).inner_text().strip() +``` + +(d) 空列表兜底 **已存在**（L389-392），无需新增。 + +--- + +### Task 6：安能 anneng.py（CDP） + +**Files:** Modify `inbound_verify/sites/anneng.py` + +(a) `anneng_expected_download(force=False)` + impl 加 force（actual 同理）； + +**(b) 主流程加载 existing**（`anneng_expected_download_impl` 内、L907 收集 `target_ids` 前，与 `export_times = []`（L882）并列，同 Task 4b，站点"安能"）； + +**(c) 主循环命中跳过**（L919 `for` 内、L920 print 之后、L921 `activate_tab` 之前）： +```python + for i, ewbs_no in enumerate(target_ids, start=1): + print(f" ⏳ [{i}/{len(target_ids)}] 交接单号 {ewbs_no}") + if ewbs_no in existing: + print(f" ⏭️ 交接单号 {ewbs_no} 已落库，跳过。") + continue + activate_tab(tab_cdp, "交接单信息") + ... +``` + +**(d) 空列表兜底**（主循环后、进入 `poll_and_download_tasks` 之前）： +```python + if not export_times: + print(">> 本次无新交接单需导出（全部已落库或无数据），结束。") + return +``` + +--- + +### Task 7：顺心 shunxin.py（方式1 + 实勘） + +顺心流程：L315 点"运单列表" → L316 等"运单查询"label（**运单列表界面**） → L319 点"导出"。方式1 在 L316 之后、L319 之前读交接单号。 + +**Step 1（实勘，不改业务逻辑）：** 确认运单列表界面里**交接单号的 DOM 选择器**（哪个元素/列）。实勘方式二选一（待用户同意）： +- 方式 A：临时在 L316 后加调试打印（dump 运单列表界面关键 DOM 文本），跑一次顺心应到，从日志定位选择器，再删调试代码。 +- 方式 B：debug 模式（`config.yaml` debug.target_site=顺心，CDP 9223）单独挂载，用 Playwright CLI 观察。 + +**Step 2（注入，选择器 `` 确认后替换）：** + +(a) 入口与 impl 加 force： +```python +def shunxin_expected_download(pages, foreground=True, force=False): + return with_retry( + "顺心", "应到", + lambda: shunxin_expected_download_impl(pages, foreground=foreground, force=force), + lambda: shunxin_reset(pages), + ) + +def shunxin_expected_download_impl(pages, foreground=True, force=False): + ... +# shunxin_actual_download / impl 同样加 force=False 形参（actual 不用） +``` + +(b) 循环前加载 existing（同 Task 4b，站点"顺心"；两账号共享同一 `existing`）。 + +**(c) 循环内：点运单列表 → 读交接单号 → 命中则退回跳过**（L315-316 之后、L319 点导出之前）： +```python + for i in range(count): + print(f" ⏳ 正在处理第 {i+1}/{count} 个班次...") + + waybill_btns.nth(i).click() + page.locator("label[title='运单查询']").wait_for(state="visible") + + # 【方式1】运单列表界面已加载，读交接单号 → 已落库则退回列表跳过 + handover_no = page.locator("").first.inner_text().strip() + if handover_no in existing: + print(f" ⏭️ 交接单号 {handover_no} 已落库，跳过提交导出。") + page.get_by_role("tab", name="车辆点到").click() # 退回列表（复用 L336） + page.wait_for_timeout(500) + continue + + # 4. 执行导出流程 + page.get_by_role("button", name="export 导出").click() + ... +``` + +(d) 空列表兜底（L339 循环后、进入导出任务管理页轮询之前）： +```python + if not export_times: + print(">> 本次无新班次需导出（全部已落库或无数据），结束。") + return +``` + +--- + +### Task 8：前端 force 开关（dashboard） + +**Files:** Modify `dashboard/app/page.tsx`（BFF `app/api/tasks/route.ts` 是 generic 透传，**不用改**） + +**(a) 加 state（L21 附近）：** +```tsx +const [forceRedownload, setForceRedownload] = useState(false); +``` + +**(b) `trigger` 加 force 形参并写入 body（L27-50）：** +```tsx + const trigger = useCallback( + async (site: string, kind: string, label: string, force?: boolean) => { + ... + body: JSON.stringify({ site, kind, force: !!force }), + ... + }, + [refreshTasks], + ); +``` + +**(c) `triggerPrimary` 透传（L58-63）：** +```tsx + const triggerPrimary = useCallback( + async (cfg: SiteConfig) => { + await trigger(cfg.name, cfg.primaryKind, `${cfg.name}·获取未到`, forceRedownload); + }, + [trigger, forceRedownload], + ); +``` + +**(d) UI：在配置区/顶部加 checkbox（默认不勾）：** +```tsx + +``` + +> 仅"获取未到"主按钮透传 force；周期抓取不经前端、恒不 force。 + +**验证：** 前端勾选 → 触发 → 网络面板看到 POST `/api/tasks` body 含 `force:true`；后端日志 `强制重下，跳过去重`。 + +## 6. 关键注意事项（陷阱） + +- **跳过的交接单号绝不 `export_times.append`**：否则 `len(export_times)` > 实际提交数 → 下载数校验失败。所有 `continue` 都在 append 之前。 +- **全部跳过时必须 `return`**：`export_times` 为空时不进导出任务管理页轮询。 +- **顺心方式1跳过要退回**：点进运单列表后命中已落库，需点"车辆点到"tab 退回再 `continue`（复用 L336）。 +- **PG 降级只防漏不防重**：查询失败 = 空集 = 当作未存在 = 继续提交；UPSERT 兜底。 +- **不动 `export_times` 时间容差匹配**。 +- actual/百世 `download` 只加 `force` 形参兼容，impl 不加去重。 + +## 7. 验证（无 pytest） + +1. **单站联调**（`config.yaml` debug 单站）：首次新单正常下载+入库；再触发同范围 → 已入库的全部 `⏭️ 跳过`，`export_times` 空，直接 return。 +2. **force 开关**：勾选"强制重下" → 已落库的也重新提交导出（日志 `强制重下，跳过去重`）。 +3. **DB 核对**：`SELECT site, handover_no, COUNT(*) FROM expected_record GROUP BY site, handover_no` 无翻倍。 +4. **cpolar 降级**：断 PG → `get_existing_handover_nos` 返回空集 + 告警，流程仍正常下载（不漏）。 +5. **Black + py_compile**；前端 `npm run build` 或 dev 热更无类型错。 + +## 8. 实施顺序与依赖 + +Task 1 → 2 → 3（骨架，此时各站 download 的 force 形参在 4-7 补）→ 4/5/6/7（各站，连续做完让链路自洽）→ 8（前端）。顺心 Task 7 的 Step1 实勘需在运行的服务上操作，实施时与用户协调时机。 diff --git a/inbound_verify/cli/server.py b/inbound_verify/cli/server.py index 709978b..1ae282d 100644 --- a/inbound_verify/cli/server.py +++ b/inbound_verify/cli/server.py @@ -187,6 +187,9 @@ app = FastAPI(title="InboundVerify 服务端", lifespan=lifespan) class TaskRequest(BaseModel): site: str kind: str + force: bool = ( + False # 强制重下：忽略已落库去重，重新提交所有班次/交接单的导出任务（默认关） + ) @app.post("/tasks") @@ -200,7 +203,7 @@ def create_task(req: TaskRequest): if (req.site, req.kind) not in TASK_HANDLERS: raise HTTPException(status_code=400, detail=f"无效任务: {req.site}/{req.kind}") task_id = state_store.create_task(req.site, req.kind) - task_queue.put((task_id, {"site": req.site, "kind": req.kind})) + task_queue.put((task_id, {"site": req.site, "kind": req.kind, "force": req.force})) return {"task_id": task_id} diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index 19ffe81..c7de688 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -454,14 +454,14 @@ def _web_handler(site, download_func): 焦点；交互模式置顶便于调试。顺心是 page 列表，置顶标志透传给 shunxin_download。 """ - def handler(ctx): + def handler(ctx, force=False): pg = ctx.pages_map[site] if isinstance(pg, list): # 顺心双账号：置顶与否交给 shunxin_download 在逐账号循环里按 foreground 决定 - return download_func(pg, foreground=ctx.foreground) + return download_func(pg, foreground=ctx.foreground, force=force) if ctx.foreground: pg.bring_to_front() - return download_func(pg) + return download_func(pg, force=force) return handler @@ -470,11 +470,13 @@ def _site_undelivered_handler(site): """4 站未到：下应到+实到 → 比对写 downloads/<站>-未到数据.xlsx。 任一下载失败 → 清掉旧未到文件、返回 False（前端不展示陈旧未到）。""" - def handler(ctx): + def handler(ctx, force=False): # 各站下载入口约定返回 True/False；顺心历史返回 None（视为成功，与 dispatch 一致） - exp_ok = TASK_HANDLERS[(site, "expected")](ctx) is not False + exp_ok = TASK_HANDLERS[(site, "expected")](ctx, force) is not False act_ok = ( - (TASK_HANDLERS[(site, "actual")](ctx) is not False) if exp_ok else False + (TASK_HANDLERS[(site, "actual")](ctx, force) is not False) + if exp_ok + else False ) if exp_ok and act_ok: return compare.write_site_file(site) @@ -502,10 +504,14 @@ TASK_HANDLERS = { ("韵达", "expected"): _web_handler("韵达", yunda.yunda_expected_download), ("韵达", "actual"): _web_handler("韵达", yunda.yunda_actual_download), ("韵达", "undelivered"): _site_undelivered_handler("韵达"), - ("安能", "expected"): lambda ctx: anneng.anneng_expected_download(), - ("安能", "actual"): lambda ctx: anneng.anneng_actual_download(), + ("安能", "expected"): lambda ctx, force=False: anneng.anneng_expected_download( + force=force + ), + ("安能", "actual"): lambda ctx, force=False: anneng.anneng_actual_download( + force=force + ), ("安能", "undelivered"): _site_undelivered_handler("安能"), - ("__compare__", "compare"): lambda ctx: (compare.main() or True), + ("__compare__", "compare"): lambda ctx, force=False: (compare.main() or True), } @@ -594,7 +600,7 @@ def dispatch_task(ctx, task_spec): if handler is None: return (state_store.TASK_FAILED, f"未知任务: {site}/{kind}") try: - ret = handler(ctx) + ret = handler(ctx, bool(task_spec.get("force", False))) if ret is False: return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）") _record_business_date(site, kind) diff --git a/inbound_verify/sites/anneng.py b/inbound_verify/sites/anneng.py index 4475acc..02781c1 100644 --- a/inbound_verify/sites/anneng.py +++ b/inbound_verify/sites/anneng.py @@ -855,13 +855,15 @@ def _load_query_days(): return max(1, days) -def anneng_expected_download(): +def anneng_expected_download(force=False): """安能：应到货物数据下载（内部含异常兜底重试，路由层无感）。""" - return with_retry("安能", "应到", anneng_expected_download_impl, anneng_reset) + return with_retry( + "安能", "应到", lambda: anneng_expected_download_impl(force=force), anneng_reset + ) -def anneng_expected_download_impl(): +def anneng_expected_download_impl(force=False): """安能：应到货物数据（运单信息）下载，完整流程（单次执行，无重试；供自动化测试用）。""" print("\n▶ 开始执行【安能 - 应到货物数据下载】任务 ...") download_dir = DOWNLOAD_DIR @@ -880,6 +882,20 @@ def anneng_expected_download_impl(): main_cdp = find_main_page_cdp() export_times = [] + + # 【去重】加载本站已落库交接单号；force=True 或查询失败时 existing=空集（不去重） + if force: + existing = set() + print(">> [去重] 强制重下，跳过去重。") + else: + try: + from inbound_verify import store + + existing = store.get_existing_handover_nos("安能") + except Exception as _e: + existing = set() + print(f">> [去重] 加载已落库交接单号失败，本次不去重: {_e}") + try: # 1) 确认主页就绪并导航到“进站交接单查询” wait_home_ready(main_cdp) @@ -918,6 +934,10 @@ def anneng_expected_download_impl(): for i, ewbs_no in enumerate(target_ids, start=1): print(f" ⏳ [{i}/{len(target_ids)}] 交接单号 {ewbs_no}") + # 【去重】已落库则跳过：不双击、不导出、不 append export_times + if ewbs_no in existing: + print(f" ⏭️ 交接单号 {ewbs_no} 已落库，跳过。") + continue activate_tab(tab_cdp, "交接单信息") time.sleep(0.3) if not dblclick_jiaojie_dan_row(tab_cdp, ewbs_no): @@ -936,6 +956,11 @@ def anneng_expected_download_impl(): close_tab_by_label(main_cdp, "进站交接单查询") time.sleep(0.8) + # 【去重兜底】全部已落库/无数据 → 无导出任务，查询 tab 已关，跳过下载段 + if not export_times: + print(">> 本次无新交接单需导出（全部已落库或无数据），结束。") + return True + # 5) 打开导出下载 tab，轮询并下载 print(">> 打开【导出下载】tab ...") export_cdp = ensure_tab_open(main_cdp, "导出下载", EXPORT_TAB_URL_HINT) @@ -1233,7 +1258,7 @@ def _save_actual(rows, download_dir): print("====================================================") -def anneng_actual_download(): +def anneng_actual_download(force=False): """安能：实到数据下载（内部含异常兜底重试，路由层无感）。""" return with_retry("安能", "实到", anneng_actual_download_impl, anneng_reset) diff --git a/inbound_verify/sites/baishi.py b/inbound_verify/sites/baishi.py index d0b145e..7027d06 100644 --- a/inbound_verify/sites/baishi.py +++ b/inbound_verify/sites/baishi.py @@ -137,7 +137,7 @@ def _close_tab(page, tab_name): print(f" ⚠️ 关闭标签页【{tab_name}】时出错: {e}") -def baishi_download_undelivered_data(page): +def baishi_download_undelivered_data(page, force=False): """百世：一键提取应到未到（当日未扫）数据（内部含异常兜底重试，路由层无感）。""" return with_retry( diff --git a/inbound_verify/sites/shunxin.py b/inbound_verify/sites/shunxin.py index d78511c..bca5418 100644 --- a/inbound_verify/sites/shunxin.py +++ b/inbound_verify/sites/shunxin.py @@ -149,7 +149,7 @@ def shunxin_merge_final(kind, tags): pass -def shunxin_expected_download(pages, foreground=True): +def shunxin_expected_download(pages, foreground=True, force=False): """顺心：应到货物数据下载（双账号/双归属地，内部含异常兜底重试与数据融合）。 pages 为该站点的 page 列表（双账号在同一窗口的各一个标签页）。 @@ -175,7 +175,9 @@ def shunxin_expected_download(pages, foreground=True): ok = with_retry( f"顺心-{tag}", "应到", - lambda p=pg, t=tag: shunxin_expected_download_impl(p, out_tag=t), + lambda p=pg, t=tag, f=force: shunxin_expected_download_impl( + p, out_tag=t, force=f + ), lambda p=pg: shunxin_reset(p), ) if not ok: @@ -185,7 +187,7 @@ def shunxin_expected_download(pages, foreground=True): return True -def shunxin_expected_download_impl(page, out_tag=""): +def shunxin_expected_download_impl(page, out_tag="", force=False): """顺心：应到货物数据下载（单次执行，无重试；供自动化测试探测原始结果用）。 out_tag 为归属地标签时，合并产物命名为「顺心-{out_tag}-应到货物数据.xlsx」， @@ -309,12 +311,44 @@ def shunxin_expected_download_impl(page, out_tag=""): count = waybill_btns.count() print(f">> 共发现 {count} 个班次需要导出。") + # 【去重】加载本站已落库交接单号；force=True 或查询失败时 existing=空集（不去重）。 + # 两账号共享同一集合（班次号/交接单号跨归属地不重叠）。 + if force: + existing = set() + print(">> [去重] 强制重下，跳过去重。") + else: + try: + from inbound_verify import store + + existing = store.get_existing_handover_nos("顺心") + except Exception as _e: + existing = set() + print(f">> [去重] 加载已落库交接单号失败，本次不去重: {_e}") + for i in range(count): print(f" ⏳ 正在处理第 {i+1}/{count} 个班次...") waybill_btns.nth(i).click() page.locator("label[title='运单查询']").wait_for(state="visible") + # 【方式1】运单列表界面已加载，读交接单号（RTS 开头）→ 已落库则退回列表跳过。 + # 交接单号格式 RTS\d{3}WJ\d+（如 RTS023WJ374837），用 [A-Z0-9]+ 连续匹配整段。 + # 读不到（DOM 变动/未渲染）则 handover_no 为空 → 不跳过（安全降级，继续导出）。 + handover_no = "" + try: + _txt = page.locator("text=/RTS\\d+/").first.inner_text(timeout=3000) + _m = re.search(r"RTS[A-Z0-9]+", _txt) + if _m: + handover_no = _m.group(0) + except Exception: + pass + print(f" -> 运单列表交接单号：{handover_no or '(未读到，不去重)'}") + if handover_no and handover_no in existing: + print(f" ⏭️ 交接单号 {handover_no} 已落库，跳过提交导出。") + page.get_by_role("tab", name="车辆点到").click() + page.wait_for_timeout(500) + continue + # 4. 执行导出流程 page.get_by_role("button", name="export 导出").click() @@ -342,6 +376,11 @@ def shunxin_expected_download_impl(page, out_tag=""): _close_tab(page, "运单列表") _close_tab(page, "车辆点到") + # 【去重兜底】全部已落库/无数据 → 无导出任务，标签页已关，跳过下载轮询 + if not export_times: + print(">> 本次无新班次需导出（全部已落库或无数据），结束。") + return True + # 6. 前往数据导出页面去下载 print(">> 正在前往【数据导出】界面...") page.locator("a[href='/dataExport']").click() @@ -477,7 +516,7 @@ def shunxin_expected_download_impl(page, out_tag=""): return False -def shunxin_actual_download(pages, foreground=True): +def shunxin_actual_download(pages, foreground=True, force=False): """顺心：实到货物数据下载（双账号/双归属地，内部含异常兜底重试与数据融合）。 与 shunxin_expected_download 同构：读归属地 → 去重校验 → 顺序各账号下载 → diff --git a/inbound_verify/sites/yunda.py b/inbound_verify/sites/yunda.py index c6e1a89..9245e17 100644 --- a/inbound_verify/sites/yunda.py +++ b/inbound_verify/sites/yunda.py @@ -169,18 +169,18 @@ def yunda_smart_menu_click(page, menu_path): page.wait_for_timeout(1000) -def yunda_expected_download(page): +def yunda_expected_download(page, force=False): """韵达：应到货物数据下载（内部含异常兜底重试，路由层无感）。""" return with_retry( "韵达", "应到", - lambda: yunda_expected_download_impl(page), + lambda: yunda_expected_download_impl(page, force=force), lambda: yunda_reset(page), ) -def yunda_expected_download_impl(page): +def yunda_expected_download_impl(page, force=False): """韵达：应到货物数据下载（单次执行，无重试；供自动化测试探测原始结果用）。""" print("\n▶ 开始执行【韵达 - 应到货物数据下载】任务...") @@ -283,6 +283,19 @@ def yunda_expected_download_impl(page): row_count = main_rows.count() print(f">> 当前视窗共捕获到活跃交接单记录: {row_count} 条") + # 【去重】加载本站已落库交接单号；force=True 或查询失败时 existing=空集（不去重） + if force: + existing = set() + print(">> [去重] 强制重下，跳过去重。") + else: + try: + from inbound_verify import store + + existing = store.get_existing_handover_nos("韵达") + except Exception as _e: + existing = set() + print(f">> [去重] 加载已落库交接单号失败，本次不去重: {_e}") + # 5. 逐行双击并提交导出 for i in range(row_count): print(f" ⏳ 正在处理第 {i+1}/{row_count} 个交接单模块...") @@ -290,6 +303,11 @@ def yunda_expected_download_impl(page): raw_no = current_row.locator("td").nth(1).inner_text().strip() + # 【去重】已落库的交接单号不再提交导出任务 + if raw_no in existing: + print(f" ⏭️ 交接单号 {raw_no} 已落库，跳过提交导出。") + continue + # 跳过已绑定的交接单 bind_status = current_row.locator("td").nth(2).inner_text().strip() print(f" -> 交接单号: {raw_no} [绑定状态: {bind_status}]") @@ -404,7 +422,7 @@ def yunda_expected_download_impl(page): return False -def yunda_actual_download(page): +def yunda_actual_download(page, force=False): """韵达：实到货物数据下载（内部含异常兜底重试，路由层无感）。""" return with_retry( diff --git a/inbound_verify/sites/zto.py b/inbound_verify/sites/zto.py index efc0d9f..3d89b0f 100644 --- a/inbound_verify/sites/zto.py +++ b/inbound_verify/sites/zto.py @@ -109,18 +109,18 @@ def zto_smart_menu_click(page, menu_path): page.wait_for_timeout(1000) -def zto_expected_download(page): +def zto_expected_download(page, force=False): """中通：应到货物数据下载（内部含异常兜底重试，路由层无感）。""" return with_retry( "中通", "应到", - lambda: zto_expected_download_impl(page), + lambda: zto_expected_download_impl(page, force=force), lambda: zto_reset(page), ) -def zto_expected_download_impl(page): +def zto_expected_download_impl(page, force=False): """中通：应到货物数据下载（单次执行，无重试；供自动化测试探测原始结果用）。""" print("\n▶ 开始执行【中通 - 应到货物数据下载】任务...") @@ -240,6 +240,19 @@ def zto_expected_download_impl(page): count = main_rows.count() print(f">> 共发现 {count} 个交接单需要导出。") + # 【去重】加载本站已落库交接单号；force=True 或查询失败时 existing=空集（不去重） + if force: + existing = set() + print(">> [去重] 强制重下，跳过去重。") + else: + try: + from inbound_verify import store + + existing = store.get_existing_handover_nos("中通") + except Exception as _e: + existing = set() + print(f">> [去重] 加载已落库交接单号失败，本次不去重: {_e}") + for i in range(count): print(f" ⏳ 正在处理第 {i+1}/{count} 个交接单...") row = ewb_frame.locator( @@ -251,6 +264,11 @@ def zto_expected_download_impl(page): handover_no = match.group(0) if match else raw_text.strip() print(f" -> 当前交接单号：{handover_no}") + # 【去重】已落库的交接单号不再提交导出任务（不双击、不 append export_times） + if handover_no in existing: + print(f" ⏭️ 交接单号 {handover_no} 已落库，跳过提交导出。") + continue + row.dblclick() ewb_frame.locator("#datagrid2").get_by_text("运单号").wait_for( @@ -316,6 +334,11 @@ def zto_expected_download_impl(page): except Exception as e: print(f" ⚠️ 关闭【进站交接单查询】标签页时出错: {e}") + # 【去重兜底】全部已落库/无数据 → 无导出任务，标签页已关，直接结束不进轮询 + if not export_times: + print(">> 本次无新交接单需导出（全部已落库或无数据），结束。") + return True + # 交由统一的轮询下载流程处理 _zto_poll_and_download_tasks( page, @@ -330,7 +353,7 @@ def zto_expected_download_impl(page): return False -def zto_actual_download(page): +def zto_actual_download(page, force=False): """中通：实到货物数据下载（内部含异常兜底重试，路由层无感）。""" return with_retry( diff --git a/inbound_verify/store.py b/inbound_verify/store.py index 3c73d66..508978c 100644 --- a/inbound_verify/store.py +++ b/inbound_verify/store.py @@ -422,6 +422,25 @@ def ingest_task(site, kind): return total +def get_existing_handover_nos(site): + """查该站点已落库的交接单号集合（expected_record.handover_no）。 + 供"提交导出任务前"去重：已落库的交接单号不再重复提交导出任务。 + PG 不可用（cpolar 抖动等）时返回空集 + 告警，调用方按"未确认存在"处理 + （继续提交导出，UPSERT 兜底，绝不因去重查询失败而漏数据）。""" + try: + with _connect(_load_pg_config()["dbname"]) as conn: + with conn.cursor() as cur: + cur.execute( + "SELECT handover_no FROM expected_record " + "WHERE site=%s AND handover_no IS NOT NULL AND handover_no <> ''", + (site,), + ) + return {str(r[0]).strip() for r in cur.fetchall()} + except Exception as e: + print(f">> [去重] 查询已落库交接单号失败（{site}），本次不去重: {e}") + return set() + + # ============================== 命令行 ============================== diff --git a/schema.sql b/schema.sql index 862dee4..7c516a0 100644 --- a/schema.sql +++ b/schema.sql @@ -24,6 +24,7 @@ CREATE TABLE IF NOT EXISTS expected_record ( 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 TABLE IF NOT EXISTS actual_record (