Files
pp-qa-km/tools/diag_refs.py
yellowadmin b90727fc11 feat: diag_refs 對帳診斷 + ingest 進度可見性/Ctrl-C 安全 (Closes #2) (#5)
## 摘要

本 PR 含兩個獨立關注點,各一 commit。

### 1. `diag_refs` — source_ref↔manifest 對帳診斷 (`04f4fd5`) — 回應 #4

lint 的「source_ref 不在 manifest」只報對不上、不辨成因。新增 `tools/diag_refs.py`,把每個對不上的 source_ref 分成五類(hash 不符/路徑字串岔開/檔名 hash 後綴岔開/檔在磁碟未登記/完全無對應),各附修法,讓「補帳本(§14.3)」與「修規則(§12)」的相反處置不再混為一談。純唯讀、結束碼 `0/1`,可當搬機或改 manifest 後的閘門。附 selfcheck 驗五型別分類、結束碼與零寫檔。README §3.3/§6/§7/§8 同步。

### 2. ingest 進度可見性 + Ctrl-C 安全回復 (`4add7fa`) — Closes #2

`--all-pending` 長時間本地 Ollama 攝入全程靜默,使用者不確定有沒有在跑、又不敢中斷:

- **進度可見**:每份來源 `[i/N]`、長文件逐段 `分段 k/n`、每個 item `↳ 新增/更新`、完成 `✓`,全 `flush=True` 即時顯示。
- **Ctrl-C 安全**:獨立攔 `KeyboardInterrupt`(它不是 `Exception` 子類,原本會略過善後)。因 commit 在迴圈之後,**main 必然未受影響**;中斷時印可照做的回復指令並以結束碼 130 收場。

## 測試

- `python tools/selfcheck_diag_refs.py` — ALL PASS
- `python tools/selfcheck_ingest.py` — ALL PASS(含進度標記、中斷後 main 未動、回復指令實測可回乾淨 main)
- README markdownlint 0 issues

Closes #2

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-authored-by: LittleYellow <crazytea@gmail.com>
Reviewed-on: #5
2026-07-24 05:42:35 +00:00

135 lines
5.6 KiB
Python
Raw Permalink 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.
"""source_ref ↔ manifest 對帳診斷AGENTS.md §7 schema / §14 對帳的輔助工具)。
lint.py 對「source_ref 不在 manifest」只報一句無法區分成因本工具把每個
對不上的 source_ref 分類成可據以修復的型別(見 CAT_TITLE / FIX讓「帳本掉
條目」「路徑字串岔開」「檔名 hash 後綴岔開」「檔真的遺失」彼此不再混為一談。
純唯讀——不改任何檔(含 manifest有對不上以結束碼 1 表示(可當搬機/改
manifest 後的閘門)。分類邏輯與 lint 一致:以最後一個 '#' 切出 path 與 sha8。
用法python tools/diag_refs.py [--root DIR]
結束碼0 = 全部解得開1 = 有對不上。
"""
import argparse
import json
import pathlib
import re
import sys
sys.path.insert(0, str(pathlib.Path(__file__).parent))
import kb
CAT_TITLE = {
"hash-mismatch": "Hash 不符(帳本有此路徑,但 sha256 與 ref 不同)",
"normalize": "路徑字串岔開(正規化後與帳本 key 相同:斜線/前綴/大小寫)",
"stem": "檔名 hash 後綴岔開帳本有同檔名幹、hash 後綴不同的 key",
"unregistered": "檔在磁碟、帳本無此條目",
"absent": "完全無對應(帳本、磁碟、檔名幹皆無)",
}
FIX = {
"hash-mismatch": "raw 被就地改動或重轉(違反 §1.4);還原檔案,或連同 source_ref 走 PR§14.2)。",
"normalize": "修產生岔開字串的那一個環節(路徑正規化),別逐頁改 source_ref§12 change the ruler",
"stem": "source_ref 指到不同 hash 後綴的檔名核對正確檔名別改帳本去遷就§14.2)。",
"unregistered": "帳本掉了條目、檔還在;走 branch+PR 補登 manifest§14.3),別動 wiki 頁。",
"absent": "source_ref 可能攝入時寫歪,或原始檔真的遺失;上報 HITL 據實記錄§14.2)。",
}
# 分類輸出順序(愈可機械修復的排愈前)
ORDER = ["hash-mismatch", "normalize", "stem", "unregistered", "absent"]
def _norm(path):
"""正規化路徑字串以偵測「同一檔、不同寫法」:反斜線→斜線、去開頭 ./、轉小寫。"""
s = path.replace("\\", "/")
if s.startswith("./"):
s = s[2:]
return s.lower()
def _stem(path):
"""取檔名幹:去目錄、去 .md、去結尾的 -<8 碼 hex>convert 的檔名 hash 後綴)。"""
name = path.replace("\\", "/").rsplit("/", 1)[-1]
name = re.sub(r"\.md$", "", name, flags=re.I)
return re.sub(r"-[0-9a-f]{8}$", "", name, flags=re.I).lower()
def classify_ref(source_ref, files, exists):
"""把單一 source_ref 分類。filesmanifest['files']path→entry
exists(path)→bool 判磁碟是否有該檔。回傳 (category, detail)。
category 為 "resolved" 或 CAT_TITLE 的任一鍵。"""
path, _, sha8 = str(source_ref).rpartition("#")
entry = files.get(path)
if entry is not None:
if entry["sha256"].startswith(sha8):
return "resolved", path
return "hash-mismatch", f"帳本 sha256={entry['sha256'][:8]} ≠ ref #{sha8}"
norm = _norm(path)
for k in files:
if _norm(k) == norm:
return "normalize", f"帳本 key = '{k}'"
stem = _stem(path)
for k, e in files.items():
if e.get("kind") == "converted" and _stem(k) == stem:
return "stem", f"帳本 key = '{k}'"
if exists(path):
return "unregistered", f"磁碟有 '{path}',帳本無此 key"
return "absent", f"'{path}' 帳本無、磁碟無、無同名幹 key"
def collect(root, files, exists):
"""掃全庫 wiki 頁,回傳 (findings, n_refs)。findingscategory→[(page, ref, detail)]。"""
findings = {c: [] for c in ORDER}
n_refs = 0
for p in kb.iter_pages(root):
rel = p.relative_to(root).as_posix()
try:
meta, _ = kb.parse_page(p.read_text(encoding="utf-8"))
except Exception as e:
findings["absent"].append((rel, "(無法解析頁面)", str(e)))
continue
for s in meta.get("sources") or []:
n_refs += 1
cat, detail = classify_ref(s, files, exists)
if cat != "resolved":
findings[cat].append((rel, str(s), detail))
return findings, n_refs
def format_report(findings, n_refs):
total = sum(len(v) for v in findings.values())
lines = [f"檢查 source_ref{n_refs},對不上:{total}", ""]
if total == 0:
lines.append("全部解得開source_ref 與 manifest 一致)。")
return "\n".join(lines)
for cat in ORDER:
items = findings[cat]
if not items:
continue
lines.append(f"## {CAT_TITLE[cat]}{len(items)}")
lines.append(f"→ 修法:{FIX[cat]}")
for page, ref, detail in items:
lines.append(f" - [{page}] {ref}")
lines.append(f" {detail}")
lines.append("")
return "\n".join(lines)
def main(argv=None):
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--root", default=str(kb.ROOT))
a = ap.parse_args(argv)
root = pathlib.Path(a.root).resolve()
mpath = root / "raw" / "manifest.json"
try:
files = json.loads(mpath.read_text(encoding="utf-8"))["files"]
except (OSError, KeyError, json.JSONDecodeError) as e:
raise SystemExit(f"讀不到 / 解析不了 manifest{mpath}{e}")
findings, n_refs = collect(root, files, lambda p: (root / p).is_file())
print(format_report(findings, n_refs))
if sum(len(v) for v in findings.values()):
raise SystemExit(1)
if __name__ == "__main__":
main()