diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..a27987f --- /dev/null +++ b/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 薄殼 diff --git a/tools/convert/.gitkeep b/tools/convert/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/tools/convert/convert.py b/tools/convert/convert.py new file mode 100644 index 0000000..953f91b --- /dev/null +++ b/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() diff --git a/tools/convert/from_docx.py b/tools/convert/from_docx.py new file mode 100644 index 0000000..c02ecb5 --- /dev/null +++ b/tools/convert/from_docx.py @@ -0,0 +1,96 @@ +"""Word (.docx) → Markdown。保留標題層級與表格結構,圖片抽出為附檔並留佔位引用。 + +可獨立執行:python tools/convert/from_docx.py [-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() diff --git a/tools/convert/from_pdf.py b/tools/convert/from_pdf.py new file mode 100644 index 0000000..c450ff0 --- /dev/null +++ b/tools/convert/from_pdf.py @@ -0,0 +1,70 @@ +"""PDF → Markdown。無文字層(掃描件)回報 needs_ocr 並跳過,不擅自呼叫 OCR。 + +可獨立執行:python tools/convert/from_pdf.py [-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"\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() diff --git a/tools/convert/from_pptx.py b/tools/convert/from_pptx.py new file mode 100644 index 0000000..28ac585 --- /dev/null +++ b/tools/convert/from_pptx.py @@ -0,0 +1,74 @@ +"""PowerPoint (.pptx) → Markdown。每張投影片一節,含表格、圖片佔位與講者備註。 + +可獨立執行:python tools/convert/from_pptx.py [-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() diff --git a/tools/convert/from_web.py b/tools/convert/from_web.py new file mode 100644 index 0000000..393da4f --- /dev/null +++ b/tools/convert/from_web.py @@ -0,0 +1,53 @@ +"""網頁 URL → 完整 HTML 快照 + Markdown 正文萃取(trafilatura,全程本地處理)。 + +可獨立執行:python tools/convert/from_web.py [-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() diff --git a/tools/convert/from_xlsx.py b/tools/convert/from_xlsx.py new file mode 100644 index 0000000..8a77dca --- /dev/null +++ b/tools/convert/from_xlsx.py @@ -0,0 +1,56 @@ +"""Excel (.xlsx) → Markdown。每個工作表一節,內容輸出為 Markdown 表格。 + +可獨立執行:python tools/convert/from_xlsx.py [-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() diff --git a/tools/convert/selfcheck.py b/tools/convert/selfcheck.py new file mode 100644 index 0000000..dfd1ce3 --- /dev/null +++ b/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( + "

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