Phase 2: 轉換管線 — 五個本地轉換器 + convert.py 統一入口 + selfcheck

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

requirements.txt

@@ -0,0 +1,9 @@
+# 全部為本地程式庫，無任何雲端轉換服務（AGENTS.md §1.1）
+pyyaml           # config/models.yaml
+python-docx      # Word → Markdown
+openpyxl         # Excel → Markdown
+pymupdf          # PDF → Markdown
+python-pptx     # PowerPoint → Markdown
+trafilatura      # 網頁快照 → Markdown 正文萃取
+rank-bm25        # tools/search.py BM25（第一版不用向量庫）
+fastmcp          # mcp/server.py 薄殼

tools/convert/convert.py

@@ -0,0 +1,188 @@
+"""統一轉換入口：依副檔名/URL 分派轉換器，登記 SHA-256 至 raw/manifest.json。
+
+流程（AGENTS.md §6.1、§9）：
+  原始檔存入 raw/originals/ 並登記 hash → 轉換 → Markdown 存入 raw/converted/
+  並登記 hash 與原始檔對應 → 之後攝入管線只讀 converted 層。
+
+用法：python tools/convert/convert.py <檔案路徑或URL>... [--dry-run] [--root DIR]
+"""
+import argparse
+import datetime
+import hashlib
+import json
+import pathlib
+import re
+import sys
+import urllib.parse
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import from_docx
+import from_pdf
+import from_pptx
+import from_web
+import from_xlsx
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+CONVERTERS = {
+    ".docx": (from_docx.convert, "from_docx", "docx"),
+    ".xlsx": (from_xlsx.convert, "from_xlsx", "xlsx"),
+    ".pdf": (from_pdf.convert, "from_pdf", "pdf"),
+    ".pptx": (from_pptx.convert, "from_pptx", "pptx"),
+}
+
+
+def _now():
+    return datetime.datetime.now().astimezone().isoformat(timespec="seconds")
+
+
+def _slug(s):
+    s = re.sub(r"[^a-z0-9一-鿿]+", "-", s.lower()).strip("-")
+    return s or "page"
+
+
+def load_manifest(root):
+    p = root / "raw" / "manifest.json"
+    m = json.loads(p.read_text(encoding="utf-8"))
+    if m.get("version") != 1 or "files" not in m:
+        raise SystemExit(f"manifest 結構不符：{p}")
+    return m
+
+
+def save_manifest(root, m):
+    p = root / "raw" / "manifest.json"
+    tmp = p.with_suffix(".json.tmp")
+    tmp.write_text(json.dumps(m, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    tmp.replace(p)  # 原子替換，避免寫壞 manifest
+
+
+def _find_by_sha(m, sha):
+    for path, e in m["files"].items():
+        if e["kind"] == "original" and e["sha256"] == sha:
+            return path
+    return None
+
+
+def _dest_name(dir_, name, sha):
+    """既有同名檔且內容不同時，附 hash 前 8 碼避免覆蓋（raw 不可變）。"""
+    p = dir_ / name
+    if p.exists() and hashlib.sha256(p.read_bytes()).hexdigest() != sha:
+        stem, ext = pathlib.Path(name).stem, pathlib.Path(name).suffix
+        return f"{stem}-{sha[:8]}{ext}"
+    return name
+
+
+def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dry):
+    if not dry:
+        md_path = root / md_rel
+        md_path.write_text(md_text, encoding="utf-8")
+        m["files"][orig_rel] = orig_entry
+        m["files"][md_rel] = {
+            "kind": "converted",
+            "sha256": hashlib.sha256(md_text.encode("utf-8")).hexdigest(),
+            "original_path": orig_rel,
+            "original_sha256": orig_entry["sha256"],
+            "converter": converter,
+            "converted_at": _now(),
+        }
+    print(f"converted: {orig_rel} -> {md_rel}" + (" [dry-run]" if dry else ""))
+
+
+def process_local(root, m, path, dry):
+    src = pathlib.Path(path).resolve()
+    if not src.is_file():
+        raise FileNotFoundError(src)
+    ext = src.suffix.lower()
+    if ext in (".html", ".htm"):
+        conv, conv_name, media = (
+            lambda p, assets_dir=None, md_dir=None: from_web.extract(
+                pathlib.Path(p).read_text(encoding="utf-8", errors="replace")),
+            "from_web", "html")
+    elif ext in CONVERTERS:
+        conv, conv_name, media = CONVERTERS[ext]
+    else:
+        raise ValueError(f"不支援的格式：{src.name}")
+
+    sha = hashlib.sha256(src.read_bytes()).hexdigest()
+    if _find_by_sha(m, sha):
+        print(f"skip: {src.name} 已登記（hash 相同）")
+        return
+
+    originals = root / "raw" / "originals"
+    if src.parent == originals.resolve():
+        dest = src
+    else:
+        dest = originals / _dest_name(originals, src.name, sha)
+        if not dry and not dest.exists():
+            dest.write_bytes(src.read_bytes())
+    orig_rel = f"raw/originals/{dest.name}"
+    orig_entry = {"kind": "original", "sha256": sha, "media_type": media,
+                  "added_at": _now(), "status": "converted"}
+
+    md_name = f"{dest.stem}-{sha[:8]}.md"
+    assets = None if dry else root / "raw" / "converted" / "assets" / f"{dest.stem}-{sha[:8]}"
+    md_text, status = conv(str(src), assets_dir=assets, md_dir=root / "raw" / "converted")
+    if status == "needs_ocr":
+        orig_entry["status"] = "needs_ocr"
+        if not dry:
+            m["files"][orig_rel] = orig_entry
+        print(f"needs_ocr: {orig_rel} 無文字層，已登記並跳過（不擅自 OCR）"
+              + (" [dry-run]" if dry else ""))
+        return
+    _register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}",
+                   md_text, conv_name, dry)
+
+
+def process_url(root, m, url, dry):
+    html = from_web.fetch(url)
+    sha = hashlib.sha256(html.encode("utf-8")).hexdigest()
+    if _find_by_sha(m, sha):
+        print(f"skip: {url} 已登記（快照 hash 相同）")
+        return
+    u = urllib.parse.urlparse(url)
+    slug = _slug(f"{u.netloc}{u.path}")[:80]
+    originals = root / "raw" / "originals"
+    name = _dest_name(originals, f"{slug}.html", sha)
+    orig_rel = f"raw/originals/{name}"
+    orig_entry = {"kind": "original", "sha256": sha, "media_type": "html",
+                  "added_at": _now(), "status": "converted",
+                  "source_url": url, "fetched_at": _now()}
+    if not dry:
+        (originals / name).write_text(html, encoding="utf-8")  # 完整快照（來源可能消失）
+    md_text, status = from_web.extract(html)
+    if status != "ok":
+        orig_entry["status"] = "pending"
+        if not dry:
+            m["files"][orig_rel] = orig_entry
+        print(f"pending: {url} 萃取不到正文，僅保存快照" + (" [dry-run]" if dry else ""))
+        return
+    md_name = f"{pathlib.Path(name).stem}-{sha[:8]}.md"
+    _register_pair(m, root, orig_rel, orig_entry, f"raw/converted/{md_name}",
+                   md_text, "from_web", dry)
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("inputs", nargs="+", help="檔案路徑或 http(s) URL")
+    ap.add_argument("--dry-run", action="store_true")
+    ap.add_argument("--root", default=str(ROOT), help="專案根目錄（測試用）")
+    a = ap.parse_args(argv)
+    root = pathlib.Path(a.root).resolve()
+    m = load_manifest(root)
+    failed = []
+    for item in a.inputs:
+        try:
+            if item.startswith(("http://", "https://")):
+                process_url(root, m, item, a.dry_run)
+            else:
+                process_local(root, m, item, a.dry_run)
+        except Exception as e:  # 單檔失敗不中斷批次，最後彙總報錯
+            failed.append((item, e))
+            print(f"error: {item}: {e}", file=sys.stderr)
+    if not a.dry_run:
+        save_manifest(root, m)
+    if failed:
+        raise SystemExit(1)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_docx.py

