"""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 成、且目標檔存在。只斷言 "![" 在不在會讓損毀連結矇混過關——含空格/不平衡括號的檔名會產出語法有效但解析失敗的連結，圖檔明明在磁碟上卻整行退化成純文字（§13.4）。 """ import re import urllib.parse from markdown_it import MarkdownIt want = md.count("![") html = MarkdownIt("commonmark").render(md) got = html.count("

AML 測試指引

" + "

可疑交易監控測試需涵蓋大額交易與分散交易兩種樣態。

" * 8 + "

", 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 "" 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()