---
title: 数据摄入
description: PAC 怎么把任何宿主的数据,经统一管道标准化成可信的患者事实 —— 传输、数据源、标准化各层、匹配规则、canonical 字典与可信保证、幂等。
icon: Import
---

摄入只有**两种传输方向**(Pull / Push),再叠加正交的**存量 / 增量**。
所有入口取回 raw 后,都汇入**同一条标准化管道**,口径一致;最终产物 `fact.content` 是下游(画像 / 召回 / AI)的**唯一真理源**,下游不再碰原始数据。

总览见 [架构总览](/docs/architecture/overview)。

---

## Pull —— PAC 主动取

按计划 / 手动去宿主取数,**三种源**任选(取回的 raw 下游完全无感知):

| 源 | 怎么取 | 状态 |
|---|---|---|
| **数据库直连** | 直连宿主数仓(目前 ClickHouse,manifest `sql_source`) | ✅ |
| **文件**(CSV / JSON) | 读宿主导出文件(manifest `tables[]`) | ✅ |
| **API 接口** | HTTP 拉取适配器(api-key / login / oauth2) | ⛔ 仅 mock,真实适配器归档待接 |

> **双源真等价**:直连用的 SQL 被刻意限制成「等价于一张 CSV 单表 dump」的朴素 SELECT(只列选 + WHERE 过滤,**禁止** JOIN / JSONExtract / multiIf 等形态改造 —— 那些归 transforms)。
> 于是宿主用直连、CSV、还是 HTTP 给数据,**下游一行都不用改**,可随时切换。

**存量 ↔ 增量**(同一条 pull 路径,靠游标切换):

- **存量**(首次,无游标)→ 全量拉。
- **增量**(后续)→ 读上次 `cursor_after`,注入 `WHERE cursor_column > 上次值`,跑完写新游标;中途失败保留上次水位,下次重跑。

---

## Push —— 宿主主动推

宿主在事件发生时 webhook 推给 PAC(`POST /push/events`,**HMAC-SHA256 验签** + 密钥轮换、±5 分钟时间窗)。
天然是事件增量流。因为 event 自带动作语义,push **可以跳过 transforms**,直接进 assembler / 落库。✅

---

## 统一标准化管道

所有入口的 raw 都过同一条管道。**前两步(transforms + assembler)是 raw → canonical,配置都在该 host 的 manifest 里(per-host 差异);后三步是集团统一内核,所有入口共用**:

```mermaid
flowchart LR
  RAW["raw 行<br/>(宿主原生形态)"]
  --> T["① transforms<br/>形态改造(6 算子)"]
  --> A["② assembler<br/>命名 + 枚举翻译"]
  --> S["③ 事务合成<br/>写 patient_transactions"]
  --> P["④ parser + Zod<br/>从账本衍生 fact"]
  --> F["⑤ fact 落地<br/>写 patient_facts"]
  style RAW fill:#eee,stroke:#999
  style F fill:#e8f5e9,stroke:#2e7d32,stroke-width:2px
```

| 层 | 做什么 | 关键 |
|---|---|---|
| **① transforms** =「**形态改造**」*(manifest,per-host)* | 拆行 / 派生 / 关键词分流 / 多列推断 / 行过滤 | **6 个白名单算子**(`project` / `split_json_array` / `derive` / `route_by_pattern` / `pick_first_nonzero` / `filter`),无副作用纯函数,**不允许任意 JS / eval** |
| **② assembler** =「**词汇翻译**」*(manifest,per-host)* | 字段名翻译(宿主 → canonical)+ 枚举码翻译(闭集字典 + `_default` 兜底)+ `emits` 推断 | manifest + yaml 驱动;**名字虽叫 assembler,本质是翻译器** |
| **③ 事务合成** | 把 canonical 行写成 `patient_transactions`(append-only 账本) | `raw_payload` 留底(宿主原文,reparse 的输入);`source_event_id` 幂等键;`event_seq` 单调水位 |
| **④ parser + Zod** =「**类型 + 语义 + 校验**」 | 从 transaction 把扁平行**构造成带类型的 `fact.content`**:单位 / 牙位 / 文本归一、code 纠偏(如 K00 误标→K08)、`FactContentSchema` 强校验 | **唯一会抛错拦脏数据的强校验闸**,字段漂移即拒 |
| **⑤ fact 落地** | 写 `patient_facts`(版本流,supersede 旧版) | 唯一写入口 `FactWriter`,evidence 反指 transaction |

**「标准化」分三个 facet,各管一层、不重复**:

