kg-agent/scripts/export_meta_schema.py

"""导出元知识图谱 Schema 为前端展示用 JSON。

输出：
  output/meta_graph_schema.json             # 完整 schema（含规划中节点）
  output/meta_graph_schema_display.json     # 仅已启用节点/关系，适合前端展示

内容包括：
  - nodes: 节点/实体类型，含属性列表、部门、是否中心节点、是否建议节点、是否已启用
  - relations: 节点间连接关系，含源节点、目标节点、关系名、关联键、基数、说明、是否已启用
"""

from __future__ import annotations

import json
import re
import sys
from datetime import date
from pathlib import Path

from step2_graph_building.meta.schema import ENTITIES, RELATIONS


sys.stdout.reconfigure(encoding="utf-8")
ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / "output" / "meta_graph_schema.json"
OUT_DISPLAY = ROOT / "output" / "meta_graph_schema_display.json"


def split_attributes(desc: str) -> list[str]:
    """从 EntitySpec.desc 中提取属性名。

    desc 基本是“属性1 / 属性2 / ...”格式；括号内的斜杠不拆分。
    """
    if not desc:
        return []
    # 先保护括号内的“/”，例如 合同状况(线上/线下)、服务状态(服务中/历史)
    protected = re.sub(
        r"([（(][^）)]*)/([^）)]*[）)])",
        lambda m: m.group(0).replace("/", "__SLASH__"),
        desc,
    )
    parts = re.split(r"\s*/\s*|；|;", protected)
    result: list[str] = []
    for p in parts:
        p = p.strip().replace("__SLASH__", "/")
        if not p:
            continue
        # 去掉解释性括号，但保留括号内容作为属性说明
        result.append(re.sub(r"\s+", "", p))
    return result


def build_payload() -> dict:
    active_names = {e.name for e in ENTITIES if e.active}

    nodes = []
    for e in ENTITIES:
        dept = [e.department] if isinstance(e.department, str) else list(e.department)
        nodes.append({
            "id": e.name,
            "name": e.name,
            "department": dept,
            "is_hub": e.is_hub,
            "suggested": e.suggested,
            "active": e.active,
            "attributes": split_attributes(e.desc),
            "description": e.desc,
        })

    relations = []
    for r in RELATIONS:
        relations.append({
            "id": f"{r.source}__{r.rel_type}__{r.target}",
            "source": r.source,
            "target": r.target,
            "type": r.rel_type,
            "key": r.key,
            "cardinality": r.cardinality,
            "description": r.desc,
            "active": r.source in active_names and r.target in active_names,
        })

    return {
        "meta": {
            "title": "申勤物业元知识图谱 Schema",
            "source": "src/step2_graph_building/meta/schema.py",
            "generated_at": date.today().isoformat(),
            "entity_count": len(nodes),
            "active_entity_count": sum(1 for n in nodes if n["active"]),
            "relation_count": len(relations),
            "active_relation_count": sum(1 for r in relations if r["active"]),
        },
        "nodes": nodes,
        "relations": relations,
    }


def main() -> None:
    from step2_graph_building.meta.production import load_production_schema, export_schema
    production = load_production_schema()
    if production is not None:
        export_schema(production, OUT.parent)
        print("已重新导出当前生产元图谱；没有使用历史固定 Schema 覆盖。")
        return
    payload = build_payload()
    OUT.parent.mkdir(parents=True, exist_ok=True)
    OUT.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")

    display_nodes = [n for n in payload["nodes"] if n["active"]]
    display_relations = [r for r in payload["relations"] if r["active"]]
    display = {
        "meta": {
            **payload["meta"],
            "title": "申勤物业元知识图谱 Schema（仅已启用节点/关系）",
            "entity_count": len(display_nodes),
            "active_entity_count": len(display_nodes),
            "relation_count": len(display_relations),
            "active_relation_count": len(display_relations),
        },
        "nodes": display_nodes,
        "relations": display_relations,
    }
    OUT_DISPLAY.write_text(json.dumps(display, ensure_ascii=False, indent=2), encoding="utf-8")

    print("完整 schema:", OUT)
    print(f"  节点: {payload['meta']['entity_count']}（启用 {payload['meta']['active_entity_count']}）")
    print(f"  关系: {payload['meta']['relation_count']}（启用 {payload['meta']['active_relation_count']}）")
    print("展示版 schema:", OUT_DISPLAY)
    print(f"  节点: {len(display_nodes)}")
    print(f"  关系: {len(display_relations)}")


if __name__ == "__main__":
    main()