|
|
@@ -0,0 +1,181 @@
|
|
|
+#!/usr/bin/env python3
|
|
|
+"""派生 MD 视图：以知识库 JSON 为唯一事实源，生成 Markdown 表格视图。
|
|
|
+
|
|
|
+用法：
|
|
|
+ python3 render_kb_md.py [知识库目录] [输出目录]
|
|
|
+ python3 render_kb_md.py --check [知识库目录] [输出目录]
|
|
|
+
|
|
|
+输出：<输出目录>/views/{supplements,clinical_indicators,exercises,interactions,protocols}.md
|
|
|
+约定：JSON 为单一事实源，MD 为派生视图；改动数据只改 JSON，随后重跑本脚本。
|
|
|
+--check 模式：重生成到临时目录并与现视图对比，不一致退出码 1（供 CI/提交流程）。
|
|
|
+"""
|
|
|
+import json
|
|
|
+import sys
|
|
|
+import tempfile
|
|
|
+from pathlib import Path
|
|
|
+
|
|
|
+DEFAULT_KB = "docs/参考资料/知识库"
|
|
|
+
|
|
|
+
|
|
|
+def load(path: Path):
|
|
|
+ return json.loads(path.read_text(encoding="utf-8"))
|
|
|
+
|
|
|
+
|
|
|
+def md_table(headers, rows):
|
|
|
+ out = ["| " + " | ".join(headers) + " |", "| " + " | ".join(["---"] * len(headers)) + " |"]
|
|
|
+ for r in rows:
|
|
|
+ cells = [str(c).replace("|", "\\|").replace("\n", " ") for c in r]
|
|
|
+ out.append("| " + " | ".join(cells) + " |")
|
|
|
+ return "\n".join(out)
|
|
|
+
|
|
|
+
|
|
|
+def render_supplements(doc):
|
|
|
+ rows = []
|
|
|
+ for s in doc["supplements"]:
|
|
|
+ ings = ", ".join(f"{i['name']} {i['amount']}" for i in s["key_ingredients"])
|
|
|
+ rows.append([s["id"], s["name"], s.get("brand") or "—", s["category"], s["form"],
|
|
|
+ ings, s["dose_reference"], s["timing"], s.get("interaction_notes") or "—",
|
|
|
+ s.get("contraindications") or "—", s.get("evidence_level", "—"),
|
|
|
+ "✔" if s.get("enabled", True) else "✘"])
|
|
|
+ return ("# 补充剂库（视图）\n\n> 由 supplements.json 派生，勿手工编辑。\n\n"
|
|
|
+ + md_table(["id", "名称", "品牌", "类别", "剂型", "关键成分", "参考剂量", "时段",
|
|
|
+ "相互作用注意", "禁忌", "证据", "启用"], rows))
|
|
|
+
|
|
|
+
|
|
|
+def render_indicators(doc):
|
|
|
+ rows = []
|
|
|
+ for ind in doc["indicators"]:
|
|
|
+ nr = ind.get("normal_range", {})
|
|
|
+ lo, hi = nr.get("low"), nr.get("high")
|
|
|
+ rng = f"{lo}~{hi}" if lo is not None and hi is not None else (str(lo or hi))
|
|
|
+ rt = ind.get("risk_thresholds", {})
|
|
|
+ warn = "; ".join(x for x in [f"低<{rt['low_warn']}" if rt.get("low_warn") else "",
|
|
|
+ f"高>{rt['high_warn']}" if rt.get("high_warn") else "",
|
|
|
+ rt.get("note") or ""] if x)
|
|
|
+ rows.append([ind["id"], ind["name"], ind.get("english_name", ""), ind["unit"], rng,
|
|
|
+ warn, ind.get("monitoring_frequency_hint") or "—",
|
|
|
+ ind.get("related_condition") or "—", ind.get("population_note") or "—"])
|
|
|
+ return ("# 临床监测指标参考库（视图）\n\n> 由 clinical_indicators.json 派生，勿手工编辑。\n\n"
|
|
|
+ + md_table(["id", "指标", "英文", "单位", "参考范围", "干预阈值", "复查频率",
|
|
|
+ "相关疾病", "人群差异"], rows))
|
|
|
+
|
|
|
+
|
|
|
+def render_exercises(doc):
|
|
|
+ rows = []
|
|
|
+ for e in doc["exercises"]:
|
|
|
+ rows.append([e["id"], e["name"], "/".join(e.get("aliases", [])), e["type"], e["intensity"],
|
|
|
+ e["duration_min"], e["frequency"], e.get("timing_pref") or "—",
|
|
|
+ e.get("suitable_population") or "—", e.get("safety_notes") or "—"])
|
|
|
+ return ("# 运动处方库（视图）\n\n> 由 exercises.json 派生，勿手工编辑。\n\n"
|
|
|
+ + md_table(["id", "名称", "别名", "类型", "强度", "时长(分)", "频率", "时段偏好",
|
|
|
+ "适宜人群", "安全注意"], rows))
|
|
|
+
|
|
|
+
|
|
|
+def render_interactions(doc):
|
|
|
+ def subj(s):
|
|
|
+ return s.get("supplement_id") or s.get("substance", "")
|
|
|
+ rows = []
|
|
|
+ for ix in doc["interactions"]:
|
|
|
+ rows.append([ix["id"], ix["type"], subj(ix["subjects_a"]), subj(ix["subjects_b"]),
|
|
|
+ ix["rule"], ix["detail"], ix["severity"], ix.get("source") or "—"])
|
|
|
+ return ("# 补剂相互作用库（视图）\n\n> 由 interactions.json 派生，勿手工编辑。\n\n"
|
|
|
+ + md_table(["id", "类型", "主体A", "主体B", "规则", "说明", "等级", "来源"], rows))
|
|
|
+
|
|
|
+
|
|
|
+def render_protocols(docs, supplement_names):
|
|
|
+ out = ["# 干预方案库（视图）\n", "> 由 protocols/*.json 派生，勿手工编辑。\n"]
|
|
|
+ for doc in docs:
|
|
|
+ out.append(f"\n## {doc['name']}（`{doc['id']}`）\n")
|
|
|
+ out.append(f"> 来源文档：{doc.get('source_doc', '—')}\n")
|
|
|
+ cycles = {c["id"]: c for c in doc.get("cycles", [])}
|
|
|
+ out.append(md_table(["周期", "阶段", "起", "止"], [[cid, c.get("name"), c.get("start"), c.get("end")]
|
|
|
+ for cid, c in cycles.items()]))
|
|
|
+ for member in doc.get("members", []):
|
|
|
+ out.append(f"\n### 成员：{member['name']}（family {member['family_id']}，user {member['user_id']}，手机 {member['phone']}）\n")
|
|
|
+ for phase in member.get("phases", []):
|
|
|
+ cyc = cycles.get(phase["cycle_id"], {})
|
|
|
+ out.append(f"\n#### {cyc.get('name', phase['cycle_id'])} 补剂\n")
|
|
|
+ rows = []
|
|
|
+ for slot in phase.get("time_slots", []):
|
|
|
+ names = "、".join(supplement_names.get(sid, sid) for sid in slot.get("supplement_ids", []))
|
|
|
+ rows.append([slot["time"], slot["label"], names, slot.get("dose", ""), slot.get("notes", "")])
|
|
|
+ out.append(md_table(["时段", "标签", "补剂", "剂量", "备注"], rows))
|
|
|
+ dm = member.get("daily_monitoring", [])
|
|
|
+ if dm:
|
|
|
+ out.append("\n#### 每日监测\n")
|
|
|
+ out.append(md_table(["时间", "内容", "类型"],
|
|
|
+ [[d["time"], d["name"], d.get("kind", "")] for d in dm]))
|
|
|
+ wm = member.get("weekly_monitoring", [])
|
|
|
+ if wm:
|
|
|
+ out.append("\n#### 每周监测\n")
|
|
|
+ out.append(md_table(["周期", "内容", "关联指标"],
|
|
|
+ [[w["weekday"], w["name"], "、".join(w.get("indicator_ids", []))] for w in wm]))
|
|
|
+ fu = member.get("followups", [])
|
|
|
+ if fu:
|
|
|
+ out.append("\n#### 复查节点\n")
|
|
|
+ out.append(md_table(["周期", "项目", "关联指标", "说明"],
|
|
|
+ [[f["week"], f["name"], "、".join(f.get("indicator_ids", [])), f.get("note", "")] for f in fu]))
|
|
|
+ ex = member.get("exercises", [])
|
|
|
+ if ex:
|
|
|
+ out.append("\n#### 运动\n")
|
|
|
+ out.append("、".join(ex) + "\n")
|
|
|
+ return "\n".join(out)
|
|
|
+
|
|
|
+
|
|
|
+def main():
|
|
|
+ argv = sys.argv[1:]
|
|
|
+ check = False
|
|
|
+ if argv and argv[0] == "--check":
|
|
|
+ check = True
|
|
|
+ argv = argv[1:]
|
|
|
+ kb_dir = Path(argv[0]) if argv else Path(DEFAULT_KB)
|
|
|
+ out_dir = Path(argv[1]) if len(argv) > 1 else kb_dir
|
|
|
+
|
|
|
+ supplements = load(kb_dir / "supplements.json")
|
|
|
+ indicators = load(kb_dir / "clinical_indicators.json")
|
|
|
+ exercises = load(kb_dir / "exercises.json")
|
|
|
+ interactions = load(kb_dir / "interactions.json")
|
|
|
+ protocols = [load(p) for p in sorted((kb_dir / "protocols").glob("*.json"))]
|
|
|
+ names = {s["id"]: s["name"] for s in supplements["supplements"]}
|
|
|
+
|
|
|
+ rendered = {
|
|
|
+ "supplements.md": render_supplements(supplements),
|
|
|
+ "clinical_indicators.md": render_indicators(indicators),
|
|
|
+ "exercises.md": render_exercises(exercises),
|
|
|
+ "interactions.md": render_interactions(interactions),
|
|
|
+ "protocols.md": render_protocols(protocols, names),
|
|
|
+ }
|
|
|
+
|
|
|
+ target = out_dir / "views"
|
|
|
+ if check:
|
|
|
+ with tempfile.TemporaryDirectory() as tmp:
|
|
|
+ tmp_dir = Path(tmp) / "views"
|
|
|
+ tmp_dir.mkdir(parents=True, exist_ok=True)
|
|
|
+ for name, text in rendered.items():
|
|
|
+ (tmp_dir / name).write_text(text + "\n", encoding="utf-8")
|
|
|
+ diff_count = 0
|
|
|
+ for name in rendered:
|
|
|
+ existing = target / name
|
|
|
+ if not existing.exists():
|
|
|
+ print(f"[DIFF] {name} 缺失")
|
|
|
+ diff_count += 1
|
|
|
+ continue
|
|
|
+ a = (tmp_dir / name).read_text(encoding="utf-8").splitlines()
|
|
|
+ b = existing.read_text(encoding="utf-8").splitlines()
|
|
|
+ if a != b:
|
|
|
+ diffs = list(difflib.unified_diff(b, a, fromfile=name, tofile=name, lineterm=""))
|
|
|
+ print(f"[DIFF] {name}: {len(diffs)} 行差异（JSON 与视图不同步）")
|
|
|
+ print("\n".join(diffs[:20]))
|
|
|
+ diff_count += 1
|
|
|
+ print("RESULT:", "FAIL" if diff_count else "PASS")
|
|
|
+ sys.exit(1 if diff_count else 0)
|
|
|
+ else:
|
|
|
+ target.mkdir(parents=True, exist_ok=True)
|
|
|
+ for name, text in rendered.items():
|
|
|
+ (target / name).write_text(text + "\n", encoding="utf-8")
|
|
|
+ print(f"wrote {target / name}")
|
|
|
+ print("RESULT: PASS")
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ main()
|
