#!/usr/bin/env python3 # -*- coding: utf-8 -*- """资源索引 · 双向对账机检（技能包内通用） 用途 ---- 每个技能包在 `references/manifest.md` 里维护一张「资源索引」表： | 资源 | 落点 | 干什么 | 被谁引用 | 许可字面量 | 口径 | 入仓 | 本脚本对这张表做**双向对账**，四个闸门： G1 落点存在 —— 每行 `落点` 必须在磁盘上（`入仓=否` 的行本机缺只给 ⚠，属正常） G2 引用为真 —— `被谁引用` 里登记的每个文件必须存在、且真的提到该资源（⛔ 不许登记假引用点； `许可字面量` 指向的文件也必须存在，⛔ 不许开空头许可） G3 字面量收敛 —— `口径=只写一次` 的行：该资源的路径字面量只许出现在清单本身与 `许可字面量` 登记的文件里， 出现在别处 ⇒ 🔴。这条就是「避免资源到处引用」的机检（⛔ 漏一处必报，不静默放行） G4 孤儿 / 未登记 —— `被谁引用` 为空的行 ⇒ 🔴；`assets/` 与 `references/stage-*/assets/` 下 磁盘上有一级子项而表里没有 ⇒ 🔴 🔴 `被谁引用` 与 `许可字面量` 是**两件事，别合并**：前者是「谁在用」（可由 `--sync-refs` 自动回写）， 后者是「谁被允许写路径字面量」（人工开的口子，⛔ 自动回写会把它冲成"谁都许可"⇒ G3 变空闸门）。 「被谁引用」怎么认：既认**路径形态**（含该资源路径尾两段），也认**名字形态**（含资源名或路径末段） —— 因为收敛之后散文里就只剩名字了，只认路径会把收敛好的资源误判成孤儿。 用法 ---- python check_resource_index.py --pkg <技能包根> python check_resource_index.py --pkg <技能包根> --index references/01-文档索引.md python check_resource_index.py --pkg <技能包根> --sync-refs # 用扫描结果回写「被谁引用」列 🔴 索引文件必须放在该技能包**入库（git 跟踪）**的那份文件里 —— 实测踩过： `session-mechanism/references/manifest.md` 被 `.gitignore` 挡着（它是 `install.py --manifest` 在本机生成、随机器变）⇒ 索引写那里＝别台机器取仓后**看不到**，机制静默失效。 该包的索引改放 `references/01-文档索引.md`（入库），用 `--index` 指过去。 退出码：0 全过 ｜ 1 有 🔴 ｜ 2 用法或清单结构错（找不到表头 ⇒ 停手，⛔ 不猜） """ from __future__ import annotations import argparse import re import sys from pathlib import Path try: # 出口一律声明编码（本机铁律） sys.stdout.reconfigure(encoding="utf-8", errors="replace") sys.stderr.reconfigure(encoding="utf-8", errors="replace") except Exception: pass HEADER = "| 资源 |" COLS = ["资源", "落点", "干什么", "被谁引用", "许可字面量", "口径", "入仓"] COLLAPSE = "只写一次" # 口径取值：只写一次 / 可多次 SYNC_COL = 3 # 「被谁引用」在第几列（0 起） SYNC_CAP = 12 # 回写「被谁引用」时最多列几个文件，其余只记个数 TEXT_EXT = {".md", ".py", ".js", ".mjs", ".cjs", ".json", ".html", ".htm", ".css", ".sh", ".bash", ".tpl", ".txt", ".yaml", ".yml"} SKIP_DIR = {"_留痕", "__pycache__", "logs", "node_modules", ".git", "env", ".obsidian", "marks"} SKIP_FILE = {"install.log"} def is_scannable(p: Path, pkg: Path) -> bool: rel = p.relative_to(pkg) if any(part in SKIP_DIR for part in rel.parts): return False if p.name in SKIP_FILE or p.suffix not in TEXT_EXT: return False return True def scan_files(pkg: Path) -> list[Path]: out = [p for p in pkg.rglob("*") if p.is_file() and is_scannable(p, pkg)] return sorted(out, key=lambda q: q.relative_to(pkg).as_posix()) def read_text(p: Path) -> str: try: return p.read_text(encoding="utf-8", errors="replace") except Exception: return "" def cell(s: str) -> str: return s.strip().strip("`").strip() def split_refs(raw: str) -> list[str]: """`被谁引用` / `许可字面量` 列 → 条目清单（按 `、；;,` 分隔）。 🔴 原样保留条目文本，⛔ 不在这一步剥括号 —— 真实文件名里可能就带全角括号 （实测踩过：`references/作业规矩/00-作业总规矩（原 agent-operating-rules）.md` 被剥成 `…/00-作业总规矩.md` ⇒ 不存在 ⇒ 报假 🔴）。括号交给 `resolve()` 按「原样 → 去括号」两候选试。 """ if not raw or raw.strip() in ("—", "-", "（无）", "(无)"): return [] out: list[str] = [] for it in re.split(r"[；;、,，]", raw): s = cell(it) if s: out.append(s) return out def resolve(pkg: Path, item: str) -> Path | None: """把一条登记项解析成真实路径：先按原样，再按「剥掉括号说明」的样子各试一次。""" if item.startswith("<技能根>/"): cands = [same_root_dir(pkg, item)] else: cands = [pkg / item, pkg / re.sub(r"（[^）]*）", "", item).strip(), pkg / re.sub(r"\([^)]*\)", "", item).strip()] for c in cands: if c.exists(): return c return None class Row: def __init__(self, raw: str): c = raw.strip().strip("|").split("|") c += [""] * (len(COLS) - len(c)) self.name = cell(c[0]) self.loc = cell(c[1]) self.what = cell(c[2]) self.refs_raw = c[3].strip() self.allow_raw = c[4].strip() # 许可字面量：允许写该资源路径字面量的文件（G3 用） self.mode = cell(c[5]) self.inrepo = cell(c[6]) segs = [s for s in self.loc.replace("<技能根>/", "").split("/") if s] self.leaf = segs[-1] if segs else self.loc # 路径末段 self.key = "/".join(segs[-2:]) if len(segs) >= 2 else self.leaf # 路径尾两段 if self.loc.startswith("<技能根>/") and len(segs) == 1: self.key = segs[0] def load_table(man: Path) -> tuple[list[Row], int]: lines = man.read_text(encoding="utf-8").replace("\r\n", "\n").split("\n") head = next((i for i, ln in enumerate(lines) if ln.startswith(HEADER)), None) if head is None: raise SystemExit(f"🔴 {man} 里找不到 `{HEADER}` 表头 ⇒ 停手（⛔ 不猜、不追加）") rows: list[Row] = [] i = head + 2 while i < len(lines) and lines[i].startswith("|"): if cell(lines[i].strip().strip("|").split("|")[0]): rows.append(Row(lines[i])) i += 1 if not rows: raise SystemExit(f"🔴 {man} 的表里没有任何数据行 ⇒ 停手") return rows, head def path_hits(r: Row, pkg: Path, files: list[Path], man: Path) -> list[tuple[Path, int, str]]: """G3 用：找**路径形态**（含资源路径尾两段字面量）的出现。""" pat = re.compile(re.escape(r.key)) out: list[tuple[Path, int, str]] = [] for f in files: if f == man: continue for n, ln in enumerate(read_text(f).split("\n"), 1): if r.key in ln and pat.search(ln): out.append((f, n, ln.strip()[:120])) return out def ref_hits(r: Row, pkg: Path, files: list[Path], man: Path) -> list[str]: """「被谁引用」用：路径形态 或 名字形态（资源名 / 路径末段）都算。""" toks = {t for t in (r.key, r.leaf, r.name) if t} hits: list[str] = [] for f in files: if f == man: continue txt = read_text(f) if any(t in txt for t in toks): hits.append(f.relative_to(pkg).as_posix()) return hits def same_root_dir(pkg: Path, loc: str) -> Path: name = loc.replace("<技能根>/", "").strip("/").split("/")[0] return pkg.parent / name def main() -> int: ap = argparse.ArgumentParser(description="资源索引 · 双向对账机检") ap.add_argument("--pkg", required=True, help="技能包根（如 E:/.../skills/product-planning）") ap.add_argument("--index", default="references/manifest.md", help="资源索引表所在文件（默认 references/manifest.md；" "某包的清单若不入库，就把索引放进该包**入库**的那份索引文件，再指过来）") ap.add_argument("--sync-refs", action="store_true", help="用扫描结果回写「被谁引用」列（逗号分隔的文件清单）") ap.add_argument("--max-list", type=int, default=10, help="每个闸门最多列几条明细") args = ap.parse_args() pkg = Path(args.pkg).expanduser().resolve() if not pkg.is_dir(): print(f"🔴 技能包根不存在：{pkg}") return 2 man = pkg / args.index if not man.is_file(): print(f"🔴 找不到资源索引文件 {man}（用 --index 指定；默认 references/manifest.md）") return 2 rows, head = load_table(man) files = scan_files(pkg) print(f"═══ 资源索引对账 · {pkg.name} ═══") print(f"清单：{man.relative_to(pkg).as_posix()} ｜ 表行：{len(rows)} ｜ 扫描文件：{len(files)}\n") g1: list[str] = [] g2: list[str] = [] g3: list[str] = [] g4: list[str] = [] computed: dict[str, list[str]] = {} for r in rows: # ── G1 落点存在 ──────────────────────────────────────────────── if r.loc.startswith("<技能根>/"): cand = same_root_dir(pkg, r.loc) if not cand.is_dir() and r.inrepo != "否": g1.append(f"「{r.name}」落点（同根技能）不存在：{cand}") else: cand = pkg / r.loc.rstrip("/") if not cand.exists(): if r.inrepo == "否": print(f" ⚠ 「{r.name}」落点本机缺（入仓=否 ⇒ 正常，新机器取仓后这一层为空）：{r.loc}") else: g1.append(f"「{r.name}」落点不存在：{r.loc}") # ── 引用扫描（G2 / G3 / --sync-refs 共用）────────────────────── computed[r.name] = ref_hits(r, pkg, files, man) declared = split_refs(r.refs_raw) # ── G2 引用为真 ──────────────────────────────────────────────── for d in declared: dp = resolve(pkg, d) if dp is None: g2.append(f"「{r.name}」登记的引用点不存在：{d}") elif not dp.is_file(): pass # 同根技能目录：只判存在 elif r.name not in read_text(dp) and r.leaf not in read_text(dp) \ and r.key not in read_text(dp): g2.append(f"「{r.name}」登记的引用点其实没提到它：{d}") # ── G3 字面量收敛（仅 `只写一次`）────────────────────────────── if r.mode == COLLAPSE: allow: set[str] = set() for d in split_refs(r.allow_raw): if resolve(pkg, d) is None: g2.append(f"「{r.name}」`许可字面量` 指向的文件不存在：{d}") # ⛔ 不许开空头许可 allow.add(d.replace("\\", "/")) for f, n, t in path_hits(r, pkg, files, man): rel = f.relative_to(pkg).as_posix() if rel in allow: continue g3.append(f"「{r.name}」字面量出现在未登记处：{rel}:{n} ⇒ {t}") # ── G4 另一半：孤儿登记 ＋ 磁盘上有而表里没有 ──────────────────── for r in rows: if not split_refs(r.refs_raw): g4.append(f"「{r.name}」`被谁引用` 为空 ⇒ 没人引用的登记项") registered = {r.loc.rstrip("/") for r in rows} for d in [pkg / "assets"] + sorted(pkg.glob("references/stage-*/assets")): if not d.is_dir(): continue for sub in sorted(d.iterdir()): if sub.name.startswith(".") or sub.name.startswith("_"): continue rel = sub.relative_to(pkg).as_posix().rstrip("/") if not any(rl.endswith(rel) or rel.endswith(rl) for rl in registered): g4.append(f"磁盘上有、清单里没有：{rel}") # ── 汇总 ──────────────────────────────────────────────────────────── gates = [("G1 落点存在", g1), ("G2 引用为真", g2), ("G3 字面量收敛", g3), ("G4 孤儿 / 未登记", g4)] bad = 0 for title, items in gates: if items: bad += 1 print(f"🔴 {title}：{len(items)} 处") for x in items[: args.max_list]: print(f" · {x}") if len(items) > args.max_list: print(f" … 余 {len(items) - args.max_list} 处") else: print(f"✅ {title}：0") if args.sync_refs: raw = man.read_bytes() crlf = b"\r\n" in raw lines = raw.decode("utf-8").replace("\r\n", "\n").split("\n") for i in range(head + 2, len(lines)): if not lines[i].startswith("|"): break c = lines[i].strip().strip("|").split("|") if len(c) < len(COLS) or not cell(c[0]): continue lst = computed.get(cell(c[0]), []) shown = lst[:SYNC_CAP] txt = "、".join(shown) if len(lst) > SYNC_CAP: txt += f"（共 {len(lst)} 个）" # 括号里的计数会被 split_refs 剥掉，不影响 G2 校验 c[3] = " " + (txt if txt else "—") + " " lines[i] = "|" + "|".join(c) + "|" out = "\n".join(lines) man.write_bytes((out.replace("\n", "\r\n") if crlf else out).encode("utf-8")) print(f"\n✍️ 已用扫描结果回写「被谁引用」列：{man.name}") print() if bad: print(f"结论：🔴 不通过 —— {bad} 个闸门有命中（⛔ 漏一处必报，不静默放行）") return 1 print("结论：✅ 通过 —— 四个闸门全 0") return 0 if __name__ == "__main__": raise SystemExit(main())