#!/usr/bin/env python # -*- coding: utf-8 -*- """ populate-relations.py — Phase 3 P1: 批量填充 wiki 页面的 relations 字段 策略（自动推断，无 LLM 依赖）： 1. **Source-grouping**：共享同一 raw source 的页面群，互为 part_of 关系 - 每组中被引用最多的页面成为 hub - 其余页面 part_of → hub 2. **Type-based**： - entity (人物) → example_of 对应 concept（如 毛泽东 example_of 教员的终极答案） - place → part_of 上级地点（如 西安 part_of 陕西省） 3. **Cross-link inference**：页面正文中的 [[wikilink]] 若目标存在， 且源页面 type=concept，推断 extends 关系（仅取 top 2 避免噪声） 安全措施： - 幂等：已有 relations 不重复添加 - 仅处理 wiki/*.md（排除 index/log/archive） - --dry-run 预览，--apply 执行 - 每页最多添加 5 条 relations（防爆裂） 用法： python tools/scripts/populate-relations.py --dry-run # 预览 python tools/scripts/populate-relations.py --apply # 执行 python tools/scripts/populate-relations.py --apply --max-per-page 3 """ import argparse import json import re import sys from collections import defaultdict from pathlib import Path try: import yaml except ImportError: print("ERROR: pyyaml required"); sys.exit(1) VAULT = Path(r"D:\Applications\app\kepano-obsidian-main") WIKI = VAULT / "wiki" EXCLUDE = {"index.md", "log.md"} MAX_RELATIONS_DEFAULT = 5 def read_frontmatter(fp: Path) -> tuple[dict, str]: content = fp.read_text(encoding="utf-8") m = re.match(r"^---\n(.*?)\n---", content, re.DOTALL) if not m: return {}, content try: front = yaml.safe_load(m.group(1)) or {} except Exception: return {}, content return front, content def write_frontmatter(fp: Path, front: dict, body: str): # Strip any existing frontmatter from body to avoid duplication body_clean = re.sub(r"^---\n.*?\n---\n?", "", body, count=1, flags=re.DOTALL).lstrip() new_yaml = yaml.dump(front, allow_unicode=True, default_flow_style=False, sort_keys=False) fp.write_text(f"---\n{new_yaml}---\n\n{body_clean}", encoding="utf-8") def get_source_key(front: dict) -> str: """Extract a normalized source key for grouping""" source = front.get("source", "") if not source: return "" s = str(source).strip().strip('"').strip("'") # Remove [[ ]] and raw/ prefix s = re.sub(r"\[\[(?:raw/)?(.+?)(?:\|.*)?\]\]", r"\1", s) # Remove .md extension s = re.sub(r"\.md$", "", s) # Remove path prefixes s = s.split("/")[-1] return s def get_inbound_counts() -> dict[str, int]: """Count inbound wikilinks for each wiki page""" counts = defaultdict(int) all_content = "" for fp in VAULT.rglob("*.md"): if ".git" in str(fp): continue try: all_content += fp.read_text(encoding="utf-8") + "\n" except Exception: continue for m in re.finditer(r"\[\[([^\]|#]+?)(?:[#\]|]|$)", all_content): name = m.group(1).strip() if name and not name.startswith("http"): counts[name] += 1 return counts def extract_wikilinks(content: str) -> list[str]: """Extract wikilink targets from page body (not frontmatter)""" # Remove frontmatter body = re.sub(r"^---\n.*?\n---", "", content, count=1, flags=re.DOTALL) links = [] for m in re.finditer(r"\[\[([^\]|#]+?)(?:[#\]|])", body): target = m.group(1).strip() if target and not target.startswith("http") and not target.startswith("raw/"): links.append(target) return links def dedup_existing(front: dict) -> set[tuple[str, str]]: """Get existing relation (type, target) pairs""" existing = set() for r in front.get("relations", []) or []: if isinstance(r, dict): t = r.get("target", "").strip("[]") existing.add((r.get("type", ""), t)) return existing def main(): parser = argparse.ArgumentParser(description="Batch populate relations for wiki pages") parser.add_argument("--dry-run", action="store_true", help="Preview without writing") parser.add_argument("--apply", action="store_true", help="Write changes to files") parser.add_argument("--max-per-page", type=int, default=MAX_RELATIONS_DEFAULT) args = parser.parse_args() if not args.dry_run and not args.apply: args.dry_run = True # default to dry-run for safety # Phase 1: Read all pages pages = {} for fp in sorted(WIKI.glob("*.md")): if fp.name in EXCLUDE: continue front, content = read_frontmatter(fp) if not front: continue pages[fp.stem] = {"fp": fp, "front": front, "content": content} print(f"Loaded {len(pages)} wiki pages") # Phase 2: Compute inbound counts inbound = get_inbound_counts() print(f"Computed inbound counts for {len(inbound)} link targets") # Phase 3: Group by source source_groups = defaultdict(list) for name, info in pages.items(): src = get_source_key(info["front"]) if src: source_groups[src].append(name) # Phase 4: Infer relations additions = defaultdict(list) # page -> list of {type, target, ...} # 4a: Source-grouping → part_of hub multi_page_groups = {s: g for s, g in source_groups.items() if len(g) > 1} print(f"\nSource groups with >1 page: {len(multi_page_groups)}") for src, group in multi_page_groups.items(): # Hub = most referenced page in group hub = max(group, key=lambda p: inbound.get(p, 0)) for page in group: if page == hub: continue additions[page].append({ "type": "part_of", "target": f"[[{hub}]]", "description": f"同源：{src}", "confidence": 3, }) # 4b: Single-source pages → extends the source concept if source is a wiki page single_source = {s: g[0] for s, g in source_groups.items() if len(g) == 1} for src, page in single_source.items(): info = pages.get(page) if not info: continue page_type = info["front"].get("type", "") if isinstance(page_type, list): page_type = page_type[0] if page_type else "" page_type = str(page_type).lower() # If source matches a wiki page name, add extends if src in pages and src != page: src_type = pages[src]["front"].get("type", "") if isinstance(src_type, list): src_type = src_type[0] if src_type else "" src_type = str(src_type).lower() if page_type == "entity" and src_type in ("concept", "reference"): additions[page].append({ "type": "example_of", "target": f"[[{src}]]", "confidence": 3, }) elif page_type == "concept" and src_type == "concept": additions[page].append({ "type": "extends", "target": f"[[{src}]]", "confidence": 3, }) elif page_type in ("place",) and src_type in ("place", "concept"): additions[page].append({ "type": "part_of", "target": f"[[{src}]]", "confidence": 3, }) # 4c: Cross-link inference for concept pages (top 2 wikilinks) for name, info in pages.items(): page_type = info["front"].get("type", "") if isinstance(page_type, list): page_type = page_type[0] if page_type else "" if str(page_type).lower() not in ("concept", "method"): continue links = extract_wikilinks(info["content"]) seen = set() added = 0 for target in links: if added >= 2: break if target == name or target in seen: continue if target not in pages: continue seen.add(target) additions[name].append({ "type": "extends", "target": f"[[{target}]]", "confidence": 2, }) added += 1 # Phase 5: Apply (single pass — dry-run prints, apply writes) total_added = 0 pages_modified = 0 for name in sorted(pages.keys()): pending = additions.get(name, []) if not pending: continue info = pages[name] front = info["front"] existing = dedup_existing(front) new_rels = list(front.get("relations", []) or []) added = 0 for rel in pending: key = (rel["type"], rel["target"].strip("[]")) if key in existing: continue if len(new_rels) >= args.max_per_page: break new_rels.append(rel) existing.add(key) added += 1 if added == 0: continue front["relations"] = new_rels pages_modified += 1 total_added += added if args.dry_run: print(f" [[{name}]]: +{added}") for rel in new_rels[-added:]: print(f" --{rel['type']}--> {rel['target']}") mode = "DRY RUN" if args.dry_run else "APPLYING" print(f"\n=== {mode} ===") print(f" Pages to modify: {pages_modified}") print(f" Relations to add: {total_added}") if args.apply and pages_modified > 0: written = 0 for name in sorted(pages.keys()): info = pages[name] front = info["front"] rels = front.get("relations", []) if not rels: continue # Only write if relations were newly populated (not pre-existing) # Check by seeing if this page was in our additions map if name not in additions: continue write_frontmatter(info["fp"], front, info["content"]) written += 1 print(f" Written to {written} files") if __name__ == "__main__": main()