"""MonkeyOCR transport: 双端点协议细节(M3 设计 §4;不含任何治理)。 蓝本: CHS `invokers.py:427-552`(两段协议 + 数值防御,逐段保真)与 VT `adapters/ocr.py`(/ocr/text 请求形态)。已在设计声明的偏离: 全元素 提取、`success!=true` 附 status_code=200、429 细分放弃、健康判定收紧 2xx、 `content` 显式校验。服务无鉴权——不带 Authorization 头,api_key 是占位。 """ from __future__ import annotations import asyncio import json import math import zipfile from contextlib import contextmanager from io import BytesIO from typing import TYPE_CHECKING, Any import httpx from polygateway.errors import ( RequestRejectedError, ResultInvalidError, SourceDeadError, TransientError, ) from polygateway.types import ( OcrLayoutElement, OcrLayoutTransportResult, OcrTextTransportResult, ) if TYPE_CHECKING: from collections.abc import Callable, Iterator from polygateway.types import SourceConfig _HEALTH_TIMEOUT_S = 5.0 # 探测常量(同 jitter 系数先例,不进配置) def _default_client_factory(source: SourceConfig) -> httpx.AsyncClient: return httpx.AsyncClient( base_url=source.base_url, timeout=httpx.Timeout(source.timeout_s), trust_env=source.trust_env, # VT LAN 直连绕代理场景配 false(R9) ) @contextmanager def _translate_http_errors(source_name: str, operation: str) -> Iterator[None]: """httpx 异常 → 错误四分类(CHS `_translate_http_errors` 语义主体保真)。 429 细分(insufficient_quota/Retry-After)有意放弃——MonkeyOCR 无鉴权 无计费(设计 §10.2);`from exc` 保留 __cause__ 供 RetryMW 归类 timeout。 """ try: yield except httpx.TimeoutException as exc: raise TransientError( f"{source_name} OCR {operation} 超时: {exc}", source_name=source_name, operation=operation, ) from exc except httpx.HTTPStatusError as exc: raise _classify_status(exc, source_name, operation) from exc except httpx.HTTPError as exc: raise TransientError( f"{source_name} OCR {operation} 网络错误: {exc}", source_name=source_name, operation=operation, ) from exc def _classify_status( exc: httpx.HTTPStatusError, source_name: str, operation: str ) -> TransientError | SourceDeadError | RequestRejectedError: status = exc.response.status_code ctx: dict[str, Any] = { "source_name": source_name, "status_code": status, "operation": operation, } message = f"{source_name} OCR {operation} HTTP {status}" if status >= 500 or status == 429: return TransientError(message, **ctx) if status in (401, 403): return SourceDeadError(message, **ctx) 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 无效") x1, y1, x2, y2 = (_finite_number(item) for item in bbox) if x2 <= x1 or y2 <= y1: raise ValueError("bbox 边界顺序无效") # 库返回 float 原生 bbox;整数化退化校验专为 CHS shim 的 int() 裁剪路径 # 兜底(亚像素宽的框裁剪即空图)——不是死代码,勿删(设计 §4) if int(x2) <= int(x1) or int(y2) <= int(y1): raise ValueError("bbox 整数化后退化") return (x1, y1, x2, y2) def _parse_block(block: object, page_index: int) -> OcrLayoutElement: 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 OcrLayoutElement( type=block_type, bbox=_parse_bbox(block.get("bbox")), page_index=page_index ) def _parse_pages(payload: object) -> tuple[list[OcrLayoutElement], list[tuple[float, float]]]: if not isinstance(payload, dict) or not isinstance(payload.get("pdf_info"), list): raise ValueError("_middle.json 缺少 pdf_info 列表") elements: list[OcrLayoutElement] = [] page_sizes: list[tuple[float, float]] = [] for fallback_index, page in enumerate(payload["pdf_info"]): if not isinstance(page, dict): raise ValueError("页面结构无效") page_sizes.append(_parse_page_size(page)) raw_index = page.get("page_idx", fallback_index) # bool 是 int 子类、负页号无意义——与 _finite_number 拒 bool 的口径一致 valid_idx = ( isinstance(raw_index, int) and not isinstance(raw_index, bool) and raw_index >= 0 ) page_index = raw_index if valid_idx else fallback_index blocks = page.get("para_blocks", []) if not isinstance(blocks, list): raise ValueError("para_blocks 无效") elements.extend(_parse_block(block, page_index) for block in blocks) return elements, page_sizes def _parse_middle_json( zip_bytes: bytes, ) -> tuple[list[OcrLayoutElement], list[tuple[float, float]]]: """ZIP → (elements, page_sizes);一切形态异常归 ResultInvalid(坏图≠坏服务)。 数值防御全量下沉自 CHS `_parse_table_result`(invokers.py:437-479), 提取面从"首表"泛化为 para_blocks 全元素(设计 §1.2 取证证明无损)。 """ try: with zipfile.ZipFile(BytesIO(zip_bytes)) as archive: middle_files = [n for n in archive.namelist() if n.endswith("_middle.json")] if not middle_files: raise ValueError("结果包缺少 _middle.json") with archive.open(middle_files[0]) as stream: payload = json.load(stream) return _parse_pages(payload) except (ValueError, TypeError, KeyError, json.JSONDecodeError, zipfile.BadZipFile) as exc: raise ResultInvalidError(f"OCR 结果包内容无效: {exc}") from exc class MonkeyOcrTransport: """MonkeyOCR transport: 每源一个预配 httpx client,懒创建,aclose 统一释放。""" def __init__( self, *, client_factory: Callable[[SourceConfig], httpx.AsyncClient] | None = None, ) -> None: self._client_factory = client_factory or _default_client_factory self._clients: dict[str, httpx.AsyncClient] = {} def _client_for(self, source: SourceConfig) -> httpx.AsyncClient: client = self._clients.get(source.name) if client is None: client = self._client_factory(source) self._clients[source.name] = client return client @staticmethod def _files(image: bytes) -> dict[str, tuple[str, bytes, str]]: # 固定文件名/类型(CHS invokers.py:496 原样;服务按内容处理,35 样本实测) return {"file": ("image.jpg", image, "image/jpeg")} def _response_json(self, resp: httpx.Response, source: SourceConfig, operation: str) -> dict: """响应体 JSON 化;坏 JSON/非 dict → Transient(设计 §4: 服务侧瞬时异常)。""" try: data = resp.json() except (json.JSONDecodeError, ValueError) as exc: raise TransientError( f"{source.name} OCR {operation} 响应非合法 JSON", source_name=source.name, operation=operation, ) from exc if not isinstance(data, dict): raise TransientError( f"{source.name} OCR {operation} JSON 结构无效", source_name=source.name, operation=operation, ) return data async def recognize_text( self, *, image: bytes, source: SourceConfig, call_id: str ) -> OcrTextTransportResult: """POST /ocr/text → 多行纯文本;content 缺失/非 str → ResultInvalid。""" del call_id # 协议无关,治理侧遥测使用 client = self._client_for(source) with _translate_http_errors(source.name, "ocr_text"): resp = await client.post("/ocr/text", files=self._files(image)) resp.raise_for_status() data = self._response_json(resp, source, "ocr_text") content = data.get("content") if not isinstance(content, str): # VT 的 .get("content","") 静默兜底有意替换为显式校验(P5) raise ResultInvalidError( f"{source.name} /ocr/text 响应 content 缺失或非字符串", source_name=source.name, operation="ocr_text", ) return OcrTextTransportResult( text=content, raw={k: v for k, v in data.items() if k != "content"} ) async def parse_layout( self, *, image: bytes, source: SourceConfig, call_id: str ) -> OcrLayoutTransportResult: """两段协议: POST /parse → GET download_url → ZIP → 全元素提取。""" del call_id client = self._client_for(source) with _translate_http_errors(source.name, "parse"): resp = await client.post("/parse", files=self._files(image)) resp.raise_for_status() data = self._response_json(resp, source, "parse") if data.get("success") is not True: # 设计 §10.2 有意修复: 200 响应确证服务活着,附 status_code # 让治理侧按"响应即健康"记熔断成功(CHS 原样会误走本地拒绝分支) raise RequestRejectedError( f"{source.name} OCR 解析失败: {data.get('message')}", source_name=source.name, status_code=resp.status_code, operation="parse", ) download_url = data.get("download_url") if not isinstance(download_url, str) or not download_url.strip(): raise TransientError( f"{source.name} /parse 响应缺少 download_url", source_name=source.name, operation="parse", ) with _translate_http_errors(source.name, "download_result"): # 实测 download_url 为相对路径(/static/*.zip),相对 base_url 解析; # httpx 对绝对 URL 会忽略 base_url,两种形态天然兼容 result_resp = await client.get(download_url) result_resp.raise_for_status() elements, page_sizes = _parse_middle_json(result_resp.content) return OcrLayoutTransportResult( elements=elements, page_sizes=page_sizes, raw={"success": True, "download_url": download_url}, ) async def check_health(self, *, source: SourceConfig) -> bool: """GET /health 探测: 2xx → True,失败 → False,取消穿透(R10)。""" client = self._client_for(source) try: resp = await client.get("/health", timeout=_HEALTH_TIMEOUT_S) except asyncio.CancelledError: raise except Exception: # noqa: BLE001 — 探测不是调用,失败一律视为不健康 return False return resp.is_success async def aclose(self) -> None: """幂等释放全部 per-source client。""" clients, self._clients = self._clients, {} for client in clients.values(): await client.aclose()