@@ -0,0 +1,96 @@
+"""Word (.docx) → Markdown。保留標題層級與表格結構，圖片抽出為附檔並留佔位引用。
+
+可獨立執行：python tools/convert/from_docx.py <file.docx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+import re
+
+import docx
+from docx.oxml.ns import qn
+from docx.table import Table
+from docx.text.paragraph import Paragraph
+
+
+def _heading_level(par):
+    # ponytail: 只認 "Heading N" / "標題 N" 樣式名；其他語系的 Word 樣式名不涵蓋，
+    # 升級路徑：改讀 w:outlineLvl。
+    m = re.match(r"(?:Heading|標題)\s*(\d)", par.style.name or "")
+    return int(m.group(1)) if m else None
+
+
+def _table_md(tbl):
+    rows = [
+        [c.text.strip().replace("\n", " ").replace("|", "\\|") for c in row.cells]
+        for row in tbl.rows
+    ]
+    if not rows:
+        return ""
+    width = max(len(r) for r in rows)
+    rows = [r + [""] * (width - len(r)) for r in rows]
+    out = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+    out += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+    return "\n".join(out)
+
+
+def _par_md(par, doc, assets_dir, md_dir, counter):
+    parts = []
+    for run in par.runs:
+        for blip in run._element.findall(".//" + qn("a:blip")):
+            rid = blip.get(qn("r:embed"))
+            if not rid:
+                continue
+            part = doc.part.related_parts[rid]
+            ext = pathlib.Path(part.partname).suffix or ".png"
+            counter[0] += 1
+            name = f"img{counter[0]}{ext}"
+            if assets_dir is not None:
+                assets_dir.mkdir(parents=True, exist_ok=True)
+                (assets_dir / name).write_bytes(part.blob)
+                rel = (assets_dir / name).relative_to(md_dir).as_posix()
+            else:
+                rel = name  # dry-run：僅佔位
+            parts.append(f"![{name}]({rel})")
+        parts.append(run.text)
+    text = "".join(parts).strip()
+    lvl = _heading_level(par)
+    if lvl and text:
+        return "#" * lvl + " " + text
+    return text
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'（docx 必有文字層）。"""
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    doc = docx.Document(str(src))
+    counter = [0]
+    blocks = []
+    for child in doc.element.body.iterchildren():
+        if child.tag == qn("w:p"):
+            blocks.append(_par_md(Paragraph(child, doc), doc, assets_dir, md_dir, counter))
+        elif child.tag == qn("w:tbl"):
+            blocks.append(_table_md(Table(child, doc)))
+    md = "\n\n".join(b for b in blocks if b)
+    return md, "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars, "
+              f"{md.count(chr(10)) + 1} lines")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, _ = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_pdf.py

