249 lines
9.8 KiB
Python
249 lines
9.8 KiB
Python
"""Probe: 方舟"文档理解"能不能直读扫描件 PDF(路线 A 验证,不动线上代码)。
|
|
|
|
跑法: .venv/Scripts/python.exe scripts/probe_ark_doc.py [真实PDF路径]
|
|
依赖 .env 里 ARK_API_KEY。**会真调豆包 seed-2.0-lite,产生 < ¥0.05 费用**。
|
|
|
|
验证点:
|
|
1. 合成一份 3 页"扫描件"PDF(PIL 纯图页,无文本层,每页埋魔术串 + 中文段落/表格)
|
|
2. markitdown 对它抽不出文字(证明现状确实是死路)
|
|
3. chat/completions 的 file 内容块用哪种 JSON 形状能被接受(候选格式挨个试,
|
|
400 报错会带字段名,本身就是探针产出)
|
|
4. OCR 保真度:三页魔术串是否全命中(页覆盖)、中文/表格内容是否读出
|
|
5. usage tokens → 每页成本口径
|
|
|
|
传真实 PDF 路径时跳过合成,直接测该文件(页数/体积上限、真实扫描件质量)。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
# Windows 控制台默认 GBK,打印中文/特殊符号会崩 → 强制 stdout UTF-8
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8", errors="replace") # type: ignore[attr-defined]
|
|
except Exception:
|
|
pass
|
|
|
|
# 读 .env(同 smoke_look_at_image)
|
|
env_file = ROOT / ".env"
|
|
if env_file.exists():
|
|
for line in env_file.read_text(encoding="utf-8").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, _, v = line.partition("=")
|
|
os.environ.setdefault(k.strip(), v.strip())
|
|
|
|
from PIL import Image, ImageDraw, ImageFont
|
|
|
|
from core.ark_client import ArkClient, ArkConfig, ArkError
|
|
|
|
def _magic(i: int) -> str:
|
|
"""第 i 页(0 起)的魔术串,确定性可复算。"""
|
|
return f"ZCBOT-DOC-P{i + 1}-{7391 + i * 613}"
|
|
|
|
# 每页正文:标题 + 段落 + 第 2 页一个小表格,模拟真实扫描标准的版面
|
|
PAGE_LINES = [
|
|
[
|
|
"水泥胶砂强度检验方法(模拟扫描件 第1页)",
|
|
"本方法规定了水泥胶砂抗压强度与抗折强度的测定步骤。",
|
|
"试验室温度应保持在 20 ± 2 摄氏度,相对湿度不低于 50%。",
|
|
"校验码: {magic}",
|
|
],
|
|
[
|
|
"第2页 配合比与龄期",
|
|
"胶砂配比 水泥 450g 标准砂 1350g 水 225g",
|
|
"龄期(d) 3 7 28",
|
|
"抗压(MPa) 22.5 35.0 52.5",
|
|
"校验码: {magic}",
|
|
],
|
|
[
|
|
"第3页 结果处理",
|
|
"以三条试体抗折结果的算术平均值作为试验结果。",
|
|
"当三个值中有超出平均值 ±10% 时,应剔除后重新计算。",
|
|
"校验码: {magic}",
|
|
],
|
|
]
|
|
|
|
_FONT_CANDIDATES = [
|
|
r"C:\Windows\Fonts\msyh.ttc",
|
|
r"C:\Windows\Fonts\simhei.ttf",
|
|
r"C:\Windows\Fonts\simsun.ttc",
|
|
]
|
|
|
|
|
|
def _load_font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont:
|
|
for p in _FONT_CANDIDATES:
|
|
if Path(p).exists():
|
|
return ImageFont.truetype(p, size)
|
|
return ImageFont.load_default()
|
|
|
|
|
|
def make_scanned_pdf(dest: Path, n_pages: int = 3) -> None:
|
|
"""N 页纯图 PDF(A4 150dpi,无文本层)—— 模拟扫描件。
|
|
|
|
前 3 页用真实版面(段落 + 表格),之后的页是短页(标题 + 魔术串),
|
|
专测页数上限 / 页覆盖。1-bit 模式控体积(黑字白底扫描件本就近似双色)。
|
|
"""
|
|
font = _load_font(36)
|
|
pages = []
|
|
for i in range(n_pages):
|
|
img = Image.new("RGB", (1240, 1754), (255, 255, 255))
|
|
d = ImageDraw.Draw(img)
|
|
lines = (PAGE_LINES[i] if i < len(PAGE_LINES)
|
|
else [f"第{i + 1}页 附录条款", "本页为附录占位内容。", "校验码: {magic}"])
|
|
y = 120
|
|
for line in lines:
|
|
d.text((100, y), line.format(magic=_magic(i)), fill=(0, 0, 0), font=font)
|
|
y += 90
|
|
# 1-bit 模式绕开本环境 PIL 缺 JPEG 编码器的问题(RGB 页会走 DCT/JPEG),
|
|
# 且体积最小(P 模式实测一页 4MB+,多页测试撑爆请求)
|
|
pages.append(img.convert("1", dither=Image.Dither.NONE))
|
|
# resolution=150:页物理尺寸=px/150 英寸 ≈ A4。缺省 72dpi 会把页标成 A4 两倍大,
|
|
# 方舟按固定 dpi 栅格化 PDF 页,超尺寸页撞"单页 3600 万像素"上限(probe 实测报错)
|
|
pages[0].save(dest, save_all=True, append_images=pages[1:], resolution=150.0)
|
|
|
|
|
|
def check_markitdown_dead_end(pdf: Path) -> None:
|
|
"""现状对照:markitdown 对纯图 PDF 应抽不出正文。"""
|
|
exe = ROOT / ".venv" / "Scripts" / "markitdown.exe"
|
|
cmd = [str(exe) if exe.exists() else "markitdown", str(pdf)]
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=120,
|
|
encoding="utf-8", errors="replace")
|
|
text = (r.stdout or "").strip()
|
|
print(f"[markitdown] exit={r.returncode} 抽出正文 {len(text)} 字符"
|
|
+ (f" → 非空?! 前 200 字: {text[:200]!r}" if text else " → 空(证实扫描件死路)"))
|
|
except Exception as e:
|
|
print(f"[markitdown] 跑不了({type(e).__name__}: {e}),跳过对照")
|
|
|
|
|
|
# file 内容块候选形状:方舟文档说 file_id/file_data/file_url 三选一,但块的外层
|
|
# JSON 没抓到 → 挨个试,400 报错信息(带字段名)也是探针产出
|
|
def _candidate_blocks(b64: str, filename: str) -> list[tuple[str, dict]]:
|
|
data_url = f"data:application/pdf;base64,{b64}"
|
|
return [
|
|
("openai_file_data_url",
|
|
{"type": "file", "file": {"filename": filename, "file_data": data_url}}),
|
|
("openai_file_raw_b64",
|
|
{"type": "file", "file": {"filename": filename, "file_data": b64}}),
|
|
("ark_file_url_data",
|
|
{"type": "file_url", "file_url": {"url": data_url}}),
|
|
]
|
|
|
|
|
|
QUESTION = (
|
|
"这是一份多页 PDF 文档。请逐页把其中的文字完整 OCR 出来,"
|
|
"每页以「== 第N页 ==」开头,保留表格数据与换行,不要总结不要遗漏。"
|
|
)
|
|
|
|
# 多页模式只要校验码清单:把"页覆盖上限"和"输出 token 上限"两个变量拆开测
|
|
QUESTION_MAGIC_ONLY = (
|
|
"这是一份多页 PDF 文档,每页都有一行「校验码: ZCBOT-DOC-...」。"
|
|
"请按页序把每页的校验码逐行列出(格式:第N页 <校验码>),只要校验码,别的不用输出。"
|
|
)
|
|
|
|
|
|
def main() -> int:
|
|
cfg = ArkConfig.load()
|
|
if cfg is None:
|
|
print("[SKIP] ARK_API_KEY 未设(或 doubao.yaml 缺失)")
|
|
return 0
|
|
vcfg = (cfg.raw.get("vision") or {}).get("seed_2_lite") or {}
|
|
model_id = vcfg.get("model_id", "doubao-seed-2-0-lite-260428")
|
|
print(f"[setup] model={model_id} base={cfg.base_url}")
|
|
|
|
n_pages = 3
|
|
if len(sys.argv) > 2 and sys.argv[1] == "--pages":
|
|
n_pages = int(sys.argv[2])
|
|
pdf = None
|
|
elif len(sys.argv) > 1:
|
|
pdf = Path(sys.argv[1])
|
|
else:
|
|
pdf = None
|
|
|
|
if pdf is not None:
|
|
synthetic = False
|
|
print(f"[setup] 使用真实 PDF: {pdf}")
|
|
else:
|
|
pdf = Path(tempfile.mkdtemp(prefix="zcbot_probe_")) / "scanned_probe.pdf"
|
|
make_scanned_pdf(pdf, n_pages)
|
|
synthetic = True
|
|
print(f"[setup] 合成扫描件 PDF: {pdf}"
|
|
f"({pdf.stat().st_size} bytes, {n_pages} 页, 无文本层)")
|
|
|
|
check_markitdown_dead_end(pdf)
|
|
|
|
b64 = base64.b64encode(pdf.read_bytes()).decode()
|
|
print(f"[setup] base64 体积 {len(b64) / 1024:.0f} KB")
|
|
|
|
question = QUESTION_MAGIC_ONLY if (synthetic and n_pages > 5) else QUESTION
|
|
resp = None
|
|
accepted = None
|
|
for name, block in _candidate_blocks(b64, pdf.name):
|
|
body = {
|
|
"model": model_id,
|
|
"messages": [{
|
|
"role": "user",
|
|
"content": [{"type": "text", "text": question}, block],
|
|
}],
|
|
}
|
|
print(f"[try] 格式 {name} ...")
|
|
try:
|
|
with ArkClient(cfg, timeout_s=300) as client:
|
|
resp = client.post_json("/chat/completions", body, timeout_s=300)
|
|
accepted = name
|
|
print(f"[OK] 格式 {name} 被接受")
|
|
break
|
|
except ArkError as e:
|
|
print(f"[reject] {name}: {e}")
|
|
|
|
if resp is None:
|
|
print("\n[FAIL] 所有候选格式都被拒 —— 看上面报错定位正确字段名,"
|
|
"或该模型版本不支持 file 输入(需查文档理解模型列表)")
|
|
return 2
|
|
|
|
content = ((resp.get("choices") or [{}])[0].get("message") or {}).get("content") or ""
|
|
if isinstance(content, list):
|
|
content = "\n".join(c.get("text", "") for c in content if isinstance(c, dict))
|
|
usage = resp.get("usage") or {}
|
|
tin = int(usage.get("prompt_tokens", 0) or 0)
|
|
tout = int(usage.get("completion_tokens", 0) or 0)
|
|
cost = (tin * float(vcfg.get("price_cny_per_mtoken_input", 0.6))
|
|
+ tout * float(vcfg.get("price_cny_per_mtoken_output", 3.6))) / 1e6
|
|
|
|
print(f"\n[usage] tokens={tin}+{tout} cost≈¥{cost:.4f} accepted_format={accepted}")
|
|
print(f"[response]\n{content}\n")
|
|
|
|
if synthetic:
|
|
flat = content.replace(" ", "").replace("-", "")
|
|
magics = [_magic(i) for i in range(n_pages)]
|
|
hits = [m for m in magics if m.replace("-", "") in flat]
|
|
missed = [m for m in magics if m not in hits]
|
|
print(f"[verify] 魔术串命中 {len(hits)}/{n_pages}"
|
|
+ (f" 漏: {missed[:5]}{'...' if len(missed) > 5 else ''}" if missed else ""))
|
|
if n_pages <= 5:
|
|
cn_hit = "标准砂" in content and "抗压" in content
|
|
print(f"[verify] 中文表格关键词(标准砂/抗压)命中: {cn_hit}")
|
|
else:
|
|
cn_hit = True
|
|
if not missed and cn_hit:
|
|
print(f"\n[PASS] {n_pages} 页全覆盖:file 输入 + 扫描件 OCR 验证通过")
|
|
return 0
|
|
print("\n[WARN] 部分未命中 —— 人工核对上面 response 判断保真度")
|
|
return 1
|
|
print("[DONE] 真实 PDF 模式:人工核对上面 response")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|