docs: add ingest-hook design spec and implementation plan

Spec for mounting PostgreSQL ingest as a best-effort, kind-level, synchronous hook on runtime.dispatch_task after a successful download, plus the 5-task implementation plan. Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-24 10:27:00 +08:00
parent f61f52bcbe
commit 81dea38310
2 changed files with 812 additions and 0 deletions
--- a/docs/superpowers/specs/2026-07-24-ingest-hook-design.md
+++ b/docs/superpowers/specs/2026-07-24-ingest-hook-design.md
@@ -0,0 +1,290 @@
+# 下载后自动入库钩子设计
+
+- **日期**:2026-07-24
+- **方案**:在 `runtime.dispatch_task` 下载成功分支挂一个**同步、kind 级、尽力而为**的入库钩子，调用新增的 `store.ingest_task(site, kind)`，仅入库本次刚下载的文件
+- **力度**:后端最小闭环(钩子 + `ingest_task` + `ingest_state` 表 + `/status` 字段 + 配置开关/超时);不碰 dashboard UI、不上连接池/后台线程
+- **状态**:已与用户对齐(4 个关键决策已逐项确认),待 spec 评审
+
+---
+
+## 1. 背景与目标
+
+各站点的下载流程与 PostgreSQL 持久化模块(`store.py`)**都已实现**，但入库动作目前只能靠 CLI
+(`inbound-verify-db ingest` / `python -m inbound_verify.store ingest`)手动触发，**没有挂到任何自动钩子上**。
+
+本设计把入库动作挂到"数据完成下载之后"——具体挂在所有下载任务的唯一汇聚点
+`runtime.dispatch_task` 的成功分支，与既有的"下载后写业务日期"钩子 `_record_business_date` 并列。
+
+**目标**
+
+1. 下载成功后**自动**把刚下载的那份数据 UPSERT 进 PostgreSQL，无需手动跑 CLI。
+2. 入库是**尽力而为**：失败只告警、绝不影响下载任务的成功判定(下载成功 = 任务成功)。
+3. 入库结果可查：写入 `state_store`，经 `/api/status` 暴露，便于发现"入库坏了好几天"。
+4. 可门控、可降级：配置开关 + PG 连接/语句超时，无 PG/cpolar 的开发机可整体跳过。
+
+**非目标(本 spec 不做)**
+
+- dashboard 前端展示 ingest 态(Next.js 侧，API 字段已就绪待消费，单独小任务)。
+- 4 站 `-未到数据.xlsx` 入库(既有设计：仅百世未到入库，4 站未到只给汇总报表)。
+- 连接池、后台入库线程、失败补入重试(YAGNI)。
+- 任何与下载流程本身、比对逻辑相关的改动。
+
+---
+
+## 2. 现状(决策依据)
+
+### 2.1 汇聚点与既有钩子先例
+
+- `runtime.dispatch_task(ctx, {"site","kind"})` 是全部 9 个下载任务 + 比对任务的**唯一执行入口**;
+  交互模式(`cli/router`)与服务模式(`cli/server` worker)都走它。
+- 它的成功分支**已经有一个"下载后"钩子**:`_record_business_date(site, kind)`——写业务日期到
+  `state.db`，best-effort,失败仅告警。新钩子天然挂在它旁边，house style 完全一致。
+- 4 站 `undelivered` 任务的 handler(`_site_undelivered_handler`)内部**直接调**
+  `TASK_HANDLERS[(site,"expected"/"actual")](ctx)`(不经 dispatch_task)，故只触发**一次**外层
+  dispatch_task 钩子——`ingest_task(site,"undelivered")` 需据此同时入 expected+actual。
+
+### 2.2 持久化模块现状
+
+- `store.ingest(site=None)`:读 `downloads/` 现有 xlsx,幂等 UPSERT 到 PG。**站点级**:对顺心/中通/
+  韵达/安能同时入该站 expected+actual;百世只入 undelivered。**不感知 kind**。
+- 三张表:`expected_record`(运单级) / `actual_record`(扫描件级) / `undelivered_record`(仅百世)。
+- `_ingest_expected(cur, site, business_date)` / `_ingest_actual(cur, site)` /
+  `_ingest_undelivered_baishi(cur)` 为内部助手,直接复用,不改。
+- `_connect(dbname)`:每次新建一条连接,`options=-c search_path=<schema>`。**经 cpolar 隧道**
+  (`5.tcp.cpolar.top:10364`),潜在延迟/抖动。
+
+### 2.3 线程模型
+
+- dispatch_task 跑在 Playwright 所属线程(router=主线程;server=单 worker 线程,串行消费
+  task_queue、空闲跑心跳)。**同步做 PG I/O 会阻塞这条线程**——故入库必须快、可超时、可降级。
+
+### 2.4 数据流定位
+
+- dashboard **不直接读 PG**:全部经 Next.js `/api/*` 代理到 InboundVerify FastAPI(读 state.db +
+  文件 + 汇总报表)。故自动入库的直接受益者是 **PG 这个下游数仓**(BI/长期归档/未来报表),
+  不是当前前端。这支撑了"同步内联、不上复杂调度"的判断。
+
+### 2.5 state_store 现状(决定表设计)
+
+- `site_status` 是**每站一行**、按 kind 展开列(`{kind}_ready/_generated_at/_business_date`)。
+- `_upsert` 是**手写枚举列**的 read-modify-write(脆)。往里塞 ingest 列(4 列 × 3 kind = 12 列)
+  会很丑且易错——故选**独立 `ingest_state` 表**(用户选项里也提过"或一张很小的入库记录表")。
+
+---
+
+## 3. 四个关键决策(均已与用户确认)
+
+| 决策 | 选定 | 理由 |
+|---|---|---|
+| 执行模型 | **同步内联** | 与 `_record_business_date` 一致、零新线程,贴合"刻意不优化结构"风格;cpolar 风险用超时+try/except 兜底 |
+| 入库粒度 | **kind 级**(`ingest_task(site,kind)`) | 只入本次刚下载的文件,阻塞最小;"下了什么入什么";CLI `ingest(site)` 保留不动 |
+| 配置门控 | **开关 + 连接超时** | `auto_ingest`(默认开)+ `connect_timeout_seconds`(默认 5);无 PG 开发机可关 |
+| 失败可见性 | **stdout + state_store** | 仅 stdout 会在服务模式静默失败多天;落库后 `/api/status` 可查 |
+
+---
+
+## 4. 组件设计
+
+### 4.1 `store.py` — 新增 `ingest_task(site, kind)` + 连接/配置增强
+
+**新函数 `ingest_task(site, kind) -> int`**
+
+单连接、单事务,复用现有 `_ingest_*`,返回总条数。路由表:
+
+| site | kind | 调用 |
+|---|---|---|
+| `__compare__` | * | 无 → 返回 0 |
+| 顺心/中通/韵达/安能 | `expected` | `_ingest_expected(cur, site, dates.get(site))` |
+| 顺心/中通/韵达/安能 | `actual` | `_ingest_actual(cur, site)` |
+| 顺心/中通/韵达/安能 | `undelivered` | `_ingest_expected` + `_ingest_actual` |
+| 百世 | `undelivered` | `_ingest_undelivered_baishi(cur)` |
+| 百世 | `expected`/`actual` | (无此任务)防御性返回 0 |
+
+`dates = _read_business_dates()`(既有)。骨架:
+
+```python
+def ingest_task(site, kind):
+    """按 (site, kind) 入库本次刚下载的文件(幂等 UPSERT)。返回总条数。
+    与 ingest(site) 的区别:只入本次刷新的那一类,避免重读写另一类文件。"""
+    if site == "__compare__":
+        return 0
+    dates = _read_business_dates()
+    total = 0
+    with _connect(_load_pg_config()["dbname"]) as conn:
+        with conn.cursor() as cur:
+            if kind == "expected":
+                total += _ingest_expected(cur, site, dates.get(site))
+            elif kind == "actual":
+                total += _ingest_actual(cur, site)
+            elif kind == "undelivered":
+                if site == "百世":
+                    total += _ingest_undelivered_baishi(cur)
+                else:  # 4 站:handler 内部连带下了 expected+actual
+                    total += _ingest_expected(cur, site, dates.get(site))
+                    total += _ingest_actual(cur, site)
+            # 其它组合(如 百世/expected):防御性 0
+        conn.commit()
+    return total
+```
+
+**`_load_pg_config()` 增键**:`auto_ingest`(默认 `True`)、`connect_timeout_seconds`(默认 `5`)。
+
+**`_connect(dbname)` 增强**:
+
+- `psycopg.connect(..., connect_timeout=c["connect_timeout_seconds"])`。
+- 连上后 `cur.execute("SET statement_timeout = '30s'")`(固定值,带注释说明可按需 knob 化;
+  connect_timeout 守隧道死连,statement_timeout 守失控查询,小 UPSERT 极少触发)。
+- `create_database` / `init_schema` / `ingest` 共用此连接,30s 对它们无影响。
+
+**新薄封装 `ingest_enabled() -> bool`**:读 `_load_pg_config()["auto_ingest"]`。
+供 runtime 钩子判定开关,避免钩子伸手进 store 私有函数。
+
+**CLI `ingest-one`**(便于脱离下载单测路由):`python -m inbound_verify.store ingest-one 韵达 expected`
+→ 直接调 `ingest_task(site, kind)`。`main()` 增该分支。
+
+**不改**:既有 `ingest(site)` / `_ingest_*` / SQL / `domain`。
+
+### 4.2 `runtime.py` — 新增 `_persist_to_db(site, kind)`,挂到 `dispatch_task`
+
+**`dispatch_task` 成功分支**(`_record_business_date` 之后)新增一行:
+
+```python
+ret = handler(ctx)
+if ret is False:
+    return (state_store.TASK_FAILED, "任务执行失败(重试耗尽)")
+_record_business_date(site, kind)
+_persist_to_db(site, kind)   # 新增:尽力而为,绝不外抛,不影响任务判定
+return (state_store.TASK_SUCCESS, None)
+```
+
+**新函数 `_persist_to_db(site, kind) -> None`**:
+
+- `store` **懒导入**(函数内 `from inbound_verify import store`),彻底回避 import 顺序/成环,冷路径无性能影响。
+- `if site == "__compare__": return`
+- 读开关:`if not store.ingest_enabled(): print(">> [入库] 已关闭,跳过"); return`。
+- 主逻辑(所有写库/写状态都包 try/except,绝不外抛):
+
+```python
+try:
+    count = store.ingest_task(site, kind)
+    state_store.set_ingest_state(site, kind, ok=True, count=count)
+    print(f">> [入库] {site}/{kind} 成功,{count} 条")
+except Exception as e:
+    print(f">> [warn] 入库失败 {site}/{kind}: {e}")
+    try:
+        state_store.set_ingest_state(site, kind, ok=False, error=str(e))
+    except Exception as e2:
+        print(f">> [warn] 写入库状态也失败: {e2}")
+```
+
+### 4.3 `state_store.py` — 新增独立表 `ingest_state` + 读写
+
+**`init_db()` 增表(幂等)**:
+
+```sql
+CREATE TABLE IF NOT EXISTS ingest_state (
+    site        TEXT,
+    kind        TEXT,
+    ok          INTEGER,       -- 1=成功 0=失败
+    ingested_at TEXT,
+    count       INTEGER,       -- 入库条数
+    error       TEXT,          -- 失败原因(成功则 '')
+    PRIMARY KEY (site, kind)
+)
+```
+
+**新函数**:
+
+- `set_ingest_state(site, kind, ok, count=0, error=None)`:UPSERT(短连接,`ingested_at=_now()`,
+  与现有 `set_*` 风格一致)。
+- `get_all_ingest_state() -> {site: {kind: {ok, ingested_at, count, error}}}`(库不存在返回 `{}`)。
+
+**心跳不刷 `ingest_state`**(只由钩子写)。澄清:复用的是 state_store 管线 + `/api/status` 通道,
+不是心跳循环。**不碰 `site_status` / `_upsert`**。
+
+### 4.4 `cli/server.py` — `/status` 暴露 ingest 态
+
+`GET /status` 现有返回 `get_all_status()`;追加一个 `ingest` 键 = `state_store.get_all_ingest_state()`。
+一行改动,前端可按需消费。
+
+### 4.5 配置 — `config.example.yaml` + `config.yaml`
+
+`postgres` 段新增(两文件都加;`config.yaml` 已 gitignore):
+
+```yaml
+postgres:
+  # ...既有 host/port/user/password/dbname/schema...
+  auto_ingest: true            # 下载成功后自动入库;false=跳过(无 PG/cpolar 的开发机)
+  connect_timeout_seconds: 5   # PG 连接超时(秒);cpolar 抖动兜底
+```
+
+顺手把 `config.yaml` 注释里过期的 `db_store.py` 改为 `store.py`(小清理)。
+
+---
+
+## 5. 数据流
+
+```
+dispatch_task 成功
+  → _record_business_date(site, kind)        # 既有:SQLite 业务日期,best-effort
+  → _persist_to_db(site, kind)               # 新增
+       auto_ingest? ─ no ─→ print 跳过,return
+                  └ yes ─→ 懒导 store
+                            store.ingest_task(site, kind)
+                              _connect(connect_timeout) → SET statement_timeout
+                              路由 _ingest_* → commit → 返回 count
+                            成功 → set_ingest_state(ok,count) + print
+                            异常 → set_ingest_state(fail,error) + warn
+  → return TASK_SUCCESS(始终)
+```
+
+---
+
+## 6. 错误处理矩阵
+
+| 情形 | 行为 | 任务判定 |
+|---|---|---|
+| `auto_ingest=false` | print 跳过,return | SUCCESS(不变) |
+| 文件缺失(下载刚成功却无文件,罕见) | `_ingest_*` 已 `[跳过]` 返回 0;ingest_task 返回 0;记 ok/count=0 | SUCCESS |
+| PG 不可达 / connect_timeout / statement_timeout | psycopg 异常 → 捕获 → fail + warn | SUCCESS |
+| 表未初始化(UndefinedTable) | 捕获 → fail + warn(提示"请先 `inbound-verify-db init`") | SUCCESS |
+| `set_ingest_state` 自身失败 | 再包 try/except,绝不外抛(呼应 `_record_business_date`) | SUCCESS |
+
+**核心不变式:入库的任何失败都不改变下载任务的成功判定。**
+
+---
+
+## 7. 测试
+
+无 pytest(house 约定)。验证手段:
+
+1. **路由单测(新 CLI)**:`.venv/Scripts/python.exe -m inbound_verify.store ingest-one 韵达 expected`
+   → 确认只入韵达应到,不动韵达实到;`ingest-one 百世 undelivered` → 只入百世未到;
+   `ingest-one 顺心 undelivered` → 入顺心 expected+actual。
+2. **端到端**:dashboard/API 触发一次下载 → 看 stdout `[入库]` 行 → 查 PG 行数 →
+   查 `GET /api/status` 的 `ingest` 字段反映 ok/count/ingested_at。
+3. **降级**:`auto_ingest=false` → 下载仍 SUCCESS、无 `[入库]` 行;临时封掉 cpolar 端口 →
+   下载仍 SUCCESS、`[warn] 入库失败`、`ingest_state` 记 fail、`/status` 可见。
+4. **回归**:`py_compile inbound_verify` + `black inbound_verify` + `compileall` 导入冒烟;
+   确认未改 `ingest(site)` 行为(手动跑一次 `store ingest` 全站入库仍正常)。
+
+---
+
+## 8. 已确认约束
+
+- InboundVerify 是 git **子模块**;改动在子模块内,父仓库仅跟踪指针。
+- **不加测试套件**;验证靠编译/导入冒烟 + 手动端到端。
+- 改完 Python **必须跑 Black**。
+- **不自动提交/推送**:本仓库约定改动后等用户明确说"提交"再 commit/push(覆盖全局 auto-push 默认,
+  亦覆盖 brainstorming 默认的"写完即提交")。
+
+---
+
+## 9. 实现顺序提示(供 writing-plans 展开)
+
+1. `store.py`:`_load_pg_config` 加键 + `_connect` 加超时/语句超时 → `ingest_task` → CLI `ingest-one`。
+2. `state_store.py`:`ingest_state` 建表 + `set_ingest_state` + `get_all_ingest_state`。
+3. `runtime.py`:`_persist_to_db` + 挂到 `dispatch_task`。
+4. `cli/server.py`:`/status` 加 `ingest` 字段。
+5. 配置:`config.example.yaml` + `config.yaml` 加键 + 修过期注释。
+6. Black + 编译/导入冒烟 + 手动端到端验证。