@@ -0,0 +1,70 @@
+"""PDF → Markdown。無文字層（掃描件）回報 needs_ocr 並跳過，不擅自呼叫 OCR。
+
+可獨立執行：python tools/convert/from_pdf.py <file.pdf> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import fitz  # pymupdf
+
+# ponytail: 掃描件偵測用「平均每頁字元數 < 25」的 naive heuristic，
+# 混合型 PDF（部分頁掃描）會整份判為有文字層；升級路徑：逐頁判定 + 逐頁標記。
+_MIN_CHARS_PER_PAGE = 25
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status ∈ {'ok', 'needs_ocr'}。
+
+    ponytail: 表格輸出為純文字行（不重建表格結構），升級路徑：page.find_tables()。
+    """
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    doc = fitz.open(str(src))
+    pages = [page.get_text("text").strip() for page in doc]
+    total = sum(len(p) for p in pages)
+    if total < _MIN_CHARS_PER_PAGE * max(len(pages), 1):
+        doc.close()
+        return "", "needs_ocr"
+
+    counter = 0
+    blocks = []
+    for i, page in enumerate(doc):
+        blocks.append(f"<!-- page {i + 1} -->\n\n{pages[i]}")
+        if assets_dir is None:
+            continue
+        for img in page.get_images(full=True):
+            xref = img[0]
+            pix = fitz.Pixmap(doc, xref)
+            if pix.n > 4:  # CMYK 等 → 轉 RGB
+                pix = fitz.Pixmap(fitz.csRGB, pix)
+            counter += 1
+            name = f"img{counter}.png"
+            assets_dir.mkdir(parents=True, exist_ok=True)
+            pix.save(str(assets_dir / name))
+            rel = (assets_dir / name).relative_to(md_dir).as_posix()
+            blocks.append(f"![{name}]({rel})")
+    doc.close()
+    return "\n\n".join(blocks), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, status = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    if status == "needs_ocr":
+        print(f"needs_ocr: {a.src} 無文字層，已跳過（不擅自 OCR）")
+        raise SystemExit(2)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_pptx.py

