cfc/docs/scripts/validate_kb.py

#!/usr/bin/env python3
"""知识库质量校验：JSON Schema 结构 + 跨库引用完整性。

用法：
    python3 validate_kb.py [知识库目录]

自 v3.0 起，所有子库已并入「统一知识库.json」（整体知识库基础）。本脚本改为从
统一知识库内的对应 section 读取并校验，id 前缀与枚举规则保持不变。

校验清单：
  1. 各子库 JSON 可解析、顶层字段存在；
  2. id 唯一、id 前缀模式正确（sup-/ind-/ex-/ix-/protocol-）；
  3. 枚举值合法（category/timing/form/type/intensity/severity 等）；
  4. 跨库引用完整性：
     - 干预方案库: supplement_ids->补充剂库, exercise_ids->运动处方库,
       indicator_ids->临床监测指标库
     - 补剂相互作用库: supplement_id->补充剂库
"""
import json
import re
import sys
from pathlib import Path

KB_DIR = Path(sys.argv[1] if len(sys.argv) > 1 else "docs/参考资料/知识库")
UNIFIED = KB_DIR / "统一知识库.json"

# 原独立文件名 -> (统一库 section, 内部数组字段, schema 相对路径)
SECTION_MAP = {
    "supplements.json": ("补充剂库", "supplements", "schema/supplement.schema.json"),
    "clinical_indicators.json": ("临床监测指标库", "indicators", "schema/clinical_indicator.schema.json"),
    "exercises.json": ("运动处方库", "exercises", "schema/exercise.schema.json"),
    "interactions.json": ("补剂相互作用库", "interactions", "schema/interaction.schema.json"),
}
PROTOCOLS_SECTION = "干预方案库"

ID_PATTERNS = {
    "supplements.json": ("supplements", "id", re.compile(r"^sup-[a-z0-9-]+$")),
    "clinical_indicators.json": ("indicators", "id", re.compile(r"^ind-[a-z0-9-]+$")),
    "exercises.json": ("exercises", "id", re.compile(r"^ex-[a-z0-9-]+$")),
    "interactions.json": ("interactions", "id", re.compile(r"^ix-[a-z0-9-]+$")),
    "protocols": ("members/phases", "protocol", re.compile(r"^protocol-[a-z0-9-]+$")),
}

CATEGORY_ENUM = {"益生菌", "益生元", "氨基酸", "维生素", "矿物质", "脂肪酸", "植物提取物"}
TIMING_ENUM = {"晨起空腹", "早餐后", "午餐后", "晚餐后", "睡前", "可选"}
FORM_ENUM = {"粉剂", "胶囊", "片剂", "滴剂", "颗粒", "液体"}
EX_TYPE_ENUM = {"有氧", "抗阻", "柔韧", "平衡", "其他"}
EX_INTENSITY_ENUM = {"低", "中低", "中", "中高", "高"}
IX_TYPE_ENUM = {"协同", "拮抗", "时序禁忌", "安全警示"}
IX_SEVERITY_ENUM = {"提示", "警示", "禁忌"}


def load_unified():
    try:
        return json.loads(UNIFIED.read_text(encoding="utf-8"))
    except Exception as e:  # noqa: BLE001
        raise SystemExit(f"统一知识库解析失败: {UNIFIED} -> {e}")


def load(path: Path):
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:  # noqa: BLE001
        raise SystemExit(f"JSON 解析失败: {path} -> {e}")


def check_enum(field, value, allowed, errors, where):
    if value not in allowed:
        errors.append(f"{where}.{field}={value!r} 不在枚举 {sorted(allowed)} 内")


def validate_list_ids(path, items, item_key, id_pattern, errors):
    seen = set()
    for i, item in enumerate(items):
        where = f"{path}[{i}]"
        if not isinstance(item, dict):
            errors.append(f"{where} 不是对象")
            continue
        iid = item.get(item_key)
        if not iid:
            errors.append(f"{where} 缺 {item_key}")
            continue
        if iid in seen:
            errors.append(f"{where} 重复 {item_key}: {iid}")
        seen.add(iid)
        if not id_pattern.match(iid):
            errors.append(f"{where} {item_key}={iid!r} 不匹配 {id_pattern.pattern}")


