cfc/docs/scripts/render_kb_md.py

#!/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("description", ""), 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.get("time", ""), w["name"], w.get("description", ""),
                                      "、".join(w.get("indicator_ids", []))] for w in wm]))
            fu = member.get("followups", [])
            if fu:
                out.append("\n#### 复查节点\n")
                out.append(md_table(["周期", "日期", "项目", "关联指标", "说明"],
                                    [[f.get("week", ""), f.get("due", ""), f["name"],
                                      "、".join(f.get("indicator_ids", [])), f.get("note", "")] for f in fu]))
            ex = member.get("exercise_tasks", [])
            if ex:
                out.append("\n#### 运动任务\n")
                out.append(md_table(["时间", "任务", "描述", "引用运动"],
                                    [[e["time"], e["title"], e.get("description", ""),
                                      "、".join(e.get("exercise_ids", []))] for e in ex]))
    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()