@@ -0,0 +1,74 @@
+"""PowerPoint (.pptx) → Markdown。每張投影片一節，含表格、圖片佔位與講者備註。
+
+可獨立執行：python tools/convert/from_pptx.py <file.pptx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+from pptx import Presentation
+from pptx.enum.shapes import MSO_SHAPE_TYPE
+
+
+def _table_md(tbl):
+    rows = [
+        [c.text.strip().replace("|", "\\|").replace("\n", " ") for c in row.cells]
+        for row in tbl.rows
+    ]
+    if not rows:
+        return ""
+    width = len(rows[0])
+    out = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+    out += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+    return "\n".join(out)
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'。"""
+    src = pathlib.Path(src)
+    md_dir = pathlib.Path(md_dir) if md_dir else src.parent
+    prs = Presentation(str(src))
+    counter = 0
+    sections = []
+    for i, slide in enumerate(prs.slides, 1):
+        blocks = [f"## Slide {i}"]
+        for shape in slide.shapes:
+            if shape.has_text_frame and shape.text_frame.text.strip():
+                blocks.append(shape.text_frame.text.strip())
+            elif shape.has_table:
+                blocks.append(_table_md(shape.table))
+            elif shape.shape_type == MSO_SHAPE_TYPE.PICTURE:
+                counter += 1
+                name = f"img{counter}.{shape.image.ext}"
+                if assets_dir is not None:
+                    assets_dir.mkdir(parents=True, exist_ok=True)
+                    (assets_dir / name).write_bytes(shape.image.blob)
+                    rel = (assets_dir / name).relative_to(md_dir).as_posix()
+                else:
+                    rel = name
+                blocks.append(f"![{name}]({rel})")
+        if slide.has_notes_slide:
+            notes = slide.notes_slide.notes_text_frame.text.strip()
+            if notes:
+                blocks.append(f"### 講者備註\n\n{notes}")
+        sections.append("\n\n".join(blocks))
+    return "\n\n".join(sections), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    if a.dry_run:
+        md, status = convert(a.src)
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    md, _ = convert(a.src, assets_dir=out.parent / f"{out.stem}-assets", md_dir=out.parent)
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_web.py

