diff --git a/docs/superpowers/plans/2026-07-24-ingest-hook.md b/docs/superpowers/plans/2026-07-24-ingest-hook.md new file mode 100644 index 0000000..02d5615 --- /dev/null +++ b/docs/superpowers/plans/2026-07-24-ingest-hook.md @@ -0,0 +1,522 @@ +# 下载后自动入库钩子 Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 把 PostgreSQL 入库动作挂到 `runtime.dispatch_task` 下载成功分支,使每次下载成功后自动、尽力而为、kind 级地把刚下载的数据 UPSERT 进 PG。 + +**Architecture:** 新增 `store.ingest_task(site, kind)`(kind 级路由,复用现有 `_ingest_*`)+ `runtime._persist_to_db(site, kind)`(同步内联钩子,懒导入 store,绝不外抛)+ `state_store.ingest_state` 表(结果可查,经 `/status` 暴露)+ 配置开关/超时。详见 spec `docs/superpowers/specs/2026-07-24-ingest-hook-design.md`。 + +**Tech Stack:** Python 3.10+、psycopg(v3)、sqlite3、Playwright(不动)、FastAPI(仅 `/status` 加字段)、pyyaml。包以可编辑模式安装(`pip install -e .`),命令走 `.venv/Scripts/python.exe -m inbound_verify...`。 + +## Global Constraints + +- **Python 一律走项目虚拟环境**:`.venv/Scripts/python.exe`(全局 CLAUDE.md 协议)。命令默认在 `InboundVerify/` 根目录执行。 +- **无 pytest 测试套件**(本仓库约定,覆盖 writing-plans 默认的 TDD-pytest 步骤):每个任务的验证用 `py_compile` + `black` + 导入冒烟 + 功能/手动校验,不写 pytest。 +- **改完任何 .py 必须跑 Black**:`.venv/Scripts/python.exe -m black inbound_verify`。 +- **不自动提交/推送**(本仓库约定,覆盖 writing-plans 默认的"每任务即提交"):每个任务的 Commit 步骤**仅在用户明确说"提交"/"commit"时执行**;否则完成任务后停在待提交态,告知用户。 +- **commit message 一律英文**(全局 CLAUDE.md)。 +- **不改下载流程/比对逻辑**;4 站 `-未到数据.xlsx` 不入库(既有设计);不上连接池/后台线程/补入重试。 +- 导入约束:`runtime` 已 import `compare`,`store` 也 import `compare`;钩子里对 `store` **懒导入**以回避成环。 + +--- + +## File Structure + +| 文件 | 责任 | 本计划改动 | +|---|---|---| +| `inbound_verify/store.py` | PG 持久化(建库/建表/入库 CLI) | 改 `_load_pg_config`/`_connect`;加 `ingest_enabled()`、`ingest_task()`;CLI 加 `ingest-one` | +| `inbound_verify/state_store.py` | SQLite 状态持久化 | `init_db` 加 `ingest_state` 表;加 `set_ingest_state()`、`get_all_ingest_state()` | +| `inbound_verify/runtime.py` | 两种模式共享核心(含 dispatch_task) | 加 `_persist_to_db()`;在 `dispatch_task` 成功分支调用 | +| `inbound_verify/cli/server.py` | FastAPI 服务模式 | `/status` 返回加 `ingest` 字段 | +| `config.example.yaml` | 配置模板(提交) | `postgres` 段加 `auto_ingest`/`connect_timeout_seconds`;修过期 `db_store.py`→`store.py` 注释 | +| `config.yaml` | 真实配置(gitignored) | 同上两键 + 修注释 | +| `README.md` | 用户文档 | DB CLI 命令列表加 `ingest-one` | + +任务依赖:Task 2 依赖 Task 1;Task 4 依赖 Task 1+2+3;Task 5 依赖 Task 3。Task 3 独立。建议顺序 1→2→3→4→5。 + +--- + +### Task 1: store.py 配置键 + 连接超时层 + +**Files:** +- Modify: `inbound_verify/store.py:48-78`(`_load_pg_config`、`_connect`) +- Modify: `config.example.yaml:72-89`、`config.yaml`(postgres 段) + +**Interfaces:** +- Consumes: 无(配置层根基) +- Produces: `_load_pg_config()` 返回新增 `auto_ingest: bool`、`connect_timeout_seconds: int`;`_connect(dbname)` 连接带 `connect_timeout` 且会话级 `statement_timeout=30s`;新函数 `ingest_enabled() -> bool`。后续任务依赖 `ingest_enabled()` 与超时连接。 + +- [ ] **Step 1: 改 `_load_pg_config` 加两键** + +`inbound_verify/store.py` 的 `_load_pg_config` 返回 dict,在 `schema` 之后追加两键: + +```python + return { + "host": pg.get("host", "127.0.0.1"), + "port": int(pg.get("port", 5432)), + "user": pg.get("user", "postgres"), + "password": pg.get("password", ""), + "dbname": pg.get("dbname", "CQHXDB"), + "schema": pg.get("schema", "inbound_verify"), + "auto_ingest": bool(pg.get("auto_ingest", True)), + "connect_timeout_seconds": int(pg.get("connect_timeout_seconds", 5)), + } +``` + +- [ ] **Step 2: 改 `_connect` 加 connect_timeout + statement_timeout** + +把 `_connect` 替换为(用 `options` 一次性设 search_path + statement_timeout,等价于 spec 的 SET 但不引入额外事务): + +```python +def _connect(dbname): + """用关键字参数连接(避开 conninfo 对密码特殊字符的解析)。 + options 设 search_path 到专用 schema + 会话级 statement_timeout=30s + (cpolar 隧道上防失控查询;connect_timeout 守连接阶段)。""" + c = _load_pg_config() + return psycopg.connect( + host=c["host"], + port=c["port"], + dbname=dbname, + user=c["user"], + password=c["password"], + options=f"-c search_path={c['schema']} -c statement_timeout=30s", + connect_timeout=c["connect_timeout_seconds"], + ) +``` + +- [ ] **Step 3: 加 `ingest_enabled()` 薄封装** + +在 `_connect` 之后、`# 建库 / 建表` 分节注释之前插入: + +```python +def ingest_enabled(): + """是否启用下载后自动入库(config.yaml postgres.auto_ingest,默认 True)。 + 供 runtime 钩子判定开关,避免它伸手进 _load_pg_config。""" + return _load_pg_config()["auto_ingest"] +``` + +- [ ] **Step 4: 更新 store.py 模块 docstring 的命令列表** + +把文件顶部 docstring 的命令行小节,在 `ingest` 行后补一行 `ingest-one`(Task 2 会实现该命令,docstring 先行): + +``` + python -m inbound_verify.store ingest [site] 入库全站或单站(幂等 UPSERT) + python -m inbound_verify.store ingest-one 仅入库指定站/类(钩子同款路由) + python -m inbound_verify.store all createdb → init → 全站 ingest 一条龙 +``` + +- [ ] **Step 5: config.example.yaml 加两键 + 修过期注释** + +`config.example.yaml` 的 postgres 段:把 `# PostgreSQL 数据持久化(到货核销数据入库,详见 db_store.py)` 改为 `... 详见 store.py)`;命令注释里的 `python db_store.py ...` 改为 `python -m inbound_verify.store ...`;在 `schema: inbound_verify` 之后追加: + +```yaml + schema: inbound_verify + # 下载成功后自动入库(钩子,见 runtime._persist_to_db);false=跳过(无 PG/cpolar 的开发机)。 + auto_ingest: true + # PG 连接超时(秒);cpolar 抖动时快速失败,不拖垮下载 worker。 + connect_timeout_seconds: 5 +``` + +- [ ] **Step 6: config.yaml 同步(gitignored 本地文件)** + +对 `config.yaml` 做同样两键追加,并把注释里的 `db_store.py` 改为 `store.py`。保留其真实凭据不动。 + +- [ ] **Step 7: 编译 + Black** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify/store.py +.venv/Scripts/python.exe -m black inbound_verify/store.py +``` +Expected: py_compile 无输出;black 报 `reformatted` 或 `left unchanged`。 + +- [ ] **Step 8: 回归既有 CLI 不破** + +``` +.venv/Scripts/python.exe -c "from inbound_verify import store; c=store._load_pg_config(); assert c['auto_ingest'] is True and c['connect_timeout_seconds']==5; assert store.ingest_enabled() is True; print('config ok')" +``` +Expected: `config ok`。 + +- [ ] **Step 9: Commit(gated)** + +仅当用户说"提交"时执行: + +```bash +git add inbound_verify/store.py config.example.yaml +git commit -m "feat(store): add auto_ingest config + pg connect/statement timeouts" +``` +(`config.yaml` 已 gitignore,不入提交。) + +--- + +### Task 2: store.ingest_task + ingest-one CLI + +**Files:** +- Modify: `inbound_verify/store.py`(加 `ingest_task`;`main()` 加 `ingest-one` 分支) + +**Interfaces:** +- Consumes: Task 1 的 `_connect`(超时)、`_load_pg_config`、既有 `_ingest_expected/_ingest_actual/_ingest_undelivered_baishi`、`_read_business_dates`、`ALL_SITES`、`_site_cfg`。 +- Produces: `ingest_task(site: str, kind: str) -> int`(返回入库总条数;`__compare__`/不支持组合返回 0)。 + +- [ ] **Step 1: 加 `ingest_task`** + +在 `ingest(site)` 函数之后、`# 命令行` 分节之前插入: + +```python +def ingest_task(site, kind): + """按 (site, kind) 入库本次刚下载的文件(幂等 UPSERT),返回总条数。 + 与 ingest(site) 的区别:只入本次刷新的那一类,避免重读写另一类文件(同步钩子里减少阻塞)。 + kind 路由: + expected/actual 各入其列; + undelivered 百世 入未到; + undelivered 4 站 _site_undelivered_handler 内部连带下了 expected+actual,故入两者; + __compare__ / 其它组合 返回 0。 + """ + 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: # 顺心/中通/韵达/安能 + total += _ingest_expected(cur, site, dates.get(site)) + total += _ingest_actual(cur, site) + # 其它组合(如 百世/expected,正常不经钩子触发):防御性返回 0 + conn.commit() + return total +``` + +- [ ] **Step 2: `main()` 加 `ingest-one` 分支** + +在 `main()` 的 `elif cmd == "all":` 分支之后、`else:` 之前插入: + +```python + elif cmd == "ingest-one": + kind = sys.argv[3] if len(sys.argv) > 3 else None + if not site or kind not in ("expected", "actual", "undelivered"): + print("用法: python -m inbound_verify.store ingest-one ") + sys.exit(1) + total = ingest_task(site, kind) + print(f">> [ingest-one] {site}/{kind} 入库 {total} 条") +``` + +- [ ] **Step 3: 编译 + Black** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify/store.py +.venv/Scripts/python.exe -m black inbound_verify/store.py +``` +Expected: 无编译错误。 + +- [ ] **Step 4: 路由正确性(手动,需 PG 已 createdb+init 且 downloads/ 有数据)** + +逐条验证 kind 级只入对应文件: + +``` +.venv/Scripts/python.exe -m inbound_verify.store ingest-one 韵达 expected +``` +Expected: stdout 只出现 `[应到] 韵达:N 条运单`,**不**出现 `[实到] 韵达` 行;结尾 `[ingest-one] 韵达/expected 入库 N 条`。 + +``` +.venv/Scripts/python.exe -m inbound_verify.store ingest-one 顺心 undelivered +``` +Expected: stdout 同时出现 `[应到] 顺心` 与 `[实到] 顺心`(undelivered→两者)。 + +``` +.venv/Scripts/python.exe -m inbound_verify.store ingest-one 百世 undelivered +``` +Expected: stdout 出现 `[未到] 百世`。 + +> 若某站 downloads/ 无文件:`_ingest_*` 打印 `[跳过] ... 文件不存在`,ingest_task 返回 0,属正常(非错误)。 + +- [ ] **Step 5: 无 PG 时的降级(手动,可选)** + +临时把 `config.yaml` 的 `host` 改成不可达地址,重跑 Step 4 任一命令:应在 `connect_timeout_seconds`(默认 5s)内报 psycopg 连接错误并退出(非 hang)。验证后改回真实 host。 + +- [ ] **Step 6: Commit(gated)** + +```bash +git add inbound_verify/store.py +git commit -m "feat(store): add kind-level ingest_task + ingest-one CLI subcommand" +``` + +--- + +### Task 3: state_store ingest_state 表 + 读写函数 + +**Files:** +- Modify: `inbound_verify/state_store.py:32-113`(`init_db` 加表)、`:219-247`(加两函数) + +**Interfaces:** +- Consumes: 无(独立叶子,仅依赖 paths + sqlite3 + datetime)。 +- Produces: `set_ingest_state(site, kind, ok, count=0, error=None) -> None`;`get_all_ingest_state() -> dict`(形状 `{site: {kind: {ok, ingested_at, count, error}}}`)。Task 4 与 Task 5 依赖这两个。 + +- [ ] **Step 1: `init_db` 加 `ingest_state` 表** + +在 `init_db` 内、`site_settings` 表 CREATE 之后(`conn.commit()` 之前)插入: + +```python + conn.execute(""" + CREATE TABLE IF NOT EXISTS ingest_state ( + site TEXT, + kind TEXT, + ok INTEGER, + ingested_at TEXT, + count INTEGER, + error TEXT, + PRIMARY KEY (site, kind) + ) + """) +``` + +- [ ] **Step 2: 加 `set_ingest_state` 与 `get_all_ingest_state`** + +在 `get_all_status()` 函数之后插入: + +```python +def set_ingest_state(site, kind, ok, count=0, error=None): + """记录一次入库结果(UPSERT)。ok: bool;count: 入库条数;error: 失败原因或 None。""" + with sqlite3.connect(STATE_DB_PATH) as conn: + conn.execute( + "INSERT INTO ingest_state (site, kind, ok, ingested_at, count, error) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT(site, kind) DO UPDATE SET " + "ok=excluded.ok, ingested_at=excluded.ingested_at, " + "count=excluded.count, error=excluded.error", + (site, kind, 1 if ok else 0, _now(), int(count or 0), error or ""), + ) + conn.commit() + + +def get_all_ingest_state(): + """返回 {site: {kind: {ok, ingested_at, count, error}}};库不存在返回 {}。""" + if not os.path.exists(STATE_DB_PATH): + return {} + with sqlite3.connect(STATE_DB_PATH) as conn: + rows = conn.execute( + "SELECT site, kind, ok, ingested_at, count, error FROM ingest_state" + ).fetchall() + out = {} + for site, kind, ok, ingested_at, count, error in rows: + out.setdefault(site, {})[kind] = { + "ok": bool(ok), + "ingested_at": ingested_at or "", + "count": int(count or 0), + "error": error or "", + } + return out +``` + +- [ ] **Step 3: 编译 + Black** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify/state_store.py +.venv/Scripts/python.exe -m black inbound_verify/state_store.py +``` +Expected: 无错误。 + +- [ ] **Step 4: set/get 往返冒烟** + +``` +.venv/Scripts/python.exe -c "from inbound_verify import state_store as s; s.init_db(); s.set_ingest_state('韵达','expected',True,count=42); s.set_ingest_state('韵达','expected',False,error='boom'); d=s.get_all_ingest_state(); r=d['韵达']['expected']; assert r['ok'] is False and r['count']==42 and r['error']=='boom' and r['ingested_at']; print('ingest_state ok')" +``` +Expected: `ingest_state ok`(验证 UPSERT 覆盖:第二次写把 ok 改 False,count 保留 42,error 写入)。 + +- [ ] **Step 5: Commit(gated)** + +```bash +git add inbound_verify/state_store.py +git commit -m "feat(state_store): add ingest_state table + set/get helpers" +``` + +--- + +### Task 4: runtime._persist_to_db 钩子 + 挂到 dispatch_task + +**Files:** +- Modify: `inbound_verify/runtime.py`(加 `_persist_to_db`,位置在 `_record_business_date` 之后、`dispatch_task` 之前;改 `dispatch_task` 成功分支 `:552-557`) + +**Interfaces:** +- Consumes: Task 1 `store.ingest_enabled()`、Task 2 `store.ingest_task(site, kind)`、Task 3 `state_store.set_ingest_state(...)`;既有 `state_store`(runtime 已 import)。 +- Produces: `_persist_to_db(site, kind)` 被 `dispatch_task` 在下载成功后调用;下载任务的成功判定**不变**。 + +- [ ] **Step 1: 加 `_persist_to_db`** + +在 `_record_business_date` 函数之后(`def dispatch_task` 之前)插入: + +```python +def _persist_to_db(site, kind): + """下载成功后把本次数据入库 PostgreSQL(尽力而为,绝不外抛,不影响任务判定)。 + - __compare__ 无源数据,跳过。 + - auto_ingest=false 时跳过(无 PG/cpolar 的开发机)。 + - 懒导入 store 以回避 import 顺序(store↔compare 与 runtime↔compare 共存)。 + - 结果写 state_store.ingest_state,供 /api/status 反映入库健康。 + 所有写库/写状态都包 try/except:失败仅告警,绝不改变 dispatch_task 的 SUCCESS 判定。""" + if site == "__compare__": + return + try: + from inbound_verify import store # 懒导入:冷路径(每下载一次),回避成环 + except Exception as e: + print(f">> [warn] 入库模块不可用: {e}") + return + if not store.ingest_enabled(): + print(">> [入库] 已关闭 (auto_ingest=false),跳过") + return + 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}") +``` + +- [ ] **Step 2: 挂到 `dispatch_task` 成功分支** + +把 `dispatch_task` 内的成功分支: + +```python + _record_business_date(site, kind) + return (state_store.TASK_SUCCESS, None) +``` + +改为: + +```python + _record_business_date(site, kind) + _persist_to_db(site, kind) + return (state_store.TASK_SUCCESS, None) +``` + +(`_persist_to_db` 绝不外抛,故不会被外层 `except Exception` 误判为任务失败。) + +- [ ] **Step 3: 编译 + Black + 导入冒烟** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify/runtime.py +.venv/Scripts/python.exe -m black inbound_verify/runtime.py +.venv/Scripts/python.exe -c "from inbound_verify.runtime import dispatch_task, _persist_to_db; print('runtime import ok')" +``` +Expected: `runtime import ok`。 + +- [ ] **Step 4: 端到端(手动,需站点已登录)** + +任选一种模式触发一次真实下载并观察钩子: + +- 服务模式:`POST /tasks` `{"site":"韵达","kind":"expected"}`(或经 dashboard 触发),下载完成后看 worker stdout: + - 成功:`>> [入库] 韵达/expected 成功,N 条`。 + - 失败(如 PG 未 init):`>> [warn] 入库失败 ...`,且任务本身仍 `success`(`GET /tasks/{id}` 验证)。 +- 交互模式:菜单 `[6]` 韵达应到,完成后看同样的 `[入库]` 行。 + +并查 `state/state.db`: +``` +.venv/Scripts/python.exe -c "from inbound_verify import state_store as s; print(s.get_all_ingest_state())" +``` +Expected: 含 `韵达`/`expected` 的记录,`ok` 与 stdout 一致。 + +- [ ] **Step 5: 关开关降级(手动)** + +把 `config.yaml` 的 `auto_ingest` 改 `false`,再触发一次下载:stdout 应出现 `>> [入库] 已关闭 (auto_ingest=false),跳过`,且不连 PG;任务仍 SUCCESS。验证后改回 `true`。 + +- [ ] **Step 6: Commit(gated)** + +```bash +git add inbound_verify/runtime.py +git commit -m "feat(runtime): auto-ingest hook after successful download" +``` + +--- + +### Task 5: /status 暴露 ingest 态 + README + +**Files:** +- Modify: `inbound_verify/cli/server.py:189-196`(`get_status`) +- Modify: `README.md`(DB CLI 命令列表) + +**Interfaces:** +- Consumes: Task 3 `state_store.get_all_ingest_state()`。 +- Produces: `GET /status` 返回体新增 `ingest` 字段。 + +- [ ] **Step 1: `/status` 加 `ingest` 字段** + +把 `get_status` 的返回 dict: + +```python + return { + "worker_ready": worker_state["ready"], + "worker_error": worker_state["error"], + "sites": state_store.get_all_status(), + } +``` + +改为: + +```python + return { + "worker_ready": worker_state["ready"], + "worker_error": worker_state["error"], + "sites": state_store.get_all_status(), + "ingest": state_store.get_all_ingest_state(), + } +``` + +并把 docstring 顺手补一句(可选):`"""各站登录态 + 数据态 + 入库态(前端状态盘用),另含 worker 就绪状态。"""` + +- [ ] **Step 2: README DB CLI 命令列表加 `ingest-one`** + +`README.md` 第五章"运行"下、DB CLI 注释行(`# 或:inbound-verify-db createdb|init|ingest|all` 附近)补一行说明自动入库 + 新命令: + +``` +# DB CLI:建库 / 初始化 / 灌数据 / 全流程 / 单站单类 +.venv/Scripts/python.exe -m inbound_verify.store createdb # 或 init | ingest | ingest-one | all +# 注:下载成功后会自动入库(postgres.auto_ingest,默认开);ingest-one 用于手动重灌指定站/类。 +``` + +- [ ] **Step 3: 编译 + Black** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify/cli/server.py +.venv/Scripts/python.exe -m black inbound_verify/cli/server.py +``` +Expected: 无错误。 + +- [ ] **Step 4: `/status` 含 ingest 字段(手动,服务模式已启动)** + +``` +curl -s http://127.0.0.1:8000/status | python -m json.tool +``` +(或浏览器 `http://127.0.0.1:8000/docs` 试 `/status`。)Expected: 返回体含 `"ingest": {...}` 键(无入库记录时为 `{}`,Task 4 跑过后会有值)。经 dashboard 的 `GET /api/status`(代理)同样可见。 + +- [ ] **Step 5: 全量回归冒烟** + +``` +.venv/Scripts/python.exe -m py_compile inbound_verify +.venv/Scripts/python.exe -m black inbound_verify +.venv/Scripts/python.exe -c "import inbound_verify.store, inbound_verify.state_store, inbound_verify.runtime, inbound_verify.cli.server; print('all imports ok')" +``` +Expected: `all imports ok`,black 无 diff。 + +- [ ] **Step 6: Commit(gated)** + +```bash +git add inbound_verify/cli/server.py README.md +git commit -m "feat(server): expose ingest state in /status; doc auto-ingest in README" +``` + +--- + +## Self-Review 结论 + +- **Spec 覆盖**:spec §4.1 → Task 1+2;§4.2 → Task 4;§4.3 → Task 3;§4.4 → Task 5;§4.5 配置 → Task 1(config 两文件);§7 测试手段 → 各任务手动步(ingest-one 路由、端到端、降级、回归)。无遗漏。 +- **占位符**:无 TBD/TODO;每步含完整代码与确切命令。 +- **类型/命名一致**:`ingest_task(site, kind)`、`ingest_enabled()`、`set_ingest_state(site, kind, ok, count=0, error=None)`、`get_all_ingest_state()`、`_persist_to_db(site, kind)` 在各任务间签名一致;`/status` 字段名 `ingest` 与 `get_all_ingest_state` 返回一致。 +- **已标注偏离**:(a) 无 pytest → 用 py_compile/black/冒烟/手动代替(Global Constraints);(b) 不自动提交 → Commit 步骤 gated(Global Constraints + 每任务注明);(c) `_connect` 用 `options` 设 statement_timeout 取代 spec 的 `SET`(等价、无额外事务,Task 1 Step 2 注释说明)。 diff --git a/docs/superpowers/specs/2026-07-24-ingest-hook-design.md b/docs/superpowers/specs/2026-07-24-ingest-hook-design.md new file mode 100644 index 0000000..0051346 --- /dev/null +++ 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=`。**经 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 + 编译/导入冒烟 + 手动端到端验证。