P4 Gotham

多源融合：把散落的名单汇成一张图

CSV 人员名单、JSON 工商信息、数据库登记表、手工预置记录——四类数据源接入 Gotham，Ingest 真实执行、按映射规则变成图节点，fingerprint 去重防重复，批次记录全程留痕。看完这 4 个故事，你就能把一堆散文件变成可分析的统一实体池。

情报分析员数据接入工程师 file_csv file_json database fingerprint 清洗规则质量评估共 5 个故事

能 / 不能速览

✅ 这个主题能做

注册四类数据源：file_csv / file_json / database / manual
点"运行导入"真实执行：读 CSV / 读 JSON / 连数据库查询 / 读预置记录
按 mapping_config 把记录映射为图节点（或边），属性字段自由指定
fingerprint 去重键：同源同 fingerprint 重复导入走更新，不重复插入
清洗规则引擎：trim / date_iso / phone / 全半角 / 正则替换 / 枚举映射等 7 类规则，strict/skip/tolerant 三策略
导入前四维质量评估：完整度 / 唯一性 / 一致性 / 时效性，报告挂在批次 Quality
SSE 进度推送 + 暂停 + 断点恢复：每 500 行一个事务，暂停落库 paused，恢复续导
批次全程留痕：running / success / partial / failed + 坏行 error_log
融合实体统一落库（fused_entities），供实体解析作业消费
V5 行级图谱化导入：database 源配置图谱映射后可"一行一节点"直达图上（含外键生成边），绕过"候选池→解析"两步（见图谱化导入主题）

⛔ 这个主题做不了

文件源路径创建时不强制存在，Ingest 运行时才校验，路径错会跑失败
fusion 只产"融合实体候选"，不做实体合并；合并交给实体解析（resolution）作业
数据库源走 connector 工厂 allow-list，默认仅放行 MYSQL / SQLSERVER / POSTGRESQL
CSV 要求第一行表头；JSON 要求数组（每元素一条记录）

适用角色

本主题面向两类角色：

情报分析员：把办案中收集的名单、记录接入系统，形成统一实体池。
数据接入工程师：配置数据源、映射规则、排查批次失败原因，保证数据管线顺畅。

接入后的融合实体是实体解析（entity-resolution 主题）的输入，两条链路经常连着用。

能力速览（能做什么）

四类数据源

file_csv（第一行表头）、file_json（数组记录）、database（connector 工厂连库查询）、manual（手工预置记录）。

映射规则

节点型映射：node_type + id_field + label_field + property_fields；边型映射：edge_type + source_field + target_field。

Ingest 真实执行

点"运行导入"就真读文件 / 真连库 / 真融合，返回批次对象（success / partial / failed 与计数）。

去重与留痕

fingerprint（sha256）作去重键，重复导入走更新；批次记录 total / imported / failed 与 error_log 全程可查。

清洗与质量评估

7 类清洗规则（trim / date_iso / phone / 全半角 / 正则 / 枚举映射）三策略参与 Ingest；导入前四维质量报告（完整度 / 唯一性 / 一致性 / 时效性）挂在批次。

进度与断点

SSE 逐批推送进度（text/event-stream）；暂停为"请求式"（500 行批边界落库 paused），恢复从 ResumeOffset 续导不重复。

调整指南（怎么调整）

改映射：想导入成船（ship）就改 node_type，想导入关系就配置边型映射（source_field / target_field 指向记录里的字段）。
改去重：fingerprint 由 id_field + node_type（节点）或 source+target+edge_type（边）计算；同源同 fingerprint 重复导入是更新不是新增。
改导入范围：先小样验证映射，再全量跑；manual 源用 records 数组直接预置，适合补几条情报。
改失败处理：批次 partial 时看 error_log 的坏行明细；文件路径写错、JSON 格式不对都会导致失败，先修配置再重跑。

做得好的场景

多源融合最擅长"收拢散料"：

名单批量入库：一份 CSV 人员名单几分钟变成图上候选实体，替代手工录入几小时。
多源归一：CSV、JSON、数据库、手工四类来源进入同一张实体池，为后续去重合并打底。
重复导入不慌：fingerprint 去重保证同一条记录再跑一次不会复制出双份。
失败可追：批次状态与 error_log 把"哪几条坏行"记下来，排查不靠猜。

限制与不足

以下是明确的边界，使用前先知道：

不做实体合并：fusion 只产出统一实体候选与去重键，同名同人的归并交给实体解析（resolution）作业完成。
文件源运行时才校验：路径创建时不强制存在，Ingest 时路径错 / 文件格式错都会使批次失败。
数据库源受白名单约束：connector 工厂默认仅放行 MYSQL / SQLSERVER / POSTGRESQL，其他类型需显式授权。
格式要求硬：CSV 第一行必须是表头，JSON 必须是数组且每元素一条记录；坏行计入 failed。
V5 图谱化导入是并行路径：database 源配置图谱映射后可"一行一节点"直接上图（含外键生成边），不走候选池与解析作业；但"融合导入只产候选不做合并"的边界依旧成立，图谱化节点与解析产物并存，归并仍由图存储 MergeNode 与实体解析承担。

