diff --git a/src/polygateway/transports/_monkey_document.py b/src/polygateway/transports/_monkey_document.py
new file mode 100644
index 0000000..295297c
--- /dev/null
+++ b/src/polygateway/transports/_monkey_document.py
@@ -0,0 +1,293 @@
+"""MonkeyOCR 结果包解析: bytes → 文档元素的纯函数(#34 设计 §5/§5.1)。
+
+本模块**无 HTTP、无治理、无全局状态**,所以能在 `asyncio.to_thread` 里运行——
+同步解析大对象会占住事件循环,使取消与 `call_deadline_s`(基于 `asyncio.timeout`)
+无法及时生效。
+
+内容只取自 `_middle.json`(bbox/类型/页码/阅读序的权威来源,spans 里直接带
+`content`/`html`)与配对的 `.md`。**有意不读 `_content_list.json`**: 它是展示格式,
+文本里 `* \\` ~ $` 会被转义成 `\\$` 之类(实测 134 个 text/title 块中 23 个含 `\\$`,
+取它会污染 LaTeX),且无 bbox、类型词汇与 middle 不同,逐块对位还会引入版本漂移
+时整份失败的新故障面。
+
+与 layout 路径(`monkey_ocr._parse_middle_json`,医院线在用,保持原样)的三处差异:
+零页拒绝(成功解析的文档至少 1 页)、`page_idx` 必须等于页序(layout 路径回退枚举
+下标,会让 page_sizes 按 page_index 索引的契约静默错位)、bbox 只要求有限且有序
+(整数化退化校验专为 CHS shim 的 `int()` 裁剪,文档消费者不裁剪,一个退化框不该
+让数百个块的整份文档失败)。
+"""
+
+from __future__ import annotations
+
+import json
+import math
+import zipfile
+import zlib
+from io import BytesIO
+from typing import TYPE_CHECKING
+
+from polygateway.errors import ResultInvalidError
+from polygateway.types import OcrDocumentElement
+
+if TYPE_CHECKING:
+ from collections.abc import Callable
+
+MAX_MIDDLE_BYTES = 64 * 1024 * 1024
+"""`_middle.json` 声明大小上限。样本约 72KB/页,64MiB 足够数百页;超限即拒,
+防止一个畸形/恶意包把解析线程的内存吃满。"""
+
+MAX_MARKDOWN_BYTES = 16 * 1024 * 1024
+"""`.md` 声明大小上限(样本约 3KB/页)。"""
+
+_MIDDLE_SUFFIX = "_middle.json"
+
+
+def _finite_number(value: object) -> float:
+ """JSON 数值 → 有限浮点;布尔与非有限数拒绝(CHS invokers.py:427 逐字)。"""
+ if isinstance(value, bool) or not isinstance(value, (int, float)):
+ raise ValueError("字段不是数值")
+ number = float(value)
+ if not math.isfinite(number):
+ raise ValueError("字段不是有限数值")
+ return number
+
+
+def _parse_page_size(page: dict) -> tuple[float, float]:
+ page_size = page.get("page_size")
+ if not isinstance(page_size, list) or len(page_size) != 2:
+ raise ValueError("page_size 无效")
+ width, height = (_finite_number(item) for item in page_size)
+ if width <= 0 or height <= 0:
+ raise ValueError("page_size 必须为正数")
+ return (width, height)
+
+
+def parse_document_package(
+ zip_bytes: bytes,
+) -> tuple[list[OcrDocumentElement], list[tuple[float, float]], str]:
+ """ZIP → (elements, page_sizes, markdown);一切形态异常归 `ResultInvalidError`。
+
+ 外层异常边界**同时盖住「读取」与「全部结构解析」**: `_finite_number` 里
+ `float(10**400)` 会抛 `OverflowError`,它发生在读取之后。放错位置的后果是原生
+ 异常漏出——`OcrClient._call` 只接 `PolyGatewayError`,于是既不写终态遥测行,又让
+ 持有的半开探针不被归还。`CancelledError` 是 `BaseException`,天然不在其内。
+ """
+ try:
+ payload, markdown = _read_members(zip_bytes)
+ elements, page_sizes = _parse_pages(payload)
+ except ResultInvalidError:
+ raise
+ except (
+ ValueError, # 含 JSONDecodeError / UnicodeDecodeError
+ TypeError,
+ KeyError,
+ OverflowError,
+ RecursionError,
+ zipfile.BadZipFile,
+ zipfile.LargeZipFile,
+ ) as exc:
+ raise ResultInvalidError(f"OCR 文档结果包内容无效: {exc}") from exc
+ return elements, page_sizes, markdown
+
+
+def _read_members(zip_bytes: bytes) -> tuple[object, str]:
+ """取首个 `*_middle.json` 与按名配对的 `.md`,返回 (middle 解码对象, md 文本)。"""
+ with zipfile.ZipFile(BytesIO(zip_bytes)) as archive:
+ names = archive.namelist()
+ middle_names = [n for n in names if n.endswith(_MIDDLE_SUFFIX)]
+ if not middle_names:
+ raise ResultInvalidError("OCR 文档结果包缺少 _middle.json")
+ middle_name = middle_names[0]
+ # 精确配对而非"取第一个 .md": 包里可能有多个 .md,取错就是把别的文档当本文档
+ markdown_name = middle_name[: -len(_MIDDLE_SUFFIX)] + ".md"
+ if markdown_name not in names:
+ raise ResultInvalidError(f"OCR 文档结果包缺少与 {middle_name} 配对的 {markdown_name}")
+ middle_raw = _read_member(archive, middle_name, MAX_MIDDLE_BYTES)
+ markdown_raw = _read_member(archive, markdown_name, MAX_MARKDOWN_BYTES)
+ return json.loads(middle_raw), markdown_raw.decode("utf-8-sig")
+
+
+def _read_member(archive: zipfile.ZipFile, name: str, cap: int) -> bytes:
+ """读一个成员:先查**声明大小**再解压。
+
+ `zipfile` 读取不会超过声明大小(超出即 CRC 失败),所以声明大小检查足以界定内存。
+ 窄 `try` 只包解压调用本身: 加密成员抛 `RuntimeError`、未知压缩方法抛
+ `NotImplementedError`,这两个是标准库对"外部数据不可读"的约定用法,宽泛捕获
+ `RuntimeError` 会掩盖编程错误,故不放到外层边界。
+ """
+ info = archive.getinfo(name)
+ if info.file_size > cap:
+ raise ResultInvalidError(
+ f"OCR 文档结果包成员 {name} 声明大小 {info.file_size} 超过上限 {cap}"
+ )
+ try:
+ with archive.open(info) as stream:
+ return stream.read()
+ except (RuntimeError, NotImplementedError, zlib.error, EOFError) as exc:
+ raise ResultInvalidError(f"OCR 文档结果包成员 {name} 无法解压: {exc}") from exc
+
+
+def _parse_pages(
+ payload: object,
+) -> tuple[list[OcrDocumentElement], list[tuple[float, float]]]:
+ if not isinstance(payload, dict) or not isinstance(payload.get("pdf_info"), list):
+ raise ValueError("_middle.json 缺少 pdf_info 列表")
+ pages = payload["pdf_info"]
+ if not pages:
+ raise ResultInvalidError("OCR 文档结果包 pdf_info 为空: 成功解析的文档至少应有 1 页")
+ elements: list[OcrDocumentElement] = []
+ page_sizes: list[tuple[float, float]] = []
+ for position, page in enumerate(pages):
+ page_elements, page_size = _parse_page(page, position)
+ elements.extend(page_elements)
+ page_sizes.append(page_size)
+ return elements, page_sizes
+
+
+def _parse_page(
+ page: object, position: int
+) -> tuple[list[OcrDocumentElement], tuple[float, float]]:
+ if not isinstance(page, dict):
+ raise ValueError("页面结构无效")
+ page_size = _parse_page_size(page)
+ page_index = _page_index(page, position)
+ blocks = page.get("para_blocks", []) # 缺失 = 空页(空白页/无可识别块)是合法"无内容"
+ if not isinstance(blocks, list):
+ raise ValueError("para_blocks 无效")
+ return [e for block in blocks for e in _block_elements(block, page_index)], page_size
+
+
+def _page_index(page: dict, position: int) -> int:
+ """`page_idx` 必须是等于页序的整数(非 bool)。
+
+ 不沿用 layout 路径的"非法则回退枚举下标": 回退会掩盖服务端跳页/乱序,而
+ `page_sizes` 是按 page_index 索引的,静默错位比响亮失败更糟。
+ """
+ raw = page.get("page_idx")
+ if isinstance(raw, bool) or not isinstance(raw, int) or raw != position:
+ raise ValueError(f"page_idx 无效或与页序不符: {raw!r} != {position}")
+ return raw
+
+
+def _block_type(block: object) -> str:
+ if not isinstance(block, dict):
+ raise ValueError("块结构无效")
+ block_type = block.get("type")
+ if not isinstance(block_type, str) or not block_type.strip():
+ raise ValueError("块缺少 type")
+ return block_type
+
+
+def _loose_bbox(raw: object) -> tuple[float, float, float, float]:
+ """长度 4 的有限数值且有序(允许零宽/零高);差异理由见模块 docstring。"""
+ if not isinstance(raw, list) or len(raw) != 4:
+ raise ValueError("bbox 无效")
+ x1, y1, x2, y2 = (_finite_number(item) for item in raw)
+ if x2 < x1 or y2 < y1:
+ raise ValueError("bbox 边界顺序无效")
+ return (x1, y1, x2, y2)
+
+
+def _sub_blocks(block: dict) -> list[dict]:
+ subs = block.get("blocks", [])
+ if not isinstance(subs, list):
+ raise ValueError("blocks 无效")
+ for sub in subs:
+ _block_type(sub) # 顺带校验子块是 dict 且 type 合法
+ return subs
+
+
+def _is_body(sub: dict) -> bool:
+ return sub["type"].endswith("_body")
+
+
+def _block_elements(block: object, page_index: int) -> list[OcrDocumentElement]:
+ """父块 → 元素,外加其非 `*_body` 的嵌套子块(紧随父块、保持服务顺序)。"""
+ block_type = _block_type(block)
+ assert isinstance(block, dict) # _block_type 已校验;仅为类型收窄(内部不变量)
+ parent = OcrDocumentElement(
+ type=block_type,
+ bbox=_loose_bbox(block.get("bbox")),
+ page_index=page_index,
+ text=_block_text(block_type, block),
+ )
+ return [parent, *_extra_elements(block, page_index)]
+
+
+def _extra_elements(block: dict, page_index: int) -> list[OcrDocumentElement]:
+ """caption/footnote 一类子块:默认部署不产生,出现时有文字就独立成元素,不静默丢弃。"""
+ extras: list[OcrDocumentElement] = []
+ for sub in _sub_blocks(block):
+ if _is_body(sub):
+ continue
+ text = _lines_text(sub.get("lines", []))
+ if text:
+ extras.append(
+ OcrDocumentElement(
+ type=sub["type"],
+ bbox=_loose_bbox(sub.get("bbox")),
+ page_index=page_index,
+ text=text,
+ )
+ )
+ return extras
+
+
+def _block_text(block_type: str, block: dict) -> str:
+ if block_type == "table":
+ return _table_text(_sub_blocks(block))
+ if block_type == "image":
+ return "" # 图内文字服务不识别;span 只有 image_path
+ if "lines" in block:
+ return _lines_text(block["lines"])
+ # 未知容器类型: 其 `*_body` 子块的文字并入父元素(非 body 子块已由 _extra_elements 输出)
+ return "\n".join(
+ _lines_text(sub.get("lines", [])) for sub in _sub_blocks(block) if _is_body(sub)
+ )
+
+
+def _table_text(subs: list[dict]) -> str:
+ bodies = [sub for sub in subs if sub["type"] == "table_body"]
+ if not bodies:
+ raise ValueError("table 块缺少 table_body 子块")
+ return "\n".join(_lines_text(body.get("lines"), _table_span_text) for body in bodies)
+
+
+def _table_span_text(span: dict) -> str:
+ """表体 span 取 html;上游另有 latex 分支。皆无则响亮失败——表格"无内容"与
+ 合法空串在结果里无法区分,静默补空会让下游把缺失当成空表。"""
+ for key in ("html", "latex"):
+ value = span.get(key)
+ if isinstance(value, str):
+ return value
+ raise ValueError("table span 缺少 html/latex")
+
+
+def _lines_text(lines: object, span_text: Callable[[dict], str] | None = None) -> str:
+ """行内 span 按 "" 拼、行间按 "\\n" 拼。
+
+ 样本全是 1 行 1 span;多行/多 span 路径无真实样本,拼接规则由构造 fixture 钉住,
+ 文档注明"未经实测"。
+ """
+ if not isinstance(lines, list):
+ raise ValueError("lines 无效")
+ return "\n".join(_line_text(line, span_text or _span_text) for line in lines)
+
+
+def _line_text(line: object, span_text: Callable[[dict], str]) -> str:
+ if not isinstance(line, dict) or not isinstance(line.get("spans"), list):
+ raise ValueError("行结构无效")
+ parts = []
+ for span in line["spans"]:
+ if not isinstance(span, dict):
+ raise ValueError("span 结构无效")
+ parts.append(span_text(span))
+ return "".join(parts)
+
+
+def _span_text(span: dict) -> str:
+ """文本 span 取 `content`(必须是 str);`inline_equation` 包 `$…$`(上游惯例)。"""
+ content = span.get("content")
+ if not isinstance(content, str):
+ raise ValueError("span 缺少字符串 content")
+ return f"${content}$" if span.get("type") == "inline_equation" else content
diff --git a/src/polygateway/transports/monkey_ocr.py b/src/polygateway/transports/monkey_ocr.py
index 254cc55..c492a01 100644
--- a/src/polygateway/transports/monkey_ocr.py
+++ b/src/polygateway/transports/monkey_ocr.py
@@ -10,7 +10,6 @@ from __future__ import annotations
import asyncio
import json
-import math
import zipfile
from contextlib import contextmanager
from io import BytesIO
@@ -29,6 +28,7 @@ from polygateway.transports._http_errors import (
response_body,
summarize_body,
)
+from polygateway.transports._monkey_document import _finite_number, _parse_page_size
from polygateway.types import (
OcrLayoutElement,
OcrLayoutTransportResult,
@@ -144,26 +144,6 @@ def _classify_status(
return RequestRejectedError(message, **ctx)
-def _finite_number(value: object) -> float:
- """JSON 数值 → 有限浮点;布尔与非有限数拒绝(CHS invokers.py:427 逐字)。"""
- if isinstance(value, bool) or not isinstance(value, (int, float)):
- raise ValueError("字段不是数值")
- number = float(value)
- if not math.isfinite(number):
- raise ValueError("字段不是有限数值")
- return number
-
-
-def _parse_page_size(page: dict) -> tuple[float, float]:
- page_size = page.get("page_size")
- if not isinstance(page_size, list) or len(page_size) != 2:
- raise ValueError("page_size 无效")
- width, height = (_finite_number(item) for item in page_size)
- if width <= 0 or height <= 0:
- raise ValueError("page_size 必须为正数")
- return (width, height)
-
-
def _parse_bbox(bbox: object) -> tuple[float, float, float, float]:
if not isinstance(bbox, list) or len(bbox) != 4:
raise ValueError("bbox 无效")
diff --git a/tests/unit/test_monkey_document.py b/tests/unit/test_monkey_document.py
new file mode 100644
index 0000000..f2cc4cc
--- /dev/null
+++ b/tests/unit/test_monkey_document.py
@@ -0,0 +1,534 @@
+"""MonkeyOCR 文档结果包解析测试(#34 设计 §5/§5.1)。
+
+fixtures 全部**合成**: 保留 2026-10-21 实测的结构骨架(para_blocks 的块类型、
+table/image 的嵌套 `blocks`、span 字段),一切识别文本替换为 `SYNTH-*` 占位——
+第三方论文内容许可不明、医疗数据不入库(零业务假设 + P5),真实 ZIP 不提交。
+
+契约表逐行一条,参数化列明各分支。解析函数是 bytes → 结果的纯函数,失败一律
+`ResultInvalidError`,**不得漏出原生异常**(漏出会让 `OcrClient._call` 既不写终态遥测
+行也不归还半开探针)。
+"""
+
+import io
+import json
+import struct
+import zipfile
+
+import pytest
+
+from polygateway.errors import ResultInvalidError
+from polygateway.transports import _monkey_document
+from polygateway.transports._monkey_document import parse_document_package
+from polygateway.types import OcrDocumentElement
+
+_BBOX = (10.0, 20.0, 110.0, 40.0)
+
+
+def _span(content="SYNTH-TEXT", type_="text"):
+ return {"type": type_, "content": content, "bbox": list(_BBOX), "score": 1.0}
+
+
+def _text_block(content="SYNTH-TEXT", *, type_="text", bbox=_BBOX):
+ return {
+ "type": type_,
+ "bbox": list(bbox),
+ "index": 0,
+ "lines": [{"bbox": list(bbox), "spans": [_span(content)]}],
+ }
+
+
+def _body(type_, spans, bbox=_BBOX):
+ return {"type": type_, "bbox": list(bbox), "lines": [{"bbox": list(bbox), "spans": spans}]}
+
+
+def _table_block(html="
", extra_subs=()):
+ body = _body("table_body", [{"type": "table", "html": html, "image_path": "synth.jpg"}])
+ return {"type": "table", "bbox": list(_BBOX), "index": 1, "blocks": [*extra_subs, body]}
+
+
+def _image_block():
+ body = _body("image_body", [{"type": "image", "image_path": "synth.jpg"}])
+ return {"type": "image", "bbox": list(_BBOX), "index": 2, "blocks": [body]}
+
+
+def _page(blocks=None, *, page_idx=0, page_size=(612.0, 792.0), **extra):
+ page = {"page_idx": page_idx, "page_size": list(page_size), "discarded_blocks": []}
+ if blocks is not None:
+ page["para_blocks"] = blocks
+ page.update(extra)
+ return page
+
+
+def _package(
+ pages=None,
+ *,
+ md="# synth\n",
+ middle=None,
+ middle_name="doc_middle.json",
+ md_name=None,
+ extra=(),
+):
+ """组装结果包。`middle` 给原始 bytes 时原样写入;`md_name` 默认由 middle 名配对。"""
+ if middle is None:
+ middle = json.dumps({"pdf_info": pages if pages is not None else [_page([])]}).encode()
+ if md_name is None:
+ md_name = middle_name[: -len("_middle.json")] + ".md"
+ buf = io.BytesIO()
+ with zipfile.ZipFile(buf, "w") as zf:
+ zf.writestr(middle_name, middle)
+ if md is not None:
+ zf.writestr(md_name, md if isinstance(md, bytes) else md.encode())
+ for name, data in extra:
+ zf.writestr(name, data)
+ return buf.getvalue()
+
+
+def _parse(pages=None, **kwargs):
+ return parse_document_package(_package(pages, **kwargs))
+
+
+def _invalid(pages=None, **kwargs):
+ with pytest.raises(ResultInvalidError):
+ _parse(pages, **kwargs)
+
+
+class TestHappyPath:
+ def test_every_block_type_yields_text_per_contract(self):
+ pages = [
+ _page(
+ [
+ _text_block("SYNTH-TITLE", type_="title"),
+ _text_block("SYNTH-BODY"),
+ _text_block("x^{2}", type_="interline_equation"),
+ _table_block(""),
+ _image_block(),
+ ]
+ )
+ ]
+ elements, page_sizes, markdown = _parse(pages)
+ assert [(e.type, e.text) for e in elements] == [
+ ("title", "SYNTH-TITLE"),
+ ("text", "SYNTH-BODY"),
+ ("interline_equation", "x^{2}"),
+ ("table", ""),
+ ("image", ""),
+ ]
+ assert all(isinstance(e, OcrDocumentElement) for e in elements)
+ assert all(e.bbox == _BBOX and e.page_index == 0 for e in elements)
+ assert page_sizes == [(612.0, 792.0)]
+ assert markdown == "# synth\n"
+
+ def test_reading_order_and_page_index_across_pages(self):
+ pages = [
+ _page([_text_block("A")], page_idx=0),
+ _page([_text_block("B"), _text_block("C")], page_idx=1, page_size=(595.0, 842.0)),
+ ]
+ elements, page_sizes, _ = _parse(pages)
+ assert [(e.page_index, e.text) for e in elements] == [(0, "A"), (1, "B"), (1, "C")]
+ assert page_sizes == [(612.0, 792.0), (595.0, 842.0)]
+
+ def test_content_list_and_images_are_ignored(self):
+ """内容只来自 middle: content_list 把 `$` 转义成 `\\$`(展示格式),取它会污染 LaTeX。"""
+ garbage = json.dumps([{"type": "text", "text": "GARBAGE \\$"}]).encode()
+ elements, _, _ = parse_document_package(
+ _package(
+ [_page([_text_block("SYNTH $x$")])],
+ extra=[("doc_content_list.json", garbage), ("images/x.jpg", b"\xff\xd8")],
+ )
+ )
+ assert [e.text for e in elements] == ["SYNTH $x$"]
+
+ def test_discarded_blocks_are_not_in_the_result(self):
+ pages = [_page([_text_block("KEEP")], discarded_blocks=[_text_block("DROPPED")])]
+ elements, _, _ = _parse(pages)
+ assert [e.text for e in elements] == ["KEEP"]
+
+ def test_need_drop_pages_are_neither_read_nor_rejected(self):
+ pages = [_page([_text_block("KEEP")], need_drop=True, drop_reason=["synthetic"])]
+ elements, _, _ = _parse(pages)
+ assert [e.text for e in elements] == ["KEEP"]
+
+
+class TestTextAssembly:
+ def test_multiple_lines_and_spans_are_joined(self):
+ """样本全是 1 行 1 span;多行/多 span 路径未经实测,只由构造 fixture 钉住拼接规则。"""
+ block = {
+ "type": "text",
+ "bbox": list(_BBOX),
+ "lines": [
+ {"spans": [_span("A"), _span("B")]},
+ {"spans": [_span("C")]},
+ ],
+ }
+ elements, _, _ = _parse([_page([block])])
+ assert elements[0].text == "AB\nC"
+
+ def test_inline_equation_spans_are_wrapped_in_dollars(self):
+ """上游惯例;样本中从未出现该 span 类型,未经实测。"""
+ block = {
+ "type": "text",
+ "bbox": list(_BBOX),
+ "lines": [{"spans": [_span("see "), _span("a+b", "inline_equation"), _span(".")]}],
+ }
+ elements, _, _ = _parse([_page([block])])
+ assert elements[0].text == "see $a+b$."
+
+ def test_empty_lines_list_is_empty_text(self):
+ block = {"type": "text", "bbox": list(_BBOX), "lines": []}
+ elements, _, _ = _parse([_page([block])])
+ assert elements[0].text == ""
+
+ @pytest.mark.parametrize(
+ "mutate",
+ [
+ pytest.param(lambda b: b.update(lines="not-a-list"), id="lines-not-list"),
+ pytest.param(lambda b: b.update(lines=["not-a-dict"]), id="line-not-dict"),
+ pytest.param(lambda b: b["lines"][0].update(spans="x"), id="spans-not-list"),
+ pytest.param(lambda b: b["lines"][0].update(spans=["x"]), id="span-not-dict"),
+ pytest.param(lambda b: b["lines"][0]["spans"][0].pop("content"), id="content-missing"),
+ pytest.param(lambda b: b["lines"][0]["spans"][0].update(content=7), id="content-int"),
+ pytest.param(
+ lambda b: b["lines"][0]["spans"][0].update(content=None), id="content-null"
+ ),
+ ],
+ )
+ def test_malformed_text_structure_is_rejected_not_silently_empty(self, mutate):
+ block = _text_block()
+ mutate(block)
+ _invalid([_page([block])])
+
+
+class TestTableAndImage:
+ def test_table_falls_back_to_latex_when_no_html(self):
+ body = _body("table_body", [{"type": "table", "latex": "\\begin{tabular}SYNTH"}])
+ block = {"type": "table", "bbox": list(_BBOX), "blocks": [body]}
+ elements, _, _ = _parse([_page([block])])
+ assert elements[0].text == "\\begin{tabular}SYNTH"
+
+ @pytest.mark.parametrize(
+ "block",
+ [
+ pytest.param(
+ {
+ "type": "table",
+ "bbox": list(_BBOX),
+ "blocks": [_body("table_body", [{"type": "table"}])],
+ },
+ id="span-has-neither-html-nor-latex",
+ ),
+ pytest.param(
+ {"type": "table", "bbox": list(_BBOX), "blocks": []}, id="no-body-sub-block"
+ ),
+ pytest.param({"type": "table", "bbox": list(_BBOX)}, id="no-blocks-key"),
+ ],
+ )
+ def test_table_without_content_is_rejected_not_silently_empty(self, block):
+ """表格无内容与"合法空串"无法在结果里区分,故响亮失败。"""
+ _invalid([_page([block])])
+
+ def test_image_text_is_empty_even_with_path_only_span(self):
+ elements, _, _ = _parse([_page([_image_block()])])
+ assert (elements[0].type, elements[0].text) == ("image", "")
+
+ def test_non_body_sub_blocks_become_independent_elements_after_their_parent(self):
+ """默认部署不产生 caption/footnote 子块(未经实测);出现时不得静默丢弃。"""
+ caption = _body("table_caption", [_span("SYNTH-CAPTION")], bbox=(1.0, 2.0, 3.0, 4.0))
+ footnote = _body("table_footnote", [_span("SYNTH-NOTE")], bbox=(5.0, 6.0, 7.0, 8.0))
+ block = _table_block(extra_subs=[caption])
+ block["blocks"].append(footnote)
+ elements, _, _ = _parse([_page([block])])
+ assert [(e.type, e.text) for e in elements] == [
+ ("table", ""),
+ ("table_caption", "SYNTH-CAPTION"),
+ ("table_footnote", "SYNTH-NOTE"),
+ ]
+ assert elements[1].bbox == (1.0, 2.0, 3.0, 4.0)
+ assert all(e.page_index == 0 for e in elements)
+
+ def test_empty_non_body_sub_block_is_not_emitted(self):
+ caption = _body("table_caption", [_span("")])
+ elements, _, _ = _parse([_page([_table_block(extra_subs=[caption])])])
+ assert [e.type for e in elements] == ["table"]
+
+ def test_unknown_type_without_lines_or_blocks_keeps_type_and_empty_text(self):
+ block = {"type": "synthetic_widget", "bbox": list(_BBOX)}
+ elements, _, _ = _parse([_page([block])])
+ assert (elements[0].type, elements[0].text) == ("synthetic_widget", "")
+
+ def test_unknown_container_keeps_its_body_text(self):
+ """未知容器类型的 `*_body` 子块文字并入父元素,不被静默丢弃。"""
+ block = {
+ "type": "synthetic_code",
+ "bbox": list(_BBOX),
+ "blocks": [_body("synthetic_code_body", [_span("SYNTH-CODE")])],
+ }
+ elements, _, _ = _parse([_page([block])])
+ assert (elements[0].type, elements[0].text) == ("synthetic_code", "SYNTH-CODE")
+
+
+class TestPageStructure:
+ @pytest.mark.parametrize(
+ "payload",
+ [
+ pytest.param({}, id="pdf_info-missing"),
+ pytest.param({"pdf_info": "x"}, id="pdf_info-not-list"),
+ pytest.param({"pdf_info": []}, id="zero-pages"),
+ pytest.param([], id="payload-not-dict"),
+ ],
+ )
+ def test_document_level_shape(self, payload):
+ """零页被拒: 成功解析的文档至少 1 页(与 layout 路径"elements 空合法"不同)。"""
+ _invalid(middle=json.dumps(payload).encode())
+
+ def test_page_not_a_dict(self):
+ _invalid(["not-a-dict"])
+
+ @pytest.mark.parametrize(
+ "page_size",
+ [
+ pytest.param(None, id="missing"),
+ pytest.param([612.0], id="one-element"),
+ pytest.param([612.0, 792.0, 1.0], id="three-elements"),
+ pytest.param([0, 792.0], id="zero-width"),
+ pytest.param([612.0, -1], id="negative-height"),
+ pytest.param([True, 792.0], id="bool"),
+ pytest.param(["612", 792.0], id="string"),
+ pytest.param([10**400, 792.0], id="huge-int"),
+ ],
+ )
+ def test_page_size_invalid(self, page_size):
+ page = _page([])
+ if page_size is None:
+ del page["page_size"]
+ else:
+ page["page_size"] = page_size
+ _invalid([page])
+
+ @pytest.mark.parametrize(
+ "page_idx",
+ [
+ pytest.param("missing", id="missing"),
+ pytest.param("0", id="string"),
+ pytest.param(0.0, id="float"),
+ pytest.param(True, id="bool"),
+ pytest.param(-1, id="negative"),
+ pytest.param(3, id="out-of-position"),
+ ],
+ )
+ def test_page_idx_must_equal_position(self, page_idx):
+ """layout 路径对非法 `page_idx` 回退枚举下标,会让 page_sizes 按 page_index 的索引契约静默错位。"""
+ page = _page([])
+ if page_idx == "missing":
+ del page["page_idx"]
+ else:
+ page["page_idx"] = page_idx
+ _invalid([page])
+
+ def test_second_page_must_not_claim_the_first_index(self):
+ _invalid([_page([], page_idx=0), _page([], page_idx=0)])
+
+ def test_para_blocks_missing_is_a_legal_empty_page(self):
+ elements, page_sizes, _ = _parse([_page()])
+ assert elements == [] and page_sizes == [(612.0, 792.0)]
+
+ def test_para_blocks_not_a_list_is_rejected(self):
+ _invalid([_page("x")])
+
+
+class TestBlockStructure:
+ @pytest.mark.parametrize(
+ "block",
+ [
+ pytest.param("not-a-dict", id="block-not-dict"),
+ pytest.param({"bbox": list(_BBOX)}, id="type-missing"),
+ pytest.param({"type": 7, "bbox": list(_BBOX)}, id="type-not-str"),
+ pytest.param({"type": "", "bbox": list(_BBOX)}, id="type-empty"),
+ pytest.param({"type": " ", "bbox": list(_BBOX)}, id="type-whitespace"),
+ ],
+ )
+ def test_block_shape(self, block):
+ _invalid([_page([block])])
+
+ @pytest.mark.parametrize(
+ "bbox",
+ [
+ pytest.param(None, id="missing"),
+ pytest.param("1,2,3,4", id="string"),
+ pytest.param([1, 2, 3], id="length-3"),
+ pytest.param([1, 2, 3, 4, 5], id="length-5"),
+ pytest.param([1, 2, "3", 4], id="string-element"),
+ pytest.param([1, 2, True, 4], id="bool-element"),
+ pytest.param([1, 2, float("nan"), 4], id="nan"),
+ pytest.param([1, 2, float("inf"), 4], id="inf"),
+ pytest.param([10, 2, 5, 4], id="x-reversed"),
+ pytest.param([1, 20, 5, 4], id="y-reversed"),
+ pytest.param([0, 0, 10**400, 1], id="huge-int"),
+ ],
+ )
+ def test_bbox_invalid(self, bbox):
+ block = _text_block()
+ if bbox is None:
+ del block["bbox"]
+ else:
+ block["bbox"] = bbox
+ # NaN/Infinity 不是合法 JSON,但 Python 的 json 模块默认接受并往返——真实服务可能发出
+ _invalid([_page([block])])
+
+ @pytest.mark.parametrize("bbox", [(5, 5, 5, 20), (5, 5, 50, 5), (5, 5, 5, 5)])
+ def test_zero_width_or_height_boxes_are_legal(self, bbox):
+ """整数化退化校验专为 CHS shim 的 int() 裁剪;文档消费者不裁剪,一个退化框不该让整份失败。"""
+ elements, _, _ = _parse([_page([_text_block(bbox=bbox)])])
+ assert elements[0].bbox == tuple(float(v) for v in bbox)
+
+ def test_out_of_page_boxes_are_returned_as_is(self):
+ """越界与否是图内语义,归调用方(ARCHITECTURE §5.1 边界);库不裁剪。"""
+ elements, _, _ = _parse([_page([_text_block(bbox=(0, 0, 9999, 9999))])])
+ assert elements[0].bbox == (0.0, 0.0, 9999.0, 9999.0)
+
+
+class TestMembers:
+ def test_missing_middle_json(self):
+ buf = io.BytesIO()
+ with zipfile.ZipFile(buf, "w") as zf:
+ zf.writestr("doc.md", "x")
+ with pytest.raises(ResultInvalidError):
+ parse_document_package(buf.getvalue())
+
+ def test_missing_paired_markdown(self):
+ _invalid(md=None)
+
+ def test_markdown_is_paired_by_name_not_first_match(self):
+ elements, _, markdown = parse_document_package(
+ _package(
+ [_page([_text_block("A")])],
+ md="PAIRED",
+ extra=[("aaa_other.md", "DISTRACTOR")],
+ )
+ )
+ assert markdown == "PAIRED"
+
+ def test_unpaired_markdown_alone_is_not_accepted(self):
+ _invalid(md=None, extra=[("other.md", "DISTRACTOR")])
+
+ def test_first_middle_json_wins_and_pairs_with_its_own_markdown(self):
+ pages_a = json.dumps({"pdf_info": [_page([_text_block("A")])]}).encode()
+ pages_b = json.dumps({"pdf_info": [_page([_text_block("B")])]}).encode()
+ buf = io.BytesIO()
+ with zipfile.ZipFile(buf, "w") as zf:
+ zf.writestr("a_middle.json", pages_a)
+ zf.writestr("b_middle.json", pages_b)
+ zf.writestr("a.md", "MD-A")
+ zf.writestr("b.md", "MD-B")
+ elements, _, markdown = parse_document_package(buf.getvalue())
+ assert [e.text for e in elements] == ["A"] and markdown == "MD-A"
+
+ def test_markdown_bom_is_stripped(self):
+ _, _, markdown = _parse(md=b"\xef\xbb\xbf# synth")
+ assert markdown == "# synth"
+
+ def test_empty_markdown_is_legal(self):
+ _, _, markdown = _parse(md="")
+ assert markdown == ""
+
+ def test_non_utf8_markdown_is_rejected(self):
+ _invalid(md=b"\xff\xfe\x00bad")
+
+ @pytest.mark.parametrize(
+ "middle",
+ [
+ pytest.param(b"not json at all", id="not-json"),
+ pytest.param(b"", id="empty"),
+ pytest.param(b"\xff\xfe", id="not-utf8"),
+ ],
+ )
+ def test_middle_json_unparseable(self, middle):
+ _invalid(middle=middle)
+
+ def test_not_a_zip(self):
+ with pytest.raises(ResultInvalidError):
+ parse_document_package(b"definitely not a zip")
+
+
+def _patch_first_member_compression(zip_bytes: bytes, method: int) -> bytes:
+ """把第一个成员的压缩方法号改成 `method`(本地头偏移 8,中央目录偏移 10)。"""
+ data = bytearray(zip_bytes)
+ local = data.index(b"PK\x03\x04")
+ central = data.index(b"PK\x01\x02")
+ struct.pack_into(" bytes:
+ buf = io.BytesIO()
+ with zipfile.ZipFile(buf, "w") as zf:
+ info = zipfile.ZipInfo("doc_middle.json")
+ info.flag_bits |= 0x1 # 加密位:读取时 zipfile 要求密码
+ zf.writestr(info, b'{"pdf_info": []}')
+ zf.writestr("doc.md", "x")
+ return buf.getvalue()
+
+
+class TestExternalDataExceptionBoundary:
+ """外部字节引起的一切已知异常都归 `ResultInvalidError`,绝不漏出原生异常。
+
+ 漏出的后果(`OcrClient._call` 只接 PolyGatewayError): 不写终态遥测行,持有的半开探针
+ 也不归还。边界必须同时盖住「读取」与「全部结构解析」两段。
+ """
+
+ def test_encrypted_member(self):
+ with pytest.raises(ResultInvalidError):
+ parse_document_package(_encrypted_zip())
+
+ def test_unknown_compression_method(self):
+ patched = _patch_first_member_compression(_package([_page([])]), 99)
+ with pytest.raises(ResultInvalidError):
+ parse_document_package(patched)
+
+ def test_crc_corruption(self):
+ corrupted = _package([_page([])]).replace(b'"pdf_info"', b'"pdf_infq"', 1)
+ with pytest.raises(ResultInvalidError):
+ parse_document_package(corrupted)
+
+ def test_huge_integer_inside_page_structure_is_not_an_overflow_error(self):
+ """`float(10**400)` 抛 OverflowError——发生在结构解析段,不在读取段。"""
+ block = _text_block()
+ block["bbox"] = [0, 0, 10**400, 1]
+ _invalid([_page([block])])
+
+ def test_deeply_nested_json_is_not_a_recursion_error(self):
+ deep = b'{"pdf_info": ' + b"[" * 100_000 + b"]" * 100_000 + b"}"
+ _invalid(middle=deep)
+
+ @pytest.mark.parametrize("member", ["middle", "markdown"])
+ def test_declared_size_over_cap_is_rejected(self, member, monkeypatch):
+ monkeypatch.setattr(
+ _monkey_document, "MAX_MIDDLE_BYTES" if member == "middle" else "MAX_MARKDOWN_BYTES", 8
+ )
+ _invalid([_page([_text_block("A")])], md="# a markdown longer than eight bytes")
+
+ def test_existing_result_invalid_messages_pass_through_unchanged(self):
+ with pytest.raises(ResultInvalidError, match="至少应有 1 页"):
+ _parse([])
+
+
+class TestLayoutPathUntouched:
+ """文档路径更严/更宽的口径不得外溢到医院线在用的 layout 解析。"""
+
+ def test_layout_parser_still_rejects_degenerate_int_boxes(self):
+ from polygateway.transports.monkey_ocr import _parse_middle_json
+
+ block = {"type": "text", "bbox": [5, 5, 5.4, 20], "lines": []}
+ zip_bytes = _package([_page([block])])
+ with pytest.raises(ResultInvalidError):
+ _parse_middle_json(zip_bytes)
+
+ def test_layout_parser_keeps_its_lenient_page_idx_handling(self):
+ """layout 路径: 合法的非负整数 `page_idx` 原样接受(哪怕与页序不符),非法值才回退
+ 枚举下标——这正是文档路径改为严格的原因,但医院线的既有行为不得被连带改动。"""
+ from polygateway.transports.monkey_ocr import _parse_middle_json
+
+ accepted, _ = _parse_middle_json(_package([_page([_text_block()], page_idx=7)]))
+ assert accepted[0].page_index == 7
+ fallback, _ = _parse_middle_json(_package([_page([_text_block()], page_idx="x")]))
+ assert fallback[0].page_index == 0
diff --git a/tests/unit/test_monkey_ocr.py b/tests/unit/test_monkey_ocr.py
index 21ae776..05ee84e 100644
--- a/tests/unit/test_monkey_ocr.py
+++ b/tests/unit/test_monkey_ocr.py
@@ -190,7 +190,9 @@ class TestSniffMedia:
pytest.param(b"GIF87a" + b"0" * 20, "image", id="gif87a"),
pytest.param(b"RIFF\x00\x00\x00\x00WEBPVP8 ", "image", id="webp"),
# 图片魔数优先: 含 XMP 元数据的图片可能在 1KB 内带 `%PDF-` 字样
- pytest.param(b"\xff\xd8\xff\xe0" + b"%PDF-", "image", id="jpeg-mentions-pdf"),
+ pytest.param(
+ b"\xff\xd8\xff\xe0" + b"%PDF-", "image", id="jpeg-mentions-pdf"
+ ),
pytest.param(b"paywall", "unknown", id="html"),
pytest.param(b"GARBAGE-NOT-AN-IMAGE", "unknown", id="garbage"),
pytest.param(b"", "unknown", id="empty"),
@@ -205,7 +207,9 @@ class TestSniffMedia:
assert _upload_part(_PDF_BODY) == ("doc.pdf", _PDF_BODY, "application/pdf")
assert _upload_part(_JUNK + _PDF_BODY)[0] == "doc.pdf"
- @pytest.mark.parametrize("data", [b"\xff\xd8\xff\xe0abc", b"\x89PNG\r\n\x1a\nabc", b"junk", b""])
+ @pytest.mark.parametrize(
+ "data", [b"\xff\xd8\xff\xe0abc", b"\x89PNG\r\n\x1a\nabc", b"junk", b""]
+ )
def test_everything_else_keeps_the_legacy_label(self, data):
"""非 PDF 路径字节级不变: 服务对图片按内容解码(2026-10-21 实测 PNG/TIFF/BMP 标 image.jpg 成功)。"""
assert _upload_part(data) == ("image.jpg", data, "image/jpeg")