|
|
@@ -1,10 +1,17 @@
|
|
|
#!/usr/bin/env python3
|
|
|
-"""知识库质量校验:JSON Schema + 引用完整性。
|
|
|
+"""知识库质量校验:JSON Schema 结构 + 跨库引用完整性。
|
|
|
|
|
|
用法:
|
|
|
python3 validate_kb.py [知识库目录]
|
|
|
-默认校验 docs/参考资料/知识库/ 下新增的 3 个子库及其 schema。
|
|
|
-无 jsonschema 依赖时降级为内置结构校验。
|
|
|
+
|
|
|
+校验清单:
|
|
|
+ 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
|
|
|
@@ -17,12 +24,17 @@ 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 = {"益生菌", "益生元", "氨基酸", "维生素", "矿物质", "脂肪酸", "植物提取物"}
|
|
|
@@ -30,6 +42,8 @@ TIMING_ENUM = {"晨起空腹", "早餐后", "午餐后", "晚餐后", "睡前",
|
|
|
FORM_ENUM = {"粉剂", "胶囊", "片剂", "滴剂", "颗粒", "液体"}
|
|
|
EX_TYPE_ENUM = {"有氧", "抗阻", "柔韧", "平衡", "其他"}
|
|
|
EX_INTENSITY_ENUM = {"低", "中低", "中", "中高", "高"}
|
|
|
+IX_TYPE_ENUM = {"协同", "拮抗", "时序禁忌", "安全警示"}
|
|
|
+IX_SEVERITY_ENUM = {"提示", "警示", "禁忌"}
|
|
|
|
|
|
|
|
|
def load(path: Path):
|
|
|
@@ -39,94 +53,207 @@ def load(path: Path):
|
|
|
raise SystemExit(f"JSON 解析失败: {path} -> {e}")
|
|
|
|
|
|
|
|
|
-def class_check(field, value, allowed, errors, where):
|
|
|
+def check_enum(field, value, allowed, errors, where):
|
|
|
if value not in allowed:
|
|
|
errors.append(f"{where}.{field}={value!r} 不在枚举 {sorted(allowed)} 内")
|
|
|
|
|
|
|
|
|
-def validate_doc(path: Path, schema_path, kind):
|
|
|
- errors = []
|
|
|
- doc = {"doc": load(path), "schema": load(schema_path)}
|
|
|
- root = doc["doc"]
|
|
|
- items = root.get(kind)
|
|
|
- if items is None:
|
|
|
- raise SystemExit(f"{path} 缺少顶层数组字段 {kind}")
|
|
|
-
|
|
|
+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
|
|
|
- id_key = ID_PATTERNS[path.name][1]
|
|
|
- if id_key not in item:
|
|
|
- errors.append(f"{where} 缺 id")
|
|
|
- else:
|
|
|
- iid = item[id_key]
|
|
|
- if iid in seen:
|
|
|
- errors.append(f"{where} 重复 id: {iid}")
|
|
|
- seen.add(iid)
|
|
|
- pat = ID_PATTERNS[path.name][2]
|
|
|
- if not pat.match(iid):
|
|
|
- errors.append(f"{where} id={iid!r} 不匹配 {pat.pattern}")
|
|
|
-
|
|
|
- if path.name == "supplements.json":
|
|
|
- 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}")
|
|
|
- class_check("category", item.get("category"), CATEGORY_ENUM, errors, where)
|
|
|
- class_check("timing", item.get("timing"), TIMING_ENUM, errors, where)
|
|
|
- class_check("form", item.get("form"), FORM_ENUM, errors, where)
|
|
|
- for ing in item.get("key_ingredients", []):
|
|
|
- if ing.get("name") is None or ing.get("amount") is None:
|
|
|
- errors.append(f"{where}.key_ingredients 缺 name/amount")
|
|
|
- elif path.name == "clinical_indicators.json":
|
|
|
- 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")
|
|
|
- elif path.name == "exercises.json":
|
|
|
- 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}")
|
|
|
- class_check("type", item.get("type"), EX_TYPE_ENUM, errors, where)
|
|
|
- class_check("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 需为正整数")
|
|
|
-
|
|
|
- return errors
|
|
|
+ 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
|
|
|
- schema_path = KB_DIR / schema_rel
|
|
|
if not data_path.exists():
|
|
|
print(f"[SKIP] {data_path} 不存在")
|
|
|
- failed = True
|
|
|
continue
|
|
|
- if not schema_path.exists():
|
|
|
- print(f"[FAIL] schema 缺失: {schema_path}")
|
|
|
+ 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]
|
|
|
- errors = validate_doc(data_path, schema_path, kind)
|
|
|
+ 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[:15]:
|
|
|
+ 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 ] {data_name} 校验通过")
|
|
|
+ print(f"[ OK ] {proto.name}")
|
|
|
|
|
|
print("RESULT:", "FAIL" if failed else "PASS")
|
|
|
sys.exit(1 if failed else 0)
|