场景故事

故事 1 接入一份 CSV 人员名单，跑一次 Ingest 变成图节点

场景：CSV 接入角色：情报分析员耗时：约 5 分钟

背景: 周警官从合作单位拿到一份"远海船务人员名单.csv"，第一行是表头：entity_id,name,org,role,amount，里面是 5 名船务相关人员的记录。他要把它接进 Gotham，变成图上可查的 person 节点候选。

传统做法对比: 以前收到名单先手工录入 Excel、再逐条核对、再找开发写导入脚本，一份 5 条的名单也要折腾大半天。现在建个源、配好映射、点运行，几分钟完成。

角色: 周警官（情报分析员，具备数据接入权限；登录 admin / admin1，端口 18083）。

操作步骤

进入数据接入页 /gotham/ingestion，点"新建数据源"
填写 name=haiyuan_personnel、source_type=file_csv
connection_config 里填 Path（CSV 绝对路径）
mapping_config：node_type=person、id_field=entity_id、label_field=name、property_fields=["org","role","amount"]
保存后在数据源列表点"运行导入"

系统响应

返回批次对象：

{
  "id": 12,
  "source_id": 7,
  "status": "success",
  "total_records": 5,
  "imported_records": 5,
  "failed_records": 0,
  "error_log": null
}

融合实体列表（/ingestion/entities?source_id=7）出现 5 条 person 候选。

结果洞察: 5 条记录全部导入成功，entity_id 作为 external_id，name 作为 label，org / role / amount 落入 properties；每条实体带 fingerprint 去重键与 status=pending（等待实体解析）。周警官的名单第一次进入系统化的实体池。

调整建议: 先小样验证映射再全量跑；CSV 第一行必须是表头；如果想导入成 ship / org，改 mapping_config 的 node_type 即可。

动手试一试: 操作：新建 file_csv 源，指向一份含表头的名单文件，配置 person 映射后运行。预期结果：批次返回 success、imported_records=记录数；再点一次"运行导入"，imported_records 不变（同 fingerprint 走更新），这就是去重的直观效果。

限制提示: 文件路径创建时不强制存在，运行时才校验，路径错会批次失败；CSV 需第一行表头，坏行计入 failed_records 并写入 error_log；同源同 fingerprint 重复导入走更新路径，不会产生重复记录。

故事 2 JSON 工商信息 + 数据库船舶登记表，多源进入同一张图

场景：多源接入角色：数据接入工程师耗时：约 8 分钟

背景: 数据接入工程师李诚要把两类新源接进来：一份公司工商信息的 JSON 文件（数组，每元素一条记录），和一张 SQLite 库里的船舶登记表。多源数据要汇进同一张图，供后续统一分析。

传统做法对比: 以前每接一个源写一套 ETL 脚本，JSON 一套、数据库一套，脚本没人维护就烂尾；现在统一建源、统一映射、统一跑批，四类源同一套操作。

角色: 李诚（数据接入工程师，配置映射与排查失败）。

操作步骤

新建 file_json 源：connection_config.path 指向工商 JSON 文件
mapping_config：node_type=org、id_field=org_code、label_field=name
新建 database 源：connector_type=SQLITE、Database=库文件路径、Query=SELECT * FROM ships
mapping_config：node_type=ship、id_field=ship_id、label_field=ship_name
两个源分别点"运行导入"

系统响应

每个源有独立 source_id，各自返回批次结果：

{ "source_id": 8, "status": "success", "imported_records": 3 }
{ "source_id": 9, "status": "success", "imported_records": 2 }

fusion 实体按 entity_type 区分：org、ship 候选分别入库。

结果洞察: JSON 导入 3 条 org、数据库导入 2 条 ship，与之前的 CSV person 源共同进入统一实体池。同一个组织名可能同时出现在 JSON 与 CSV 两个源里——这正是"多源归一"要解决的重复问题，交给实体解析处理。

调整建议: 边型映射用 edge_type + source_field + target_field，可把"哪艘船归哪个公司管"也一并导成边候选；manual 源可以快速补几条临时记录。

动手试一试: 操作：新建 file_json 源并运行。预期结果：/ingestion/entities?entity_type=org 出现 JSON 里的组织记录；再新建 database 源（SQLITE + Query）运行，entity_type=ship 出现船舶记录。

