Files
llm_wiki/tools/scripts/sync_home_wiki.py
T
hehaiguang1123 a6f05ab2d5 Phase 0-2: Schema cleanup, typed relations, event-driven automation
- Phase 0: AGENTS.md cleanup (dedup quotes, renumber sections, merge qmd)
- Phase 1: typed relations (manage-relations.py, graph-search.py, check-staleness.py, detect-conflicts.py)
- Phase 2: frontmatter validator, weekly lint, knowledge promotion, git hooks
- Fix .gitignore to track tools/ and .githooks/
- Fix git remote URL (remove plaintext token)
- New wiki pages: 504 pages, 34 raw sources
2026-07-01 08:05:43 +08:00

613 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
sync_home_wiki.py — home-wiki 知识页面 → kepano LLM Wiki 同步脚本
用途(定期执行):
将 D:\\TC_UP\\home-wiki 的 concepts/ entities/ syntheses/ 知识页面,
同步到 D:\\Applications\\app\\kepano-obsidian-main\\wiki\\
执行 frontmatter 规范转换,幂等更新;可选自动维护 index.md 与 lint 体检。
同步范围:
concepts/ + entities/ + syntheses/ (排除 index.md
Daily/ MyNotes/ reports/ 不在知识同步范围。
转换规则(home-wiki frontmatter → kepano Wiki 规范):
categories: 加 [[LLM Wiki]]entity/person 额外加 [[People]]
tags: 前缀 [wiki, {people|concept}]subtype→concept/{subtype}(仅 concept),追加原 tags
created: 保留
source: sources[0] → 文件路径取 stem 做 wikilink;URL/标识符原样;空则留空
type: 保留(concept/entity/synthesis
aliases: title 或 name
正文: 保留,移除 home-wiki 特有的 openclaw 自动段落(## Related、注释标记行)
index 自动维护(--index):
用标记块 <!-- BEGIN/END home-wiki-sync --> 界定 index.md 的「概念页」「实体页」
两个表,脚本依据同步页面元数据 + 简介 JSON 缓存自动重生成。
简介缓存 tools/data/home-wiki-summaries.json:首次从 index.md 现有表格种子化,
保留人工优化;新增页面自动提取(description → 顶部引用块 → 概述首句)。
综合报告表因混合其他来源,保持手动维护。
用法:
python tools/scripts/sync_home_wiki.py --dry-run # 仅诊断差异
python tools/scripts/sync_home_wiki.py # 同步(写入变化的页面)
python tools/scripts/sync_home_wiki.py --index --log # 一站式:同步+更新index+记日志
python tools/scripts/sync_home_wiki.py --lint # 体检(孤儿/断链/source空值)
退出码:0 成功;1 源目录缺失;2 有错误。
"""
import argparse
import json
import re
import sys
from datetime import date
from pathlib import Path
try:
import yaml
except ImportError:
sys.stderr.write("ERROR: PyYAML 未安装,请运行 pip install pyyaml\n")
sys.exit(2)
# ---------------------------------------------------------------------------
# 路径配置(两库位于不同位置,必须使用绝对路径)
# ---------------------------------------------------------------------------
SRC_ROOT = Path(r"D:\TC_UP\home-wiki")
DST_WIKI = Path(r"D:\Applications\app\kepano-obsidian-main\wiki")
DST_INDEX = DST_WIKI / "index.md"
DST_LOG = DST_WIKI / "log.md"
SUMMARIES_CACHE = Path(__file__).resolve().parent.parent / "data" / "home-wiki-summaries.json"
SRC_DIRS = ["concepts", "entities", "syntheses"]
# home-wiki 完整文件名 → kepano 简短名(仅含副标题、过长标题的需要映射)
NAME_MAP = {
"顶级人生三重境:道家驭势、佛家修心、儒家立身": "顶级人生三重境",
}
INDEX_BEGIN = "<!-- BEGIN home-wiki-sync -->"
INDEX_END = "<!-- END home-wiki-sync -->"
# ---------------------------------------------------------------------------
# 文档解析
# ---------------------------------------------------------------------------
def _extract_list_field(text, key):
"""从 frontmatter 文本中提取列表字段(兼容 flow 与 block 两种写法)。"""
m = re.search(rf"^[ \t]*{re.escape(key)}[ \t]*:\s*\[(.*)\][ \t]*$", text, re.M)
if m:
return [t.strip().strip("\"'") for t in m.group(1).split(",") if t.strip()]
m = re.search(rf"^[ \t]*{re.escape(key)}[ \t]*:\s*\n((?:[ \t]+-.+\n?)+)", text, re.M)
if m:
items = re.findall(r"^[ \t]+-[ \t]+(.+?)[ \t]*$", m.group(1), re.M)
return [it.strip().strip("\"'") for it in items]
return []
def _robust_parse_fm(fm_text):
"""健壮解析 frontmatter:先 yaml.safe_load,失败则正则逐字段回退。
home-wiki frontmatter 常含 `related: [[wikilink]]` 这类 Obsidian wikilink
`[[` 会触发 YAML flow sequence 解析异常,故需要回退。
"""
try:
fm = yaml.safe_load(fm_text)
if isinstance(fm, dict) and fm.get("type"):
return fm
except yaml.YAMLError:
pass
fm = {}
for key in ("type", "subtype", "title", "name", "created", "updated",
"confidence", "field", "nameEn", "description"):
m = re.search(rf"^[ \t]*{re.escape(key)}[ \t]*:[ \t]*(.+?)[ \t]*$", fm_text, re.M)
if m and m.group(1).strip() not in ("", "[]"):
fm[key] = m.group(1).strip().strip("\"'")
for key in ("tags", "sources"):
vals = _extract_list_field(fm_text, key)
if vals:
fm[key] = vals
return fm
def parse_doc(text):
"""分离 frontmatter 与正文。返回 (fm_dict, body_str)。"""
m = re.match(r"^---\s*\n(.*?)\n---\s*\n?(.*)$", text, re.S)
if not m:
return {}, text
return _robust_parse_fm(m.group(1)), m.group(2)
def strip_openclaw(body):
"""移除 home-wiki 特有的 openclaw 自动段落,保留实质内容。
- 移除 `## Related` 及其 openclaw:wiki:related 包裹块
- 移除所有 `<!-- openclaw:...:...:(start|end) -->` 注释标记行(保留被包裹的内容)
"""
body = re.sub(
r"\n##\s*Related\s*\n<!--\s*openclaw:wiki:related:start\s*-->.*?<!--\s*openclaw:wiki:related:end\s*-->",
"\n",
body,
flags=re.S,
)
body = re.sub(r"<!--\s*openclaw:[a-z]+:[a-z]+:(?:start|end)\s*-->\s*\n?", "", body)
return body.rstrip() + "\n"
def norm_body(body):
"""规范化正文用于对比:移除 openclaw 段,丢弃空行与行尾空白。"""
b = strip_openclaw(body)
lines = [ln.rstrip() for ln in b.splitlines() if ln.strip()]
return "\n".join(lines)
# ---------------------------------------------------------------------------
# frontmatter 转换
# ---------------------------------------------------------------------------
def _as_list(v):
if v is None:
return []
if isinstance(v, list):
return v
if isinstance(v, str):
s = v.strip()
if s.startswith("[") and s.endswith("]"):
s = s[1:-1]
return [t.strip() for t in s.split(",") if t.strip()]
return [v]
def convert_frontmatter(fm, default_name):
"""home-wiki frontmatter → kepano Wiki frontmatter(有序)。"""
fm_type = str(fm.get("type", "concept")).strip()
subtype = fm.get("subtype")
if isinstance(subtype, list):
subtype = subtype[0] if subtype else None
subtype = str(subtype).strip() if subtype else None
title = fm.get("title") or fm.get("name") or default_name
categories = ["[[LLM Wiki]]"]
type_tag = "concept"
if fm_type == "entity":
if subtype == "person":
categories.append("[[People]]")
type_tag = "people"
else:
type_tag = "entity"
elif fm_type == "synthesis":
type_tag = "concept"
tags = ["wiki", type_tag]
# 仅 concept 类型把 subtype 转为 concept/{subtype} 标签
# entity/person 用 people 标签 + [[People]] category,不加 concept/person
if subtype and fm_type == "concept":
tags.append(f"concept/{subtype}")
for t in _as_list(fm.get("tags")):
t = str(t).strip()
if t and t not in tags:
tags.append(t)
sources = _as_list(fm.get("sources"))
src0 = str(sources[0]) if sources else ""
if ("/" in src0 or "\\" in src0) and not src0.startswith("http"):
src_val = "[[" + Path(src0).stem + "]]"
elif src0:
src_val = src0
else:
src_val = ""
return [
("categories", categories),
("tags", tags),
("created", str(fm.get("created", ""))),
("source", src_val),
("type", fm_type),
("aliases", [str(title)]),
]
def _yaml_quote(val):
"""如果值包含 YAML 特殊字符(如 [[ wikilink 的方括号),加双引号。"""
s = str(val)
if not s:
return '""'
if '[' in s or ']' in s or '{' in s or '}' in s or ':' in s or '#' in s:
return f'"{s}"'
return s
def dump_frontmatter(pairs):
out = ["---"]
for k, v in pairs:
if isinstance(v, list):
out.append(f"{k}:")
for item in v:
out.append(f" - {_yaml_quote(item)}")
else:
out.append(f"{k}: {_yaml_quote(v)}")
out.append("---")
return "\n".join(out) + "\n"
# ---------------------------------------------------------------------------
# 同步主流程
# ---------------------------------------------------------------------------
def target_name(stem):
"""home-wiki 文件名(stem) → kepano 目标文件名。"""
return NAME_MAP.get(stem, stem)
def collect_sources():
"""收集所有待同步源文件(排除 index.md)。返回 [(src_path, stem)]。"""
items = []
for d in SRC_DIRS:
sdir = SRC_ROOT / d
if not sdir.is_dir():
continue
for p in sorted(sdir.glob("*.md")):
if p.stem.lower() == "index":
continue
items.append((p, p.stem))
return items
def build_page(src_path, stem):
"""读取源文件,生成 kepano 页面内容 (frontmatter_str + body)。"""
text = src_path.read_text(encoding="utf-8")
fm, body = parse_doc(text)
pairs = convert_frontmatter(fm, stem)
body_clean = strip_openclaw(body)
# 确保正文与 frontmatter 间有空行
return dump_frontmatter(pairs) + "\n" + body_clean.lstrip("\n")
def _norm_full(text):
"""规范化整篇(frontmatter+正文)用于对比:去行尾空白、去空行。"""
return "\n".join(ln.rstrip() for ln in text.splitlines() if ln.strip())
def get_page_meta():
"""返回 {target_stem: {"type":..., "src_dir":..., "src_stem":...}}。"""
meta = {}
for src_path, stem in collect_sources():
fm, _ = parse_doc(src_path.read_text(encoding="utf-8"))
meta[target_name(stem)] = {
"type": str(fm.get("type", "concept")),
"src_dir": src_path.parent.name,
"src_stem": stem,
}
return meta
def sync(dry_run=False):
if not SRC_ROOT.is_dir():
sys.stderr.write(f"ERROR: 源目录不存在: {SRC_ROOT}\n")
sys.exit(1)
if not DST_WIKI.is_dir():
sys.stderr.write(f"ERROR: 目标 wiki 目录不存在: {DST_WIKI}\n")
sys.exit(1)
items = collect_sources()
created, updated, unchanged = [], [], []
errors = []
for src_path, stem in items:
try:
dst_path = DST_WIKI / f"{target_name(stem)}.md"
new_content = build_page(src_path, stem)
if not dst_path.exists():
created.append(stem)
if not dry_run:
dst_path.write_text(new_content, encoding="utf-8")
continue
old_content = dst_path.read_text(encoding="utf-8")
# 对比完整生成内容(frontmatter+正文),确保 frontmatter 损坏也会被修复
if _norm_full(new_content) == _norm_full(old_content):
unchanged.append(stem)
continue
updated.append(stem)
if not dry_run:
dst_path.write_text(new_content, encoding="utf-8")
except Exception as e: # noqa
errors.append(f"{stem}: {e}")
return {
"total": len(items),
"created": created,
"updated": updated,
"unchanged": unchanged,
"errors": errors,
}
# ---------------------------------------------------------------------------
# 简介提取与缓存(index 自动维护用)
# ---------------------------------------------------------------------------
def _wikilink_to_text(s):
"""将简介里的 wikilink 转为纯文本显示:[[a|b]]→b[[a]]→a。"""
s = re.sub(r"\[\[([^\]|]+?)\|([^\]]+?)\]\]", r"\2", s)
s = re.sub(r"\[\[([^\]]+?)\]\]", r"\1", s)
return s
def _trunc(s, n=80):
s = s.strip()
if len(s) > n:
s = s[:n].rstrip() + ""
return s
def extract_summary(body, fm):
"""从页面提取一句话简介。优先级:description → 顶部首个引用块 → 概述/定义首句。"""
if fm.get("description"):
return _trunc(_wikilink_to_text(str(fm["description"])))
lines = body.splitlines()
i, n = 0, len(lines)
while i < n and not lines[i].lstrip().startswith(">"):
i += 1
if i < n:
quotes = []
while i < n and lines[i].lstrip().startswith(">"):
content = lines[i].lstrip()[1:].strip()
if content and not re.match(r"^[—\-]+", content):
quotes.append(content)
i += 1
if quotes:
return _trunc(_wikilink_to_text(" ".join(quotes)))
m = re.search(r"^##\s*(?:概述|定义|简介)\s*\n\s*(.+)$", body, re.M)
if m:
first = m.group(1).strip().splitlines()[0]
if first:
return _trunc(_wikilink_to_text(first))
return ""
def seed_from_index():
"""首次运行:从 index.md 现有跨库表格解析人工简介作为初始缓存。"""
cache = {}
if not DST_INDEX.exists():
return cache
text = DST_INDEX.read_text(encoding="utf-8")
# 只解析来源列含 [[home-wiki/ 的行,避免误抓其他来源页面
pattern = re.compile(
r"^\|\s*\[\[([^\]|]+?)(?:\|[^\]]+)?\]\]\s*\|\s*([^|]+?)\s*\|\s*\[\[home-wiki/",
re.M,
)
for m in pattern.finditer(text):
stem = m.group(1).strip()
summary = m.group(2).strip()
if stem and summary:
cache[stem] = {"summary": summary}
return cache
def load_summaries():
if SUMMARIES_CACHE.exists():
try:
return json.loads(SUMMARIES_CACHE.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError):
pass
return seed_from_index()
def save_summaries(cache):
SUMMARIES_CACHE.parent.mkdir(parents=True, exist_ok=True)
SUMMARIES_CACHE.write_text(
json.dumps(cache, ensure_ascii=False, indent=2, sort_keys=True),
encoding="utf-8",
)
# ---------------------------------------------------------------------------
# index.md 自动维护
# ---------------------------------------------------------------------------
def _build_table(stems, cache, meta, title):
"""生成单个分类表格(Markdown 文本)。"""
lines = [f"### {title}{len(stems)}", "", "| 页面 | 简介 | 来源 |", "|------|------|------|"]
for s in stems:
summ = cache.get(s, {}).get("summary", "")
src = f"[[home-wiki/{meta[s]['src_dir']}/{s}]]"
lines.append(f"| [[{s}]] | {summ} | {src} |")
return "\n".join(lines)
def update_index(meta):
"""用标记块自动维护 index.md 的概念/实体表,并刷新「最后更新」日期。"""
cache_existed = SUMMARIES_CACHE.exists()
cache = load_summaries()
# 为缓存缺失的页面提取简介
dirty = False
for stem, m in meta.items():
if not cache.get(stem, {}).get("summary"):
dst = DST_WIKI / f"{stem}.md"
if dst.exists():
fm, body = parse_doc(dst.read_text(encoding="utf-8"))
cache[stem] = {"summary": extract_summary(body, fm), "type": m["type"]}
dirty = True
# 首次运行(缓存文件不存在)或有新提取,都持久化(JSON 是简介权威来源)
if dirty or not cache_existed:
save_summaries(cache)
concepts = sorted(s for s, m in meta.items() if m["type"] == "concept")
entities = sorted(s for s, m in meta.items() if m["type"] == "entity")
block_body = [
"## 跨库同步(home-wiki",
"",
"> 来自 `D:\\TC_UP\\home-wiki` 的知识内容,聚焦高等教育 AI、系统理论、哲学思想等方向。",
"",
_build_table(concepts, cache, meta, "概念页"),
"",
_build_table(entities, cache, meta, "实体页"),
]
generated = INDEX_BEGIN + "\n" + "\n".join(block_body) + "\n" + INDEX_END
text = DST_INDEX.read_text(encoding="utf-8")
if INDEX_BEGIN in text and INDEX_END in text:
new_text = re.sub(
re.escape(INDEX_BEGIN) + r".*?" + re.escape(INDEX_END),
lambda _: generated,
text,
flags=re.S,
)
else:
# 首次:替换现有手动「跨库同步」章节(到「综合报告」之前)为标记块
new_text = re.sub(
r"## 跨库同步(home-wiki.*?(?=### 综合报告)",
lambda _: generated + "\n\n",
text,
flags=re.S,
)
# 刷新「最后更新」日期(pattern 匹配整格含结尾 |replacement 给完整行,避免管道符累加)
today = date.today().isoformat()
new_text = re.sub(
r"\|\s*最后更新\s*\|\s*[^\n|]*\|",
f"| 最后更新 | {today} (home-wiki 跨库同步) |",
new_text,
)
if new_text != text:
DST_INDEX.write_text(new_text, encoding="utf-8")
return True
return False
# ---------------------------------------------------------------------------
# lint 体检
# ---------------------------------------------------------------------------
def lint():
meta = get_page_meta()
synced = list(meta.keys())
allfiles = {p.stem for p in DST_WIKI.glob("*.md")}
# aliases 反向映射(MarkItDown 等大小写变体)
aliases_map = {}
for p in DST_WIKI.glob("*.md"):
fm, _ = parse_doc(p.read_text(encoding="utf-8"))
for a in _as_list(fm.get("aliases")):
aliases_map[str(a).strip()] = p.stem
# 孤儿:同步页面在 wiki 的入站链接数(排除自身)
inbound = {s: 0 for s in synced}
for p in DST_WIKI.glob("*.md"):
t = p.read_text(encoding="utf-8")
for s in synced:
if p.stem == s:
continue
if re.search(rf"\[\[{re.escape(s)}[\]|\]]", t):
inbound[s] += 1
orphans = [s for s, c in inbound.items() if c == 0]
# source 空值
empty_source = []
for s in synced:
t = (DST_WIKI / f"{s}.md").read_text(encoding="utf-8")
m = re.search(r"^source:[ \t]*(.*)$", t, re.M)
if m and not m.group(1).strip():
empty_source.append(s)
# 断链:同步页面引用的 [[x]] 在 wiki 是否存在(排除 raw 来源类、aliases、category
valid = allfiles | set(aliases_map.keys()) | set(synced)
skip = {"People", "LLM Wiki", "wikilink"}
broken = {}
for s in synced:
t = (DST_WIKI / f"{s}.md").read_text(encoding="utf-8")
for m in re.findall(r"\[\[([^\]|]+?)(?:\|[^\]]+)?\]\]", t):
tgt = m.strip()
if not tgt or tgt.startswith("http") or "/" in tgt or "\\" in tgt:
continue
if tgt in valid or tgt in skip:
continue
broken.setdefault(tgt, []).append(s)
# index 覆盖检查:同步的概念/实体是否都在 index 跨库章节
idx_text = DST_INDEX.read_text(encoding="utf-8") if DST_INDEX.exists() else ""
not_in_index = [s for s in synced if meta[s]["type"] in ("concept", "entity")
and not re.search(rf"\[\[{re.escape(s)}[\]|\]]", idx_text)]
print("== home-wiki 同步体检 ==")
print(f"同步页面: {len(synced)} | wiki 总文件: {len(allfiles)}")
print(f"\n[孤儿] {len(orphans)} 个(0 入站链接)")
for s in orphans:
print(f" - {s}")
print(f"\n[source 空值] {len(empty_source)} 个(违反 wiki 层 source 必填)")
for s in empty_source:
print(f" - {s}")
print(f"\n[断链] {len(broken)} 个目标(同步页面引用、kepano 无对应页)")
for tgt in sorted(broken):
print(f" - {tgt}{', '.join(sorted(broken[tgt]))}")
print(f"\n[未入 index] {len(not_in_index)} 个(概念/实体未在跨库章节)")
for s in not_in_index:
print(f" - {s}")
issues = len(orphans) + len(empty_source) + len(not_in_index)
print(f"\n硬性问题(孤儿+空source+未入index: {issues}")
print(f"软性问题(断链,多为待创建概念/raw来源): {len(broken)}")
return issues
# ---------------------------------------------------------------------------
# 日志
# ---------------------------------------------------------------------------
def append_log(result):
today = date.today().isoformat()
lines = [
f"\n## [{today}] sync | home-wiki 跨库同步",
"",
f"**来源**: `{SRC_ROOT}`",
"",
"**操作**:",
f"- 扫描 concepts/entities/syntheses 共 {result['total']} 个页面",
]
if result["updated"]:
lines.append(f"- 更新 {len(result['updated'])} 页: {', '.join(result['updated'])}")
if result["created"]:
lines.append(f"- 新建 {len(result['created'])} 页: {', '.join(result['created'])}")
if result["unchanged"]:
lines.append(f"- 未变 {len(result['unchanged'])}")
if result["errors"]:
lines.append(f"- 错误 {len(result['errors'])} 项: {'; '.join(result['errors'])}")
lines += ["", f"**涉及页面**: {len(result['updated'])} 更新,{len(result['created'])} 新建", ""]
with DST_LOG.open("a", encoding="utf-8") as f:
f.write("\n".join(lines))
# ---------------------------------------------------------------------------
def main():
ap = argparse.ArgumentParser(description="home-wiki → kepano LLM Wiki 同步")
ap.add_argument("--dry-run", action="store_true", help="仅诊断,不写入")
ap.add_argument("--log", action="store_true", help="同步后追加 wiki/log.md")
ap.add_argument("--index", action="store_true", help="同步后自动维护 wiki/index.md 跨库章节")
ap.add_argument("--lint", action="store_true", help="仅体检(孤儿/断链/source空值),不同步")
args = ap.parse_args()
if args.lint:
issues = lint()
sys.exit(0 if issues == 0 else 0) # lint 不以问题数为错误码
res = sync(dry_run=args.dry_run)
verb = "诊断" if args.dry_run else "同步"
print(f"== home-wiki {verb}报告 ==")
print(f"扫描: {res['total']}")
print(f"需更新: {len(res['updated'])} -> {res['updated']}")
print(f"需新建: {len(res['created'])} -> {res['created']}")
print(f"未变化: {len(res['unchanged'])} -> {res['unchanged']}")
if res["errors"]:
print(f"错误: {res['errors']}")
sys.exit(2)
if not args.dry_run:
if args.log and (res["updated"] or res["created"]):
append_log(res)
print("已追加 wiki/log.md")
if args.index:
meta = get_page_meta()
changed = update_index(meta)
print(f"wiki/index.md 跨库章节: {'已更新' if changed else '无变化'}")
print("完成。")
if __name__ == "__main__":
main()