def validate_supplements(path, items, errors):
    for i, item in enumerate(items):
        where = f"{path}[{i}]"
        for req in ("name", "category", "key_ingredients", "form", "dose_reference", "timing"):
            if req not in item:
                errors.append(f"{where} 缺必需字段 {req}")
        check_enum("category", item.get("category"), CATEGORY_ENUM, errors, where)
        check_enum("timing", item.get("timing"), TIMING_ENUM, errors, where)
        check_enum("form", item.get("form"), FORM_ENUM, errors, where)
        for ing in item.get("key_ingredients", []):
            if not isinstance(ing, dict) or ing.get("name") is None or ing.get("amount") is None:
                errors.append(f"{where}.key_ingredients 缺 name/amount")


def validate_clinical(path, items, errors):
    for i, item in enumerate(items):
        where = f"{path.name if hasattr(path, 'name') else path}[{i}]"
        for req in ("name", "unit", "normal_range"):
            if req not in item:
                errors.append(f"{where} 缺必需字段 {req}")
        nr = item.get("normal_range")
        if not isinstance(nr, dict) or "low" not in nr or "high" not in nr:
            errors.append(f"{where}.normal_range 需含 low/high")


def validate_exercises(path, items, errors):
    for i, item in enumerate(items):
        where = f"{path}[{i}]"
        for req in ("name", "type", "intensity", "duration_min", "frequency"):
            if req not in item:
                errors.append(f"{where} 缺必需字段 {req}")
        check_enum("type", item.get("type"), EX_TYPE_ENUM, errors, where)
        check_enum("intensity", item.get("intensity"), EX_INTENSITY_ENUM, errors, where)
        if not isinstance(item.get("duration_min"), int) or item.get("duration_min") < 1:
            errors.append(f"{where}.duration_min 需为正整数")


def validate_interactions(path, items, errors):
    for i, item in enumerate(items):
        where = f"{path}[{i}]"
        for req in ("type", "subjects_a", "subjects_b", "rule", "detail", "severity"):
            if req not in item:
                errors.append(f"{where} 缺必需字段 {req}")
        check_enum("type", item.get("type"), IX_TYPE_ENUM, errors, where)
        check_enum("severity", item.get("severity"), IX_SEVERITY_ENUM, errors, where)
        for part in ("subjects_a", "subjects_b"):
            s = item.get(part)
            if not isinstance(s, dict):
                errors.append(f"{where}.{part} 需为对象")
                continue
            if "supplement_id" not in s and "substance" not in s:
                errors.append(f"{where}.{part} 需含 supplement_id 或 substance")


def validate_protocol(doc, errors, refs):
    root = doc
    cycles = dict((c["id"], c) for c in root.get("cycles", []) if isinstance(c, dict) and c.get("id"))
    for i, member in enumerate(root.get("members", [])):
        where = f"方案{member.get('name','?')} 成员[{i}]"
        for req in ("family_id", "name", "user_id", "family_member_id", "phone", "phases",
                    "daily_monitoring", "weekly_monitoring", "followups", "exercise_tasks"):
            if req not in member:
                errors.append(f"{where} 缺必需字段 {req}")
        for phase in member.get("phases", []):
            ph = f"{where}.phases"
            if not isinstance(phase, dict) or "cycle_id" not in phase or "time_slots" not in phase:
                errors.append(f"{ph} 缺 cycle_id/time_slots")
                continue
            if phase["cycle_id"] not in cycles:
                errors.append(f"{ph}.cycle_id={phase['cycle_id']!r} 未在 cycles 中定义")
            for slot in phase.get("time_slots", []):
                sl = f"{ph}.time_slots"
                if not isinstance(slot, dict) or "supplement_ids" not in slot:
                    errors.append(f"{sl} 缺 supplement_ids")
                    continue
                for sid in slot["supplement_ids"]:
                    if sid not in refs["supplements"]:
                        errors.append(f"{sl} 引用不存在的补充剂: {sid}")
        for wm in member.get("weekly_monitoring", []):
            for iid in wm.get("indicator_ids", []):
                if iid not in refs["indicators"]:
                    errors.append(f"{where}.weekly_monitoring 引用不存在的指标: {iid}")
        for fu in member.get("followups", []):
            for iid in fu.get("indicator_ids", []):
                if iid not in refs["indicators"]:
                    errors.append(f"{where}.followups 引用不存在的指标: {iid}")
        for eid in member.get("exercise_ids", []):
            if eid not in refs["exercises"]:
                errors.append(f"{where}.exercise_ids 引用不存在的运动: {eid}")
        for et in member.get("exercise_tasks", []):
            if not isinstance(et, dict) or "title" not in et or "exercise_ids" not in et:
                errors.append(f"{where}.exercise_tasks 缺 title/exercise_ids")
                continue
            for eid in et.get("exercise_ids", []):
                if eid not in refs["exercises"]:
                    errors.append(f"{where}.exercise_tasks 引用不存在的运动: {eid}")


