petstore-docs/graph/validate_ontology.py

#!/usr/bin/env python3
"""
宠小它 知识本体校验脚本。

校验 docs/graph/ontology.jsonl 的：
1. JSONL 格式合法（每行一个 JSON 对象）
2. ID 唯一且符合命名规则（entity:/action:/event:/rule:/rel:）
3. 引用完整性：每个被引用的 ID（subject/object/inputs/outputs/appliesTo/appliesRules/emittedBy/appliesTo）都存在
4. 证据强度分布统计
5. 类型必填字段检查（Entity/Action/Event/Rule/Relation 各自的最小字段集）

用法：
    python3 docs/graph/validate_ontology.py [docs_root]

默认 docs_root = docs（相对当前工作目录）。
退出码：0 = 全部通过；1 = 有错误。
"""

import json
import os
import re
import sys
from collections import Counter

VALID_EVIDENCE = {"anchored", "documented", "implemented", "gap"}
VALID_TYPES = {"Entity", "Action", "Event", "Rule", "Relation"}
ID_PREFIX_BY_TYPE = {
    "Entity": "entity:",
    "Action": "action:",
    "Event": "event:",
    "Rule": "rule:",
    "Relation": "rel:",
}

REQUIRED_FIELDS_BY_TYPE = {
    "Entity": {"id", "type", "name", "fields", "evidence"},
    "Action": {"id", "type", "name", "entrypoints", "evidence"},
    "Event": {"id", "type", "name", "domain", "emittedBy", "evidence"},
    "Rule": {"id", "type", "name", "appliesTo", "evidence"},
    "Relation": {"id", "type", "subject", "predicate", "object", "evidence"},
}

# 引用字段：这些字段里看起来像本体 ID 的字符串（以 entity:/action:/event:/rule:/rel: 开头）必须存在
# 原始字段名（如 phone、code、token）不视为引用，跳过
REFERENCE_FIELDS = {
    "inputs", "outputs", "appliesTo", "appliesRules", "emittedBy",
    "subject", "object",
}
ID_PREFIXES = ("entity:", "action:", "event:", "rule:", "rel:")


def is_id_like(s):
    return isinstance(s, str) and s.startswith(ID_PREFIXES)


def load_jsonl(path):
    entries = []
    errors = []
    with open(path, "r", encoding="utf-8") as f:
        for lineno, line in enumerate(f, 1):
            line = line.strip()
            if not line:
                continue
            try:
                obj = json.loads(line)
                obj["_lineno"] = lineno
                entries.append(obj)
            except json.JSONDecodeError as e:
                errors.append(f"line {lineno}: invalid JSON - {e}")
    return entries, errors


def validate_ids(entries):
    errors = []
    ids = set()
    for e in entries:
        eid = e.get("id")
        if not eid:
            errors.append(f"line {e['_lineno']}: missing id")
            continue
        if eid in ids:
            errors.append(f"line {e['_lineno']}: duplicate id {eid}")
        ids.add(eid)
        etype = e.get("type")
        expected_prefix = ID_PREFIX_BY_TYPE.get(etype)
        if expected_prefix and not eid.startswith(expected_prefix):
            errors.append(f"line {e['_lineno']}: id {eid} should start with {expected_prefix} for type {etype}")
    return ids, errors


def validate_required_fields(entries):
    errors = []
    for e in entries:
        etype = e.get("type")
        if etype not in VALID_TYPES:
            errors.append(f"line {e['_lineno']}: unknown type {etype}")
            continue
        required = REQUIRED_FIELDS_BY_TYPE.get(etype, set())
        missing = required - set(e.keys())
        if missing:
            errors.append(f"line {e['_lineno']}: {etype} {e.get('id')} missing fields {sorted(missing)}")
    return errors


def validate_references(entries, all_ids):
    errors = []
    for e in entries:
        eid = e.get("id", "?")
        for field in REFERENCE_FIELDS:
            if field not in e:
                continue
            val = e[field]
            if val is None:
                continue
            if isinstance(val, str):
                refs = [val]
            elif isinstance(val, list):
                refs = val
            else:
                errors.append(f"line {e['_lineno']}: {eid}.{field} must be string or list, got {type(val).__name__}")
                continue
            for ref in refs:
                if not ref:
                    continue
                # 只校验看起来像本体 ID 的引用；原始字段名（phone/code/token）跳过
                if not is_id_like(ref):
                    continue
                if ref not in all_ids:
                    errors.append(f"line {e['_lineno']}: {eid}.{field} references unknown id {ref}")
    return errors


def validate_evidence(entries):
    errors = []
    for e in entries:
        ev = e.get("evidence")
        if ev is None:
            errors.append(f"line {e['_lineno']}: {e.get('id')} missing evidence")
            continue
        if ev not in VALID_EVIDENCE:
            errors.append(f"line {e['_lineno']}: {e.get('id')} invalid evidence '{ev}', must be one of {VALID_EVIDENCE}")
    return errors


def print_stats(entries):
    type_counts = Counter(e.get("type", "?") for e in entries)
    evidence_counts = Counter(e.get("evidence", "?") for e in entries)
    print("\n=== 本体统计 ===")
    print(f"总条目数: {len(entries)}")
    print("\n按类型:")
    for t in sorted(type_counts):
        print(f"  {t:10s} {type_counts[t]}")
    print("\n按证据强度:")
    for ev in sorted(evidence_counts):
        print(f"  {ev:12s} {evidence_counts[ev]}")
    total = len(entries)
    if total:
        anchored = evidence_counts.get("anchored", 0)
        print(f"\nanchored 占比: {anchored}/{total} = {anchored/total*100:.1f}%")


def main():
    docs_root = sys.argv[1] if len(sys.argv) > 1 else "docs"
    jsonl_path = os.path.join(docs_root, "graph", "ontology.jsonl")
    if not os.path.exists(jsonl_path):
        print(f"ERROR: {jsonl_path} not found")
        sys.exit(1)

    entries, errors = load_jsonl(jsonl_path)
    if errors:
        print("=== JSONL 解析错误 ===")
        for err in errors:
            print(f"  {err}")
        sys.exit(1)

    all_ids, id_errors = validate_ids(entries)
    field_errors = validate_required_fields(entries)
    ref_errors = validate_references(entries, all_ids)
    ev_errors = validate_evidence(entries)

    all_errors = id_errors + field_errors + ref_errors + ev_errors
    if all_errors:
        print("=== 校验失败 ===")
        for err in all_errors:
            print(f"  {err}")
        print(f"\n共 {len(all_errors)} 个错误")
        print_stats(entries)
        sys.exit(1)

    print("=== 校验通过 ===")
    print(f"  {jsonl_path}: {len(entries)} 条目，引用完整，证据强度合法")
    print_stats(entries)
    sys.exit(0)


if __name__ == "__main__":
    main()