"""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 = ("壓測報告.docx", "測試案例清單.xlsx", "規格說明.pdf", "掃描件無文字層.pdf") PPTX, HTML, TXT, DOC = ("結案簡報.pptx", "指引快照.html", "不支援筆記.txt", "舊版報告.doc") def make_fixtures(d): 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.save(d / DOCX) import openpyxl wb = openpyxl.Workbook() ws = wb.active ws.title = "測項" ws.append(["編號", "說明"]) ws.append(["TC-001", "逾時 | 重試"]) wb.save(d / XLSX) import fitz pdf = fitz.open() page = pdf.new_page() page.insert_text((72, 72), "Payment gateway timeout defect reproduced under load test.") pdf.save(d / PDF) pdf.close() scanned = fitz.open() scanned.new_page() # 無文字層 → 應判 needs_ocr scanned.save(d / SCAN) scanned.close() from pptx import Presentation prs = Presentation() slide = prs.slides.add_slide(prs.slide_layouts[1]) slide.shapes.title.text = "結案報告" slide.notes_slide.notes_text_frame.text = "備註：法遵項目全數通過" prs.save(d / PPTX) (d / HTML).write_text( "

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 / SCAN), 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) == 3 and len(convs) == 2, m 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 對應") # 冪等：同檔再跑一次 → 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 跳過）") # 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 print("PASS: convert.py --vision 端到端（needs_ocr → 本地 vision 轉錄）") print("ALL PASS") finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": main()