292 lines
10 KiB
Python
292 lines
10 KiB
Python
#!/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()
|