#!/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", "exercises"): 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("exercises", []): if eid not in refs["exercises"]: errors.append(f"{where}.exercises 引用不存在的运动: {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()