|
|
@@ -0,0 +1,853 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""
|
|
|
+kb.py — 做课系统知识库工具链
|
|
|
+
|
|
|
+所有"客观事实"由本脚本负责：文件指纹、差异计算、关系图、索引重建、健康检查。
|
|
|
+所有"主观判断"由 skill（LLM）负责：抽取、归纳、写页、写摘要。
|
|
|
+
|
|
|
+设计原则：
|
|
|
+ 1. 脚本不碰内容语义，只维护结构。
|
|
|
+ 2. 幂等：任何命令重复执行结果一致。
|
|
|
+ 3. 可回滚：删除先进 _archive，不物理销毁。
|
|
|
+
|
|
|
+用法：
|
|
|
+ python kb.py status
|
|
|
+ python kb.py scan [--json]
|
|
|
+ python kb.py mark <源相对路径> --pages a.md,b.md
|
|
|
+ python kb.py retract <源相对路径>
|
|
|
+ python kb.py move <旧路径> <新路径>
|
|
|
+ python kb.py graph
|
|
|
+ python kb.py lint [--json]
|
|
|
+"""
|
|
|
+import sys, os, json, re, hashlib, shutil, argparse, subprocess
|
|
|
+from pathlib import Path
|
|
|
+from datetime import datetime, timezone, timedelta
|
|
|
+
|
|
|
+if hasattr(sys.stdout, "reconfigure"):
|
|
|
+ sys.stdout.reconfigure(encoding="utf-8", errors="replace")
|
|
|
+
|
|
|
+ROOT = Path(__file__).resolve().parent.parent
|
|
|
+RAW = ROOT / "00-原材料"
|
|
|
+WIKI = ROOT / "10-知识库"
|
|
|
+META = WIKI / "_meta"
|
|
|
+ARCHIVE = WIKI / "_archive"
|
|
|
+MANIFEST = META / "manifest.json"
|
|
|
+DERIV = META / "derivations.jsonl"
|
|
|
+CHANGELOG = META / "changelog.jsonl"
|
|
|
+CHANGELOG_MD = META / "CHANGELOG.md"
|
|
|
+REBUILD = META / "rebuild_queue.jsonl"
|
|
|
+
|
|
|
+CST = timezone(timedelta(hours=8))
|
|
|
+DIRS = ["sources", "topics", "entities", "modules"]
|
|
|
+
|
|
|
+# 原料目录里需要无视的噪声
|
|
|
+NOISE_NAMES = {"desktop.ini", "thumbs.db", ".ds_store", ".gitkeep"}
|
|
|
+NOISE_TAIL = (".tmp", ".crdownload", ".part", ".download", ".swp", ".lock")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- 基础工具
|
|
|
+
|
|
|
+def now() -> str:
|
|
|
+ return datetime.now(CST).isoformat(timespec="seconds")
|
|
|
+
|
|
|
+
|
|
|
+def rel(p: Path, base: Path = None) -> str:
|
|
|
+ """统一成 POSIX 风格相对路径，作为全库唯一键。"""
|
|
|
+ return p.relative_to(base or RAW).as_posix()
|
|
|
+
|
|
|
+
|
|
|
+def sha256(p: Path, chunk: int = 1 << 20) -> str:
|
|
|
+ h = hashlib.sha256()
|
|
|
+ with p.open("rb") as f:
|
|
|
+ for blk in iter(lambda: f.read(chunk), b""):
|
|
|
+ h.update(blk)
|
|
|
+ return h.hexdigest()
|
|
|
+
|
|
|
+
|
|
|
+def short_id(key: str) -> str:
|
|
|
+ return "src_" + hashlib.sha1(key.encode("utf-8")).hexdigest()[:8]
|
|
|
+
|
|
|
+
|
|
|
+def is_noise(p: Path) -> bool:
|
|
|
+ n = p.name
|
|
|
+ if n.startswith(".") or n.startswith("~$"):
|
|
|
+ return True
|
|
|
+ if n.lower() in NOISE_NAMES:
|
|
|
+ return True
|
|
|
+ return n.lower().endswith(NOISE_TAIL)
|
|
|
+
|
|
|
+
|
|
|
+def load_json(p: Path, default):
|
|
|
+ if p.exists():
|
|
|
+ try:
|
|
|
+ return json.loads(p.read_text(encoding="utf-8"))
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ return default
|
|
|
+
|
|
|
+
|
|
|
+def dump_json(p: Path, obj):
|
|
|
+ p.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ p.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
+
|
|
|
+
|
|
|
+def append_jsonl(p: Path, row: dict):
|
|
|
+ p.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ with p.open("a", encoding="utf-8") as f:
|
|
|
+ f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
|
|
+
|
|
|
+
|
|
|
+def read_jsonl(p: Path) -> list:
|
|
|
+ if not p.exists():
|
|
|
+ return []
|
|
|
+ out = []
|
|
|
+ for line in p.read_text(encoding="utf-8").splitlines():
|
|
|
+ line = line.strip()
|
|
|
+ if line:
|
|
|
+ try:
|
|
|
+ out.append(json.loads(line))
|
|
|
+ except Exception:
|
|
|
+ pass
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def log_human(op: str, subject: str, detail: str = ""):
|
|
|
+ """人读日志，与 changelog.jsonl 双写。
|
|
|
+
|
|
|
+ 统一前缀是刻意设计：`grep '^## \\[' CHANGELOG.md | tail -5` 就能取最近 5 条，
|
|
|
+ 不需要任何工具。这是 Karpathy llm-wiki 里 log.md 的做法，被验证过很好用。
|
|
|
+ """
|
|
|
+ if not CHANGELOG_MD.exists():
|
|
|
+ CHANGELOG_MD.write_text(
|
|
|
+ "# 变更日志\n\n"
|
|
|
+ "> 只增不改。统一前缀便于 shell 检索：\n"
|
|
|
+ "> `grep '^## \\[' CHANGELOG.md | tail -5`\n\n",
|
|
|
+ encoding="utf-8")
|
|
|
+ stamp = datetime.now(CST).strftime("%Y-%m-%d %H:%M")
|
|
|
+ block = f"## [{stamp}] {op} | {subject}\n"
|
|
|
+ if detail:
|
|
|
+ block += detail.rstrip() + "\n"
|
|
|
+ with CHANGELOG_MD.open("a", encoding="utf-8") as f:
|
|
|
+ f.write(block + "\n")
|
|
|
+
|
|
|
+
|
|
|
+# ------------------------------------------------------------ frontmatter
|
|
|
+
|
|
|
+def split_fm(text: str):
|
|
|
+ """返回 (dict, body)。不依赖 pyyaml，只支持本库用到的子集。"""
|
|
|
+ if not text.startswith("---"):
|
|
|
+ return {}, text
|
|
|
+ m = re.match(r"^---\r?\n(.*?)\r?\n---\r?\n?", text, re.S)
|
|
|
+ if not m:
|
|
|
+ return {}, text
|
|
|
+ raw, body = m.group(1), text[m.end():]
|
|
|
+ data, key = {}, None
|
|
|
+ for line in raw.splitlines():
|
|
|
+ if not line.strip() or line.lstrip().startswith("#"):
|
|
|
+ continue
|
|
|
+ if re.match(r"^\s+-\s+", line) and key:
|
|
|
+ if not isinstance(data.get(key), list):
|
|
|
+ data[key] = []
|
|
|
+ data[key].append(_scalar(line.strip()[2:]))
|
|
|
+ continue
|
|
|
+ km = re.match(r"^([A-Za-z_][\w-]*)\s*:\s*(.*)$", line)
|
|
|
+ if km:
|
|
|
+ key, val = km.group(1), km.group(2).strip()
|
|
|
+ if val == "":
|
|
|
+ data[key] = []
|
|
|
+ elif val.startswith("[") and val.endswith("]"):
|
|
|
+ inner = val[1:-1].strip()
|
|
|
+ data[key] = [_scalar(x) for x in inner.split(",")] if inner else []
|
|
|
+ else:
|
|
|
+ data[key] = _scalar(val)
|
|
|
+ return data, body
|
|
|
+
|
|
|
+
|
|
|
+def _scalar(s: str):
|
|
|
+ s = s.strip()
|
|
|
+ if len(s) >= 2 and s[0] == s[-1] and s[0] in "\"'":
|
|
|
+ return s[1:-1]
|
|
|
+ if s.lower() in ("true", "false"):
|
|
|
+ return s.lower() == "true"
|
|
|
+ return s
|
|
|
+
|
|
|
+
|
|
|
+def dump_fm(fm: dict) -> str:
|
|
|
+ order = ["id", "type", "title", "summary", "evidence", "tags", "sources",
|
|
|
+ "related", "status", "stale", "stale_reason", "created", "updated"]
|
|
|
+ keys = [k for k in order if k in fm] + [k for k in fm if k not in order]
|
|
|
+ lines = ["---"]
|
|
|
+ for k in keys:
|
|
|
+ v = fm[k]
|
|
|
+ if isinstance(v, list):
|
|
|
+ if not v:
|
|
|
+ lines.append(f"{k}: []")
|
|
|
+ else:
|
|
|
+ lines.append(f"{k}:")
|
|
|
+ lines += [f" - {json.dumps(str(x), ensure_ascii=False)}" for x in v]
|
|
|
+ elif isinstance(v, bool):
|
|
|
+ lines.append(f"{k}: {'true' if v else 'false'}")
|
|
|
+ else:
|
|
|
+ lines.append(f"{k}: {json.dumps(str(v), ensure_ascii=False)}")
|
|
|
+ lines.append("---")
|
|
|
+ return "\n".join(lines) + "\n"
|
|
|
+
|
|
|
+
|
|
|
+def read_page(p: Path):
|
|
|
+ text = p.read_text(encoding="utf-8", errors="replace")
|
|
|
+ fm, body = split_fm(text)
|
|
|
+ return fm, body
|
|
|
+
|
|
|
+
|
|
|
+def write_page(p: Path, fm: dict, body: str):
|
|
|
+ p.parent.mkdir(parents=True, exist_ok=True)
|
|
|
+ p.write_text(dump_fm(fm) + body, encoding="utf-8")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- scan
|
|
|
+
|
|
|
+def manifest_skeleton():
|
|
|
+ return {"version": 2, "root": "00-原材料", "updated_at": None,
|
|
|
+ "files": {}, "tombstones": []}
|
|
|
+
|
|
|
+
|
|
|
+def collect_raw() -> dict:
|
|
|
+ """返回 {relpath: {size, mtime_ns, sha256}}，大文件用 size+mtime 短路。"""
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ known = man.get("files", {})
|
|
|
+ out = {}
|
|
|
+ if not RAW.exists():
|
|
|
+ return out
|
|
|
+ for p in sorted(RAW.rglob("*")):
|
|
|
+ if not p.is_file() or is_noise(p):
|
|
|
+ continue
|
|
|
+ r = rel(p)
|
|
|
+ st = p.stat()
|
|
|
+ old = known.get(r, {})
|
|
|
+ if old.get("size") == st.st_size and old.get("mtime_ns") == st.st_mtime_ns and old.get("sha256"):
|
|
|
+ out[r] = {"size": st.st_size, "mtime_ns": st.st_mtime_ns,
|
|
|
+ "sha256": old["sha256"], "fast": True}
|
|
|
+ else:
|
|
|
+ out[r] = {"size": st.st_size, "mtime_ns": st.st_mtime_ns,
|
|
|
+ "sha256": sha256(p), "fast": False}
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def cmd_scan(args):
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ known = man.get("files", {})
|
|
|
+ cur = collect_raw()
|
|
|
+
|
|
|
+ new_keys = sorted(set(cur) - set(known))
|
|
|
+ gone_keys = sorted(set(known) - set(cur))
|
|
|
+ common = sorted(set(cur) & set(known))
|
|
|
+ changed = [k for k in common if cur[k]["sha256"] != known[k].get("sha256")]
|
|
|
+
|
|
|
+ # 移动识别：内容 hash 相同、路径不同
|
|
|
+ gone_by_hash = {}
|
|
|
+ for k in gone_keys:
|
|
|
+ h = known[k].get("sha256")
|
|
|
+ if h:
|
|
|
+ gone_by_hash.setdefault(h, []).append(k)
|
|
|
+ new_by_hash = {}
|
|
|
+ for k in new_keys:
|
|
|
+ new_by_hash.setdefault(cur[k]["sha256"], []).append(k)
|
|
|
+
|
|
|
+ moved, consumed_new, consumed_gone = [], set(), set()
|
|
|
+ for h, news in new_by_hash.items():
|
|
|
+ gones = gone_by_hash.get(h, [])
|
|
|
+ if len(news) == 1 and len(gones) == 1: # 只在能唯一配对时认定移动
|
|
|
+ moved.append({"from": gones[0], "to": news[0], "sha256": h})
|
|
|
+ consumed_new.add(news[0]); consumed_gone.add(gones[0])
|
|
|
+
|
|
|
+ result = {
|
|
|
+ "generated_at": now(),
|
|
|
+ "new": [k for k in new_keys if k not in consumed_new],
|
|
|
+ "changed": changed,
|
|
|
+ "deleted": [k for k in gone_keys if k not in consumed_gone],
|
|
|
+ "moved": moved,
|
|
|
+ "unchanged": len(common) - len(changed),
|
|
|
+ "details": {k: {"size": cur[k]["size"]} for k in
|
|
|
+ [*new_keys, *changed] if k in cur},
|
|
|
+ }
|
|
|
+
|
|
|
+ if args.json:
|
|
|
+ print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
|
+ return
|
|
|
+
|
|
|
+ print(f"扫描 {RAW.name}/ @{result['generated_at']}")
|
|
|
+ print(f" 新增 {len(result['new'])} · 变更 {len(result['changed'])} · "
|
|
|
+ f"删除 {len(result['deleted'])} · 移动 {len(result['moved'])} · "
|
|
|
+ f"未变 {result['unchanged']}")
|
|
|
+ for tag, key in (("NEW", "new"), ("CHANGED", "changed"), ("DELETED", "deleted")):
|
|
|
+ for k in result[key]:
|
|
|
+ print(f" [{tag}] {k}")
|
|
|
+ for m in result["moved"]:
|
|
|
+ print(f" [MOVED] {m['from']} -> {m['to']}")
|
|
|
+ if not any(result[k] for k in ("new", "changed", "deleted", "moved")):
|
|
|
+ print(" 库已是最新，无需同步。")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- mark
|
|
|
+
|
|
|
+def cmd_mark(args):
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ cur = collect_raw()
|
|
|
+ r = args.path
|
|
|
+ if r not in cur:
|
|
|
+ print(f"[ERR] 原材料中不存在：{r}")
|
|
|
+ return 1
|
|
|
+ pages = [x.strip().replace("\\", "/") for x in (args.pages or "").split(",") if x.strip()]
|
|
|
+ rec = man["files"].get(r, {})
|
|
|
+ rec.update({"size": cur[r]["size"], "mtime_ns": cur[r]["mtime_ns"],
|
|
|
+ "sha256": cur[r]["sha256"], "id": rec.get("id") or short_id(r),
|
|
|
+ "status": "ingested", "ingested_at": now(),
|
|
|
+ "pages": sorted(set(rec.get("pages", []) + pages))})
|
|
|
+ man["files"][r] = rec
|
|
|
+ man["updated_at"] = now()
|
|
|
+ dump_json(MANIFEST, man)
|
|
|
+ append_jsonl(CHANGELOG, {"ts": now(), "op": "ingest", "source": r,
|
|
|
+ "pages": pages, "sha256": rec["sha256"][:12]})
|
|
|
+ log_human("ingest", r, "\n".join(f"- 派生 `{x}`" for x in pages))
|
|
|
+ print(f"[OK] 已登记 {r} -> {len(pages)} 页")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- retract
|
|
|
+
|
|
|
+def _backlinks(target_keys: set) -> dict:
|
|
|
+ """扫描全库，返回 {被链页面: [引用它的页面]}。用于级联失效。"""
|
|
|
+ links = {}
|
|
|
+ for d in DIRS:
|
|
|
+ for p in (WIKI / d).rglob("*.md"):
|
|
|
+ if p.name.startswith("_"):
|
|
|
+ continue
|
|
|
+ fm, body = read_page(p)
|
|
|
+ for name in set(re.findall(r"\[\[([^\]|#]+)", body)):
|
|
|
+ links.setdefault(name.strip(), []).append(rel(p, WIKI))
|
|
|
+ return links
|
|
|
+
|
|
|
+
|
|
|
+def cmd_retract(args):
|
|
|
+ """源文件被删除后回滚知识库。三种情形：独占页归档 / 共享页降级 / 级联标记。"""
|
|
|
+ src = args.path.replace("\\", "/")
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ if src not in man["files"]:
|
|
|
+ print(f"[ERR] manifest 中无此源：{src}")
|
|
|
+ return 1
|
|
|
+
|
|
|
+ archived, degraded, cascade = [], [], []
|
|
|
+ touched_stems = set()
|
|
|
+
|
|
|
+ for d in DIRS:
|
|
|
+ for p in sorted((WIKI / d).rglob("*.md")):
|
|
|
+ if p.name.startswith("_"):
|
|
|
+ continue
|
|
|
+ fm, body = read_page(p)
|
|
|
+ srcs = fm.get("sources") or []
|
|
|
+ if not isinstance(srcs, list):
|
|
|
+ srcs = [srcs]
|
|
|
+ if src not in srcs:
|
|
|
+ continue
|
|
|
+ stamped = datetime.now(CST).strftime("%Y%m%d-%H%M%S")
|
|
|
+ if len(srcs) == 1:
|
|
|
+ ARCHIVE.mkdir(parents=True, exist_ok=True)
|
|
|
+ dest = ARCHIVE / f"{p.stem}.{stamped}.md"
|
|
|
+ fm["status"] = "archived"
|
|
|
+ fm["stale_reason"] = f"源文件已删除：{src}"
|
|
|
+ fm["updated"] = now()
|
|
|
+ write_page(dest, fm, body)
|
|
|
+ p.unlink()
|
|
|
+ archived.append(rel(dest, WIKI))
|
|
|
+ touched_stems.add(p.stem)
|
|
|
+ else:
|
|
|
+ fm["sources"] = [s for s in srcs if s != src]
|
|
|
+ fm["stale"] = True
|
|
|
+ fm["stale_reason"] = f"少了一个来源（{src}），需重新合成"
|
|
|
+ fm["updated"] = now()
|
|
|
+ write_page(p, fm, body)
|
|
|
+ degraded.append(rel(p, WIKI))
|
|
|
+ append_jsonl(REBUILD, {"ts": now(), "page": rel(p, WIKI),
|
|
|
+ "reason": "source_removed", "source": src})
|
|
|
+ touched_stems.add(p.stem)
|
|
|
+
|
|
|
+ # 级联：引用了已归档页面的页面 -> 标记待复核
|
|
|
+ if touched_stems:
|
|
|
+ back = _backlinks(touched_stems)
|
|
|
+ for stem in touched_stems:
|
|
|
+ for ref in back.get(stem, []):
|
|
|
+ fp = WIKI / ref
|
|
|
+ if not fp.exists():
|
|
|
+ continue
|
|
|
+ fm, body = read_page(fp)
|
|
|
+ if fm.get("stale"):
|
|
|
+ continue
|
|
|
+ fm["stale"] = True
|
|
|
+ fm["stale_reason"] = f"引用的页面 {stem} 已因源删除而失效"
|
|
|
+ fm["updated"] = now()
|
|
|
+ write_page(fp, fm, body)
|
|
|
+ cascade.append(ref)
|
|
|
+
|
|
|
+ rec = man["files"].pop(src, {})
|
|
|
+ man.setdefault("tombstones", []).append({"path": src, "removed_at": now(),
|
|
|
+ "sha256": rec.get("sha256"),
|
|
|
+ "pages": rec.get("pages", [])})
|
|
|
+ man["updated_at"] = now()
|
|
|
+ dump_json(MANIFEST, man)
|
|
|
+ append_jsonl(CHANGELOG, {"ts": now(), "op": "retract", "source": src,
|
|
|
+ "archived": archived, "degraded": degraded,
|
|
|
+ "cascade": sorted(set(cascade))})
|
|
|
+ log_human("retract", src,
|
|
|
+ "\n".join([f"- 归档 `{x}`" for x in archived] +
|
|
|
+ [f"- 降级待重建 `{x}`" for x in degraded] +
|
|
|
+ [f"- 级联标记 `{x}`" for x in sorted(set(cascade))]) or "- 无派生页")
|
|
|
+
|
|
|
+ print(f"[OK] 回滚 {src}")
|
|
|
+ print(f" 归档页 {len(archived)} · 降级为待重建 {len(degraded)} · 级联标记 {len(set(cascade))}")
|
|
|
+ for x in archived:
|
|
|
+ print(f" ARCHIVE {x}")
|
|
|
+ for x in degraded:
|
|
|
+ print(f" DEGRADE {x}")
|
|
|
+ for x in sorted(set(cascade)):
|
|
|
+ print(f" CASCADE {x}")
|
|
|
+ if not archived and not degraded:
|
|
|
+ print(" （该源未派生任何页面，仅清理 manifest）")
|
|
|
+
|
|
|
+
|
|
|
+def cmd_move(args):
|
|
|
+ """源文件移动：sha 相同，只改 manifest 和所有 frontmatter 引用。"""
|
|
|
+ old, new = args.old.replace("\\", "/"), args.new.replace("\\", "/")
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ if old not in man["files"]:
|
|
|
+ print(f"[ERR] manifest 中无此源：{old}")
|
|
|
+ return 1
|
|
|
+ man["files"][new] = man["files"].pop(old)
|
|
|
+ man["updated_at"] = now()
|
|
|
+ dump_json(MANIFEST, man)
|
|
|
+
|
|
|
+ n = 0
|
|
|
+ for d in DIRS:
|
|
|
+ for p in (WIKI / d).rglob("*.md"):
|
|
|
+ fm, body = read_page(p)
|
|
|
+ srcs = fm.get("sources")
|
|
|
+ if isinstance(srcs, list) and old in srcs:
|
|
|
+ fm["sources"] = [new if s == old else s for s in srcs]
|
|
|
+ fm["updated"] = now()
|
|
|
+ write_page(p, fm, body)
|
|
|
+ n += 1
|
|
|
+ append_jsonl(CHANGELOG, {"ts": now(), "op": "move", "from": old, "to": new})
|
|
|
+ log_human("move", f"{old} -> {new}", f"- 更新引用 {n} 页")
|
|
|
+ print(f"[OK] 移动登记 {old} -> {new}，更新引用 {n} 页")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- graph
|
|
|
+
|
|
|
+def all_pages():
|
|
|
+ out = []
|
|
|
+ for d in DIRS:
|
|
|
+ base = WIKI / d
|
|
|
+ if not base.exists():
|
|
|
+ continue
|
|
|
+ for p in sorted(base.rglob("*.md")):
|
|
|
+ if p.name.startswith("_"):
|
|
|
+ continue
|
|
|
+ fm, body = read_page(p)
|
|
|
+ out.append({"path": rel(p, WIKI), "stem": p.stem, "fm": fm,
|
|
|
+ "body": body, "dir": d})
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+def cmd_graph(args):
|
|
|
+ pages = all_pages()
|
|
|
+ edges = []
|
|
|
+ index_map = {p["stem"]: p for p in pages}
|
|
|
+ for p in pages:
|
|
|
+ fm = p["fm"]
|
|
|
+ for s in (fm.get("sources") or []):
|
|
|
+ edges.append({"kind": "derives", "from": str(s), "to": p["path"]})
|
|
|
+ for name in re.findall(r"\[\[([^\]|#]+)", p["body"]):
|
|
|
+ tgt = index_map.get(name.strip())
|
|
|
+ if tgt:
|
|
|
+ edges.append({"kind": "links", "from": p["path"], "to": tgt["path"]})
|
|
|
+
|
|
|
+ DERIV.write_text("".join(json.dumps(e, ensure_ascii=False) + "\n" for e in edges),
|
|
|
+ encoding="utf-8")
|
|
|
+
|
|
|
+ # 重建各目录 _index.md
|
|
|
+ ts = now()
|
|
|
+ titles = {"sources": "来源页", "topics": "主题页", "entities": "实体页",
|
|
|
+ "modules": "课程模块"}
|
|
|
+ for d in DIRS:
|
|
|
+ group = [p for p in pages if p["dir"] == d]
|
|
|
+ lines = [f"# {titles[d]}索引\n",
|
|
|
+ f"> 自动生成于 {ts}，共 {len(group)} 条。请勿手改，改页面 frontmatter 后重跑 `kb.py graph`。\n"]
|
|
|
+ if not group:
|
|
|
+ lines.append("_（暂无内容）_\n")
|
|
|
+ else:
|
|
|
+ lines.append("| 页面 | 一句话 | 标签 | 源数 | 更新 | 状态 |")
|
|
|
+ lines.append("|---|---|---|---|---|---|")
|
|
|
+ for p in sorted(group, key=lambda x: x["path"]):
|
|
|
+ fm = p["fm"]
|
|
|
+ flag = "⚠️stale" if fm.get("stale") else (fm.get("status") or "ok")
|
|
|
+ lines.append("| [{}]({}) | {} | {} | {} | {} | {} |".format(
|
|
|
+ fm.get("title") or p["stem"], p["path"],
|
|
|
+ (fm.get("summary") or "").replace("|", "/"),
|
|
|
+ ", ".join(fm.get("tags") or []) or "-",
|
|
|
+ len(fm.get("sources") or []),
|
|
|
+ str(fm.get("updated") or "")[:10], flag))
|
|
|
+ (WIKI / d / "_index.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
|
+
|
|
|
+ print(f"[OK] 关系图重建：{len(pages)} 页，{len(edges)} 条边 -> {rel(DERIV, ROOT)}")
|
|
|
+ for d in DIRS:
|
|
|
+ print(f" {d}/_index.md ({len([p for p in pages if p['dir']==d])} 条)")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- lint
|
|
|
+
|
|
|
+def cmd_lint(args):
|
|
|
+ pages = all_pages()
|
|
|
+ stems = {p["stem"] for p in pages}
|
|
|
+ aliases = set()
|
|
|
+ for p in pages:
|
|
|
+ for a in (p["fm"].get("aliases") or []):
|
|
|
+ aliases.add(str(a))
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ live_sources = set(man["files"].keys())
|
|
|
+ inbound = {}
|
|
|
+ for p in pages:
|
|
|
+ for name in re.findall(r"\[\[([^\]|#]+)", p["body"]):
|
|
|
+ inbound.setdefault(name.strip(), []).append(p["path"])
|
|
|
+
|
|
|
+ issues = []
|
|
|
+
|
|
|
+ # 源文件路径 -> 该源的 source 页所标证据等级（用于木桶校验）
|
|
|
+ # 字母序 A<B<C<D，越靠后越弱，所以「最弱」取 max
|
|
|
+ src_ev = {}
|
|
|
+ for p in pages:
|
|
|
+ if p["fm"].get("type") == "source" and p["fm"].get("evidence"):
|
|
|
+ for s in (p["fm"].get("sources") or []):
|
|
|
+ ev = p["fm"]["evidence"]
|
|
|
+ src_ev[s] = max(src_ev[s], ev) if s in src_ev else ev
|
|
|
+
|
|
|
+ for p in pages:
|
|
|
+ fm = p["fm"]
|
|
|
+ if not fm:
|
|
|
+ issues.append(("NO_FM", p["path"], "缺少 frontmatter"))
|
|
|
+ continue
|
|
|
+ for f in ("title", "type", "summary"):
|
|
|
+ if not fm.get(f):
|
|
|
+ issues.append(("MISSING_FIELD", p["path"], f"缺少字段 {f}"))
|
|
|
+ ev = fm.get("evidence")
|
|
|
+ if not ev:
|
|
|
+ issues.append(("MISSING_EVIDENCE", p["path"], "未标证据等级 A/B/C/D"))
|
|
|
+ elif ev not in ("A", "B", "C", "D"):
|
|
|
+ issues.append(("BAD_EVIDENCE", p["path"], f"证据等级非法：{ev}"))
|
|
|
+ else:
|
|
|
+ if fm.get("type") in ("topic", "module"):
|
|
|
+ lows = [src_ev[s] for s in (fm.get("sources") or []) if s in src_ev]
|
|
|
+ weakest = max(lows) if lows else None # 字母最大 = 证据最弱
|
|
|
+ if weakest and ev < weakest:
|
|
|
+ issues.append(("EVIDENCE_INFLATED", p["path"],
|
|
|
+ f"标了 {ev}，但来源最低只有 {weakest}（合成不提升证据强度）"))
|
|
|
+ if fm.get("type") == "module" and ev == "D":
|
|
|
+ issues.append(("D_IN_MODULE", p["path"], "D 级内容不许进课件"))
|
|
|
+ if len(p["body"].strip()) < 100:
|
|
|
+ issues.append(("THIN", p["path"], f"正文过短（{len(p['body'].strip())} 字）"))
|
|
|
+ for s in (fm.get("sources") or []):
|
|
|
+ if s not in live_sources:
|
|
|
+ issues.append(("DANGLING_SOURCE", p["path"], f"引用了不存在的源：{s}"))
|
|
|
+ for name in set(re.findall(r"\[\[([^\]|#]+)", p["body"])):
|
|
|
+ if name.strip() not in stems and name.strip() not in aliases:
|
|
|
+ issues.append(("DEAD_LINK", p["path"], f"[[{name.strip()}]] 无对应页"))
|
|
|
+ if not (fm.get("sources") or []) and not inbound.get(p["stem"]):
|
|
|
+ issues.append(("ORPHAN", p["path"], "无来源且无入链"))
|
|
|
+ if fm.get("stale"):
|
|
|
+ issues.append(("STALE", p["path"], fm.get("stale_reason") or "待重建"))
|
|
|
+ if fm.get("type") == "source":
|
|
|
+ src = (fm.get("sources") or [None])[0]
|
|
|
+ if src and man["files"].get(src, {}).get("sha256") and \
|
|
|
+ fm.get("source_sha") and fm["source_sha"] != man["files"][src]["sha256"]:
|
|
|
+ issues.append(("SOURCE_DRIFT", p["path"], "源文件已变更，页面需重新整理"))
|
|
|
+
|
|
|
+ pend = read_jsonl(REBUILD)
|
|
|
+ summary = {
|
|
|
+ "pages": len(pages), "sources": len(live_sources),
|
|
|
+ "issues": len(issues),
|
|
|
+ "by_kind": {},
|
|
|
+ "rebuild_queue": len(pend),
|
|
|
+ "score": max(0, 100 - len(issues) * 3 - len(pend) * 2),
|
|
|
+ }
|
|
|
+ for k, _, _ in issues:
|
|
|
+ summary["by_kind"][k] = summary["by_kind"].get(k, 0) + 1
|
|
|
+
|
|
|
+ if args.json:
|
|
|
+ print(json.dumps({"summary": summary,
|
|
|
+ "issues": [{"kind": k, "page": p, "detail": d} for k, p, d in issues]},
|
|
|
+ ensure_ascii=False, indent=2))
|
|
|
+ return
|
|
|
+
|
|
|
+ print(f"健康分 {summary['score']}/100 · {len(pages)} 页 · {len(live_sources)} 源 · "
|
|
|
+ f"{len(issues)} 问题 · {len(pend)} 项待重建")
|
|
|
+ if issues:
|
|
|
+ print("\n按类型：")
|
|
|
+ for k, v in sorted(summary["by_kind"].items(), key=lambda x: -x[1]):
|
|
|
+ print(f" {k:<18} {v}")
|
|
|
+ print("\n明细：")
|
|
|
+ for k, p, d in issues:
|
|
|
+ print(f" [{k}] {p} — {d}")
|
|
|
+ else:
|
|
|
+ print("未发现问题。")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- status
|
|
|
+
|
|
|
+def cmd_status(args):
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ pages = all_pages()
|
|
|
+ pend = read_jsonl(REBUILD)
|
|
|
+ logs = read_jsonl(CHANGELOG)
|
|
|
+ cur = collect_raw()
|
|
|
+ pending_new = set(cur) - set(man["files"])
|
|
|
+ pending_gone = set(man["files"]) - set(cur)
|
|
|
+ stale = [p["path"] for p in pages if p["fm"].get("stale")]
|
|
|
+
|
|
|
+ print("=" * 56)
|
|
|
+ print(" 做课系统 · 知识库状态")
|
|
|
+ print("=" * 56)
|
|
|
+ print(f" 根目录 {ROOT}")
|
|
|
+ print(f" 最近同步 {man.get('updated_at') or '从未'}")
|
|
|
+ print(f" 原料文件 {len(cur)} 个（已整理 {len(man['files'])}）")
|
|
|
+ print(f" 知识页 {len(pages)} 页")
|
|
|
+ for d in DIRS:
|
|
|
+ print(f" {d:<10} {len([p for p in pages if p['dir']==d])}")
|
|
|
+ print(f" 待处理新增 {len(pending_new)}")
|
|
|
+ print(f" 待处理删除 {len(pending_gone)}")
|
|
|
+ print(f" 待重建 {len(pend)}")
|
|
|
+ print(f" 陈旧页 {len(stale)}")
|
|
|
+ print(f" 历史操作 {len(logs)} 条")
|
|
|
+ if pending_new:
|
|
|
+ print("\n 未入库：")
|
|
|
+ for x in sorted(pending_new)[:12]:
|
|
|
+ print(f" + {x}")
|
|
|
+ if len(pending_new) > 12:
|
|
|
+ print(f" ... 还有 {len(pending_new)-12} 个")
|
|
|
+ if pending_gone:
|
|
|
+ print("\n 源已消失（待回滚）：")
|
|
|
+ for x in sorted(pending_gone)[:12]:
|
|
|
+ print(f" - {x}")
|
|
|
+ if stale:
|
|
|
+ print("\n 陈旧页：")
|
|
|
+ for x in stale[:12]:
|
|
|
+ print(f" ! {x}")
|
|
|
+ print("=" * 56)
|
|
|
+ if pending_new or pending_gone or pend:
|
|
|
+ print(" → 建议运行：kb-sync")
|
|
|
+ else:
|
|
|
+ print(" → 库是干净的")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- guide
|
|
|
+
|
|
|
+def cmd_guide(args):
|
|
|
+ """给 AI 的自描述入口（对应 llmwiki 的 guide 工具）。
|
|
|
+
|
|
|
+ 为什么要它：每次新会话，AI 都要重建对这套系统的认知。让它读 README 加 8 个
|
|
|
+ SKILL.md，要烧掉几万 token，读完还不一定知道"现在该干什么"。
|
|
|
+ 这个命令一次给全：这是什么、铁律是什么、当前什么状态、下一步做什么。
|
|
|
+ """
|
|
|
+ man = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ pages = all_pages()
|
|
|
+ cur = collect_raw()
|
|
|
+ pend_new = sorted(set(cur) - set(man["files"]))
|
|
|
+ pend_gone = sorted(set(man["files"]) - set(cur))
|
|
|
+ stale = [p["path"] for p in pages if p["fm"].get("stale")]
|
|
|
+ nreb = len(read_jsonl(REBUILD))
|
|
|
+
|
|
|
+ print("=" * 60)
|
|
|
+ print(" 做课系统 · 给接手者的指南")
|
|
|
+ print("=" * 60)
|
|
|
+ print("""
|
|
|
+【这是什么】
|
|
|
+ 一条单向管道：
|
|
|
+ 00-原材料/ ──你整理──▶ 10-知识库/ ──取材──▶ 20-课程产出/
|
|
|
+ （只进不改） （全 md，你维护） （交付物）
|
|
|
+ 人是资料策展人，你是维护者。人不写知识库，你写。
|
|
|
+
|
|
|
+【三条铁律】
|
|
|
+ 1. 差异问脚本，别凭记忆 —— 判断哪些文件是新的，永远跑 kb.py scan
|
|
|
+ 2. 脚本管事实，你管语义 —— 页数、hash、关系图都问脚本
|
|
|
+ 3. 补料不许编，取材必标出处 —— 这是两条方向相反的规则，见 AGENTS.md
|
|
|
+
|
|
|
+【该走哪条路】
|
|
|
+ 丢进了新资料 / 删了源文件 → 30-技能/kb-sync （先 scan 再分流）
|
|
|
+ 查点东西 → 30-技能/kb-query （好答案记得沉淀回库）
|
|
|
+ 要做一门课 → 30-技能/course-outline → course-slides
|
|
|
+ 库看着有点乱 → kb.py check （结构 + 内容一起过）
|
|
|
+ 改页面格式或加规则 → 10-知识库/_meta/约定.md（改完登记起因）
|
|
|
+
|
|
|
+【格式硬约束】（细节见 约定.md）
|
|
|
+ 每页 frontmatter 必填：type / title / summary / evidence / sources
|
|
|
+ evidence 只能是 A/B/C/D，且 topic、module 页不得高于其来源最低级
|
|
|
+ sources 填原材料相对路径，一字不差 —— 这是删除回滚的唯一依据
|
|
|
+""")
|
|
|
+ print("-" * 60)
|
|
|
+ print("【当前状态】")
|
|
|
+ print(f" 原料 {len(cur)} 个（已整理 {len(man['files'])}）· "
|
|
|
+ f"知识页 {len(pages)} 页 · 最近同步 {man.get('updated_at') or '从未'}")
|
|
|
+ for d in DIRS:
|
|
|
+ n = len([p for p in pages if p["dir"] == d])
|
|
|
+ if n:
|
|
|
+ print(f" {d:<10} {n}")
|
|
|
+ print(f" 待处理新增 {len(pend_new)} · 待处理删除 {len(pend_gone)} · "
|
|
|
+ f"待重建 {nreb} · 陈旧页 {len(stale)}")
|
|
|
+
|
|
|
+ print("\n【下一步】")
|
|
|
+ todo = []
|
|
|
+ if pend_new:
|
|
|
+ todo.append(f"有 {len(pend_new)} 个新源未入库 → kb-sync")
|
|
|
+ if pend_gone:
|
|
|
+ todo.append(f"有 {len(pend_gone)} 个源已消失 → kb-sync 的 retract 分支")
|
|
|
+ if nreb:
|
|
|
+ todo.append(f"有 {nreb} 页待重建 → 读 _meta/rebuild_queue.jsonl")
|
|
|
+ if stale:
|
|
|
+ todo.append(f"有 {len(stale)} 页被标 stale → 处理完记得清标记")
|
|
|
+ if not pages:
|
|
|
+ todo.append("库还是空的 → 把资料丢进 00-原材料/，然后说「同步知识库」")
|
|
|
+ if not todo:
|
|
|
+ todo.append("库是干净的 → 可以说「做一门课」或「查 XX」")
|
|
|
+ for t in todo:
|
|
|
+ print(f" · {t}")
|
|
|
+ print("=" * 60)
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- repair
|
|
|
+
|
|
|
+def cmd_repair(args):
|
|
|
+ """从磁盘现状 + 页面 frontmatter 反推 manifest。
|
|
|
+
|
|
|
+ 什么时候用：
|
|
|
+ 1. git 合并把 manifest.json 搞冲突了
|
|
|
+ 2. manifest 被误删、或从别处只拷了页面没拷 _meta
|
|
|
+ 3. 换机器接手
|
|
|
+
|
|
|
+ 能恢复什么：每个源的指纹、它派生了哪些页、已整理/未整理状态。
|
|
|
+ 恢复不了什么：整理的具体时间（只能留空）、删除过的源记录（tombstone 会丢）。
|
|
|
+ """
|
|
|
+ cur = collect_raw()
|
|
|
+ derived = {}
|
|
|
+
|
|
|
+ # 活跃页 + 归档页都算 —— 归档页的 sources 仍然说明"这个源整理过"
|
|
|
+ scan_roots = [WIKI / d for d in DIRS]
|
|
|
+ if ARCHIVE.exists():
|
|
|
+ scan_roots.append(ARCHIVE)
|
|
|
+
|
|
|
+ for root in scan_roots:
|
|
|
+ if not root.exists():
|
|
|
+ continue
|
|
|
+ for p in sorted(root.rglob("*.md")):
|
|
|
+ if p.name.startswith("_"):
|
|
|
+ continue
|
|
|
+ fm, _ = read_page(p)
|
|
|
+ for s in (fm.get("sources") or []):
|
|
|
+ near = p if p.parent != ARCHIVE else ARCHIVE / p.name
|
|
|
+ derived.setdefault(str(s), []).append(rel(near, WIKI))
|
|
|
+
|
|
|
+ old = load_json(MANIFEST, manifest_skeleton())
|
|
|
+ files, orphan_disk, orphan_pages = {}, [], []
|
|
|
+
|
|
|
+ for r, meta in cur.items():
|
|
|
+ ps = sorted(set(derived.pop(r, [])))
|
|
|
+ prev = old.get("files", {}).get(r, {})
|
|
|
+ files[r] = {
|
|
|
+ "size": meta["size"], "mtime_ns": meta["mtime_ns"],
|
|
|
+ "sha256": meta["sha256"], "id": prev.get("id") or short_id(r),
|
|
|
+ "status": "ingested" if ps else "pending",
|
|
|
+ "ingested_at": prev.get("ingested_at"),
|
|
|
+ "pages": ps,
|
|
|
+ }
|
|
|
+ if not ps:
|
|
|
+ orphan_disk.append(r)
|
|
|
+
|
|
|
+ # 页面引用了磁盘上不存在的源
|
|
|
+ for s, ps in derived.items():
|
|
|
+ orphan_pages.extend(ps)
|
|
|
+
|
|
|
+ man = {"version": 2, "root": "00-原材料", "updated_at": now(),
|
|
|
+ "files": files, "tombstones": old.get("tombstones", [])}
|
|
|
+ dump_json(MANIFEST, man)
|
|
|
+ log_human("repair", f"重建 manifest：{len(files)} 个源",
|
|
|
+ f"- 已整理 {len(files)-len(orphan_disk)} 个，未整理 {len(orphan_disk)} 个")
|
|
|
+
|
|
|
+ print(f"[OK] manifest 已重建：{len(files)} 个源")
|
|
|
+ print(f" 已整理 {len(files)-len(orphan_disk)} · 未整理 {len(orphan_disk)}")
|
|
|
+ if orphan_disk:
|
|
|
+ print(f"\n 磁盘上有、但没派生任何页（下次同步会当新的处理）：")
|
|
|
+ for x in orphan_disk[:10]:
|
|
|
+ print(f" · {x}")
|
|
|
+ if len(orphan_disk) > 10:
|
|
|
+ print(f" ... 还有 {len(orphan_disk)-10} 个")
|
|
|
+ if orphan_pages:
|
|
|
+ print(f"\n ⚠️ 有 {len(set(orphan_pages))} 个页面引用了磁盘上已不存在的源")
|
|
|
+ print(f" → 跑 kb.py lint 看 DANGLING_SOURCE，再决定恢复文件还是 retract")
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- check
|
|
|
+
|
|
|
+def cmd_check(args):
|
|
|
+ """聚合器：结构检查（本文件）+ 内容检查（check-content.py）。
|
|
|
+ 两个检查器各自独立可跑，这里是给"提交前一次过"用的。"""
|
|
|
+ cmd_lint(argparse.Namespace(json=False))
|
|
|
+ print("\n" + "-" * 56)
|
|
|
+ script = ROOT / "90-工具" / "check-content.py"
|
|
|
+ if not script.exists():
|
|
|
+ print("（未找到 check-content.py，跳过内容检查）")
|
|
|
+ return 0
|
|
|
+ r = subprocess.run([sys.executable, str(script)],
|
|
|
+ capture_output=True, text=True,
|
|
|
+ encoding="utf-8", errors="replace")
|
|
|
+ print(r.stdout.strip() or r.stderr.strip()[:800])
|
|
|
+ return 0 if r.returncode == 0 else 1
|
|
|
+
|
|
|
+
|
|
|
+# ---------------------------------------------------------------- main
|
|
|
+
|
|
|
+def main():
|
|
|
+ ap = argparse.ArgumentParser(prog="kb", description="做课系统知识库工具链")
|
|
|
+ sub = ap.add_subparsers(dest="cmd")
|
|
|
+
|
|
|
+ sub.add_parser("status", help="总览")
|
|
|
+ sub.add_parser("guide", help="给 AI 的自描述指南")
|
|
|
+ s = sub.add_parser("scan", help="扫描原料目录，输出差异")
|
|
|
+ s.add_argument("--json", action="store_true")
|
|
|
+
|
|
|
+ m = sub.add_parser("mark", help="登记某源已整理")
|
|
|
+ m.add_argument("path")
|
|
|
+ m.add_argument("--pages", default="")
|
|
|
+
|
|
|
+ r = sub.add_parser("retract", help="源删除后回滚")
|
|
|
+ r.add_argument("path")
|
|
|
+
|
|
|
+ mv = sub.add_parser("move", help="登记源文件移动")
|
|
|
+ mv.add_argument("old")
|
|
|
+ mv.add_argument("new")
|
|
|
+
|
|
|
+ sub.add_parser("graph", help="重建关系图与索引")
|
|
|
+
|
|
|
+ l = sub.add_parser("lint", help="结构健康检查")
|
|
|
+ l.add_argument("--json", action="store_true")
|
|
|
+
|
|
|
+ sub.add_parser("check", help="全量检查：结构 + 内容")
|
|
|
+
|
|
|
+ sub.add_parser("repair", help="从磁盘和页面反推 manifest（git 冲突后用）")
|
|
|
+
|
|
|
+ args = ap.parse_args()
|
|
|
+ fn = {"status": cmd_status, "guide": cmd_guide, "scan": cmd_scan, "mark": cmd_mark,
|
|
|
+ "retract": cmd_retract, "move": cmd_move, "graph": cmd_graph,
|
|
|
+ "lint": cmd_lint, "check": cmd_check, "repair": cmd_repair}.get(args.cmd)
|
|
|
+ if not fn:
|
|
|
+ ap.print_help()
|
|
|
+ return 0
|
|
|
+ return fn(args) or 0
|
|
|
+
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ sys.exit(main())
|