限制提示: JSON 文件必须是数组，每元素一条记录；database 源走 connector 工厂 allow-list，默认仅放行 MYSQL / SQLSERVER / POSTGRESQL，SQLITE 等需要显式授权；数据源 name 唯一，重复创建返回冲突。

故事 3 演示源里就有 5 条人员情报，同一个人出现在多个源里怎么办

场景：融合实体角色：情报分析员耗时：约 5 分钟

背景: 周警官发现 Gotham 演示源 gotham_demo_personnel（manual 类型）里已经预置了 5 条人员情报：张远·核心人物·20 万、李四·联络人、王敏·船务、吴刚·船长、赵敏·财务。他自己接的 CSV 里又有"吴刚"——同一个人的记录出现在两个源里，他要弄清系统怎么处理。

传统做法对比: 以前两个源里出现同一个人，得人肉比对姓名、职务、公司再手工归并，还要留一份"谁合并了谁"的记录。现在融合实体带 fingerprint 与 external_id，谁来自哪个源一目了然。

角色: 周警官（情报分析员，查看融合实体列表）。

操作步骤

打开 /ingestion/entities，按 entity_type=person 过滤
找到演示源的 5 条记录，核对 external_id / properties / fingerprint
对比自己 CSV 源里的同名记录，看 status 与 resolved_entity_id
确认"融合实体是候选池，落不落到图上由实体解析决定"

系统响应

演示源实体示例：

{
  "external_id": "gotham_p_004",
  "entity_type": "person",
  "properties": { "name": "吴刚", "org": "青龙航运", "role": "船长", "amount": 30000 },
  "fingerprint": "sha256:…",
  "status": "resolved",
  "resolved_entity_id": "graph:person:wugang"
}

未解析的候选 status=pending，resolved_entity_id 为空。

结果洞察: 演示源里吴刚（gotham_p_004）status=resolved、resolved_entity_id=graph:person:wugang，说明它已经被解析并落到图上；而 CSV 源里的同名记录仍可能 pending。同一个人跨源出现时，谁去重、谁合并，由 fingerprint 与实体解析作业共同决定。

调整建议: 用 fingerprint 理解去重（同源同 fingerprint 走更新）；跨源合并要看实体解析结果，先建一个 resolution 作业把 person 类候选跑一遍。

动手试一试: 操作：/ingestion/entities?entity_type=person。预期结果：看到演示源 5 条人员记录（张远 20 万 / 李四 5 万 / 王敏 8 万 / 吴刚 3 万 / 赵敏 12 万），其中至少 1~2 条 status=resolved 且带 resolved_entity_id。

限制提示: fusion 本身不做实体合并，只产出候选、去重键与解析状态；真正把"同一个人"归并成图上单一节点，需要跑实体解析（resolution）作业。resolved 状态由解析作业回填。

故事 4 回看批次记录与 fingerprint，确认每次导入的成败明细

场景：批次留痕角色：数据接入工程师耗时：约 4 分钟

背景: 李诚接到周警官的反馈："昨天那批导入好像少了两条。"他要回看历史批次记录，核对某次导入的 total / imported / failed，再看 error_log 里的坏行明细，确认是不是 CSV 里有格式问题。

传统做法对比: 以前导入脚本跑完只留个日志文件，没人维护就覆盖了，出了事查无实据。现在每条批次有独立记录，状态、计数、坏行明细都在，随时可回溯。

角色: 李诚（数据接入工程师，排查批次失败）。

操作步骤

打开 /ingestion/batches，用 source_id 过滤出目标源的历史批次
查看每条批次的 status / total_records / imported_records / failed_records
对 failed / partial 的批次查看 error_log 的坏行明细
修正源文件或映射后重新"运行导入"

系统响应

批次列表返回：

[
  { "id": 12, "source_id": 7, "status": "success", "total_records": 5,
    "imported_records": 5, "failed_records": 0, "error_log": null },
  { "id": 15, "source_id": 10, "status": "partial", "total_records": 6,
    "imported_records": 4, "failed_records": 2,
    "error_log": ["2 条坏行（解析失败）"] }
]

演示源 gotham_demo_personnel 通常有一条 success、imported_records=5 的批次。

结果洞察: 周警官"少两条"的问题定位到：那次 partial 批次 6 条里 2 条坏行，坏行只计入 failed、不产生实体，所以实体池里看起来少了。修好源文件后重跑，imported_records 补足，去重键保证不重复。

调整建议: 重跑同一条记录走 fingerprint 更新不重复插入；排查顺序：先看 error_log，再核对文件路径 / 表头 / 字段名，最后重跑。

动手试一试: 操作：/ingestion/batches（不带 source_id 看全部）。预期结果：能看到演示源的 success 批次（5 条导入）；如果自己建的源有坏行，会看到 partial / failed 批次及 error_log 明细。