- **形态**(transforms):把数据拆成对的行 / 列结构。
- **词汇**(assembler):把宿主的字段名、枚举码查表翻成 PAC 闭集字典(精确映射,不猜)。
- **类型 + 语义 + 校验**(parser):把翻译后的扁平行**构造成带类型的事实**,做归一 / 纠偏 / 强校验 —— 标准化的**最后一层 + 兜底闸**。

> **一条原则 —— 宿主差异只在 yaml**:每接一个宿主,差异**全部收敛在它的 `manifest` + `assembler yaml`**(取数 SQL/文件、形态算子、字段 / 枚举翻译)。过了标准化管道之后 —— `fact.content` 及其下游(画像 / 召回 / AI / 前端)—— **全集团一套代码、零 per-host 分支**。接新宿主 = 写一份 yaml,不改一行下游代码。

> **唯一收口**:全库只有 `TransactionSynthesizer` 能写账本、只有 `FactWriter` 能写 fact;字典只认 `canonical-codes.ts` 闭集;CI 有防漂移测试(详见下〈Canonical 字典与可信保证〉)。

### 各入口在哪一段汇入

四种取数方式接入管道的位置不同,**事务合成 → parser → 落库 这段统一内核完全共用**:

| 入口 | 取数 | transforms | assembler | 事务合成 → parser → 落库 |
|---|---|---|---|---|
| **直连 / 文件**(cold-import) | SQL / CSV | ✅ | ✅ | ✅ 共用 |
| **API 拉取** | HTTP GET | ✅ | ✅ | ✅ 共用 |
| **Push** | 宿主直推 event | ❌ 跳过 | 视形态可选 | ✅ 共用 |

---

## 一个例子:一行宽表 → 9 个 fact

> 取自一宿主一次真实接诊:宽表 `fact_emr_treatment_out` 的**一行**(`diag` / `treat_plan` 是 JSON 数组子字段)。
> 直观感受「摄入到底在干什么」:**一行 → 拆 / 译 / 校验 → 多个独立 fact**。

原始一行的 `diag` 数组里有 6 条诊断,其中 2 条只有中文 `message`、没有标准码:

```jsonc
"diag": [
  { "toothPosition": "16;17;25;26", "message": "残根" },          // ← 无 stdCode
  { "toothPosition": "27;35",       "message": "牙周炎" },         // ← 无 stdCode
  { "stdCode": "K08.103", "toothPosition": "44;45;46", "message": "后天性牙齿缺失" },
  { "stdCode": "K05.300", "toothPosition": "41;31",    "message": "慢性牙周炎" },
  { "stdCode": "K04.500", "toothPosition": "14;15",    "message": "慢性根尖牙周炎" },
  { "stdCode": "K07.303", "toothPosition": "11;12;21", "message": "牙体缺损" }
]
```

经过管道:

1. **transforms**:`split_json_array` 把 `diag` 拆成 6 行 → `filter` 只卡「**有没有诊断名 `message`**」,6 条都有 → **一条不丢** → `derive` 有码取 K 码前 3 位(`K08.103 → K08`),无码取归一后的中文名。`treat_plan` 拆出 1 行拔牙计划。
2. **assembler**:`organization_id → clinicId`、`rq → occurredAt`、`brand → tenant_id`;诊断码 —— 有 ICD 码 `K0x` 直通;**无码用中文名查字典**:`牙周炎 → K05`、`残根 → K03`(都命中)。**无 ICD 码 ≠ 翻不出**。
3. **事务合成 → parser + Zod**:每个资源先写一条账本(`patient_transactions`),再从账本逐条衍生、规范化、强校验成 fact。

**结果:DW 一行 → 9 个独立 fact**(一条都不丢):

| fact 类型 | 条数 | 内容 |
|---|---|---|
| `encounter_record` | 1 | 接诊元数据(医生 / 时间) |
| `diagnosis_record` | **6** | 4 条有 ICD 码 → K08 / K07 / K04 / K05;2 条无码按中文名查字典 → 牙周炎 K05、残根 K03(本行 6 条全部出码) |
| `treatment_record`(planned) | 1 | 拔除,牙位 27 |
| `emr_record` | 1 | 病历自由文本(留待后续抽取) |

> **无 ICD 码 ≠ 翻不出**:中文名先精确查字典(`残根→K03`、`牙周炎→K05`),精确不中再走关键词模糊,**都不中才** `code=null` 留底(诊断不丢 fact)。本行 6 条全部出码。下游只读 `fact.content`,不碰原始宽表。

---

## 匹配规则:诊断 / 建议 / 治疗 / 等自由文本

把原始信号映射成 PAC 标准码,是**三级漏斗**:精确优先,逐级兜底。

**① 关键词路由(transforms `route_by_pattern`)** —— 先把 `treat_plan` 按 `treat_name` 分流(三模式:`starts_with` / `equals` / `contains` 子串):

