#!/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()