"""統一轉換入口：依副檔名/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()