限制提示: 删除数据源时保留历史批次与融合实体（不级联删）；failed 记录不会产生实体；批次状态是 running → success / partial / failed 的终态流转，重跑会生成新批次。

故事 5 3000 行大名单：清洗规则先修数据，质量评估打分，跑一半暂停再接上

场景：清洗/质量/断点角色：数据接入工程师耗时：约 10 分钟

背景: 李诚拿到一份 3000 行的船务人员名单，里面有全角括号、座机号、空姓名等"脏数据"。他给数据源配好清洗规则（trim + fullwidth_to_halfwidth + phone），导入前先看质量评估分数，跑批中再演示一次"暂停→恢复"验证断点续导。

传统做法对比: 以前 3000 行名单要么人肉在 Excel 里逐列修格式（几小时起步），要么写一次性脚本，脚本一挂就白跑。现在清洗规则随源配置、导入时自动执行，暂停/恢复随时控制，不用重来。

角色: 李诚（数据接入工程师，配置清洗规则与断点管理）。

操作步骤

新建 file_csv 源，cleaning_config 配 trim（name 字段）+ fullwidth_to_halfwidth + phone 规则，error_strategy=skip
运行导入后立即打开 SSE 进度流看逐批推送
批次跑到一半点"暂停"，确认批次状态变 paused、ResumeOffset 记录偏移
点"恢复"，从断点续导到 success
查看批次 Quality 的四维质量报告

系统响应

暂停返回：

{"code":0,"data":{"paused":true,"batch_id":18}}

恢复返回：

{"code":0,"data":{"resumed":true,"batch_id":18}}

批次质量报告示例：

{
  "completeness": 0.97, "uniqueness": 0.99,
  "consistency": 0.95, "timeliness": 1.0, "score": 0.97
}

SSE 进度：data: {"batch_id":18,"processed":3000,"total":3000,"status":"completed"}

结果洞察: 清洗规则把全角括号、座机格式修干净，坏行按 skip 策略跳过并计入 error_log，批次以 success 收尾、imported=3000；质量评估 0.97 说明名单整体规范（唯一性 0.99 是因为 fingerprint 去重生效）；暂停是"请求式"（至多 500 行延迟），恢复从 ResumeOffset 续导，imported 承接暂停前计数不重复。

调整建议: 清洗规则先小样验证再全量；质量评估低分先查一致性与唯一性（是否同名多表头）；大批次中途想停用暂停而非强杀进程，恢复续导不重来。

动手试一试: 操作：配一条 trim 清洗规则导入含空格姓名的名单，再对 3000 行批次练习暂停→恢复。预期结果：导入后 label 无空格（清洗生效）；暂停后批次 status=paused 且 ResumeOffset>0，恢复后 success、imported_records=3000。

限制提示: 暂停是"请求式"而非"中断式"——至多等 500 行批边界才落库 paused；清洗统计只并入 error_log 文本（无独立结构化清洗报告）；质量评估为启发式（一致性=最常见类型占比、时效性=新鲜度衰减），空数据集记全 1.0。

常见问题

接入的数据会自动画到图上吗？

不会直接画上。fusion 接入后产生的是"融合实体候选"（fused_entities），status 先为 pending；只有跑实体解析（resolution）作业，代表实体才会写入图存储（resolved 状态指向图节点）。两步是先后关系。

重复导入会不会产生重复数据？

不会。每条实体按 (source_id, fingerprint) 去重：同源同 fingerprint 重复导入走更新路径，更新属性与批次信息，不插入新记录。fingerprint 由 id_field + node_type（节点）或 source+target+edge_type（边）计算。

数据库源支持哪些类型？

走平台 connector 工厂，默认 allow-list 仅放行 MYSQL / SQLSERVER / POSTGRESQL，其他类型（如 SQLITE）需要显式授权（AllowAdvancedTypes）。演示库连接需先确保已注册。

批次 failed 了怎么办？

先看批次的 error_log 明细（坏行与原因），再核对：文件路径是否存在、CSV 是否第一行表头、JSON 是否为数组、字段名是否与映射一致、数据库源连接参数与 Query 是否正确，修好后重新运行导入。

manual 源是什么？

manual 是"手工预置记录"类型：在 connection_config.records 里直接写记录数组，不需要文件或数据库。演示源 gotham_demo_personnel 就是 manual 源，预置了张远、李四、王敏、吴刚、赵敏 5 条人员情报。

主题小结

一句话：多源融合把 CSV、JSON、数据库、手工记录四类来源收进统一实体池——建源、配映射、点运行三步走，清洗规则 + 质量评估先行，fingerprint 去重防重复，批次留痕、暂停断点可恢复。记住边界：fusion 只产候选不做合并（合并归实体解析）、文件源运行时才校验、数据库源受白名单约束、CSV / JSON 格式有硬要求。