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