def validate_interaction_refs(items, errors, refs):
    for i, item in enumerate(items):
        for part in ("subjects_a", "subjects_b"):
            s = item.get(part) or {}
            sid = s.get("supplement_id")
            if sid and sid not in refs["supplements"]:
                errors.append(f"补剂相互作用库[{i}].{part} 引用不存在的补充剂: {sid}")


def build_refs(unified):
    # 仅补充剂/指标/运动作为「被引用方」加入 refs；相互作用库是引用方，不加入。
    refs = {"supplements": set(), "indicators": set(), "exercises": set()}
    for data_name, (section, inner, _schema) in SECTION_MAP.items():
        if ID_PATTERNS[data_name][0] not in refs:
            continue
        doc = unified.get(section)
        if not isinstance(doc, dict):
            continue
        for item in doc.get(inner, []):
            iid = item.get("id")
            if iid:
                refs[ID_PATTERNS[data_name][0]].add(iid)
    return refs


def main():
    if not UNIFIED.exists():
        raise SystemExit(f"未找到统一知识库: {UNIFIED}")
    unified = load_unified()
    failed = False
    refs = build_refs(unified)

    for data_name, (section, inner, schema_rel) in SECTION_MAP.items():
        if not (KB_DIR / schema_rel).exists():
            print(f"[FAIL] schema 缺失: {schema_rel}")
            failed = True
            continue
        doc = unified.get(section)
        if not isinstance(doc, dict) or inner not in doc:
            print(f"[FAIL] 统一库缺少 section `{section}.{inner}`")
            failed = True
            continue
        items = doc[inner]
        path = f"统一库.{section}.{inner}"
        errors = []
        validate_list_ids(path, items, "id", ID_PATTERNS[data_name][2], errors)
        if data_name == "supplements.json":
            validate_supplements(path, items, errors)
        elif data_name == "clinical_indicators.json":
            validate_clinical(path, items, errors)
        elif data_name == "exercises.json":
            validate_exercises(path, items, errors)
        elif data_name == "interactions.json":
            validate_interactions(path, items, errors)
            validate_interaction_refs(items, errors, refs)
        if errors:
            failed = True
            print(f"[FAIL] {section}.{inner}: {len(errors)} 个问题")
            for e in errors[:20]:
                print("   -", e)
        else:
            print(f"[ OK ] {section}.{inner} ({len(items)} 条)")

    # 干预方案库
    proto_doc = unified.get(PROTOCOLS_SECTION)
    if not isinstance(proto_doc, dict) or "protocols" not in proto_doc:
        print(f"[FAIL] 统一库缺少 section `{PROTOCOLS_SECTION}.protocols`")
        failed = True
    else:
        for doc in proto_doc["protocols"]:
            pid = doc.get("id")
            if pid is None:
                print(f"[FAIL] 方案缺 id")
                failed = True
                continue
            if not ID_PATTERNS["protocols"][2].match(pid):
                print(f"[FAIL] 方案 id={pid!r} 不匹配 {ID_PATTERNS['protocols'][2].pattern}")
                failed = True
                continue
            errors = []
            validate_protocol(doc, errors, refs)
            if errors:
                failed = True
                print(f"[FAIL] {pid}: {len(errors)} 个问题")
                for e in errors[:25]:
                    print("   -", e)
            else:
                print(f"[ OK ] {pid}")

    print("RESULT:", "FAIL" if failed else "PASS")
    sys.exit(1 if failed else 0)


if __name__ == "__main__":
    main()