f9a0cd2369
- convert.py --verify：純唯讀 re-hash 全部登記檔並掃未登記檔， 報 missing/mismatch/unregistered，不一致退出碼 1（供刪除/竄改後稽核） - 修 write_text 在 Windows 轉 \n→\r\n 使磁碟 bytes 與登記 SHA-256 不符 （converted md 與 web 快照原始檔），改 write_bytes 寫入所登記的那份 bytes - selfcheck 補 clean/missing/mismatch/unregistered 四情境（全程唯讀斷言） - AGENTS.md §14 raw 完整性與修復流程 + §1.4 hash==磁碟 bytes 不變式 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
326 lines
14 KiB
Python
326 lines
14 KiB
Python
"""統一轉換入口：依副檔名/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>... [--vision] [--dry-run] [--root DIR]
|
||
支援 .docx/.xlsx/.pdf/.pptx/.html 與 URL；舊版 .doc/.xls/.ppt 經 LibreOffice 升版；
|
||
--vision 讓掃描/圖片型 PDF 走本地 vision 模型轉錄（見 from_vision.py）。
|
||
--verify 對帳模式：re-hash 全部登記檔、掃出未登記檔，純唯讀不改檔，供刪除/竄改後稽核。
|
||
"""
|
||
import argparse
|
||
import datetime
|
||
import hashlib
|
||
import json
|
||
import pathlib
|
||
import re
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import traceback
|
||
import urllib.parse
|
||
|
||
sys.path.insert(0, str(pathlib.Path(__file__).parent))
|
||
import from_docx
|
||
import from_legacy
|
||
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"),
|
||
}
|
||
# 舊版 OLE 二進位 → 現代格式（經 LibreOffice 升版後走上表對應轉換器）。
|
||
# <!-- ASSUMPTION: media_type 記真實舊格式（doc/xls/ppt），擴充 §9 的列舉；
|
||
# converter 記為 "from_docx+libreoffice" 等，保留兩段式 provenance。 -->
|
||
LEGACY_MAP = {".doc": ".docx", ".xls": ".xlsx", ".ppt": ".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 md_text.strip():
|
||
# 空轉換結果幾乎都是失敗（如純圖片 docx）；登記前示警，避免空頁靜默入庫
|
||
print(f"warning: {orig_rel} 轉換結果為空白，請人工檢查來源", file=sys.stderr)
|
||
if not dry:
|
||
md_path = root / md_rel
|
||
# write_bytes（非 write_text）：登記的 sha256 算在 md_text.encode() 上，
|
||
# 而 write_text 在 Windows 會把 \n 轉 \r\n，使磁碟 bytes 與登記 hash 不符，
|
||
# 令日後「還原後 re-hash 比對」與 --verify 失準。寫入即登記的那份 bytes。
|
||
md_path.write_bytes(md_text.encode("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 _vision_pdf(root, pdf_path, assets, conv_name):
|
||
"""--vision fallback（AGENTS.md §3 稽核原則）：needs_ocr 的 PDF 逐頁 render →
|
||
本地 vision 模型轉錄。回傳 (markdown, status, converter)。
|
||
|
||
成功 → ('...', 'ok', 'from_vision')；vision 執行期失敗 → 降級回 needs_ocr（不遺失檔案）。
|
||
設定錯誤（未設 vision / :cloud）由 from_vision 拋 SystemExit，向上傳播不靜默降級。
|
||
"""
|
||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
|
||
import kb
|
||
import to_image
|
||
import from_vision
|
||
cfg = kb.load_config(root)
|
||
md_dir = root / "raw" / "converted"
|
||
try:
|
||
images = to_image.render_pdf_pages(pdf_path, assets)
|
||
texts, model = from_vision.transcribe(cfg, images)
|
||
except SystemExit:
|
||
raise
|
||
except Exception as e:
|
||
print(f"warning: {pathlib.Path(pdf_path).name} vision 轉錄失敗，降級為 needs_ocr："
|
||
f"{type(e).__name__}: {e}", file=sys.stderr)
|
||
return "", "needs_ocr", conv_name
|
||
blocks = []
|
||
for i, (img, text) in enumerate(zip(images, texts), 1):
|
||
rel = img.relative_to(md_dir).as_posix()
|
||
# <> 包住：檔名含空格或不平衡括號時，未包住的連結會失效
|
||
blocks.append(f"<!-- page {i} (vision: {model}) -->\n\n\n\n{text}")
|
||
return "\n\n".join(blocks), "ok", "from_vision"
|
||
|
||
|
||
def verify(root, m):
|
||
"""對帳模式：re-hash 每個 manifest 條目對應的檔案，並掃出未登記的 raw 檔。
|
||
純唯讀——不改任何檔案（含 manifest），只回報。有不一致以 SystemExit(1) 表示，
|
||
可供 CI／排程當閘門用。
|
||
|
||
偵測三類意外：
|
||
missing —— 帳本有登記，磁碟上檔案不見了（手動刪除）。
|
||
mismatch —— 檔案還在但 sha256 與登記值不符（raw 層被就地改動，違反 §1.4）。
|
||
unregistered —— raw/originals 或 raw/converted 下有檔卻不在帳本；
|
||
converted/assets/* 由 markdown 連結而非帳本追蹤，故略過，.gitkeep 亦略過。
|
||
"""
|
||
missing, mismatch = [], []
|
||
for rel, e in m["files"].items():
|
||
p = root / rel
|
||
if not p.is_file():
|
||
missing.append(rel)
|
||
elif hashlib.sha256(p.read_bytes()).hexdigest() != e["sha256"]:
|
||
mismatch.append(rel)
|
||
|
||
unregistered = []
|
||
for sub in ("originals", "converted"):
|
||
base = root / "raw" / sub
|
||
if not base.is_dir():
|
||
continue
|
||
for p in base.rglob("*"):
|
||
if not p.is_file() or p.name == ".gitkeep":
|
||
continue
|
||
if "assets" in p.relative_to(base).parts: # 由 markdown 連結，不入帳本
|
||
continue
|
||
rel = p.relative_to(root).as_posix()
|
||
if rel not in m["files"]:
|
||
unregistered.append(rel)
|
||
|
||
for rel in missing:
|
||
print(f"missing: {rel}（帳本有登記，磁碟上不見了）", file=sys.stderr)
|
||
for rel in mismatch:
|
||
print(f"mismatch: {rel}（sha256 與登記值不符，raw 層被就地改動）", file=sys.stderr)
|
||
for rel in sorted(unregistered):
|
||
print(f"unregistered: {rel}（磁碟上有檔，帳本未登記）", file=sys.stderr)
|
||
|
||
if missing or mismatch or unregistered:
|
||
print(f"\nverify: {len(m['files'])} 筆登記 → "
|
||
f"{len(missing)} missing / {len(mismatch)} mismatch / "
|
||
f"{len(unregistered)} unregistered（純唯讀，未改任何檔）", file=sys.stderr)
|
||
raise SystemExit(1)
|
||
print(f"verify: {len(m['files'])} 筆登記全部對得上，raw/ 無未登記檔案")
|
||
|
||
|
||
def process_local(root, m, path, dry, vision=False):
|
||
src = pathlib.Path(path).resolve()
|
||
if not src.is_file():
|
||
raise FileNotFoundError(src)
|
||
ext = src.suffix.lower()
|
||
|
||
sha = hashlib.sha256(src.read_bytes()).hexdigest()
|
||
if _find_by_sha(m, sha):
|
||
print(f"skip: {src.name} 已登記（hash 相同）")
|
||
return
|
||
|
||
# raw/originals 一律保存「真正的原始檔」（舊格式也是），維持 provenance（§1.4）
|
||
originals = root / "raw" / "originals"
|
||
dest = src if src.parent == originals.resolve() else originals / _dest_name(originals, src.name, sha)
|
||
orig_rel = f"raw/originals/{dest.name}"
|
||
|
||
tmpdir = None
|
||
try:
|
||
if ext in LEGACY_MAP:
|
||
if dry:
|
||
print(f"legacy: {src.name} 需 LibreOffice 升版為 {LEGACY_MAP[ext]}（dry-run 不執行）")
|
||
return
|
||
tmpdir = tempfile.mkdtemp(prefix="ppqa-legacy-")
|
||
upgraded = from_legacy.upgrade(src, tmpdir) # 找不到 soffice / 升版失敗 → RuntimeError
|
||
conv, base_name, _ = CONVERTERS[upgraded.suffix.lower()]
|
||
conv_name, media, conv_src = f"{base_name}+libreoffice", ext.lstrip("."), upgraded
|
||
elif ext in (".html", ".htm"):
|
||
conv = (lambda p, assets_dir=None, md_dir=None: from_web.extract(
|
||
pathlib.Path(p).read_text(encoding="utf-8", errors="replace")))
|
||
conv_name, media, conv_src = "from_web", "html", src
|
||
elif ext in CONVERTERS:
|
||
conv, conv_name, media = CONVERTERS[ext]
|
||
conv_src = src
|
||
else:
|
||
raise ValueError(f"不支援的格式：{src.name}")
|
||
|
||
if not dry and dest is not src and not dest.exists():
|
||
dest.write_bytes(src.read_bytes())
|
||
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(conv_src), assets_dir=assets,
|
||
md_dir=root / "raw" / "converted")
|
||
if status == "needs_ocr":
|
||
if vision and not dry:
|
||
md_text, status, conv_name = _vision_pdf(root, conv_src, assets, conv_name)
|
||
if status == "needs_ocr":
|
||
orig_entry["status"] = "needs_ocr"
|
||
if not dry:
|
||
m["files"][orig_rel] = orig_entry
|
||
print(f"needs_ocr: {orig_rel} 無文字層，已登記並跳過"
|
||
+ ("（--vision 轉錄未成功）" if vision and not dry else "（不擅自 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)
|
||
finally:
|
||
if tmpdir:
|
||
shutil.rmtree(tmpdir, ignore_errors=True)
|
||
|
||
|
||
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:
|
||
# write_bytes：同 §_register_pair，登記 hash 算在 html.encode() 上，
|
||
# 避免 Windows 換行轉換讓快照 bytes 與登記 hash 不符（完整快照，來源可能消失）
|
||
(originals / name).write_bytes(html.encode("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("--verify", action="store_true",
|
||
help="對帳模式：re-hash 全部登記檔、掃出未登記檔，純唯讀不改檔；"
|
||
"有不一致以退出碼 1 表示（不吃 inputs）")
|
||
ap.add_argument("--dry-run", action="store_true")
|
||
ap.add_argument("--vision", action="store_true",
|
||
help="掃描/圖片型 PDF（needs_ocr）改走本地 vision 模型轉錄，"
|
||
"而非跳過（需 config/models.yaml 設定 tasks.vision）")
|
||
ap.add_argument("--traceback", action="store_true",
|
||
help="轉換失敗時額外印出完整堆疊，供定位除錯")
|
||
ap.add_argument("--root", default=str(ROOT), help="專案根目錄（測試用）")
|
||
a = ap.parse_args(argv)
|
||
root = pathlib.Path(a.root).resolve()
|
||
m = load_manifest(root)
|
||
if a.verify:
|
||
verify(root, m)
|
||
return
|
||
if not a.inputs:
|
||
ap.error("需指定至少一個檔案/URL，或改用 --verify 對帳")
|
||
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, vision=a.vision)
|
||
except Exception as e: # 單檔失敗不中斷批次，最後彙總報錯
|
||
failed.append((item, e))
|
||
# 帶例外型別：只印訊息字串常無法判斷成因（如 KeyError 只印 key）
|
||
print(f"error: {item}: {type(e).__name__}: {e}", file=sys.stderr)
|
||
if a.traceback:
|
||
traceback.print_exc()
|
||
if not a.dry_run:
|
||
save_manifest(root, m)
|
||
if failed:
|
||
# 尾端彙總，避免失敗行淹沒在大批次輸出中而漏看
|
||
print(f"\n{len(failed)} 個項目轉換失敗：", file=sys.stderr)
|
||
for item, e in failed:
|
||
print(f" - {item}: {type(e).__name__}: {e}", file=sys.stderr)
|
||
if not a.traceback:
|
||
print("（加 --traceback 可印完整堆疊定位）", file=sys.stderr)
|
||
raise SystemExit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|