本项目是一个专为医院信息集成平台（如基于 Apache Camel、Spring Integration 或东方通 TongE2G/TongWeb 的 ESB 系统）设计的消息血缘追踪工具。它不依赖特定厂商中间件接口，而是从原始运行日志出发，自动构建消息在各服务节点间的流转路径，形成可查询、可回放、可导出的结构化血缘档案。核心机制是：以 message ID 和 correlation ID 为关联主键，将非结构化日志统一解析为带时序与上下文的消息条目，再通过 NetworkX 构建有向无环图（DAG），并内置循环检测、最短路径计算、字段级差异比对与敏感数据脱敏能力。交付形态覆盖 CLI 命令行（Python Typer + TypeScript/Node.js 双实现）、SQLite 本地存储、交互式 HTML 可视化（vis-network 渲染）、FastAPI HTTP API 及 Docker 容器化部署。技术栈明确限定为 Python 3.10+（负责解析、建模、存储）与 TypeScript/Node.js 18+（负责 CLI 调用封装与前端渲染），所有功能模块均围绕医疗集成场景中真实存在的日志格式、转换规则与质控需求展开。

定位与能力范围

我们不做通用日志分析平台，也不对接数据库变更日志或应用埋点。本项目的边界非常清晰：只处理医院 ESB/集成平台产生的运行期消息流转日志，目标是还原“一条患者检验申请从HIS发出，经路由、协议转换、字段映射、适配器调用，最终抵达LIS”的完整链路。这意味着它天然服务于三类人：一是集成工程师，需要快速定位某次失败消息卡在哪一跳；二是信息科质控人员，需验证转换逻辑是否符合《医院信息互联互通标准化成熟度测评》中关于数据流向可追溯的要求；三是第三方审计方，在等保或电子病历评级现场，要求提供指定业务消息的端到端血缘证据。

它不替代日志采集系统（如 Filebeat/Fluentd），也不做实时流式血缘（如 Apache Atlas），而是聚焦离线批量分析，你把当天的 Camel 日志、东方通 EASWay 日志或 JSON Lines 格式日志文件丢进来，它就给你生成一份带时间戳、节点名、输入输出快照和字段差异的血缘报告。所有能力都围绕一个事实展开：医院集成链路不是扁平服务调用，而是多层嵌套的转换流水线，而日志是唯一留存全链路证据的载体。

核心功能

本系统五大能力全部源自真实集成运维痛点，不是技术炫技：

• 多源日志统一解析

  ：原生支持 Apache Camel/Spring Integration 结构化日志、东方通 TongE2G/TongWeb 私有格式、以及通用 JSON Lines 三种输入。每种解析器均实现 detect_format 方法，支持 --format auto 自动识别。

• 血缘图谱自动构建

  ：基于 NetworkX 构建消息级 DAG，自动识别循环引用（如因配置错误导致消息在两个适配器间反复跳转），并提供 shortest_path 查询接口，用于回答“HIS 到 LIS 最少经过几步”。

• 医疗敏感字段默认脱敏

  ：在解析阶段即识别手机号、身份证号、银行卡号、金额类字段（正则匹配 + 上下文关键词辅助），输出时自动替换为 ***，保障审计过程不泄露隐私。

• 转换链路逐跳回放

  ：给定任意 message ID，即可回溯其完整流转路径，展示每一跳的输入 payload 与输出 payload，并支持 --diff 参数高亮字段级变化（例如 patient_id 未变、order_time 被格式化、test_code 被映射为新值）。

• 多格式血缘导出与可视化

  ：除 SQLite 本地存档外，支持导出为交互式 HTML（拖拽缩放搜索节点）、标准 JSON（供其他系统集成）、Mermaid 流程图（嵌入 Confluence 文档），且可选包含样本数据便于演示。

功能模块	输入依据	输出成果	典型使用场景
日志解析	.log  / .jsonl 文件	结构化消息条目（含 message_id, endpoint, timestamp, input, output）	接入新上线的东方通集成项目日志
血缘构建	解析后的消息条目集合	SQLite 中的 lineage.db，含 nodes/edges 表	搭建院内集成血缘知识库底座
转换回放	message ID + lineage.db	终端逐跳打印输入/输出及 diff 高亮	排查某条医保结算失败原因
HTML 可视化	lineage.db	lineage.html （含 vis-network 渲染）	向信息科主任汇报 HIS-LIS-RIS 三方集成拓扑
API 服务	lineage.db + FastAPI	/replay/{msg_id}  等 REST 接口	对接医院统一监控平台告警联动

