From 3db7d9979c084ee084eb7af010a15dc34cec679a Mon Sep 17 00:00:00 2001 From: iomgaa Date: Wed, 21 Oct 2026 23:35:47 -0400 Subject: [PATCH] feat: parse document result packages from middle json only Content comes from the middle json and its paired markdown; the content list is deliberately ignored because it escapes markdown characters. The external-data exception boundary covers both reading and structure parsing so no native exception escapes the OCR client. The layout parser is unchanged. --- .../transports/_monkey_document.py | 293 ++++++++++ src/polygateway/transports/monkey_ocr.py | 22 +- tests/unit/test_monkey_document.py | 534 ++++++++++++++++++ tests/unit/test_monkey_ocr.py | 8 +- 4 files changed, 834 insertions(+), 23 deletions(-) create mode 100644 src/polygateway/transports/_monkey_document.py create mode 100644 tests/unit/test_monkey_document.py 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="
SYNTH-CELL
", 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("SYNTH
"), + _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", "SYNTH
"), + ("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", "
SYNTH-CELL
"), + ("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")