| 命中 | 去向 |
|---|---|
| `starts_with`「建议」「推荐」 | **建议**(recommendation) |
| `equals` 复查 / 检查 / 暂观 / 转诊 / 缴费 …(白名单) | **复查 / 非治疗**(review) |
| 其余(`default`) | **真治疗**(treatment,`kind=actual`) |

排除法:先剥离「建议」和「复查 / 行政」,**剩下的默认当真治疗**。

**② 字典翻译(assembler)** —— 把名字 / 码翻成 PAC canonical 码,**三级**:

| 级 | 机制 | 例 |
|---|---|---|
| **精确** `enum_mapping` | 精确命中闭集字典 | `K08.103→K08`、`牙周炎→K05`、`残根→K03`、`建议种植→IMPLANT_RECOMMENDED` |
| **关键词模糊** `keyword_mapping` | 精确不中 → **子串「含任一词即命中」**;有序优先 + 排除词 `none`(语义反转)+ 先剥「条件从句」 | `「楔状 / 残根 / 隐裂」→K03`;`龋(`none`: 风险 / 可疑)→K02`;`「充填,必要时根管」剥成「充填」→ restorative` |
| **兜底** `_default` | 都不中(行为见 ③) | —— |

> **为什么要模糊层**:host 术式是 free-text,**2 万+ distinct**,精确字典盖不全;关键词「一条吃一类」治长尾。所以「不靠中文猜」要修正成:**先精确、再关键词、最后才放弃** —— 关键词层就是受控的模糊匹配。

**③ `_default` 兜底按 fact 类型不同**:

- **诊断**:翻不出 → `code=null` + 留原文,**fact 仍落**(「该患者有这问题」的事实要留)。
- **建议**:翻不出 → **直接丢**(不入 fact);只有命中、能触发召回的建议才有意义,长尾仅在病历展示。

**④ 自由文本**(`illness_desc` / 医嘱原文):

- **当前只存进 `emr_record.content` 留底,不结构化**。
- 设计中的 **Layer C**(规则 / LLM / 关键词三轨抽取 → 带 confidence 的信号、下游降权)**尚未上线**:`parser` 预留了 `extractedBy = llm / rule / doctor_input` 入口,但 **LLM signal-extractor 还没实现**。
- 所以现在的「建议」**只来自结构化 `treat_plan` 的关键词路由**,**埋在自由文本里的建议还没被挖出来**。

---

## Canonical 字典与可信保证

匹配的目标,是一套**集团统一的闭集字典**(`canonical-codes.ts`):诊断码(借 ICD-10 的 K00–K14 大类)+ PAC 业务码(种植失败 / 正畸反弹等国际标准没有、但算法要的)+ 推荐码(`IMPLANT_RECOMMENDED` …)+ 11 类治疗 category。

- **宿主随意,字典闭集**:宿主端用 ICD / 自定义码 / 中文 / 自由文本都行,`assembler` 负责翻译进字典;字典本身是小闭集(几十到一两百个码)。
- 收益:集团范围内**算法 / dashboard / 报表口径完全一致**。

`fact.content` 之所以可信,靠几道**代码层质量闸**(不靠人工审):

| 闸 | 作用 |
|---|---|
| **schema 强校验** | 每个 fact 类型一份 zod(`FactContentSchemas`),写库前必过,违反抛错 → 脏数据进不了库 |
| **唯一写入口** | 只有 `FactWriter` 能写 fact、`TransactionSynthesizer` 能写账本(lint 拦其他人直写 Prisma) |
| **字典闭集** | enum 值必须 ∈ `canonical-codes.ts`,越界值进不去 |
| **单向消费** | 算法 / 画像 / AI **只读 `fact.content`**,红线禁止读 `raw_payload`(仅 reparse CLI 例外) |
| **CI 防漂移** | 改 yaml / parser / 字典,CI 自动跑全套(schema × yaml × 字典联立),漂移即挂 |

---

## 幂等与续传

- **幂等键** `source_event_id`:由 synthesizer 统一合成 = `来源 : 资源 : source_unit(品牌) : 主体id : 更新时间标记`。同一主体、同一更新时刻 → 同键;DB 偏唯一键去重,重复直接跳过。
- **完整性** `payload_hash`:对 raw 行做 sha256,**独立字段**,做兜底校验(不参与上面那个幂等键的拼接)。
- **水位** `event_seq`:Postgres 序列单调递增,供「自上次以来新增」消费,不依赖 `created_at`。
- **游标**:增量拉记录 `cursor_after`,跑完才推进;中途失败保留上次水位,下次从断点重跑。

---

延伸:[三层模型](/docs/architecture/three-layer-model) · [数据模型](/docs/architecture/data-model) · [接入](/docs/integration/overview)
