Claude Code
da9bd53cae
兩張票放同一條分支：都是「該發生卻不會自己發生的事」，觸發點都在 session 邊界。 ## inkstone/ISEP#93 —— 逾期和掛著沒人接的事會主動叫 - scripts/isep-nag 撈三種沒人會叫的事（逾期 milestone／等 leo 的票／掉在地上的棒子） - scripts/isep-notify 發 Telegram，而且**發不出去的時候不會安靜** - hooks/overdue-nag-guard.sh SessionStart 跑一次（不輪詢、不 fan-out、不掛 Actions） - hooks/tests/overdue-nag.test.sh 35 條，全程離線 實跑撈得出票上點名的那五個逾期 milestone（08-24 三個、08-26 兩個）。 沒東西可報時會說「查過了，沒有」——安靜跟壞掉長得一模一樣。 🔴 工單補的那個限制（今天實測出來的）已經處理： 「能不能發得出去」取決於這個 session 載到的 ISEP 是哪一版 （prod-write-guard v0.10.0 才認得出 notify_leo 不是部署，而 hook 註冊路徑 在 session 啟動當下就寫死了）。所以 isep-notify 會**先拿真的要送的那一則 去問這個 session 註冊的那支閘**，把判定寫成檔（誰都查得到），然後： · 放行 ⇒ 送，並驗內層 data.data.ok（外層 200 不算送到） · 會擋 ⇒ **不繞路**，改貼回票上並把原文印在眼前 用 git 歷史裡的真跡（c263866 那一版閘）測過「會擋」那條路。 ⛔ 實測發現通道本身現在是斷的：實例上找不到 notify_leo 工作流（404）。 不是閘、不是網路、不是金鑰。詳情與修法寫在 docs/TESTING.md 最後一段。 退路兩次都走通了（inkstone/ISEP#93 comment 5182、5184）。 ## inkstone/ISEP#89 —— wiki 太長時有人整理 - scripts/wiki-compress audit／plan／apply／verify／bench 五個動詞 - hooks/wiki-size-guard.sh SessionStart 點名太長的檔；寫檔時擋「沒走流程的壓縮」 - hooks/tests/wiki-compress.test.sh 25 條，全程離線、不碰真的 wiki 設計上最重要的一條：**只搬不改**，正文一個字都不動。 「合併同類、濃縮成一行」要重寫正文，而重寫的當下沒有人會發現弄丟了什麼。 所以機器只做「搬 ＋ 目錄 ＋ 標 ×N／↻」，合併留給人。 拿現在的 mistakes.md 複本實壓過（不動真的 wiki）： 7,681 行 → 1,199 行 260 條一條都沒少（verify 用內文雜湊逐條對帳） 80 個查詢命中率 80/80，定位成本 2,956 → 131 行，**快 22.5 倍**（bench） verify 反向測過：真的弄丟一條時它抓得到（抓不到的對帳表比沒有更糟）。 ## 盤點數字 在這棵樹上當場數的，不是拿上一版加減推的： ls hooks/*.sh | wc -l → 55（was 53） grep -c '"command":' hooks.json → 71（was 68） plugin.json／README／hooks-inventory 三處同步改。 **版本號沒動**（0.11.0），待總管定版。
599 lines
27 KiB
Python
Executable File
599 lines
27 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""wiki-compress — 讀不完的必讀檔，等於沒有（inkstone/ISEP#89）
|
||
|
||
━━ 這支解的問題 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
wiki 一直長大，沒有人會回頭整理。`mistakes.md` 是「必讀」等級的檔案，
|
||
實測 **7,680 行、168 條**——**沒有人真的每次都從頭讀。**
|
||
|
||
leo 2026-08-27 說那天「失去記憶」。東西還在，只是**沒人讀得完**。
|
||
|
||
已經有的三支管的是別的事：`wiki-secret-scan`（寫進去安不安全）、
|
||
`wiki-first-search`／`wiki-first-police`（有沒有先查）。**沒有任何一支管「太大了」。**
|
||
|
||
━━ 🔴 最重要的設計決定：**只搬不改** ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
票上第 2 條驗收是「**不能弄丟任何一條教訓**（要能逐條對帳）」。
|
||
|
||
⇒ 所以本支**不改寫任何一條的正文**，一個字都不動。它只做三件事：
|
||
① 產生**目錄**（每條一行：標題＋日期＋票號＋去哪了）放回原檔開頭
|
||
② 把舊的整條**原封不動搬**到 `<檔名>-archive-YYYY-MM.md`
|
||
③ 把「同一件事的第 N 次」在目錄裡標成 `×N`，**而不是把它們合併掉**
|
||
|
||
「合併同類、濃縮成一行」聽起來更漂亮，但那要重寫正文——
|
||
**而重寫的當下沒有人會發現弄丟了什麼**（票上原話）。
|
||
只搬不改的好處是：`verify` 可以用**內文雜湊**逐條對帳，
|
||
「沒弄丟」不是宣稱，是算出來的。
|
||
|
||
━━ 五個動詞 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
wiki-compress audit [<wiki 目錄或檔>...] 哪幾個檔太大了、大在哪
|
||
wiki-compress plan <檔> 要怎麼壓：同類幾群、可搬幾條
|
||
wiki-compress apply <檔> --ticket owner/repo#N
|
||
真的壓（只搬不改），並留交付紀錄
|
||
wiki-compress verify <壓縮前的檔> <壓縮後的檔> [搬去的檔...]
|
||
逐條對帳：有沒有弄丟（票上驗收 2）
|
||
wiki-compress bench <壓縮前的檔> <壓縮後的檔> [搬去的檔...]
|
||
量「查得比壓之前快」（票上驗收 1）
|
||
|
||
🔴 `apply` **一定要 --ticket**（票上驗收 3：壓縮這件事本身要有票、有交付紀錄，
|
||
不准順手做完沒人知道）。它會把「哪一版壓掉了什麼」寫進 `.compress-log.md`。
|
||
"""
|
||
import argparse
|
||
import datetime as dt
|
||
import hashlib
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
THRESHOLD = int(os.environ.get("WIKI_COMPRESS_THRESHOLD") or 1200) # 行
|
||
LOG_NAME = ".compress-log.md"
|
||
FENCE = re.compile(r"^\s*(```|~~~)")
|
||
H2 = re.compile(r"^##\s+(?!#)(.+?)\s*$")
|
||
DATE = re.compile(r"(20\d\d)[-/](\d\d)[-/]?(\d\d)?")
|
||
TICKET = re.compile(r"\b([\w.-]+/[\w.-]+#\d+)\b")
|
||
|
||
|
||
# ══ 一、把一個 wiki 檔切成「條」═══════════════════════════════════════════
|
||
#
|
||
# 🔴 一定要跳過 code fence。`mistakes.md` 第 534 行是 ```` # 兩邊各跑一次 ````
|
||
# ——那是**程式碼註解**，不是標題。照抄 grep '^#' 會把它當成一條，
|
||
# 然後 verify 會拿一個不存在的東西去對帳，全盤失準。
|
||
def split_entries(text):
|
||
"""回 (前言, [條...])。每條 = {"title","body","line","date","tickets","hash"}。"""
|
||
lines = text.split("\n")
|
||
fenced = False
|
||
marks = []
|
||
for i, ln in enumerate(lines):
|
||
if FENCE.match(ln):
|
||
fenced = not fenced
|
||
continue
|
||
if fenced:
|
||
continue
|
||
m = H2.match(ln)
|
||
if m:
|
||
marks.append((i, m.group(1).strip()))
|
||
|
||
pre = "\n".join(lines[:marks[0][0]]) if marks else text
|
||
out = []
|
||
for n, (i, title) in enumerate(marks):
|
||
end = marks[n + 1][0] if n + 1 < len(marks) else len(lines)
|
||
body = "\n".join(lines[i:end]).rstrip()
|
||
out.append({
|
||
"title": title,
|
||
"body": body,
|
||
"line": i + 1,
|
||
"lines": end - i,
|
||
"date": entry_date(title, body),
|
||
"tickets": sorted(set(TICKET.findall(body))),
|
||
"hash": body_hash(body),
|
||
})
|
||
return pre, out
|
||
|
||
|
||
def body_hash(body):
|
||
"""對帳用的指紋：把空白正規化之後雜湊。
|
||
|
||
正規化空白（而不是逐字元比對）是為了讓「搬檔時尾端多/少一個換行」
|
||
這種**無意義的差異**不要被報成「弄丟了」——誤報會讓對帳表沒人看。
|
||
真的改了字，雜湊一定不同。
|
||
"""
|
||
return hashlib.sha256(re.sub(r"\s+", " ", body).strip().encode()).hexdigest()[:16]
|
||
|
||
|
||
def entry_date(title, body):
|
||
"""條的日期：標題裡的優先（那是人寫的），沒有才往內文找第一個。"""
|
||
for src in (title, body[:1500]):
|
||
m = DATE.search(src or "")
|
||
if m:
|
||
y, mo, d = m.group(1), m.group(2), m.group(3) or "01"
|
||
try:
|
||
return dt.date(int(y), int(mo), int(d)).isoformat()
|
||
except ValueError:
|
||
return "%s-%s-01" % (y, mo)
|
||
return ""
|
||
|
||
|
||
# ══ 二之〇、自承重複：**這一條自己說它是同一件事的第 N 次** ══════════════
|
||
#
|
||
# 為什麼另外做這一格（實測數字，不是設計偏好）：
|
||
# `mistakes.md` 260 條，用**標題相似度**分群只找得到 1 對——
|
||
# 因為同一個病每次都被寫成完全不同的句子（那正是它一犯再犯的原因）。
|
||
# 但其中 **68 條的內文自己寫著「同款第三次」「又犯一次」「同一個病」**。
|
||
# ⇒ 人下的判斷比機器算的相似度準得多。**照抄人已經寫下的結論，不要重新猜一遍。**
|
||
#
|
||
# 這 68 條就是「合併成一條、記發生過幾次」最該下手的地方，
|
||
# 但**合併要人來做**（要重寫正文，機器一動就會弄丟東西）⇒ 本支只把它們點名出來。
|
||
SELF_DUP = re.compile(r"同款|第[一二三四五六七八九十百0-9]+次|又犯|再犯|又一次|"
|
||
r"同一個病|同一形狀|同樣的錯|重蹈")
|
||
|
||
|
||
def self_declared_repeat(e):
|
||
"""回它自己講的那句話（沒有就回 None）。**引它的原話，不要自己下結論。**"""
|
||
for src in (e["title"], e["body"][:900]):
|
||
m = SELF_DUP.search(src or "")
|
||
if m:
|
||
i = max(0, m.start() - 12)
|
||
return re.sub(r"\s+", " ", (src[i:m.end() + 10]).strip())
|
||
return None
|
||
|
||
|
||
# ══ 二、同類分群（標 ×N，不合併）═════════════════════════════════════════
|
||
STOP = set("的了是在有和與及對於把被就都也還很更最一個這那我你他它們不沒要會能"
|
||
"MISTAKE mistake 方法 為什麼 怎麼".split())
|
||
|
||
|
||
def tokens(title):
|
||
t = re.sub(r"[（(].*?[)）]", " ", title)
|
||
t = re.sub(r"[^\w一-鿿]+", " ", t)
|
||
out = set()
|
||
for w in t.split():
|
||
if re.fullmatch(r"[A-Za-z0-9_.-]+", w):
|
||
if len(w) > 2 and w not in STOP:
|
||
out.add(w.lower())
|
||
else:
|
||
for i in range(len(w) - 1): # 中文用 bigram
|
||
bg = w[i:i + 2]
|
||
if bg not in STOP:
|
||
out.add(bg)
|
||
return out
|
||
|
||
|
||
def cluster(entries, thresh=0.42):
|
||
"""同一件事的第 N 次 ⇒ 分到同一群。回 [[索引...]]，單獨一條的也算一群。"""
|
||
toks = [tokens(e["title"]) for e in entries]
|
||
parent = list(range(len(entries)))
|
||
|
||
def find(x):
|
||
while parent[x] != x:
|
||
parent[x] = parent[parent[x]]
|
||
x = parent[x]
|
||
return x
|
||
|
||
for a in range(len(entries)):
|
||
for b in range(a + 1, len(entries)):
|
||
ta, tb = toks[a], toks[b]
|
||
if not ta or not tb:
|
||
continue
|
||
j = len(ta & tb) / float(len(ta | tb))
|
||
# 共用票號＝同一條線，門檻放寬（票號是人下的判斷，比字面相似可靠）
|
||
same_ticket = bool(set(entries[a]["tickets"]) & set(entries[b]["tickets"]))
|
||
if j >= thresh or (same_ticket and j >= thresh * 0.6):
|
||
parent[find(a)] = find(b)
|
||
groups = {}
|
||
for i in range(len(entries)):
|
||
groups.setdefault(find(i), []).append(i)
|
||
return sorted(groups.values(), key=lambda g: (-len(g), g[0]))
|
||
|
||
|
||
# ══ 三、audit ════════════════════════════════════════════════════════════
|
||
def iter_wiki_files(paths):
|
||
for p in paths:
|
||
if os.path.isdir(p):
|
||
for name in sorted(os.listdir(p)):
|
||
if name.endswith(".md") and not name.startswith("."):
|
||
yield os.path.join(p, name)
|
||
elif os.path.isfile(p):
|
||
yield p
|
||
|
||
|
||
def cmd_audit(args):
|
||
rows = []
|
||
for f in iter_wiki_files(args.paths):
|
||
text = open(f, encoding="utf-8", errors="replace").read()
|
||
_, entries = split_entries(text)
|
||
groups = [g for g in cluster(entries) if len(g) > 1] if entries else []
|
||
rows.append({
|
||
"file": f,
|
||
"lines": text.count("\n") + 1,
|
||
"entries": len(entries),
|
||
"dup_groups": len(groups),
|
||
"dup_entries": sum(len(g) for g in groups),
|
||
"self_dup": sum(1 for e in entries if self_declared_repeat(e)),
|
||
})
|
||
rows.sort(key=lambda r: -r["lines"])
|
||
over = [r for r in rows if r["lines"] > THRESHOLD]
|
||
|
||
print("📚 wiki 體檢（門檻 %d 行——超過就沒有人會從頭讀）" % THRESHOLD)
|
||
print()
|
||
for r in rows:
|
||
flag = "🔴 太長" if r["lines"] > THRESHOLD else " "
|
||
print("%s %-46s %6d 行 / %4d 條%s"
|
||
% (flag, os.path.basename(r["file"]), r["lines"], r["entries"],
|
||
"，%d 條自承是同款的第 N 次" % r["self_dup"] if r["self_dup"] else ""))
|
||
print()
|
||
if not over:
|
||
print("✅ 查過了，沒有超過門檻的檔。")
|
||
return 0
|
||
print("⇒ %d 個檔超過門檻。**這不是風格問題**——一個沒有人讀得完的必讀檔，"
|
||
"跟沒有那個檔的差別只在於它讓人以為有。" % len(over))
|
||
print("⇒ 下一步：`wiki-compress plan <檔>` 看怎麼壓；壓之前先開票（票上驗收 3）。")
|
||
return 1
|
||
|
||
|
||
# ══ 四、plan ═════════════════════════════════════════════════════════════
|
||
def plan(path, before=None, keep=None, threshold=None):
|
||
"""算出「哪幾條留下、哪幾條搬走」。
|
||
|
||
🔴 預設政策＝**搬到原檔掉到門檻以下為止**（最新的留下、最舊的先搬）。
|
||
|
||
為什麼不是「早於某個日期就搬」（第一版是那樣寫的，實測就被打臉）：
|
||
`mistakes.md` 260 條裡有 205 條是最近一個月的 ⇒ 用 30 天當線只搬得動 40 條，
|
||
壓完還是 6,000 行 ⇒ **一樣沒有人讀得完，等於白壓。**
|
||
而這件事的目的是「讓必讀檔可讀」，那就該讓**可讀**本身當判準。
|
||
"""
|
||
text = open(path, encoding="utf-8", errors="replace").read()
|
||
pre, entries = split_entries(text)
|
||
groups = cluster(entries)
|
||
limit = threshold or THRESHOLD
|
||
|
||
# 新舊排序：有日期的照日期，沒日期的照它在檔裡的位置（這批 wiki 都是新的在上面）
|
||
order = sorted(range(len(entries)),
|
||
key=lambda i: (entries[i]["date"] or "0000-00-00", -i),
|
||
reverse=True)
|
||
|
||
if before:
|
||
stay = {i for i in range(len(entries))
|
||
if not entries[i]["date"] or entries[i]["date"] >= before}
|
||
else:
|
||
n_keep = keep if keep is not None else None
|
||
if n_keep is None:
|
||
# 目錄一條一行 ＋ 檔頭固定幾行；一直加到再加就超過門檻為止
|
||
budget = limit - (len(entries) + 14)
|
||
n_keep, used = 0, 0
|
||
for i in order:
|
||
if used + entries[i]["lines"] > budget:
|
||
break
|
||
used += entries[i]["lines"]
|
||
n_keep += 1
|
||
stay = set(order[:max(0, n_keep)])
|
||
|
||
movable = [entries[i] for i in range(len(entries)) if i not in stay]
|
||
return {"text": text, "pre": pre, "entries": entries, "groups": groups,
|
||
"stay": stay, "movable": movable, "limit": limit,
|
||
"cutoff": before or "（門檻政策：壓到 %d 行以下）" % limit}
|
||
|
||
|
||
def cmd_plan(args):
|
||
p = plan(args.file, args.before, args.keep)
|
||
entries, groups = p["entries"], p["groups"]
|
||
dups = [g for g in groups if len(g) > 1]
|
||
selfs = [e for e in entries if self_declared_repeat(e)]
|
||
stay_lines = sum(entries[i]["lines"] for i in p["stay"])
|
||
print("📐 壓縮計畫：%s" % args.file)
|
||
print(" 現況：%d 行 / %d 條" % (p["text"].count("\n") + 1, len(entries)))
|
||
print(" 政策：%s" % p["cutoff"])
|
||
print(" 搬走：%d 條 → <檔名>-archive-YYYY-MM.md（**原封不動，一個字都不改**）"
|
||
% len(p["movable"]))
|
||
print(" 留下：%d 條、%d 行 ＋ 目錄 %d 行 ＝ 約 %d 行"
|
||
% (len(p["stay"]), stay_lines, len(entries) + 14,
|
||
stay_lines + len(entries) + 14))
|
||
print()
|
||
print(" ↻ **自承是同款第 N 次的有 %d 條**（占 %.0f%%）——"
|
||
% (len(selfs), 100.0 * len(selfs) / max(1, len(entries))))
|
||
print(" 這批是「合併成一條、記發生過幾次」最該下手的地方，")
|
||
print(" 但**合併要人來做**：機器一改正文就會弄丟東西，所以本支只點名。")
|
||
for e in selfs[:6]:
|
||
print(" · %s" % e["title"][:54])
|
||
print(" 它自己說：…%s…" % (self_declared_repeat(e) or "")[:52])
|
||
if len(selfs) > 6:
|
||
print(" · …另外還有 %d 條" % (len(selfs) - 6))
|
||
if dups:
|
||
print()
|
||
print(" ×N 標題就長得像的（%d 群）：" % len(dups))
|
||
for g in dups[:5]:
|
||
print(" ×%d %s" % (len(g), entries[g[0]]["title"][:52]))
|
||
print()
|
||
print("⇒ 真的要壓：wiki-compress apply %s --ticket owner/repo#N" % args.file)
|
||
return 0
|
||
|
||
|
||
# ══ 五、apply（只搬不改）═════════════════════════════════════════════════
|
||
def cmd_apply(args):
|
||
p = plan(args.file, args.before, args.keep)
|
||
entries, groups = p["entries"], p["groups"]
|
||
if not entries:
|
||
print("🔴 這個檔切不出任何一條（沒有 `## ` 標題），不動它。", file=sys.stderr)
|
||
return 2
|
||
|
||
where = {} # 條的索引 → 它搬去哪個檔（留下的是 None）
|
||
buckets = {}
|
||
stem, _ = os.path.splitext(args.file)
|
||
for idx, e in enumerate(entries):
|
||
if idx not in p["stay"]:
|
||
month = (e["date"] or "")[:7] or "undated"
|
||
target = "%s-archive-%s.md" % (stem, month)
|
||
buckets.setdefault(target, []).append(idx)
|
||
where[idx] = target
|
||
else:
|
||
where[idx] = None
|
||
|
||
gid = {}
|
||
for n, g in enumerate(groups):
|
||
for i in g:
|
||
gid[i] = n
|
||
|
||
# ── 目錄：每條一行。這一行就是票上說的「原位置留一份目錄」──────────
|
||
idx_lines = []
|
||
for n, g in enumerate(groups):
|
||
head = entries[g[0]]
|
||
mark = "×%d " % len(g) if len(g) > 1 else ""
|
||
for k, i in enumerate(g):
|
||
e = entries[i]
|
||
dest = where[i]
|
||
loc = "本檔" if dest is None else "→ `%s`" % os.path.basename(dest)
|
||
tick = (" " + "／".join(e["tickets"][:2])) if e["tickets"] else ""
|
||
rep = "↻ " if self_declared_repeat(e) else ""
|
||
pre = mark if k == 0 else " ↳ "
|
||
idx_lines.append("- %s%s`%s` %s（%s，%d 行）%s%s"
|
||
% (pre, rep, e["hash"], e["title"], e["date"] or "沒寫日期",
|
||
e["lines"], loc, tick))
|
||
del head
|
||
|
||
stamp = dt.datetime.now().strftime("%Y-%m-%d")
|
||
header = [
|
||
p["pre"].rstrip(),
|
||
"",
|
||
"---",
|
||
"",
|
||
"## 📇 目錄（%s 壓縮，票：%s）" % (stamp, args.ticket),
|
||
"",
|
||
"> 這份目錄是機器產生的（`scripts/wiki-compress`）。**每一條的正文一個字都沒改**——",
|
||
"> 舊的整條原封不動搬到 archive，留下的還在本檔。要對帳跑：",
|
||
"> `wiki-compress verify <壓縮前> <壓縮後> <archive...>`",
|
||
">",
|
||
"> `×N` ＝ 標題就長得像的同一群。`↻` ＝ **這一條自己寫著它是同款第 N 次**",
|
||
"> ——那是它自己下的判斷，不是機器猜的。要合併就從 `↻` 這批開始，**由人合併**。",
|
||
"",
|
||
] + idx_lines + ["", "---", ""]
|
||
|
||
kept = [entries[i]["body"] for i in range(len(entries)) if where[i] is None]
|
||
new_text = "\n".join(header) + "\n\n".join(kept) + "\n"
|
||
|
||
if args.dry_run:
|
||
print("🧪 --dry-run，什麼都沒寫。壓完會是：")
|
||
print(" %s：%d 行 → %d 行（目錄 %d 條）"
|
||
% (args.file, p["text"].count("\n") + 1, new_text.count("\n") + 1,
|
||
len(idx_lines)))
|
||
for t, ids in sorted(buckets.items()):
|
||
print(" %s：+%d 條" % (os.path.basename(t), len(ids)))
|
||
return 0
|
||
|
||
# ── 先把搬走的寫出去，再改原檔（順序不能反：中途死掉也不會弄丟）──────
|
||
for target, ids in sorted(buckets.items()):
|
||
chunk = ["# %s（自 %s 搬出，%s）" % (os.path.basename(target),
|
||
os.path.basename(args.file), stamp),
|
||
"",
|
||
"> 原封不動搬過來的，**一個字都沒改**。票：%s" % args.ticket,
|
||
"> 回原檔看目錄：`%s`" % os.path.basename(args.file),
|
||
""]
|
||
chunk += [entries[i]["body"] for i in ids]
|
||
old = ""
|
||
if os.path.exists(target):
|
||
old = open(target, encoding="utf-8").read().rstrip() + "\n\n"
|
||
with open(target, "w", encoding="utf-8") as f:
|
||
f.write(old + "\n\n".join(chunk) + "\n")
|
||
|
||
backup = args.file + ".before-compress"
|
||
with open(backup, "w", encoding="utf-8") as f:
|
||
f.write(p["text"])
|
||
with open(args.file, "w", encoding="utf-8") as f:
|
||
f.write(new_text)
|
||
|
||
# ── 交付紀錄（票上驗收 3：不准順手做完沒人知道）────────────────────
|
||
log = os.path.join(os.path.dirname(os.path.abspath(args.file)) or ".", LOG_NAME)
|
||
with open(log, "a", encoding="utf-8") as f:
|
||
f.write("\n## %s — %s（票：%s）\n\n" % (stamp, os.path.basename(args.file),
|
||
args.ticket))
|
||
f.write("- 壓縮前：%d 行 / %d 條\n" % (p["text"].count("\n") + 1, len(entries)))
|
||
f.write("- 壓縮後：%d 行（目錄 %d 條 ＋ 留下 %d 條）\n"
|
||
% (new_text.count("\n") + 1, len(idx_lines), len(kept)))
|
||
for t, ids in sorted(buckets.items()):
|
||
f.write("- 搬出 %d 條 → `%s`\n" % (len(ids), os.path.basename(t)))
|
||
f.write("- 對帳指令：`wiki-compress verify %s %s %s`\n"
|
||
% (os.path.basename(backup), os.path.basename(args.file),
|
||
" ".join(os.path.basename(t) for t in sorted(buckets))))
|
||
f.write("- 🔴 正文未改動（只搬不改）；逐條對帳結果見上面那條指令\n")
|
||
|
||
print("✅ 壓完了。%d 行 → %d 行"
|
||
% (p["text"].count("\n") + 1, new_text.count("\n") + 1))
|
||
for t, ids in sorted(buckets.items()):
|
||
print(" 搬出 %d 條 → %s" % (len(ids), t))
|
||
print(" 壓縮前的原檔留在：%s" % backup)
|
||
print(" 交付紀錄：%s" % log)
|
||
print()
|
||
print("⇒ 現在跑對帳（**沒跑過就不算壓完**）：")
|
||
print(" wiki-compress verify %s %s %s"
|
||
% (backup, args.file, " ".join(sorted(buckets))))
|
||
return 0
|
||
|
||
|
||
# ══ 六、verify（票上驗收 2：不能弄丟任何一條）════════════════════════════
|
||
def cmd_verify(args):
|
||
before = open(args.before, encoding="utf-8", errors="replace").read()
|
||
_, b_entries = split_entries(before)
|
||
|
||
after_files = [args.after] + list(args.also)
|
||
found = {}
|
||
for f in after_files:
|
||
if not os.path.exists(f):
|
||
print("🔴 找不到 %s" % f, file=sys.stderr)
|
||
return 2
|
||
_, es = split_entries(open(f, encoding="utf-8", errors="replace").read())
|
||
for e in es:
|
||
found.setdefault(e["hash"], []).append((f, e["title"]))
|
||
|
||
missing = [e for e in b_entries if e["hash"] not in found]
|
||
dupes = {h: v for h, v in found.items() if len(v) > 1}
|
||
|
||
print("🧾 逐條對帳：%s → %s" % (args.before, "＋".join(after_files)))
|
||
print(" 壓縮前 %d 條 ／ 壓縮後（含搬出去的）%d 條"
|
||
% (len(b_entries), sum(len(v) for v in found.values())))
|
||
print()
|
||
if missing:
|
||
print("🔴 **弄丟了 %d 條**（下面每一條在壓縮後找不到內文一模一樣的）：" % len(missing))
|
||
for e in missing[:40]:
|
||
print(" · 第 %d 行 `%s` %s" % (e["line"], e["hash"], e["title"][:60]))
|
||
if len(missing) > 40:
|
||
print(" · …另外還有 %d 條" % (len(missing) - 40))
|
||
print()
|
||
print("⇒ **這次壓縮不合格。** 票上第 2 條驗收就是這件事。")
|
||
print("⇒ 原檔還在 `%s.before-compress`，把它還原回去再查。" % args.after)
|
||
return 1
|
||
print("✅ **一條都沒弄丟。** %d 條全部在壓縮後的檔裡找得到內文一模一樣的。"
|
||
% len(b_entries))
|
||
if dupes:
|
||
print(" ⚠️ 有 %d 條同時出現在兩個檔（搬了但原檔沒清）：" % len(dupes))
|
||
for h, v in list(dupes.items())[:8]:
|
||
print(" · %s：%s" % (v[0][1][:40], "、".join(f for f, _ in v)))
|
||
return 0
|
||
|
||
|
||
# ══ 七、bench（票上驗收 1：查得比壓之前快）═══════════════════════════════
|
||
#
|
||
# 「快」要能量，不然它只是感覺。這裡量的是**定位成本**：
|
||
# 從檔頭讀到「確定答案在哪一條」為止，你要掃過幾行。
|
||
# 壓縮前：沒有目錄 ⇒ 只能一路掃到命中的那一行 ⇒ 成本 ＝ 命中行號
|
||
# 壓縮後：檔頭有目錄 ⇒ 在目錄裡命中就停 ⇒ 成本 ＝ 目錄裡的行號
|
||
# 同時量**命中率**——「查得到」是「查得快」的前提，快而查不到是更糟的結果。
|
||
def probes(entries, cap=80):
|
||
"""從每條的標題挑一個**在全檔獨一無二**的詞當查詢。挑不到就跳過那條。
|
||
|
||
🔴 **要在整份檔上均勻取樣**，不能只取前 cap 條。
|
||
第一版就是取前 80 條——而那批正好是最新的、壓縮後留在原檔最上面的，
|
||
**等於只考它最擅長的題目**，量出來的倍數是假的。
|
||
"""
|
||
titles = [e["title"] for e in entries]
|
||
step = max(1, len(entries) // cap)
|
||
sample = entries[::step]
|
||
out = []
|
||
for e in sample:
|
||
best = None
|
||
cands = re.findall(r"[一-鿿]{4,10}|[A-Za-z][\w.-]{4,}", e["title"])
|
||
for c in sorted(cands, key=len, reverse=True):
|
||
if sum(1 for t in titles if c in t) == 1:
|
||
best = c
|
||
break
|
||
if best:
|
||
out.append((best, e["hash"]))
|
||
if len(out) >= cap:
|
||
break
|
||
return out
|
||
|
||
|
||
def first_hit_line(path, needle):
|
||
with open(path, encoding="utf-8", errors="replace") as f:
|
||
for n, ln in enumerate(f, 1):
|
||
if needle in ln:
|
||
return n
|
||
return None
|
||
|
||
|
||
def cmd_bench(args):
|
||
before = open(args.before, encoding="utf-8", errors="replace").read()
|
||
_, b_entries = split_entries(before)
|
||
qs = probes(b_entries)
|
||
if not qs:
|
||
print("🔴 這個檔挑不出可以拿來查的詞（標題太相似）。", file=sys.stderr)
|
||
return 2
|
||
|
||
files = [args.after] + list(args.also)
|
||
hit_b = hit_a = 0
|
||
cost_b = cost_a = 0
|
||
worse = []
|
||
for q, _h in qs:
|
||
cb = first_hit_line(args.before, q)
|
||
ca = None
|
||
for f in files:
|
||
c = first_hit_line(f, q)
|
||
if c is not None and (ca is None or c < ca):
|
||
ca = c
|
||
if cb:
|
||
hit_b += 1; cost_b += cb
|
||
if ca:
|
||
hit_a += 1; cost_a += ca
|
||
if cb and ca and ca > cb:
|
||
worse.append((q, cb, ca))
|
||
|
||
n = len(qs)
|
||
print("⏱ 查得比壓之前快嗎？（%d 個查詢，每條各出一題）" % n)
|
||
print()
|
||
print(" 命中率 壓縮前 %d/%d → 壓縮後 %d/%d" % (hit_b, n, hit_a, n))
|
||
print(" 定位成本（要從檔頭掃過幾行才確定答案在哪）：")
|
||
print(" 壓縮前 平均 %.0f 行" % (cost_b / max(1, hit_b)))
|
||
print(" 壓縮後 平均 %.0f 行" % (cost_a / max(1, hit_a)))
|
||
if hit_a and hit_b:
|
||
ratio = (cost_b / hit_b) / max(1e-9, cost_a / hit_a)
|
||
print(" ⇒ **快 %.1f 倍**" % ratio)
|
||
print()
|
||
ok = True
|
||
if hit_a < hit_b:
|
||
print("🔴 **有 %d 個查詢壓縮後查不到了。** 查得快但查不到＝更糟。" % (hit_b - hit_a))
|
||
ok = False
|
||
if worse:
|
||
print(" ⚠️ 有 %d 個查詢變慢了（前 5 個）：" % len(worse))
|
||
for q, cb, ca in worse[:5]:
|
||
print(" · 「%s」 %d → %d 行" % (q, cb, ca))
|
||
if ok and hit_a == hit_b:
|
||
print("✅ **同一件事查得到，而且查得更快**（票上驗收 1）。")
|
||
return 0 if ok else 1
|
||
|
||
|
||
def main(argv):
|
||
ap = argparse.ArgumentParser(prog="wiki-compress", description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||
|
||
a = sub.add_parser("audit"); a.add_argument("paths", nargs="*", default=["."])
|
||
a.set_defaults(fn=cmd_audit)
|
||
|
||
p = sub.add_parser("plan"); p.add_argument("file")
|
||
p.add_argument("--before", help="早於這個日期的搬走（預設：壓到門檻以下為止）")
|
||
p.add_argument("--keep", type=int, help="最新的幾條留在原檔")
|
||
p.set_defaults(fn=cmd_plan)
|
||
|
||
ap2 = sub.add_parser("apply"); ap2.add_argument("file")
|
||
ap2.add_argument("--ticket", required=True,
|
||
help="owner/repo#N —— 壓縮這件事本身要有票（票上驗收 3）")
|
||
ap2.add_argument("--before")
|
||
ap2.add_argument("--keep", type=int)
|
||
ap2.add_argument("--dry-run", action="store_true")
|
||
ap2.set_defaults(fn=cmd_apply)
|
||
|
||
v = sub.add_parser("verify"); v.add_argument("before"); v.add_argument("after")
|
||
v.add_argument("also", nargs="*")
|
||
v.set_defaults(fn=cmd_verify)
|
||
|
||
b = sub.add_parser("bench"); b.add_argument("before"); b.add_argument("after")
|
||
b.add_argument("also", nargs="*")
|
||
b.set_defaults(fn=cmd_bench)
|
||
|
||
args = ap.parse_args(argv)
|
||
if args.cmd == "audit" and not args.paths:
|
||
args.paths = ["."]
|
||
if args.cmd == "apply" and not re.match(r"^[\w.-]+/[\w.-]+#\d+$", args.ticket):
|
||
print("🔴 --ticket 的寫法是 owner/repo#N（裸號跨 repo 會撞號）", file=sys.stderr)
|
||
return 2
|
||
return args.fn(args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:]))
|