轉換工具強化：失敗診斷訊息 + 舊版格式升版 + 掃描 PDF vision fallback
- convert.py：轉換失敗改印例外型別、批次尾端彙總、--traceback 旗標； 空白轉換結果示警（避免純圖片 docx 等假成功靜默入庫） - from_legacy.py（B）：.doc/.xls/.ppt 經本地 LibreOffice headless 升版為 現代格式後再走既有轉換器；raw/originals 仍保存真正的舊格式原始檔 - to_image.py + from_vision.py（A）：needs_ocr 的掃描/圖片型 PDF 逐頁 render → 本地 vision 模型轉錄（--vision）；:cloud/未設定防護， 模型設定走 config/models.yaml 的 tasks.vision（預設不啟用） - kb.ollama_chat 加 images 參數（向後相容）；models.yaml 加 vision 範例 - selfcheck 補 legacy 路由、to_image 真測、from_vision 注入測、 --vision 端到端，全數 ALL PASS Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
LittleYellow
@@ -4,7 +4,9 @@
|
||||
原始檔存入 raw/originals/ 並登記 hash → 轉換 → Markdown 存入 raw/converted/
|
||||
並登記 hash 與原始檔對應 → 之後攝入管線只讀 converted 層。
|
||||
|
||||
用法：python tools/convert/convert.py <檔案路徑或URL>... [--dry-run] [--root DIR]
|
||||
用法：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）。
|
||||
"""
|
||||
import argparse
|
||||
import datetime
|
||||
@@ -12,11 +14,15 @@ 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
|
||||
@@ -29,6 +35,10 @@ CONVERTERS = {
|
||||
".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():
|
||||
@@ -72,6 +82,9 @@ def _dest_name(dir_, name, sha):
|
||||
|
||||
|
||||
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
|
||||
md_path.write_text(md_text, encoding="utf-8")
|
||||
@@ -87,49 +100,96 @@ def _register_pair(m, root, orig_rel, orig_entry, md_rel, md_text, converter, dr
|
||||
print(f"converted: {orig_rel} -> {md_rel}" + (" [dry-run]" if dry else ""))
|
||||
|
||||
|
||||
def process_local(root, m, path, dry):
|
||||
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 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()
|
||||
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
|
||||
|
||||
# raw/originals 一律保存「真正的原始檔」（舊格式也是），維持 provenance（§1.4）
|
||||
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())
|
||||
dest = src if src.parent == originals.resolve() else originals / _dest_name(originals, src.name, sha)
|
||||
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)
|
||||
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):
|
||||
@@ -164,6 +224,11 @@ 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("--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()
|
||||
@@ -174,13 +239,22 @@ def main(argv=None):
|
||||
if item.startswith(("http://", "https://")):
|
||||
process_url(root, m, item, a.dry_run)
|
||||
else:
|
||||
process_local(root, m, item, a.dry_run)
|
||||
process_local(root, m, item, a.dry_run, vision=a.vision)
|
||||
except Exception as e: # 單檔失敗不中斷批次，最後彙總報錯
|
||||
failed.append((item, e))
|
||||
print(f"error: {item}: {e}", file=sys.stderr)
|
||||
# 帶例外型別：只印訊息字串常無法判斷成因（如 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)
|
||||
|
||||
|
||||
|
||||
91
tools/convert/from_legacy.py
Normal file
91
tools/convert/from_legacy.py
Normal file
@@ -0,0 +1,91 @@
|
||||
"""舊版 OLE 二進位格式（.doc/.xls/.ppt）→ 現代格式（.docx/.xlsx/.pptx）升版。
|
||||
|
||||
以 LibreOffice headless 在本機升版（deterministic、內容不外流，AGENTS.md §1.1），
|
||||
升版後由 convert.py 回頭走既有 from_docx / from_xlsx / from_pptx 轉換器。
|
||||
升版產物是暫時中介檔，不進 raw/；raw/originals 仍保存真正的舊格式原始檔（§1.4）。
|
||||
|
||||
可獨立執行：python tools/convert/from_legacy.py <file.doc> [--out DIR]
|
||||
"""
|
||||
import argparse
|
||||
import os
|
||||
import pathlib
|
||||
import shutil
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
|
||||
LEGACY_TARGET = {".doc": "docx", ".xls": "xlsx", ".ppt": "pptx"}
|
||||
|
||||
# 常見 Windows 安裝路徑；其他平台靠 PATH（soffice/libreoffice）或 SOFFICE_BIN 覆寫。
|
||||
_WIN_CANDIDATES = (
|
||||
r"C:\Program Files\LibreOffice\program\soffice.exe",
|
||||
r"C:\Program Files (x86)\LibreOffice\program\soffice.exe",
|
||||
)
|
||||
|
||||
|
||||
def find_soffice():
|
||||
"""定位 LibreOffice 執行檔。SOFFICE_BIN 環境變數優先（唯一允許的覆寫來源）。"""
|
||||
env = os.environ.get("SOFFICE_BIN")
|
||||
if env and pathlib.Path(env).is_file():
|
||||
return env
|
||||
for c in _WIN_CANDIDATES:
|
||||
if pathlib.Path(c).is_file():
|
||||
return c
|
||||
return shutil.which("soffice") or shutil.which("libreoffice")
|
||||
|
||||
|
||||
def upgrade(src, out_dir, timeout=180):
|
||||
"""升版 src 至 out_dir，回傳升版後檔案路徑。
|
||||
|
||||
找不到 LibreOffice、格式不支援、或升版未產出檔案時拋 RuntimeError（由
|
||||
convert.py 逐檔捕捉並回報，不中斷批次）。
|
||||
"""
|
||||
src = pathlib.Path(src)
|
||||
ext = src.suffix.lower()
|
||||
if ext not in LEGACY_TARGET:
|
||||
raise RuntimeError(f"非舊版格式：{src.name}")
|
||||
soffice = find_soffice()
|
||||
if not soffice:
|
||||
raise RuntimeError(
|
||||
"找不到 LibreOffice（soffice）；.doc/.xls/.ppt 升版需要它。"
|
||||
"請安裝 LibreOffice，或以 SOFFICE_BIN 環境變數指定執行檔路徑。")
|
||||
|
||||
out_dir = pathlib.Path(out_dir)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
target = LEGACY_TARGET[ext]
|
||||
# 每次呼叫用獨立 user profile，避免 headless 並行/殘留 profile 造成的鎖定失敗
|
||||
# （LibreOffice headless 的已知痛點，§13.3：真實環境不是規格書上的理想值）。
|
||||
profile = out_dir / ".lo-profile"
|
||||
try:
|
||||
subprocess.run(
|
||||
[soffice, f"-env:UserInstallation=file:///{profile.as_posix().lstrip('/')}",
|
||||
"--headless", "--convert-to", target, "--outdir", str(out_dir), str(src)],
|
||||
check=True, capture_output=True, text=True, timeout=timeout)
|
||||
except subprocess.CalledProcessError as e:
|
||||
raise RuntimeError(f"LibreOffice 升版失敗（exit {e.returncode}）：{e.stderr.strip()}")
|
||||
except subprocess.TimeoutExpired:
|
||||
raise RuntimeError(f"LibreOffice 升版逾時（>{timeout}s）：{src.name}")
|
||||
finally:
|
||||
shutil.rmtree(profile, ignore_errors=True)
|
||||
|
||||
out = out_dir / f"{src.stem}.{target}"
|
||||
if not out.is_file():
|
||||
raise RuntimeError(f"LibreOffice 未產出預期檔案：{out.name}")
|
||||
return out
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("src")
|
||||
ap.add_argument("--out", help="升版產物輸出目錄（預設暫存目錄）")
|
||||
a = ap.parse_args(argv)
|
||||
out_dir = pathlib.Path(a.out) if a.out else pathlib.Path(tempfile.mkdtemp(prefix="ppqa-legacy-"))
|
||||
try:
|
||||
print(upgrade(a.src, out_dir))
|
||||
except RuntimeError as e:
|
||||
print(f"error: {e}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
61
tools/convert/from_vision.py
Normal file
61
tools/convert/from_vision.py
Normal file
@@ -0,0 +1,61 @@
|
||||
"""影像 → Markdown 轉錄（本地 vision 模型，AGENTS.md §1.1 資料落地）。
|
||||
|
||||
用於 vision fallback：文字抽取失敗的頁面 render 成影像後，逐頁交本地 vision
|
||||
模型忠實轉錄為 Markdown。模型設定讀 config/models.yaml 的 tasks.vision（§1.2），
|
||||
未設定即報錯、不使用預設值；:cloud 模型一律拒絕（影像不得離開本機）。
|
||||
"""
|
||||
import base64
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parents[1])) # tools/ 供 import kb
|
||||
import kb
|
||||
|
||||
VISION_PROMPT = (
|
||||
"你是文件轉錄引擎。忠實轉錄此頁影像中的所有文字與表格內容為 Markdown："
|
||||
"表格用 Markdown table，保留標題階層。只輸出轉錄內容本身，不要加任何說明或臆測。"
|
||||
"影像中沒有可辨識內容時，只回覆「（本頁無可辨識文字）」。")
|
||||
|
||||
|
||||
def _require_vision(cfg):
|
||||
"""取得並驗證 vision 模型設定。缺設定 / :cloud 皆報錯（信任邊界，§1.1/§1.2）。"""
|
||||
spec = (cfg.get("tasks") or {}).get("vision")
|
||||
if not isinstance(spec, dict) or not spec.get("model") or spec["model"] == "CHANGE_ME":
|
||||
raise SystemExit(
|
||||
"models.yaml 未設定 tasks.vision.model；--vision 需要合規本地 vision 模型"
|
||||
"（請自行確認非中國關聯、非 :cloud，AGENTS.md §1.1/§1.3）。")
|
||||
if str(spec["model"]).endswith(":cloud"):
|
||||
raise SystemExit("tasks.vision.model 為 :cloud 模型，禁止（文件影像不得離開本機，§1.1）。")
|
||||
return spec
|
||||
|
||||
|
||||
def _b64(path):
|
||||
return base64.b64encode(pathlib.Path(path).read_bytes()).decode("ascii")
|
||||
|
||||
|
||||
def transcribe(cfg, image_paths):
|
||||
"""逐頁轉錄影像，回傳 (每頁 markdown 列表, 實際使用的 model)。
|
||||
|
||||
單頁重試 max_retries 次仍失敗即拋 RuntimeError（vision 無文字 fallback 可用——
|
||||
fallback 模型非 vision，不切換）。
|
||||
"""
|
||||
spec = _require_vision(cfg)
|
||||
model = spec["model"]
|
||||
retries = int(spec.get("max_retries", 2))
|
||||
temperature = float(spec.get("temperature", 0.1))
|
||||
pages = []
|
||||
for idx, img in enumerate(image_paths, 1):
|
||||
b64 = _b64(img)
|
||||
last = None
|
||||
for _ in range(max(1, retries)):
|
||||
try:
|
||||
text = kb.ollama_chat(cfg, model, VISION_PROMPT, temperature, images=[b64])
|
||||
if text and text.strip():
|
||||
pages.append(text.strip())
|
||||
break
|
||||
last = "空回應"
|
||||
except Exception as e:
|
||||
last = f"{type(e).__name__}: {e}"
|
||||
else:
|
||||
raise RuntimeError(f"vision 轉錄失敗（第 {idx} 頁，model={model}）：{last}")
|
||||
return pages, model
|
||||
@@ -3,19 +3,25 @@
|
||||
執行：.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__).parent))
|
||||
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
|
||||
|
||||
|
||||
def make_fixtures(d):
|
||||
@@ -118,6 +124,104 @@ def main():
|
||||
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 / "notes.txt").write_text("不支援的格式", encoding="utf-8") # → ValueError
|
||||
err = io.StringIO()
|
||||
code = None
|
||||
with contextlib.redirect_stderr(err):
|
||||
try:
|
||||
convert.main([str(fx / "report.docx"), str(fx / "notes.txt"),
|
||||
str(fx / "missing.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 / "raw/originals/report.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 / "old.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 / "old.doc"), "--root", str(root4)])
|
||||
except SystemExit as e:
|
||||
code = e.code
|
||||
log = err.getvalue()
|
||||
assert code == 1 and "old.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 / "text.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 / "scanned.pdf"), "--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)
|
||||
|
||||
48
tools/convert/to_image.py
Normal file
48
tools/convert/to_image.py
Normal file
@@ -0,0 +1,48 @@
|
||||
"""PDF 頁 → PNG 影像（pymupdf，deterministic、全程本地）。
|
||||
|
||||
供 vision fallback 使用：文字抽取失敗（needs_ocr）的掃描/圖片型 PDF 逐頁 render
|
||||
成影像後，交給 from_vision.py 以本地 vision 模型判讀。render 本身不含任何 LLM，
|
||||
維持「deterministic 工具做結構、LLM 只做語意」的分工（AGENTS.md §3 稽核原則）。
|
||||
|
||||
可獨立執行：python tools/convert/to_image.py <file.pdf> --out DIR [--dpi 150]
|
||||
"""
|
||||
import argparse
|
||||
import pathlib
|
||||
|
||||
import fitz # pymupdf
|
||||
|
||||
# ponytail: 預設 150 DPI 是「可讀性 vs 影像大小/推論成本」的折衷；掃描件文字偏小時
|
||||
# 可調高。升級路徑：依頁面尺寸/內容自適應 DPI。
|
||||
DEFAULT_DPI = 150
|
||||
|
||||
|
||||
def render_pdf_pages(src, out_dir, dpi=DEFAULT_DPI, prefix="page"):
|
||||
"""將 src 每頁 render 成 PNG，回傳影像路徑列表（依頁序）。"""
|
||||
src = pathlib.Path(src)
|
||||
out_dir = pathlib.Path(out_dir)
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
paths = []
|
||||
doc = fitz.open(str(src))
|
||||
try:
|
||||
for i, page in enumerate(doc, 1):
|
||||
pix = page.get_pixmap(dpi=dpi)
|
||||
p = out_dir / f"{prefix}{i}.png"
|
||||
pix.save(str(p))
|
||||
paths.append(p)
|
||||
finally:
|
||||
doc.close()
|
||||
return paths
|
||||
|
||||
|
||||
def main(argv=None):
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("src")
|
||||
ap.add_argument("--out", required=True, help="影像輸出目錄")
|
||||
ap.add_argument("--dpi", type=int, default=DEFAULT_DPI)
|
||||
a = ap.parse_args(argv)
|
||||
paths = render_pdf_pages(a.src, a.out, a.dpi)
|
||||
print(f"{len(paths)} 頁 → {a.out}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user