yellowadmin
356ae9367c
## 目的 `convert` 後檔案被手動刪除／就地改動時，提供正規的稽核與修復途徑；並修掉一個讓「還原後 re-hash 比對」在 Windows 失效的既有 bug。 ## 變更 - **`convert.py --verify`（純唯讀對帳）**：re-hash 全部登記檔、掃未登記檔，回報 missing / mismatch / unregistered，不一致以退出碼 1 表示（可當 CI／排程閘門）。`converted/assets/*` 與 `.gitkeep` 正確略過。 - **修 CRLF 使 hash 失準**：`write_text` 在 Windows 把 `\n`→`\r\n`，但登記的 SHA-256 算在 `\n` bytes 上 → 磁碟 bytes 與帳本永遠對不上（converted md + web 快照原始檔）。改為 `write_bytes` 寫入所登記的那份 bytes。 - **`.gitattributes`（`raw/** -text`）**：`core.autocrlf=true` 下 checkout 會在 git 層重新引入 CRLF，抵銷上一項修復；關閉 raw 的換行正規化，把「登記 hash == 磁碟 bytes」不變式延伸到 git checkin/checkout。 - **selfcheck**：補 clean / missing / mismatch / unregistered 四情境（全程唯讀斷言；clean 案例含未入帳本的 assets，順帶證明不誤報）。 - **AGENTS.md**：新增 §14「raw 完整性與修復（對帳）」，並於 §1.4 補「登記 hash == 磁碟 bytes」不變式與 `.gitattributes` 機制。 ## 驗證 `.venv/Scripts/python.exe tools/convert/selfcheck.py` → `ALL PASS`。 ## 備註 `--verify` 是唯讀稽核工具，不改任何檔（含 manifest），符合 §7「lint 絕不擅自刪改」精神。修復決策（可衍生 vs 信任根、先 `git restore` 不改帳本、動 manifest 走 PR）詳見 AGENTS.md §14。 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: LittleYellow <crazytea@gmail.com> Reviewed-on: #1
316 lines
15 KiB
Python
316 lines
15 KiB
Python
"""Phase 2 自檢：程式化產生各格式最小測試檔，驗證轉換器與 convert.py 端到端。
|
||
|
||
執行：.venv/Scripts/python tools/convert/selfcheck.py
|
||
邏輯壞掉時 assert 會失敗（AGENTS.md §13.4：一個可執行的檢查，不用框架）。
|
||
"""
|
||
import contextlib
|
||
import io
|
||
import json
|
||
import pathlib
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
|
||
sys.path.insert(0, str(pathlib.Path(__file__).parent)) # tools/convert
|
||
import convert
|
||
import from_docx
|
||
import from_pdf
|
||
import from_pptx
|
||
import from_vision
|
||
import from_web
|
||
import from_xlsx
|
||
import kb
|
||
import to_image
|
||
|
||
|
||
# 測資檔名一律用「中文 + 空格」——真實來源檔就長這樣，而 git status --porcelain
|
||
# 對這兩者都會加引號轉義。ASCII 檔名的測資曾讓 ingest 的 preflight bug 溜過（§12）。
|
||
DOCX, XLSX, PDF, SCAN = ("壓測 報告(1).docx", "測試案例 清單.xlsx",
|
||
"規格 說明.pdf", "掃描件(2 期.pdf")
|
||
PPTX, HTML, TXT, DOC = ("結案 簡報.pptx", "指引 快照.html",
|
||
"不支援 筆記.txt", "舊版 報告.doc")
|
||
|
||
|
||
def assert_links_ok(md, base):
|
||
"""每個圖片連結都要真的 render 成 <img>、且目標檔存在。
|
||
|
||
只斷言 "![" 在不在會讓損毀連結矇混過關——含空格/不平衡括號的檔名會產出
|
||
語法有效但解析失敗的連結，圖檔明明在磁碟上卻整行退化成純文字（§13.4）。
|
||
"""
|
||
import re
|
||
import urllib.parse
|
||
from markdown_it import MarkdownIt
|
||
|
||
want = md.count("![")
|
||
html = MarkdownIt("commonmark").render(md)
|
||
got = html.count("<img")
|
||
assert got == want, f"{want} 個圖片連結只有 {got} 個 render 成功：\n{md[:300]}"
|
||
for m in re.finditer(r'<img src="([^"]+)"', html):
|
||
p = base / urllib.parse.unquote(m.group(1))
|
||
assert p.is_file(), f"連結目標不存在：{p}"
|
||
return want
|
||
|
||
|
||
def make_fixtures(d):
|
||
import fitz
|
||
png = d / "img.png" # 內嵌圖片用：docx/pdf/pptx 三條抽圖路徑都要走到
|
||
pix = fitz.Pixmap(fitz.csRGB, fitz.IRect(0, 0, 8, 8))
|
||
pix.clear_with(128)
|
||
pix.save(str(png))
|
||
|
||
import docx as docxlib
|
||
doc = docxlib.Document()
|
||
doc.add_heading("支付閘道測試報告", level=1)
|
||
doc.add_paragraph("逾時缺陷於壓力測試重現。")
|
||
t = doc.add_table(rows=2, cols=2)
|
||
t.rows[0].cells[0].text = "案例"
|
||
t.rows[0].cells[1].text = "結果"
|
||
t.rows[1].cells[0].text = "TC-001"
|
||
t.rows[1].cells[1].text = "FAIL"
|
||
doc.add_picture(str(png))
|
||
doc.save(d / DOCX)
|
||
|
||
import openpyxl
|
||
wb = openpyxl.Workbook()
|
||
ws = wb.active
|
||
ws.title = "測項"
|
||
ws.append(["編號", "說明"])
|
||
ws.append(["TC-001", "逾時 | 重試"])
|
||
wb.save(d / XLSX)
|
||
|
||
pdf = fitz.open()
|
||
page = pdf.new_page()
|
||
page.insert_text((72, 72), "Payment gateway timeout defect reproduced under load test.")
|
||
page.insert_image(fitz.Rect(72, 100, 122, 150), filename=str(png))
|
||
pdf.save(d / PDF)
|
||
pdf.close()
|
||
scanned = fitz.open()
|
||
scanned.new_page() # 無文字層 → 應判 needs_ocr
|
||
scanned.save(d / SCAN)
|
||
scanned.close()
|
||
|
||
from pptx import Presentation
|
||
from pptx.util import Inches
|
||
prs = Presentation()
|
||
slide = prs.slides.add_slide(prs.slide_layouts[1])
|
||
slide.shapes.title.text = "結案報告"
|
||
slide.shapes.add_picture(str(png), Inches(1), Inches(3))
|
||
slide.notes_slide.notes_text_frame.text = "備註：法遵項目全數通過"
|
||
prs.save(d / PPTX)
|
||
|
||
(d / HTML).write_text(
|
||
"<html><body><article><h1>AML 測試指引</h1>"
|
||
+ "<p>可疑交易監控測試需涵蓋大額交易與分散交易兩種樣態。</p>" * 8
|
||
+ "</article></body></html>", encoding="utf-8")
|
||
|
||
|
||
def main():
|
||
tmp = pathlib.Path(tempfile.mkdtemp(prefix="ppqa-selfcheck-"))
|
||
try:
|
||
fx = tmp / "fx"
|
||
fx.mkdir()
|
||
make_fixtures(fx)
|
||
|
||
md, st = from_docx.convert(fx / DOCX)
|
||
assert st == "ok" and "# 支付閘道測試報告" in md and "| TC-001 | FAIL |" in md, md
|
||
md, st = from_xlsx.convert(fx / XLSX)
|
||
assert st == "ok" and "## 工作表：測項" in md and "逾時 \\| 重試" in md, md
|
||
md, st = from_pdf.convert(fx / PDF)
|
||
assert st == "ok" and "<!-- page 1 -->" in md and "timeout defect" in md, md
|
||
_, st = from_pdf.convert(fx / SCAN)
|
||
assert st == "needs_ocr", st
|
||
md, st = from_pptx.convert(fx / PPTX)
|
||
assert st == "ok" and "## Slide 1" in md and "法遵項目全數通過" in md, md
|
||
md, st = from_web.extract((fx / HTML).read_text(encoding="utf-8"))
|
||
assert st == "ok" and "可疑交易監控" in md, md
|
||
print("PASS: 5 個轉換器 + needs_ocr 偵測")
|
||
|
||
# convert.py 端到端（沙盒 root）
|
||
root = tmp / "root"
|
||
for sub in ("raw/originals", "raw/converted"):
|
||
(root / sub).mkdir(parents=True)
|
||
(root / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
|
||
|
||
convert.main([str(fx / DOCX), str(fx / XLSX), str(fx / PDF), str(fx / SCAN),
|
||
str(fx / PPTX), str(fx / HTML), "--root", str(root)])
|
||
m = json.loads((root / "raw/manifest.json").read_text(encoding="utf-8"))
|
||
origs = {k: v for k, v in m["files"].items() if v["kind"] == "original"}
|
||
convs = {k: v for k, v in m["files"].items() if v["kind"] == "converted"}
|
||
assert len(origs) == 6 and len(convs) == 5, m # SCAN 為 needs_ocr，無 converted
|
||
assert origs[f"raw/originals/{SCAN}"]["status"] == "needs_ocr"
|
||
for k, v in convs.items():
|
||
assert (root / k).is_file(), k
|
||
assert m["files"][v["original_path"]]["sha256"] == v["original_sha256"]
|
||
assert (root / f"raw/originals/{DOCX}").is_file()
|
||
print("PASS: convert.py 端到端 + manifest 對應")
|
||
|
||
# 圖片連結必須真的能 render——DOCX/PDF/PPTX 三條抽圖路徑都要驗到，
|
||
# 且測資檔名刻意含空格與不平衡括號（最容易讓連結失效的形狀）
|
||
linked = sum(assert_links_ok((root / k).read_text(encoding="utf-8"),
|
||
root / "raw/converted") for k in convs)
|
||
assert linked >= 3, f"應至少驗到 docx/pdf/pptx 各一個圖片連結，實際 {linked}"
|
||
print(f"PASS: 圖片連結可 render 且目標存在（{linked} 個，檔名含空格/括號）")
|
||
|
||
# 冪等：同檔再跑一次 → skip，manifest 不變
|
||
before = (root / "raw/manifest.json").read_text(encoding="utf-8")
|
||
convert.main([str(fx / DOCX), "--root", str(root)])
|
||
assert (root / "raw/manifest.json").read_text(encoding="utf-8") == before
|
||
print("PASS: 冪等（同 hash 跳過）")
|
||
|
||
# --verify 對帳（純唯讀）：clean 通過、三類意外都抓到、且完全不改 manifest。
|
||
# root 已有 docx/pdf/pptx 抽出的 assets（不入帳本）——clean 案例即證明 assets 不被誤報。
|
||
def run_verify(rt):
|
||
mm = json.loads((rt / "raw/manifest.json").read_text(encoding="utf-8"))
|
||
err, code = io.StringIO(), 0
|
||
with contextlib.redirect_stderr(err), contextlib.redirect_stdout(io.StringIO()):
|
||
try:
|
||
convert.verify(rt, mm)
|
||
except SystemExit as e:
|
||
code = e.code
|
||
return code, err.getvalue()
|
||
|
||
before = (root / "raw/manifest.json").read_text(encoding="utf-8")
|
||
code, _ = run_verify(root)
|
||
assert code == 0, "clean 帳本應通過對帳（含未入帳本的 assets）"
|
||
|
||
rroot = tmp / "root_del" # missing：刪掉一個 converted .md
|
||
shutil.copytree(root, rroot)
|
||
gone = next(p for p in (rroot / "raw/converted").glob("*.md"))
|
||
gone.unlink()
|
||
code, log = run_verify(rroot)
|
||
assert code == 1 and "missing" in log and gone.name in log, log
|
||
|
||
rroot = tmp / "root_mut" # mismatch：就地改動一個 original 的 bytes
|
||
shutil.copytree(root, rroot)
|
||
tampered = next(p for p in (rroot / "raw/originals").iterdir() if p.is_file())
|
||
tampered.write_bytes(tampered.read_bytes() + b"tampered")
|
||
code, log = run_verify(rroot)
|
||
assert code == 1 and "mismatch" in log, log
|
||
|
||
rroot = tmp / "root_unreg" # unregistered：raw/originals 塞入未登記檔
|
||
shutil.copytree(root, rroot)
|
||
(rroot / "raw/originals/未登記 檔.docx").write_bytes(b"stray")
|
||
code, log = run_verify(rroot)
|
||
assert code == 1 and "unregistered" in log and "未登記 檔.docx" in log, log
|
||
|
||
assert (root / "raw/manifest.json").read_text(encoding="utf-8") == before # 全程唯讀
|
||
print("PASS: --verify 對帳（clean/missing/mismatch/unregistered，純唯讀）")
|
||
|
||
# dry-run：全新沙盒不落地
|
||
root2 = tmp / "root2"
|
||
for sub in ("raw/originals", "raw/converted"):
|
||
(root2 / sub).mkdir(parents=True)
|
||
(root2 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
|
||
convert.main([str(fx / DOCX), "--dry-run", "--root", str(root2)])
|
||
m2 = json.loads((root2 / "raw/manifest.json").read_text(encoding="utf-8"))
|
||
assert m2["files"] == {} and not list((root2 / "raw/originals").iterdir())
|
||
print("PASS: --dry-run 不落地")
|
||
|
||
# 失敗回報：壞輸入不中斷批次，錯誤帶例外型別 + 尾端彙總，好檔仍轉換
|
||
root3 = tmp / "root3"
|
||
for sub in ("raw/originals", "raw/converted"):
|
||
(root3 / sub).mkdir(parents=True)
|
||
(root3 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
|
||
(fx / TXT).write_text("不支援的格式", encoding="utf-8") # → ValueError
|
||
err = io.StringIO()
|
||
code = None
|
||
with contextlib.redirect_stderr(err):
|
||
try:
|
||
convert.main([str(fx / DOCX), str(fx / TXT),
|
||
str(fx / "不存在 檔案.docx"), # 不存在 → FileNotFoundError
|
||
"--root", str(root3)])
|
||
except SystemExit as e:
|
||
code = e.code
|
||
log = err.getvalue()
|
||
assert code == 1, code
|
||
assert "ValueError" in log and "FileNotFoundError" in log, log # 錯誤帶例外型別
|
||
assert "2 個項目轉換失敗" in log, log # 尾端彙總
|
||
m3 = json.loads((root3 / "raw/manifest.json").read_text(encoding="utf-8"))
|
||
assert (root3 / f"raw/originals/{DOCX}").is_file() # 好檔仍轉換並落地
|
||
assert any(v["kind"] == "converted" for v in m3["files"].values()), m3
|
||
print("PASS: 失敗回報（例外型別 + 尾端彙總，批次不中斷）")
|
||
|
||
# B: 舊格式路由 —— .doc 應走 legacy（LibreOffice），而非被當「不支援格式」拒收
|
||
root4 = tmp / "root4"
|
||
for sub in ("raw/originals", "raw/converted"):
|
||
(root4 / sub).mkdir(parents=True)
|
||
(root4 / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
|
||
(fx / DOC).write_bytes(b"\xd0\xcf\x11\xe0legacy-stub") # 非有效 .doc，僅測路由/錯誤
|
||
err = io.StringIO()
|
||
code = None
|
||
with contextlib.redirect_stderr(err):
|
||
try:
|
||
convert.main([str(fx / DOC), "--root", str(root4)])
|
||
except SystemExit as e:
|
||
code = e.code
|
||
log = err.getvalue()
|
||
assert code == 1 and DOC in log, log
|
||
assert "不支援的格式" not in log and "RuntimeError" in log, log # 已路由 legacy，乾淨報錯
|
||
print("PASS: 舊格式路由至 legacy（soffice 缺失/升版失敗皆乾淨報錯）")
|
||
|
||
# A: to_image render（deterministic，真跑）
|
||
imgs = to_image.render_pdf_pages(fx / PDF, tmp / "imgs")
|
||
assert imgs and all(p.is_file() and p.stat().st_size > 0 for p in imgs), imgs
|
||
print(f"PASS: to_image render（{len(imgs)} 頁 PNG）")
|
||
|
||
# A: from_vision 轉錄邏輯（注入假 Ollama，不需真模型）+ 合規防護
|
||
vis_cfg = {"tasks": {"vision": {"model": "fake-vision", "temperature": 0.1,
|
||
"max_retries": 1}}}
|
||
orig_chat = kb.ollama_chat
|
||
calls = {}
|
||
|
||
def fake_chat(cfg, model, prompt, temperature, json_format=False, images=None):
|
||
calls["images"] = images # 記錄 base64 是否傳入
|
||
return "可疑交易監控：大額 12,000 元"
|
||
|
||
kb.ollama_chat = fake_chat
|
||
try:
|
||
texts, model = from_vision.transcribe(vis_cfg, imgs[:1])
|
||
finally:
|
||
kb.ollama_chat = orig_chat
|
||
assert model == "fake-vision" and texts and "可疑交易監控" in texts[0], texts
|
||
assert calls.get("images"), "vision 呼叫必須帶影像"
|
||
for bad in ({"tasks": {}}, {"tasks": {"vision": {"model": "foo:cloud"}}}):
|
||
try:
|
||
from_vision._require_vision(bad)
|
||
assert False, "應拒絕未設定 / :cloud"
|
||
except SystemExit:
|
||
pass
|
||
print("PASS: from_vision 轉錄 + :cloud/未設定防護")
|
||
|
||
# A: convert.py --vision 端到端（沙盒 config + 注入假 vision）
|
||
root_v = tmp / "root_v"
|
||
(root_v / "config").mkdir(parents=True)
|
||
for sub in ("raw/originals", "raw/converted"):
|
||
(root_v / sub).mkdir(parents=True)
|
||
(root_v / "config/models.yaml").write_text(
|
||
'ollama:\n base_url: "http://localhost:11434"\n timeout_seconds: 300\n'
|
||
'tasks:\n ingest: {model: "x", temperature: 0.2, max_retries: 1}\n'
|
||
' lint: {model: "x", temperature: 0.1, max_retries: 1}\n'
|
||
' fallback: {model: "x"}\n'
|
||
' vision: {model: "fake-vision", temperature: 0.1, max_retries: 1}\n'
|
||
'limits:\n mcp_response_max_tokens: 2000\n ingest_chunk_max_chars: 12000\n',
|
||
encoding="utf-8")
|
||
(root_v / "raw/manifest.json").write_text('{"version": 1, "files": {}}', encoding="utf-8")
|
||
kb.ollama_chat = fake_chat
|
||
try:
|
||
convert.main([str(fx / SCAN), "--vision", "--root", str(root_v)])
|
||
finally:
|
||
kb.ollama_chat = orig_chat
|
||
mv = json.loads((root_v / "raw/manifest.json").read_text(encoding="utf-8"))
|
||
conv_rel = [k for k, v in mv["files"].items() if v["kind"] == "converted"]
|
||
assert len(conv_rel) == 1 and mv["files"][conv_rel[0]]["converter"] == "from_vision", mv
|
||
md_v = (root_v / conv_rel[0]).read_text(encoding="utf-8")
|
||
assert "可疑交易監控" in md_v and "(vision" in md_v and "![page 1]" in md_v, md_v
|
||
assert_links_ok(md_v, root_v / "raw/converted") # vision 頁的連結同樣要能 render
|
||
print("PASS: convert.py --vision 端到端（needs_ocr → 本地 vision 轉錄）")
|
||
|
||
print("ALL PASS")
|
||
finally:
|
||
shutil.rmtree(tmp, ignore_errors=True)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|