llm_wiki/tools/scripts/manage-relations.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
manage-relations.py — 管理 wiki 页面的 typed relationships

用法：
    python manage-relations.py list <page>        # 查看页面 relations
    python manage-relations.py add <page> --type <type> --target "页面" [--desc "说明"] [--conf 3]
    python manage-relations.py remove <page> --type <type> --target "页面"
    python manage-relations.py graph <page>        # 输出 DOT 格式
    python manage-relations.py stats               # 统计关系网络

关系类型：
    depends_on | conflicts_with | supersedes | caused_by | supports | extends | part_of | example_of
"""
import argparse
import re
import sys
from pathlib import Path

try:
    import yaml
except ImportError:
    yaml = None

VAULT = Path(r"D:\Applications\app\kepano-obsidian-main")
WIKI = VAULT / "wiki"

VALID_TYPES = [
    "depends_on", "conflicts_with", "supersedes", "caused_by",
    "supports", "extends", "part_of", "example_of"
]


def read_page(page: str) -> tuple[dict | None, str | None, str | None]:
    """返回 (frontmatter_dict, body_text, error_msg)"""
    fp = WIKI / f"{page}.md"
    if not fp.exists():
        return None, None, f"ERROR: {page}.md not found"
    content = fp.read_text(encoding="utf-8")
    m = re.match(r"^---\n(.*?)\n---", content, re.DOTALL)
    if not m:
        return None, None, f"ERROR: {page}.md has no frontmatter"
    body = content[m.end():]
    if yaml:
        try:
            front = yaml.safe_load(m.group(1)) or {}
        except yaml.YAMLError as e:
            return None, None, f"ERROR: YAML parse error: {e}"
    else:
        return None, None, "ERROR: PyYAML not installed (pip install pyyaml)"
    return front, body, None


def write_page(page: str, front: dict, body: str) -> str | None:
    """写入页面，返回 error_msg"""
    fp = WIKI / f"{page}.md"
    new_yaml = yaml.dump(front, allow_unicode=True, default_flow_style=False, sort_keys=False)
    fp.write_text(f"---\n{new_yaml}---\n{body.lstrip()}", encoding="utf-8")
    return None


def cmd_list(page: str):
    front, body, err = read_page(page)
    if err:
        print(err); return
    rels = front.get("relations", [])
    if not rels:
        print(f"  [[{page}]]: no relations")
        return
    print(f"[[{page}]] ({len(rels)} relations):")
    for r in rels:
        desc = f" — {r.get('description', '')}" if r.get("description") else ""
        conf = f" [conf={r.get('confidence', 3)}]" if r.get("confidence", 3) != 3 else ""
        target_name = r['target'].strip('[]')
        print(f"  {r['type']} --> [[{target_name}]]{desc}{conf}")


def cmd_add(page: str, rel_type: str, target: str, desc: str | None, conf: int):
    front, body, err = read_page(page)
    if err:
        print(err); return
    if "relations" not in front:
        front["relations"] = []
    # Dedup
    for r in front["relations"]:
        if r.get("type") == rel_type and r.get("target") == target:
            print(f"  Already exists: {rel_type} [[{target}]]")
            return
    entry = {"type": rel_type, "target": f"[[{target}]]"}
    if desc:
        entry["description"] = desc
    if conf < 5:
        entry["confidence"] = conf
    front["relations"].append(entry)
    err = write_page(page, front, body)
    if err:
        print(err)
    else:
        print(f"  Added: [[{page}]] --{rel_type}--> [[{target}]]")


def cmd_remove(page: str, rel_type: str, target: str):
    front, body, err = read_page(page)
    if err:
        print(err); return
    rels = front.get("relations", [])
    before = len(rels)
    front["relations"] = [
        r for r in rels
        if not (r.get("type") == rel_type and r.get("target", "").strip("[]") == target)
    ]
    if len(front["relations"]) == before:
        print(f"  Not found: {rel_type} [[{target}]]")
        return
    if not front["relations"]:
        del front["relations"]
    err = write_page(page, front, body)
    if err:
        print(err)
    else:
        print(f"  Removed: [[{page}]] --{rel_type}--> [[{target}]]")


def cmd_graph(page: str):
    """输出 DOT 格式"""
    front, body, err = read_page(page)
    if err:
        print(err); return
    rels = front.get("relations", [])
    print(f"digraph {page} {{")
    print(f'  "{page}" [style=filled, fillcolor=lightblue];')
    for r in rels:
        target = r["target"].strip("[]")
        label = r["type"]
        print(f'  "{page}" -> "{target}" [label="{label}"];')
    print("}")


def cmd_stats():
    """统计整个 wiki 的关系网络"""
    all_pages = list(WIKI.glob("*.md"))
    total = 0
    type_counts = {}
    pages_with_rels = 0
    for fp in all_pages:
        if fp.name in ("index.md", "log.md"):
            continue
        front, _, err = read_page(fp.stem)
        if err or not front:
            continue
        rels = front.get("relations", [])
        if rels:
            pages_with_rels += 1
            total += len(rels)
            for r in rels:
                t = r["type"]
                type_counts[t] = type_counts.get(t, 0) + 1

    print(f"=== Relations Stats ===")
    print(f"  Total wiki pages: {len(all_pages) - 2}")
    print(f"  Pages with relations: {pages_with_rels}")
    print(f"  Total relations: {total}")
    print(f"  By type:")
    for t, c in sorted(type_counts.items(), key=lambda x: -x[1]):
        print(f"    {t}: {c}")


def main():
    parser = argparse.ArgumentParser(description="Manage typed relationships in wiki pages")
    sub = parser.add_subparsers(dest="cmd")

    p_list = sub.add_parser("list", help="List relations of a page")
    p_list.add_argument("page")

    p_add = sub.add_parser("add", help="Add a relation")
    p_add.add_argument("page")
    p_add.add_argument("--type", required=True, choices=VALID_TYPES)
    p_add.add_argument("--target", required=True)
    p_add.add_argument("--desc")
    p_add.add_argument("--conf", type=int, default=3)

    p_rm = sub.add_parser("remove", help="Remove a relation")
    p_rm.add_argument("page")
    p_rm.add_argument("--type", required=True, choices=VALID_TYPES)
    p_rm.add_argument("--target", required=True)

    p_g = sub.add_parser("graph", help="Output DOT graph")
    p_g.add_argument("page")

    p_s = sub.add_parser("stats", help="Stats of the relation network")

    args = parser.parse_args()
    if not args.cmd:
        parser.print_help(); return

    if args.cmd == "list":
        cmd_list(args.page)
    elif args.cmd == "add":
        cmd_add(args.page, args.type, args.target, args.desc, args.conf)
    elif args.cmd == "remove":
        cmd_remove(args.page, args.type, args.target)
    elif args.cmd == "graph":
        cmd_graph(args.page)
    elif args.cmd == "stats":
        cmd_stats()


if __name__ == "__main__":
    main()