# 下载后自动入库钩子设计

- **日期**: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 + 编译/导入冒烟 + 手动端到端验证。