@@ -0,0 +1,53 @@
+"""網頁 URL → 完整 HTML 快照 + Markdown 正文萃取（trafilatura，全程本地處理）。
+
+可獨立執行：python tools/convert/from_web.py <url> [-o out.md] [--snapshot out.html] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import trafilatura
+
+
+def fetch(url):
+    """抓取 URL，回傳完整 HTML 字串。失敗時拋 RuntimeError。"""
+    html = trafilatura.fetch_url(url)
+    if not html:
+        raise RuntimeError(f"抓取失敗：{url}")
+    return html
+
+
+def extract(html):
+    """HTML → Markdown 正文。回傳 (markdown, status)。萃取不到正文時 status='empty'。"""
+    md = trafilatura.extract(
+        html, output_format="markdown", include_tables=True, include_links=True
+    )
+    if not md or not md.strip():
+        return "", "empty"
+    return md.strip(), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("url")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--snapshot", help="HTML 快照輸出路徑")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    html = fetch(a.url)
+    md, status = extract(html)
+    if a.dry_run:
+        print(f"[dry-run] {a.url}: status={status}, snapshot {len(html)} chars, "
+              f"markdown {len(md)} chars")
+        return
+    if status != "ok":
+        print(f"empty: {a.url} 萃取不到正文")
+        raise SystemExit(2)
+    if a.snapshot:
+        pathlib.Path(a.snapshot).write_text(html, encoding="utf-8")
+    out = pathlib.Path(a.output) if a.output else pathlib.Path("page.md")
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/from_xlsx.py

@@ -0,0 +1,56 @@
+"""Excel (.xlsx) → Markdown。每個工作表一節，內容輸出為 Markdown 表格。
+
+可獨立執行：python tools/convert/from_xlsx.py <file.xlsx> [-o out.md] [--dry-run]
+"""
+import argparse
+import pathlib
+
+import openpyxl
+
+
+def _cell(v):
+    return "" if v is None else str(v).replace("|", "\\|").replace("\n", " ").strip()
+
+
+def convert(src, assets_dir=None, md_dir=None):
+    """回傳 (markdown, status)。status 恆為 'ok'。
+
+    ponytail: merged cell 只有左上角有值（openpyxl read_only 行為），不展開；
+    升級路徑：非 read_only 模式讀 merged_cells.ranges 補值。
+    """
+    wb = openpyxl.load_workbook(str(src), data_only=True, read_only=True)
+    sections = []
+    for ws in wb.worksheets:
+        rows = [[_cell(v) for v in row] for row in ws.iter_rows(values_only=True)]
+        rows = [r for r in rows if any(r)]  # 去除全空列
+        body = f"## 工作表：{ws.title}\n\n"
+        if not rows:
+            body += "（空白工作表）"
+        else:
+            width = max(len(r) for r in rows)
+            rows = [r + [""] * (width - len(r)) for r in rows]
+            lines = ["| " + " | ".join(rows[0]) + " |", "|" + " --- |" * width]
+            lines += ["| " + " | ".join(r) + " |" for r in rows[1:]]
+            body += "\n".join(lines)
+        sections.append(body)
+    wb.close()
+    return "\n\n".join(sections), "ok"
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(description=__doc__)
+    ap.add_argument("src")
+    ap.add_argument("-o", "--output")
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    md, status = convert(a.src)
+    if a.dry_run:
+        print(f"[dry-run] {a.src}: status={status}, {len(md)} chars")
+        return
+    out = pathlib.Path(a.output) if a.output else pathlib.Path(a.src).with_suffix(".md")
+    out.write_text(md, encoding="utf-8")
+    print(out)
+
+
+if __name__ == "__main__":
+    main()

tools/convert/selfcheck.py

@@ -0,0 +1,127 @@
+"""Phase 2 自檢：程式化產生各格式最小測試檔，驗證轉換器與 convert.py 端到端。
+
+執行：.venv/Scripts/python tools/convert/selfcheck.py
+邏輯壞掉時 assert 會失敗（AGENTS.md §13.4：一個可執行的檢查，不用框架）。
+"""
+import json
+import pathlib
+import shutil
+import sys
+import tempfile
+
+sys.path.insert(0, str(pathlib.Path(__file__).parent))
+import convert
+import from_docx
+import from_pdf
+import from_pptx
+import from_web
+import from_xlsx
+
+
+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 / "report.docx")
+
+    import openpyxl
+    wb = openpyxl.Workbook()
+    ws = wb.active
+    ws.title = "測項"
+    ws.append(["編號", "說明"])
+    ws.append(["TC-001", "逾時 | 重試"])
+    wb.save(d / "cases.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 / "text.pdf")
+    pdf.close()
+    scanned = fitz.open()
+    scanned.new_page()  # 無文字層 → 應判 needs_ocr
+    scanned.save(d / "scanned.pdf")
+    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 / "closing.pptx")
+
+    (d / "page.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 / "report.docx")
+        assert st == "ok" and "# 支付閘道測試報告" in md and "| TC-001 | FAIL |" in md, md
+        md, st = from_xlsx.convert(fx / "cases.xlsx")
+        assert st == "ok" and "## 工作表：測項" in md and "逾時 \\| 重試" in md, md
+        md, st = from_pdf.convert(fx / "text.pdf")
+        assert st == "ok" and "<!-- page 1 -->" in md and "timeout defect" in md, md
+        _, st = from_pdf.convert(fx / "scanned.pdf")
+        assert st == "needs_ocr", st
+        md, st = from_pptx.convert(fx / "closing.pptx")
+        assert st == "ok" and "## Slide 1" in md and "法遵項目全數通過" in md, md
+        md, st = from_web.extract((fx / "page.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 / "report.docx"), str(fx / "scanned.pdf"),
+                      str(fx / "page.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["raw/originals/scanned.pdf"]["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 / "raw/originals/report.docx").is_file()
+        print("PASS: convert.py 端到端 + manifest 對應")
+
+        # 冪等：同檔再跑一次 → skip，manifest 不變
+        before = (root / "raw/manifest.json").read_text(encoding="utf-8")
+        convert.main([str(fx / "report.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 / "report.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 不落地")
+
+        print("ALL PASS")
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()