cfc/docs/scripts/validate_kb.py

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

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

校验清单：
  1. 各子库 JSON 可解析、顶层字段存在；
  2. id 唯一、id 前缀模式正确（sup-/ind-/ex-/ix-/protocol-）；
  3. 枚举值合法（category/timing/form/type/intensity/severity 等）；
  4. 跨库引用完整性：
     - protocols: supplement_ids->supplements.json, exercise_ids->exercises.json,
       indicator_ids->clinical_indicators.json, cycle_id->cycles
     - interactions: supplement_id->supplements.json
"""
import json
import re
import sys
from pathlib import Path

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

SCHEMA_FILES = {
    "supplements.json": "schema/supplement.schema.json",
    "clinical_indicators.json": "schema/clinical_indicator.schema.json",
    "exercises.json": "schema/exercise.schema.json",
    "interactions.json": "schema/interaction.schema.json",
}

PROTOCOL_SCHEMA = "schema/protocol.schema.json"

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(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: Path, items, item_key, id_pattern, errors):
    seen = set()
    for i, item in enumerate(items):
        where = f"{path.name}[{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.name}[{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}[{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.name}[{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.name}[{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(path, 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"{path.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(path, 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"{path.name}[{i}].{part} 引用不存在的补充剂: {sid}")


def build_refs():
    refs = {"supplements": set(), "indicators": set(), "exercises": set()}
    for data_name, kind_id in (("supplements.json", 0), ("clinical_indicators.json", 0), ("exercises.json", 0)):
        p = KB_DIR / data_name
        if not p.exists():
            continue
        doc = load(p)
        kind = ID_PATTERNS[data_name][0]
        for item in doc.get(kind, []):
            iid = item.get("id")
            if iid:
                refs["supplements" if data_name == "supplements.json" else "indicators" if data_name == "clinical_indicators.json" else "exercises"].add(iid)
    return refs


def main():
    failed = False
    refs = build_refs()

    for data_name, schema_rel in SCHEMA_FILES.items():
        data_path = KB_DIR / data_name
        if not data_path.exists():
            print(f"[SKIP] {data_path} 不存在")
            continue
        if not (KB_DIR / schema_rel).exists():
            print(f"[FAIL] schema 缺失: {schema_rel}")
            failed = True
            continue
        doc = load(data_path)
        kind = ID_PATTERNS[data_name][0]
        items = doc.get(kind)
        if items is None:
            print(f"[FAIL] {data_name} 缺少顶层数组字段 {kind}")
            failed = True
            continue
        errors = []
        validate_list_ids(data_path, items, "id", ID_PATTERNS[data_name][2], errors)
        if data_name == "supplements.json":
            validate_supplements(data_path, items, errors)
        elif data_name == "clinical_indicators.json":
            validate_clinical(data_path, items, errors)
        elif data_name == "exercises.json":
            validate_exercises(data_path, items, errors)
        elif data_name == "interactions.json":
            validate_interactions(data_path, items, errors)
            validate_interaction_refs(data_path, items, errors, refs)
        if errors:
            failed = True
            print(f"[FAIL] {data_name}: {len(errors)} 个问题")
            for e in errors[:20]:
                print("   -", e)
        else:
            print(f"[ OK ] {data_name}")

    for proto in sorted((KB_DIR / "protocols").glob("*.json")):
        if not (KB_DIR / PROTOCOL_SCHEMA).exists():
            print(f"[FAIL] schema 缺失: {PROTOCOL_SCHEMA}")
            failed = True
            break
        doc = load(proto)
        if doc.get("id") is None:
            print(f"[FAIL] {proto.name} 缺 id")
            failed = True
            continue
        pat = ID_PATTERNS["protocols"][2]
        if not pat.match(doc["id"]):
            print(f"[FAIL] {proto.name} id={doc['id']!r} 不匹配 {pat.pattern}")
            failed = True
            continue
        errors = []
        validate_protocol(proto, doc, errors, refs)
        if errors:
            failed = True
            print(f"[FAIL] {proto.name}: {len(errors)} 个问题")
            for e in errors[:25]:
                print("   -", e)
        else:
            print(f"[ OK ] {proto.name}")

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


if __name__ == "__main__":
    main()