# 应到数据「提交导出任务前」去重 实施计划 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（注入，选择器 `<HANDOVER_SELECTOR>` 确认后替换）：**

(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("<HANDOVER_SELECTOR>").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
<label className="inline-flex items-center gap-1 text-xs text-amber-700">
  <input
    type="checkbox"
    checked={forceRedownload}
    onChange={(e) => setForceRedownload(e.target.checked)}
  />
  强制重新下载（忽略已落库去重）
</label>
```

> 仅"获取未到"主按钮透传 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 实勘需在运行的服务上操作，实施时与用户协调时机。