使用与配置

所有操作均通过 CLI 驱动，Python 与 TypeScript 两套入口完全等价，可根据环境自由选择。安装仅需两步依赖：

pip install -r requirements.txt
npm install

实际使用分三阶段：

第一阶段：构建血缘档案
选择对应日志格式执行 analyze，输出 lineage.db：

python -m lt_src.cli.main analyze \
    --log-file data/sample_easway.log \
    --format easway \
    --output lineage.db

成功后终端显示类似：✅ 解析 142 条消息，检测到 3 处循环路径，已写入 lineage.db。

第二阶段：按需回放诊断
用 replay 命令定位具体问题：

python -m lt_src.cli.main replay \
    --message-id msg-78901 \
    --storage-db lineage.db \
    --diff \
    --highlight order_status

输出将清晰列出：[1] HIS → Router: input.order_status=“draft”, output.order_status=“draft” → [2] Router → Adapter: input.order_status=“draft”, output.order_status=“submitted”。

第三阶段：交付与共享
用 export 生成可交付物：

python -m lt_src.cli.main export \
    --format html \
    --output lineage_report \
    --storage-db lineage.db \
    --include-samples

生成 lineage_report/ 目录，含 index.html（直接双击打开）与配套数据文件。

工程结构

项目采用清晰分层架构，Python 为血缘引擎核心，TypeScript 为跨平台胶水层：

integration-lineage-tracker/
├── lt_src/               # Python 主体：解析、建模、存储、CLI
│   ├── parsers/          # camel.py, easway.py, jsonl.py（各实现 ParserProtocol）
│   ├── lineage/          # DAG 构建、循环检测、路径查询
│   ├── storage/          # SQLite CRUD 封装，支持导入导出
│   └── cli/              # Typer 实现的 analyze/replay/export/serve 四命令
├── src/                  # TypeScript CLI：调用 Python 子进程或提供独立 Node 实现
├── templates/            # HTML 可视化模板（含 vis-network 初始化脚本）
└── data/                 # 所有示例日志与规则（sample_camel.log 等）

关键设计原则是：解析器可插拔、存储可迁移、可视化可脱离。新增东方通新版本日志格式？只需在 parsers/ 下新增一个类，实现三个方法，注册进 _detect_parser() 即可；想换 PostgreSQL？改 storage/ 下的后端实现；要嵌入现有 Web 系统？直接读取 lineage.db 或调用 export --format json 获取数据。

环境与运行

本系统对运行环境有明确约束，避免“在我机器上能跑”陷阱：

环境组件	要求	说明
Python	3.10+	低版本不支持结构化模式匹配，影响 JSON 解析健壮性
Node.js	18+	为兼容 TypeScript 5.x 与现代 fetch API
SQLite	内置	无需额外安装，开箱即用，适合单机分析场景
Docker	可选	提供 docker-compose.yml，含 lineage-cli（分析）与 lineage-api（服务）两个 profile

容器化使用极简：

docker-compose run --rm lineage-cli analyze \
    --log-file /app/sample_data/sample_camel.log \
    --format camel \
    --output /app/data/lineage.db

数据自动落盘至 lineage-data volume，本地 data/ 目录实时同步，无需手动拷贝。

数据与扩展

所有数据均源于日志文件本身，不引入外部元数据源。示例数据集 data/ 是完整可用的最小闭环：

• sample_camel.log

  ：含正常流转 + 人为注入的循环路径，用于验证检测能力

• sample_easway.log

  ：东方通典型报文头 + Base64 编码 payload

• sample_generic.jsonl

  ：每行一个 JSON 对象，含 msg_id, from, to, input, output 字段

• sample_rules.xslt

   与 sample_mapping.json：用于规则解析模块测试

扩展新日志格式只需两步：
1. 在 lt_src/parsers/ 新建 my_vendor.py，实现 parse_file(), parse_line(), detect_format()；
2. 在 lt_src/cli/main.py 的 _detect_parser() 函数中添加判断逻辑。
整个过程不改动核心引擎，不重编译，符合医疗 IT 系统对稳定性的硬性要求。

项目地址：
https://github.com/nexorin9/integration-lineage-tracker