cfc/docs/scripts/validate_kb.py

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

用法：
    python3 validate_kb.py [知识库目录]
默认校验 docs/参考资料/知识库/ 下新增的 3 个子库及其 schema。
无 jsonschema 依赖时降级为内置结构校验。
"""
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",
}

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-]+$")),
}

CATEGORY_ENUM = {"益生菌", "益生元", "氨基酸", "维生素", "矿物质", "脂肪酸", "植物提取物"}
TIMING_ENUM = {"晨起空腹", "早餐后", "午餐后", "晚餐后", "睡前", "可选"}
FORM_ENUM = {"粉剂", "胶囊", "片剂", "滴剂", "颗粒", "液体"}
EX_TYPE_ENUM = {"有氧", "抗阻", "柔韧", "平衡", "其他"}
EX_INTENSITY_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 class_check(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}")

    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


def main():
    failed = False
    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}")
            failed = True
            continue
        kind = ID_PATTERNS[data_name][0]
        errors = validate_doc(data_path, schema_path, kind)
        if errors:
            failed = True
            print(f"[FAIL] {data_name}: {len(errors)} 个问题")
            for e in errors[:15]:
                print("   -", e)
        else:
            print(f"[ OK ] {data_name} 校验通过")

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


if __name__ == "__main__":
    main()