From d4eaccd12e171a4d7581cb55e5e6d4ab0073e89a Mon Sep 17 00:00:00 2001 From: q2000s Date: Tue, 13 Oct 2026 08:10:24 +0800 Subject: [PATCH 1/7] feat(pdf): add mineru_first strategy with local fallback Adds a PDF parsing strategy that prefers MinerU and degrades to the local parser instead of failing the import. It sits between 'mineru' (hard dependency on a reachable endpoint) and 'auto' (local first): local never calls MinerU mineru always calls MinerU, failure fails the import mineru_first calls MinerU first, failure falls back to local auto calls MinerU only after local fails Because the strategy is defined as recoverable, MinerU problems never become startup problems: should_preflight_mineru now covers mineru_first so the preflight runs, but its failure stays advisory and the fallback happens per document. The result metadata records the converter that actually ran as pdf_strategy_effective ('mineru' or 'local'), so operators can audit conversion quality after the fact rather than inferring it from logs. Spelled mineru_first rather than mineru-first to match the snake_case convention used by the other enumerations in ov.conf. This spelling is used in configuration, ov.conf, logs and metadata. Tests cover the fallback path, which is the contract this strategy exists to provide, plus config validation. --- docs/en/about/02-changelog.md | 28 + docs/en/guides/01-configuration.md | 69 +- docs/zh/about/02-changelog.md | 4 + docs/zh/guides/01-configuration.md | 42 +- openviking/parse/parsers/pdf.py | 623 +++++++++++++++---- openviking/service/core.py | 2 +- openviking_cli/utils/config/parser_config.py | 27 +- tests/parse/test_pdf_strategy_dispatch.py | 465 ++++++++++++++ 8 files changed, 1142 insertions(+), 118 deletions(-) create mode 100644 tests/parse/test_pdf_strategy_dispatch.py diff --git a/docs/en/about/02-changelog.md b/docs/en/about/02-changelog.md index 7a0fdae959..7f267553fe 100644 --- a/docs/en/about/02-changelog.md +++ b/docs/en/about/02-changelog.md @@ -7,6 +7,34 @@ This changelog is automatically generated from [GitHub Releases](https://github. ## Unreleased +- **`auto` degrades cheaply when a remote parser is rate limited or down**: `429` and `5xx` + are retried once after a short backoff, permanent errors such as `404` skip the retry + and go straight to local, and a provider that fails repeatedly is circuit-broken so the + rest of the batch stops paying for a doomed call. The circuit half-opens after a + cooldown so a recovered service is picked up again. Documents that fell back are + counted in `pdf_remote_degraded`, and whether any remote parser is configured is now + `PDFConfig.remote_enabled`, expressed over the provider list rather than one service. +- **MinerU hosted API support (`mineru_token`)**: a new `pdf.mineru_token` setting carries + the bearer token for the hosted MinerU service and, because that service has no + `/file_parse` route, its presence also selects the hosted batch protocol: + `POST /file-urls/batch` to reserve an upload URL, `PUT` the PDF there, poll + `GET /extract-results/batch/{batch_id}` until it settles, then download `full_zip_url` + and take the markdown with its images. Without a token the endpoint keeps using the + existing self-hosted `/file_parse` contract, so current deployments are unaffected. + If the batch handshake is refused the parser retries once against `/file_parse`, which + keeps token-gated self-hosted deployments working. +- **PDF `auto` strategy prefers the configured remote parser (behavior change)**: `auto` now + sends every PDF to the configured remote parser and uses the local parser only as a + fallback, where it previously tried local first and only used MinerU when local parsing + failed. With `pdf.mineru_endpoint` set, expect changes in latency, quota consumption and + rate-limit exposure; under throttling the fallback makes such imports slower, not faster. + Set `strategy: local` to keep the previous remote-as-last-resort behaviour, or + `strategy: mineru` to require the remote path and fail on its errors. A remote failure + degrades a single document instead of failing the batch, and `pdf_strategy_effective` + (`mineru` or `local`) records which converter actually ran. Service-specific settings + (`mineru_endpoint`, `mineru_token`, `mineru_bodys`) stay where they are, so a future + remote parser is added as a provider rather than as another `strategy` value. + - **Working Memory defaults off (behavior change)**: Commits continue to archive raw messages and extract long-term memories, but generate no WM/checkpoint summaries by default. Update installed agent plugins separately: automatic archive injection and takeover now diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index 0d47386b31..f97b643db7 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -1033,7 +1033,7 @@ needs those platform-specific URL semantics. Add other Git hosts to ### pdf -PDF parsing configuration. Three strategies are supported: `local` (local pdfplumber), `mineru` (remote MinerU API), and `auto` (try local first, fall back to MinerU). +PDF parsing configuration. Three strategies are supported: `local` (local pdfplumber), `mineru` (remote MinerU API, failures fail the import), and `auto` (use the configured remote parser when there is one, otherwise local). ```json { @@ -1056,8 +1056,71 @@ PDF parsing configuration. Three strategies are supported: `local` (local pdfplu | `mineru_endpoint` | str | MinerU API **base URL** (e.g. `http://127.0.0.1:8000`) | | `mineru_timeout` | float | Request timeout in seconds (default `300.0`) | | `mineru_bodys` | dict | MinerU API multipart form fields | - -**MinerU protocol**: a synchronous `POST {mineru_endpoint}/file_parse` request with the PDF as the multipart `files` field; form parameters are passed through from `mineru_bodys`. +| `mineru_token` | str | Bearer token for the hosted MinerU service. Setting it also selects the hosted batch protocol (see below) | + +**MinerU protocols**: which one is used follows from whether `mineru_token` is set. + +| Config | Protocol | Flow | +|---|---|---| +| no `mineru_token` | self-hosted (v1) | `POST {mineru_endpoint}/file_parse`, markdown returned inline | +| `mineru_token` set | hosted batch (v4) | `POST /file-urls/batch` to reserve an upload URL, `PUT` the PDF there, poll `GET /extract-results/batch/{batch_id}` until it settles, then download `full_zip_url` and take the markdown plus its images | + +The hosted service has no `/file_parse` route at all, so a token is treated as the +signal for that API rather than as a credential to bolt onto the v1 call. If the batch +handshake is refused the parser retries once against `/file_parse`, which keeps a +token-gated self-hosted deployment working. + +**Choosing a strategy:** + +| `strategy` | Remote parser used? | On remote failure | `mineru_endpoint` required | +|---|---|---|---| +| `local` | no | n/a | no | +| `mineru` | always | the import fails | yes | +| `auto` | yes, if one is configured | falls back to the local parser | optional | + +`auto` (the default) uses whichever remote parser is configured and treats the local +parser as the last resort. With `mineru_endpoint` set, every PDF is converted remotely; +if that call fails — missing token, rate limit, outage — that single document falls back +to local and the import still succeeds. With nothing configured, `auto` uses the local +parser directly. `mineru` is the strict variant: MinerU is the result, and a failure +fails the import rather than degrading. + +Whether a remote parser is in play is `PDFConfig.remote_enabled`, expressed over the +provider list rather than any single service, so adding a provider does not mean +revisiting `auto`, the preflight, or this page. `mineru_enabled` remains the MinerU +provider's own check. + +To keep a degraded batch cheap, `auto` does not retry blindly: + +- `429` and `5xx` are retried once after a short backoff, since they usually clear. +- `404` and other permanent errors go straight to local; retrying cannot help. +- After several consecutive failures the provider is circuit-broken: the rest of the + batch skips it instead of paying for a doomed call per document. The circuit + half-opens after a cooldown, so a recovered service is picked up again. +- Documents that fell back are counted in `pdf_remote_degraded`, alongside + `pdf_strategy_effective`, so a batch that quietly stopped using the remote parser is + visible after the fact. + +The startup preflight follows the same rule: `mineru` fails fast if the endpoint is not +ready, `auto` only warns and leaves the retry to the first parse. + +:::warning Behavior change +`auto` previously tried the local parser **first** and only used MinerU when local +parsing failed. It now prefers the configured remote parser. If you have `mineru_endpoint` +set, every PDF is sent to that service, so expect a change in latency, quota usage and +rate-limit exposure. Set `strategy: local` to keep the previous remote-as-last-resort +behaviour, or `strategy: mineru` if you want the remote path to be mandatory. +::: + +Service-specific settings live in their own fields (`mineru_endpoint`, `mineru_token`, +`mineru_bodys`), so supporting a new remote parser is a provider change rather than a new +`strategy` value. + +:::tip `auto` degrades on any remote error +A remote failure of any kind -- `404`, `405`, `429`, a timeout, a 5xx -- falls back to the +local parser for that document, so the import still succeeds. Set `mineru_token` when +pointing `mineru_endpoint` at the hosted MinerU service: that both authenticates the call +and selects the protocol the service actually speaks. ### External parser API (`parser_api`) {#parser-api} diff --git a/docs/zh/about/02-changelog.md b/docs/zh/about/02-changelog.md index b637e623e4..3b1c58b6da 100644 --- a/docs/zh/about/02-changelog.md +++ b/docs/zh/about/02-changelog.md @@ -7,6 +7,10 @@ OpenViking 的所有重要变更都将记录在此文件中。 ## 未发布 +- **`auto` 在远程解析服务被限流或崩溃时可低成本地降级**：`429` 与 `5xx` 会在一段退避后重试一次，`404` 等永久性错误不重试而直接降级，连续失败的 provider 会被熔断，使批次剩下文档不再为每份重复付出必败的调用。熔断在一段冷却后半开重试，服务恢复后会被重新使用。已降级的文档计入 `pdf_remote_degraded`；同时，“是否有远程解析服务在场”改由 `PDFConfig.remote_enabled` 判定，它从 provider 列表表达，而不绑定单一服务。 +- **MinerU 托管服务支持（`mineru_token`）**：新增 `pdf.mineru_token` 用于携带托管版 MinerU 服务的 Bearer token；由于该服务没有 `/file_parse` 路由，配置它同时会选中托管批处理协议：`POST /file-urls/batch` 预留上传 URL → `PUT` 上传 PDF → 轮询 `GET /extract-results/batch/{batch_id}` → 下载 `full_zip_url` 并取出 markdown 与其图片。未配置 token 时仍使用原有的自托管 `/file_parse` 协议，现有部署不受影响。若批处理握手被拒，解析器会再尝试一次 `/file_parse`，以保持带 token 的自托管部署仍然可用。 +- **PDF `auto` 策略改为优先使用已配置的远程解析器（行为变更）**：`auto` 现在会将每份 PDF 发往已配置的远程解析服务，本地解析器仅作兜底；而之前是先本地、本地失败才使用 MinerU。若配置了 `pdf.mineru_endpoint`，请预期延迟、额度消耗与限流风险发生变化；在限流情况下，回退机制会让导入变慢而不是变快。若希望保持原有行为请设为 `strategy: local`；若希望远程解析必须成功、失败即失败，请设为 `strategy: mineru`。远程失败会仅降级单份文档而不使整批导入失败，并由 `pdf_strategy_effective`（`mineru` 或 `local`）记录实际使用的转换器。服务类配置（`mineru_endpoint`、`mineru_token`、`mineru_bodys`）保持当前位置，未来接入新的远程解析服务时只需增加 provider，而无需新增 `strategy` 取值。 + - **Working Memory 默认关闭（行为变更）**：commit 仍保存原文并抽取长期记忆，默认不生成 WM 或 checkpoint 摘要。用户需要单独升级已安装的 agent 插件；新版默认关闭自动归档 摘要注入和上下文接管，由宿主管理历史与压缩。VikingBot 仍默认使用 OV 管理上下文与压缩， diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index 12dd465be3..229082a9f1 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -1028,7 +1028,7 @@ ollama pull guoxuter/ov_intent_analysis_sft:v7_q8 ### pdf -PDF 解析配置。支持三种策略：`local`（本地 pdfplumber）、`mineru`（远程 MinerU API）、`auto`（先本地、失败回退 MinerU）。 +PDF 解析配置。支持三种策略：`local`（本地 pdfplumber）、`mineru`（远程 MinerU API，失败即导入失败）、`auto`（配置了远程解析服务就优先使用，否则走本地）。 ```json { @@ -1051,8 +1051,46 @@ PDF 解析配置。支持三种策略：`local`（本地 pdfplumber）、`mineru | `mineru_endpoint` | str | MinerU API **base URL**（如 `http://127.0.0.1:8000`） | | `mineru_timeout` | float | 请求超时秒数（默认 `300.0`） | | `mineru_bodys` | dict | MinerU API multipart form 参数 | +| `mineru_token` | str | 托管 MinerU 服务的 Bearer token。配置它同时会选中托管批处理协议（见下） | -**MinerU 协议**：同步调用 `POST {mineru_endpoint}/file_parse`，multipart 文件字段为 `files`，form 参数由 `mineru_bodys` 透传。 +**MinerU 协议**：使用哪种协议取决于是否配置了 `mineru_token`。 + +| 配置 | 协议 | 流程 | +|---|---|---| +| 未配 `mineru_token` | 自托管 v1 | `POST {mineru_endpoint}/file_parse`，同步返回 markdown | +| 配置了 `mineru_token` | 托管批处理 v4 | `POST /file-urls/batch` 预留上传 URL → `PUT` 上传 PDF → 轮询 `GET /extract-results/batch/{batch_id}` → 下载 `full_zip_url` 并解包 markdown 与图片 | + +托管服务根本没有 `/file_parse` 路由，因此 token 在这里是“选中托管 API”的信号，而不是简单地贴到 v1 请求上。若批处理握手被拒，解析器会再尝试一次 `/file_parse`，以保持带 token 的自托管部署仍然可用。 + +**如何选择策略：** + +| `strategy` | 是否使用远程解析 | 远程失败时 | 是否必填 `mineru_endpoint` | +|---|---|---|---| +| `local` | 否 | — | 否 | +| `mineru` | 始终使用 | 导入失败 | 是 | +| `auto` | 配置了就使用 | 回退到本地解析器 | 否 | + +`auto`（默认）使用已配置的远程解析服务，并把本地解析器作为最后兜底。若配置了 `mineru_endpoint`，每份 PDF 都会走远程转换；若该调用失败（token 缺失、限流、服务不可用），只该份文档会降级为本地解析，导入仍然成功。若什么都没配，`auto` 直接使用本地解析。`mineru` 是严格变体：远程失败就使导入失败，而不降级。 + +是否有远程解析服务在场，由 `PDFConfig.remote_enabled` 判定，它是从 provider 列表的角度表达的，而不是绑定某一个具体服务；因此新增 provider 时无需重新修改 `auto`、预检或本页。`mineru_enabled` 仍是 MinerU 这个 provider 自身的判定。 + +为避免一个已降级的批次变得明显更慢，`auto` 不会无目的地重试： + +- `429` 与 `5xx` 会在一段退避后重试一次，因为它们通常会自行恢复； +- `404` 等永久性错误直接降级，重试无意义； +- 连续失败几次后该 provider 会被熔断：批次剩下的文档直接跳过它，不再为每份文档重复付出一次必败的调用。熔断在一段冷却后会半开重试，服务恢复后会被重新使用； +- 已降级的文档会计入 `pdf_remote_degraded`，与 `pdf_strategy_effective` 并列输出，便于事后发现某个批次已经静默地改用本地解析。 + +启动预检遵循同一规则：`mineru` 在端点未就绪时快速失败，`auto` 仅告警，重试留给首次解析。 + +:::warning 行为变更 +`auto` 之前是**先本地、本地失败才用 MinerU**；现在改为**优先使用已配置的远程服务**。若你配置了 `mineru_endpoint`，每份 PDF 都会发往该服务，延迟、额度消耗与限流风险都会变化。若希望保持原有行为，设为 `strategy: local`；若希望远程解析必须成功，设为 `strategy: mineru`。 +::: + +服务类配置各自独立（`mineru_endpoint`、`mineru_token`、`mineru_bodys`），因此接入新的远程解析服务只需增加 provider，而不是新增 `strategy` 取值。 + +:::tip `auto` 在任意远程错误时都会降级 +任何远程失败——`404`、`405`、`429`、超时、5xx——都会让该份文档降级到本地解析器，导入仍然成功。当 `mineru_endpoint` 指向托管版 MinerU 时请配置 `mineru_token`：它既能完成认证，也会选中该服务实际支持的协议。 ### 外部解析 API（`parser_api`） {#parser-api} diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index 509fb55f3f..b671581e24 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -17,11 +17,12 @@ import hashlib import io import re +import tempfile import threading import time from collections import Counter, defaultdict from pathlib import Path -from typing import Any, Dict, List, Optional, Union +from typing import Any, Dict, List, Optional, Set, Tuple, Union from openviking.parse.base import ( NodeType, @@ -31,7 +32,6 @@ lazy_import, ) from openviking.parse.parsers.base_parser import BaseParser -from openviking.utils.time_utils import parse_iso_datetime from openviking_cli.utils import get_logger from openviking_cli.utils.config.parser_config import PDFConfig @@ -53,7 +53,7 @@ class PDFParser(BaseParser): Strategies: - "local": Use pdfplumber for text and table extraction - "mineru": Use MinerU API for advanced PDF processing - - "auto": Try local first, fallback to MinerU if configured + - "auto": Use the configured remote parser if there is one, else local Examples: >>> # Local parsing @@ -79,6 +79,17 @@ def __init__(self, config: Optional[PDFConfig] = None): self.config = config or PDFConfig() self.config.validate() + # Remote-parse health, shared across documents parsed by this instance so a + # batch that hits a rate limit stops paying for doomed calls. + self._remote_failures: Dict[str, int] = {} + self._remote_circuit_open: Set[str] = set() + self._remote_skip_count: Dict[str, int] = {} + self._remote_circuit_cooldown = 20 + self._remote_circuit_threshold = 3 + self._remote_retry_on_transient = True + self._remote_retry_delay_seconds = 2.0 + self._degraded = 0 + # Lazy import MarkdownParser to avoid circular imports self._markdown_parser = None @@ -156,6 +167,16 @@ async def parse(self, source: Union[str, Path], instruction: str = "", **kwargs) result.parse_time = time.time() - start_time result.meta.update(conversion_meta) result.meta["pdf_strategy"] = self.config.strategy + # Under auto the requested strategy and the path actually taken can differ + # (the remote parser may be skipped or fail over to local), so record + # the effective one for operators. + effective = conversion_meta.get("strategy") + if effective: + result.meta["pdf_strategy_effective"] = effective + if self._degraded: + # Count documents that had to fall back, so a batch that silently + # stopped using the remote parser is visible after the fact. + result.meta["pdf_remote_degraded"] = self._degraded result.meta["intermediate_markdown_length"] = len(markdown_content) result.meta["intermediate_markdown_preview"] = markdown_content[:500] @@ -200,27 +221,91 @@ async def _convert_to_markdown( return await self._convert_local(pdf_path, resource_name=resource_name) elif self.config.strategy == "mineru": + # Strict: a MinerU failure is the result, not a reason to try something else. return await self._convert_mineru(pdf_path, resource_name=resource_name) elif self.config.strategy == "auto": - # Try local first - try: - return await self._convert_local(pdf_path, resource_name=resource_name) - except Exception as e: - logger.warning(f"Local conversion failed: {e}") - - # Fallback to MinerU if configured - if self.config.mineru_endpoint: - logger.info("Falling back to MinerU API") - return await self._convert_mineru(pdf_path, resource_name=resource_name) - else: - raise ValueError( - f"Local conversion failed and no MinerU endpoint configured: {e}" + # Remote services that are configured get tried first, in priority order, + # and the local parser is always the last resort. Adding a new external + # parser should mean adding a provider entry, not a new strategy value. + for name, convert in self._remote_converters(): + if name in self._remote_circuit_open: + # Stay off the remote for a while, then let one document through + # as a probe: a service that recovers should be used again, and + # one that is still down costs a single call, not the batch. + self._remote_skip_count[name] = self._remote_skip_count.get(name, 0) + 1 + if self._remote_skip_count[name] < self._remote_circuit_cooldown: + self._degraded += 1 + logger.debug( + f"{name} still circuit-open " + f"({self._remote_skip_count[name]}/{self._remote_circuit_cooldown})" + ) + continue + logger.info(f"{name} half-open, probing with this document") + try: + result = await self._convert_remote_with_retry( + name, convert, pdf_path, resource_name=resource_name ) + except Exception as e: + self._remote_failures[name] = self._remote_failures.get(name, 0) + 1 + if self._remote_failures[name] >= self._remote_circuit_threshold: + self._remote_circuit_open.add(name) + logger.warning( + f"{name} failed {self._remote_failures[name]} times in a row " + f"({e}); routing the rest of this batch to local" + ) + self._degraded += 1 + logger.warning(f"{name} conversion failed, trying the next option: {e}") + continue + self._remote_failures.pop(name, None) + self._remote_skip_count.pop(name, None) + self._remote_circuit_open.discard(name) + return result + return await self._convert_local(pdf_path, resource_name=resource_name) else: raise ValueError(f"Unknown strategy: {self.config.strategy}") + async def _convert_remote_with_retry( + self, + name: str, + convert, + pdf_path: Path, + resource_name: Optional[str] = None, + ): + """Call a remote converter, retrying the failures that are worth retrying. + + A 429 or a transient 5xx usually clears on its own, so one backed-off retry + is cheaper than degrading the document to local. A 404 or a bad request will + not clear, so it goes straight through and the circuit breaker takes over. + """ + attempts = 2 if self._remote_retry_on_transient else 1 + for attempt in range(attempts): + try: + return await convert(pdf_path, resource_name=resource_name) + except Exception as e: + if attempt + 1 >= attempts or not self._is_transient_remote_error(e): + raise + delay = self._remote_retry_delay_seconds + logger.info( + f"{name} transient failure (HTTP {PDFParser._http_status(e)}); retrying in {delay}s" + ) + await asyncio.sleep(delay) + raise RuntimeError(f"{name} exhausted retries") # pragma: no cover + + def _remote_converters(self) -> List[Tuple[str, Any]]: + """External converters this config enables, in the order they should be tried. + + Only services that are actually configured participate: an unset endpoint + means the service is not in play, so ``auto`` degrades to local on its own + without needing a separate strategy value. Registering another provider here + is the only change a new remote parser requires. + """ + providers: List[Tuple[str, Any]] = [] + if self.config.mineru_enabled: + providers.append(("MinerU", self._convert_mineru)) + return providers + async def _convert_local( self, pdf_path: Path, storage=None, resource_name: Optional[str] = None ) -> tuple[str, Dict[str, Any]]: @@ -686,86 +771,160 @@ def _extract_image_from_page(self, page, img_info: dict) -> Optional[bytes]: logger.debug(f"Image extraction error: {e}") return None - async def _convert_mineru( - self, - pdf_path: Path, - storage=None, - resource_name: Optional[str] = None, - ) -> tuple[str, Dict[str, Any]]: + def _mineru_headers(self) -> Dict[str, str]: + """Auth headers for the MinerU request. + + The hosted service authenticates with a bearer token; a self-hosted + deployment usually needs none, so an unset token yields no headers. """ - Convert PDF to Markdown using the MinerU /file_parse API. + if self.config.mineru_token: + return {"Authorization": f"Bearer {self.config.mineru_token}"} + return {} - Args: - pdf_path: Path to PDF file - storage: Media storage used to persist extracted images; defaults to - the configured storage if None - resource_name: Resource name under which extracted images are saved; - defaults to the PDF stem + def _mineru_base_url(self) -> str: + return self.config.mineru_endpoint.rstrip("/") - Returns: - Tuple of (markdown_content, metadata) where metadata includes - strategy, endpoint, api_version, backend, task_id, processing_time - (seconds) and images_saved + @staticmethod + def _is_transient_remote_error(exc: Exception) -> bool: + """Whether retrying could plausibly succeed. - Raises: - ValueError: If MinerU endpoint is not configured, or the task does - not complete - Exception: If the API call fails + Rate limiting and server-side faults are transient; a missing route or a + malformed request fails the same way every time, so retrying it would only + burn quota and latency. """ - httpx = lazy_import("httpx") + status = PDFParser._http_status(exc) + if status is None: + return False + return status == 429 or 500 <= status < 600 + + def _http_status(exc: Exception) -> Optional[int]: + response = getattr(exc, "response", None) + return getattr(response, "status_code", None) + + async def _mineru_post_json(self, client, url: str, payload: Dict[str, Any]) -> Dict[str, Any]: + """POST a JSON body (the hosted MinerU v4 API takes no multipart at all).""" + response = await client.post(url, json=payload, headers=self._mineru_headers()) + response.raise_for_status() + result = response.json() + code = result.get("code") + # The hosted API answers HTTP 200 with a non-zero business code on failure. + if isinstance(code, int) and code != 0: + raise RuntimeError(f"MinerU API {url} returned code={code}: {result.get('msg')!r}") + return result + + async def _mineru_upload_file(self, client, file_url: str, pdf_path: Path) -> None: + """PUT the raw PDF to the pre-signed OSS url. + + MinerU's docs are explicit: no Content-Type header on this request. + """ + with open(pdf_path, "rb") as f: + payload = f.read() + response = await client.put(file_url, content=payload, headers={"Content-Type": ""}) + response.raise_for_status() + logger.info(f"Uploaded {pdf_path.name} to MinerU staging ({len(payload)} bytes)") + + async def _mineru_submit_file_batch(self, client, base: str, pdf_path: Path) -> Dict[str, Any]: + """Hosted v4 upload handshake: reserve an OSS url, then PUT the file. + + ``POST /file-urls/batch`` (JSON) -> ``data.batch_id`` + ``data.file_urls``; + the parse task is submitted automatically once the upload lands, so the + returned body can be driven by the regular online-batch poller. + """ + payload: Dict[str, Any] = {"model_version": "vlm"} + payload.update(self.config.mineru_bodys or {}) + payload["files"] = [{"name": pdf_path.name}] + result = await self._mineru_post_json(client, f"{base}/file-urls/batch", payload) + data = self._mineru_payload_data(result) + urls = data.get("file_urls") + if not isinstance(urls, list) or not urls: + raise ValueError(f"MinerU {base}/file-urls/batch returned no file_urls: {result!r}") + await self._mineru_upload_file(client, urls[0], pdf_path) + return result - if not self.config.mineru_endpoint: - raise ValueError("MinerU endpoint not configured") + @staticmethod + def _mineru_online_result_entry(body: Dict[str, Any]) -> Optional[Dict[str, Any]]: + data = PDFParser._mineru_payload_data(body) + entries = data.get("extract_result") + if isinstance(entries, list) and entries and isinstance(entries[0], dict): + return entries[0] + return None - # mineru_endpoint is a base URL; append /file_parse unless it already ends with it - base = self.config.mineru_endpoint.rstrip("/") - url = base if base.endswith("/file_parse") else f"{base}/file_parse" + @classmethod + def _mineru_online_done(cls, body: Dict[str, Any]) -> bool: + entry = cls._mineru_online_result_entry(body) + return bool(entry) and entry.get("state") == "done" + + @classmethod + def _mineru_online_busy(cls, body: Dict[str, Any]) -> bool: + entry = cls._mineru_online_result_entry(body) + # "waiting-file" is the documented state for the file-urls/batch upload + # handshake (scanning the freshly PUT object). "converting" is the format + # export step that follows "done" parsing. Both are still in flight. + return bool(entry) and entry.get("state") in ( + "waiting", + "waiting-file", + "uploading", + "pending", + "running", + "converting", + ) - meta = { - "strategy": "mineru", - "endpoint": self.config.mineru_endpoint, - "api_version": None, - } + @classmethod + def _mineru_online_failure_detail(cls, body: Dict[str, Any]) -> str: + entry = cls._mineru_online_result_entry(body) or {} + for key in ("err_msg", "err_no", "state", "message", "msg"): + if entry.get(key): + return f"{key}={entry[key]}" + return cls._mineru_failure_detail(body) + + @classmethod + def _mineru_online_zip_url(cls, body: Dict[str, Any]) -> Optional[str]: + entry = cls._mineru_online_result_entry(body) or {} + for key in ("full_zip_url", "zip_url", "result_url"): + value = entry.get(key) + if isinstance(value, str) and value.startswith(("http://", "https://")): + return value + return None - try: - async with httpx.AsyncClient(timeout=self.config.mineru_timeout) as client: - # Prepare file upload - with open(pdf_path, "rb") as f: - files = {"files": (pdf_path.name, f, "application/pdf")} - - # Prepare Form fields - data: Dict[str, Any] = dict(self.config.mineru_bodys or {}) - - # MinerU must return extracted images for the markdown refs. - data["return_images"] = True - - # Make API request - logger.info(f"Calling MinerU API: {url}") - response = await client.post( - url, - files=files, - data=data, - ) - response.raise_for_status() - - # Parse response - result = response.json() - if result.get("status") != "completed": - raise ValueError(f"MinerU task not completed: {result.get('status')}") - - results = result.get("results") or {} - file_result = results.get(pdf_path.name) or next(iter(results.values()), {}) - markdown_content = file_result.get("md_content") or "" - - # Extract metadata from response - meta["api_version"] = result.get("version") - meta["backend"] = result.get("backend") - meta["task_id"] = result.get("task_id") - started_at, completed_at = result.get("started_at"), result.get("completed_at") - if started_at and completed_at: - meta["processing_time"] = ( - parse_iso_datetime(completed_at) - parse_iso_datetime(started_at) - ).total_seconds() + async def _mineru_download_zip(self, client, zip_url: str) -> bytes: + """Download the result archive (signed urls may already carry auth).""" + response = await client.get( + zip_url, + headers=self._mineru_headers(), + follow_redirects=True, + ) + response.raise_for_status() + return response.content + + def _mineru_unpack_zip( + self, + zip_bytes: bytes, + pdf_path: Path, + meta: Dict[str, Any], + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Unpack the zip result: locate the markdown and persist images.""" + import zipfile + + from openviking.utils.zip_safe import normalize_zip_filenames, safe_extract_zip + + with tempfile.TemporaryDirectory(prefix="mineru-") as tmp: + tmp_dir = Path(tmp) + zf = zipfile.ZipFile(io.BytesIO(zip_bytes)) + normalize_zip_filenames(zf) + safe_extract_zip(zf, tmp_dir) + zf.close() + + md_path = self._mineru_pick_markdown(tmp_dir) + if md_path is None: + raise ValueError( + f"No markdown file found in MinerU zip result extracted to {tmp_dir}" + ) + markdown_content = md_path.read_text(encoding="utf-8", errors="replace") + root_dir = md_path.parent + + meta["api_version"] = "task-zip" if not markdown_content: logger.warning(f"MinerU returned empty content for {pdf_path}") @@ -779,42 +938,284 @@ async def _convert_mineru( if resource_name is None: resource_name = pdf_path.stem - # MinerU embeds images as base64 data-URLs, referenced from markdown - # as `images/`; save them into the media store and rewrite - # the references to the stored relative paths. + # Persist every shipped image and rewrite the markdown references repl: Dict[str, str] = {} media_dir = storage.media_dir - for img_name, data_url in (file_result.get("images") or {}).items(): - try: - # data URL form: "data:image/jpeg;base64," - image_bytes = base64.b64decode(data_url.split(",", 1)[-1]) - img_path = Path(img_name) - image_path = storage.save_image( - resource_name, - image_bytes, - filename=img_path.stem, - extension=img_path.suffix or ".png", - ) - repl[f"images/{img_name}"] = image_path.relative_to(media_dir).as_posix() - except Exception as img_err: - logger.warning(f"Failed to save MinerU image {img_name}: {img_err}") + images_dir = root_dir / "images" + if images_dir.is_dir(): + for img in sorted(images_dir.iterdir()): + if not img.is_file(): + continue + try: + image_path = storage.save_image( + resource_name, + img.read_bytes(), + filename=img.stem, + extension=img.suffix or ".png", + ) + repl[f"images/{img.name}"] = image_path.relative_to(media_dir).as_posix() + except Exception as img_err: + logger.warning(f"Failed to save MinerU image {img.name}: {img_err}") if repl: - # Single pass over the markdown replaces every known reference; - # unknown ``images/...`` text is left untouched. ref_pattern = re.compile( "|".join(re.escape(ref) for ref in sorted(repl, key=len, reverse=True)) ) markdown_content = ref_pattern.sub(lambda m: repl[m.group(0)], markdown_content) meta["images_saved"] = len(repl) - logger.info(f"MinerU conversion: {len(markdown_content)} chars") + return markdown_content, meta - return markdown_content, meta + @staticmethod + def _mineru_pick_markdown(root_dir: Path) -> Optional[Path]: + """Pick the document markdown from an extracted MinerU zip.""" + candidates = [p for p in root_dir.rglob("*.md") if p.is_file()] + if not candidates: + return None + for p in candidates: + if p.name == "full.md": + return p + if len(candidates) == 1: + return candidates[0] + return min(candidates, key=lambda p: (len(p.parts), -p.stat().st_size)) + + def _mineru_file_parse_url(self) -> str: + """Upload URL for the self-hosted contract: ``/file_parse``.""" + base = self._mineru_base_url() + return base if base.endswith("/file_parse") else f"{base}/file_parse" + + async def _mineru_post( + self, client, url: str, pdf_path: Path, data: Dict[str, Any] + ) -> Dict[str, Any]: + """POST the PDF as multipart and return the decoded JSON body.""" + with open(pdf_path, "rb") as f: + files = {"files": (pdf_path.name, f, "application/pdf")} + response = await client.post( + url, files=files, data=data, headers=self._mineru_headers() + ) + response.raise_for_status() + return response.json() - except Exception as e: - logger.error(f"MinerU API call failed: {e}") - raise + async def _mineru_get_json(self, client, url: str) -> Dict[str, Any]: + response = await client.get(url, headers=self._mineru_headers()) + response.raise_for_status() + return response.json() + + @staticmethod + def _mineru_payload_data(result: Dict[str, Any]) -> Dict[str, Any]: + """Unwrap the ``data`` envelope the hosted API uses, if present.""" + if isinstance(result, dict) and isinstance(result.get("data"), dict): + return result["data"] + return result + + @staticmethod + def _mineru_sync_entry(payload: Dict[str, Any], pdf_path: Path) -> Optional[Dict[str, Any]]: + """Pull this PDF's entry out of a ``/file_parse`` response.""" + results = payload.get("results") or {} + if not isinstance(results, dict): + return None + entry = results.get(pdf_path.name) or next(iter(results.values()), None) + return entry if isinstance(entry, dict) else None + + def _mineru_extract_markdown(self, markdown: Path, root: Path) -> tuple[str, Dict[str, bytes]]: + """Read markdown from an unpacked archive and collect its images.""" + text = markdown.read_text(encoding="utf-8", errors="replace") + images: Dict[str, bytes] = {} + for candidate in sorted(root.rglob("*")): + if candidate.is_file() and candidate.suffix.lower() in ( + ".png", + ".jpg", + ".jpeg", + ".gif", + ".webp", + ".bmp", + ): + images[candidate.relative_to(root).as_posix()] = candidate.read_bytes() + return text, images + + @staticmethod + def _mineru_decode_inline_images(images: Dict[str, Any]) -> Dict[str, bytes]: + """Decode the base64 data-URL images a self-hosted reply embeds.""" + out: Dict[str, bytes] = {} + for name, value in (images or {}).items(): + try: + out[name] = base64.b64decode(str(value).split(",", 1)[-1]) + except Exception: + continue + return out + + async def _mineru_save_images( + self, + images: Dict[str, bytes], + storage, + resource_name: Optional[str], + ) -> int: + """Persist images into the media store, rewriting markdown references.""" + if not images: + return 0 + + if storage is None: + from openviking_cli.utils.storage import get_storage + + storage = get_storage() + + repl: Dict[str, str] = {} + media_dir = storage.media_dir + for name, blob in images.items(): + try: + img_path = Path(name) + stored = storage.save_image( + resource_name, + blob, + filename=img_path.stem, + extension=img_path.suffix or ".png", + ) + repl[name] = stored.relative_to(media_dir).as_posix() + except Exception as img_err: + logger.warning(f"Failed to save MinerU image {name}: {img_err}") + + return len(repl) + + async def _convert_mineru( + self, + pdf_path: Path, + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Convert a PDF to Markdown with the configured MinerU endpoint. + + A ``mineru_token`` is the signal that the endpoint is the hosted service: + that API has no synchronous ``/file_parse`` route and only speaks the + asynchronous batch contract, so we go straight to it rather than probing a + route that cannot exist. Without a token the endpoint is treated as + self-hosted and keeps using ``/file_parse``, unchanged. + + Either path raises on failure; ``strategy="auto"`` decides whether a failure + is tolerated. + """ + httpx = lazy_import("httpx") + + if not self.config.mineru_endpoint: + raise ValueError("MinerU endpoint not configured") + + meta: Dict[str, Any] = { + "strategy": "mineru", + "endpoint": self.config.mineru_endpoint, + } + + async with httpx.AsyncClient(timeout=self.config.mineru_timeout) as client: + if self.config.mineru_token: + return await self._mineru_hosted_batch( + client, + meta, + pdf_path, + storage=storage, + resource_name=resource_name, + ) + return await self._mineru_selfhosted( + client, + meta, + pdf_path, + storage=storage, + resource_name=resource_name, + ) + + async def _mineru_hosted_batch( + self, + client, + meta: Dict[str, Any], + pdf_path: Path, + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Hosted, token-authenticated MinerU: reserve url, upload, poll, unpack. + + ``POST /file-urls/batch`` -> ``PUT`` the PDF at the signed url -> + ``GET /extract-results/batch/{batch_id}`` until the state settles -> + download ``full_zip_url`` and take the markdown plus its images. + """ + base = self._mineru_base_url() + meta["api_mode"] = "v4-batch" + + try: + submitted = await self._mineru_submit_file_batch(client, base, pdf_path) + except Exception: + # A rejected or expired token can still sit in front of a working + # self-hosted endpoint, so give the legacy route one chance first. + logger.info( + "MinerU %s/file-urls/batch unavailable; trying the self-hosted route", + base, + ) + return await self._mineru_selfhosted( + client, + meta, + pdf_path, + storage=storage, + resource_name=resource_name, + ) + + batch_id = self._mineru_payload_data(submitted).get("batch_id") + meta["batch_id"] = batch_id + + waited = 0.0 + interval = 5.0 + zip_url = None + while waited < self.config.mineru_timeout: + body = await self._mineru_get_json(client, f"{base}/extract-results/batch/{batch_id}") + data = self._mineru_payload_data(body) + if self._mineru_online_done(data): + zip_url = self._mineru_online_zip_url(data) + if not zip_url: + raise ValueError( + f"MinerU batch {batch_id} finished without a result zip: " + f"{self._mineru_online_failure_detail(data)}" + ) + break + if not self._mineru_online_busy(data): + raise ValueError( + f"MinerU batch {batch_id} failed: {self._mineru_online_failure_detail(data)}" + ) + await asyncio.sleep(interval) + waited += interval + + if not zip_url: + raise TimeoutError( + f"MinerU batch {batch_id} did not finish within {self.config.mineru_timeout}s" + ) + + meta["result_url"] = zip_url + blob = await self._mineru_download_zip(client, zip_url) + text, meta = self._mineru_unpack_zip( + blob, pdf_path, meta, storage=storage, resource_name=resource_name + ) + logger.info(f"MinerU conversion: {len(text)} chars") + return text, meta + + async def _mineru_selfhosted( + self, + client, + meta: Dict[str, Any], + pdf_path: Path, + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Self-hosted MinerU: the single-shot ``/file_parse`` contract.""" + url = self._mineru_file_parse_url() + data: Dict[str, Any] = dict(self.config.mineru_bodys or {}) + # MinerU must return extracted images for the markdown refs to resolve. + data["return_images"] = True + meta["api_mode"] = "file-parse" + + result = await self._mineru_post(client, url, pdf_path, data) + entry = self._mineru_sync_entry(self._mineru_payload_data(result), pdf_path) + if entry is None: + raise ValueError(f"MinerU {url} returned no markdown for {pdf_path.name}") + + text = entry.get("md_content") or "" + images = self._mineru_decode_inline_images(entry.get("images") or {}) + meta["images_saved"] = await self._mineru_save_images(images, storage, resource_name) + logger.info(f"MinerU conversion: {len(text)} chars") + return text, meta def _format_table_markdown(self, table: List[List[Optional[str]]]) -> str: """ diff --git a/openviking/service/core.py b/openviking/service/core.py index 434ed1b64b..d585cd9b1b 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -654,7 +654,7 @@ async def initialize(self) -> None: # first PDF import. Required for strategy="mineru"; advisory for "auto". pdf_config = self._config.pdf should_preflight_mineru = pdf_config.strategy == "mineru" or ( - pdf_config.strategy == "auto" and pdf_config.mineru_endpoint is not None + pdf_config.strategy == "auto" and pdf_config.remote_enabled ) if should_preflight_mineru and pdf_config.mineru_endpoint: diff --git a/openviking_cli/utils/config/parser_config.py b/openviking_cli/utils/config/parser_config.py index 26cfcd99ce..56741c2529 100644 --- a/openviking_cli/utils/config/parser_config.py +++ b/openviking_cli/utils/config/parser_config.py @@ -135,13 +135,14 @@ class PDFConfig(ParserConfig): Supports three strategies: - "local": Use pdfplumber for local PDF→Markdown conversion - "mineru": Use MinerU API for remote PDF→Markdown conversion - - "auto": Try local first, fallback to MinerU if available + - "auto": Use the configured remote parser if there is one, else local Attributes: strategy: Parsing strategy ("local" | "mineru" | "auto") mineru_endpoint: MinerU API endpoint URL mineru_timeout: MinerU request timeout in seconds mineru_bodys: Additional MinerU API multipart form fields + mineru_token: Optional bearer token for the hosted MinerU service """ strategy: str = "auto" # "local" | "mineru" | "auto" @@ -150,6 +151,7 @@ class PDFConfig(ParserConfig): mineru_endpoint: Optional[str] = None # API endpoint URL mineru_timeout: float = 300.0 # Request timeout in seconds (5 minutes) mineru_bodys: Optional[dict] = None # Additional API multipart form fields + mineru_token: Optional[str] = None # Bearer token for the hosted MinerU service # Heading detection configuration heading_detection: str = "auto" # "bookmarks" | "font" | "auto" | "none" @@ -159,6 +161,29 @@ class PDFConfig(ParserConfig): # Image extraction configuration image_resolution: int = 300 # Rendering DPI for extracted image regions + @property + def mineru_enabled(self) -> bool: + """Whether the MinerU provider is configured. + + This is only the MinerU provider's own eligibility check. Use + :attr:`remote_enabled` when asking whether *any* remote parser is in play. + """ + return bool(self.mineru_endpoint) + + @property + def remote_enabled(self) -> bool: + """Whether at least one remote parser provider is configured. + + Written in terms of providers rather than any single service, so adding a + new remote parser does not require revisiting the ``auto`` dispatch, the + startup preflight, or the documentation. Keep this in step with the + provider list built by ``PDFParser._remote_converters``. + """ + return any( + self.mineru_endpoint, # MinerU provider + # future providers register their own config here + ) + def validate(self) -> None: """ Validate configuration. diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py new file mode 100644 index 0000000000..1f06ae3491 --- /dev/null +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -0,0 +1,465 @@ +"""Tests for PDF strategy dispatch. + +``auto`` uses whichever remote parser is configured and falls back to the local one, +so the contract worth protecting is: + +- with a remote service configured, it is tried *before* local; +- if it fails, the document is still converted, by local; +- with nothing configured, local runs and no remote call is made. + +``mineru`` stays strict: a failure is the result. +""" + +import asyncio +from pathlib import Path + +import httpx +import pytest + +from openviking.parse.parsers.pdf import PDFParser +from openviking_cli.utils.config.parser_config import PDFConfig + +ENDPOINT = "http://127.0.0.1:8000" +MINERU_MD = "# From MinerU" +LOCAL_MD = "# From local pdfplumber" + + +@pytest.fixture +def pdf_path(tmp_path) -> Path: + p = tmp_path / "doc.pdf" + p.write_bytes(b"%PDF-1.4 minimal") + return p + + +def _parser(**overrides) -> PDFParser: + defaults = {"strategy": "auto"} + defaults.update(overrides) + return PDFParser(PDFConfig(**defaults)) + + +def _stub_converters(monkeypatch, parser, *, mineru): + """Replace both conversion paths so the test exercises dispatch, not HTTP.""" + calls = {"mineru": 0, "local": 0} + + async def fake_mineru(self, pdf_path, storage=None, resource_name=None): + calls["mineru"] += 1 + if isinstance(mineru, Exception): + raise mineru + return mineru + + async def fake_local(self, pdf_path, storage=None, resource_name=None): + calls["local"] += 1 + return LOCAL_MD, {"strategy": "local"} + + monkeypatch.setattr(PDFParser, "_convert_mineru", fake_mineru) + monkeypatch.setattr(PDFParser, "_convert_local", fake_local) + return calls + + +# --------------------------------------------------------------------------- +# auto: configured remote first, local as the fallback +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_auto_prefers_configured_mineru(monkeypatch, pdf_path): + parser = _parser(mineru_endpoint=ENDPOINT) + calls = _stub_converters(monkeypatch, parser, mineru=(MINERU_MD, {"strategy": "mineru"})) + + md, meta = await parser._convert_to_markdown(pdf_path) + + assert md == MINERU_MD + assert meta["strategy"] == "mineru" + # Remote first means local is not even called when the remote succeeds. + assert calls == {"mineru": 1, "local": 0} + + +@pytest.mark.asyncio +async def test_auto_falls_back_to_local_when_mineru_raises(monkeypatch, pdf_path): + """The behaviour that matters: a remote failure degrades, it does not fail.""" + parser = _parser(mineru_endpoint=ENDPOINT) + calls = _stub_converters( + monkeypatch, parser, mineru=RuntimeError("MinerU endpoint unreachable") + ) + + md, meta = await parser._convert_to_markdown(pdf_path) + + assert md == LOCAL_MD + assert meta["strategy"] == "local" + assert calls == {"mineru": 1, "local": 1} + + +@pytest.mark.asyncio +async def test_auto_falls_back_to_local_on_timeout(monkeypatch, pdf_path): + parser = _parser(mineru_endpoint=ENDPOINT) + _stub_converters(monkeypatch, parser, mineru=TimeoutError("timed out")) + + md, meta = await parser._convert_to_markdown(pdf_path) + + assert md == LOCAL_MD + assert meta["strategy"] == "local" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "status", + [404, 405, 429, 500, 503], +) +def test_auto_falls_back_on_any_http_error(monkeypatch, pdf_path, status): + """Every remote failure mode must degrade to local, not fail the import. + + 404/405 matter in practice: the hosted MinerU service has no ``/file_parse`` + route at all (its contract is the async batch flow), so a user who points + ``mineru_endpoint`` at it under ``auto`` hits 404 on every single document. + """ + import httpx as _httpx + + parser = _parser(mineru_endpoint=ENDPOINT) + + async def boom(self, pdf_path, storage=None, resource_name=None): + raise _httpx.HTTPStatusError(f"HTTP {status}", request=None, response=None) + + async def ok_local(self, pdf_path, storage=None, resource_name=None): + return LOCAL_MD, {"strategy": "local"} + + monkeypatch.setattr(PDFParser, "_convert_mineru", boom) + monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + + md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) + + assert md == LOCAL_MD + assert meta["strategy"] == "local" + + +@pytest.mark.asyncio +async def test_auto_uses_local_when_nothing_configured(monkeypatch, pdf_path): + """No endpoint means no remote service is in play; local must not be bypassed.""" + parser = _parser() + calls = _stub_converters(monkeypatch, parser, mineru=(MINERU_MD, {"strategy": "mineru"})) + + md, meta = await parser._convert_to_markdown(pdf_path) + + assert md == LOCAL_MD + assert calls == {"mineru": 0, "local": 1} + + +def test_remote_converters_only_lists_configured_services(): + assert _parser()._remote_converters() == [] + providers = _parser(mineru_endpoint=ENDPOINT)._remote_converters() + assert [name for name, _ in providers] == ["MinerU"] + + +def test_mineru_enabled_is_the_single_source_of_truth(): + """Both the parser dispatch and the startup preflight read this one predicate.""" + assert PDFConfig(strategy="auto").mineru_enabled is False + assert PDFConfig(strategy="auto", mineru_endpoint="").mineru_enabled is False + assert PDFConfig(strategy="auto", mineru_endpoint=ENDPOINT).mineru_enabled is True + # A token alone does not enable remote parsing: it is a credential, not an endpoint. + assert PDFConfig(strategy="auto", mineru_token="t").mineru_enabled is False + + +@pytest.mark.asyncio +async def test_local_strategy_never_calls_the_remote(monkeypatch, pdf_path): + parser = _parser(strategy="local", mineru_endpoint=ENDPOINT) + calls = _stub_converters(monkeypatch, parser, mineru=(MINERU_MD, {"strategy": "mineru"})) + + md, _ = await parser._convert_to_markdown(pdf_path) + + assert md == LOCAL_MD + assert calls == {"mineru": 0, "local": 1} + + +# --------------------------------------------------------------------------- +# mineru: strict +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_mineru_strategy_does_not_fall_back(monkeypatch, pdf_path): + parser = _parser(strategy="mineru", mineru_endpoint=ENDPOINT) + calls = _stub_converters( + monkeypatch, parser, mineru=RuntimeError("MinerU endpoint unreachable") + ) + + with pytest.raises(RuntimeError): + await parser._convert_to_markdown(pdf_path) + + assert calls == {"mineru": 1, "local": 0} + + +# --------------------------------------------------------------------------- +# config validation +# --------------------------------------------------------------------------- + + +def test_strategy_is_accepted(): + for strategy in ("local", "auto"): + PDFConfig(strategy=strategy).validate() + PDFConfig(strategy="mineru", mineru_endpoint=ENDPOINT).validate() + + +def test_endpoint_is_required_for_mineru(): + with pytest.raises(ValueError, match="mineru_endpoint"): + PDFConfig(strategy="mineru").validate() + + +def test_endpoint_is_optional_for_auto(): + PDFConfig(strategy="auto").validate() + + +def test_unknown_strategy_is_rejected(): + with pytest.raises(ValueError, match="Invalid strategy"): + PDFConfig(strategy="mineru_first", mineru_endpoint=ENDPOINT).validate() + + +# --------------------------------------------------------------------------- +# hosted-service auth +# --------------------------------------------------------------------------- + + +def test_token_is_sent_as_bearer(monkeypatch, pdf_path): + """The hosted service needs a bearer token; verify it reaches the request.""" + captured = {} + + def fake_post(self, url, **kwargs): + captured["url"] = url + captured["headers"] = kwargs.get("headers") or {} + raise RuntimeError("stop here - only the request shape matters") + + monkeypatch.setattr(httpx.AsyncClient, "post", fake_post) + + parser = _parser(strategy="mineru", mineru_endpoint=ENDPOINT, mineru_token="secret-token") + assert parser._mineru_headers() == {"Authorization": "Bearer secret-token"} + + # _convert_mineru logs and re-raises transport failures; we only care that the + # call went out carrying the token. + with pytest.raises(RuntimeError, match="only the request shape matters"): + asyncio.run(parser._convert_mineru(pdf_path)) + + assert captured["headers"] == {"Authorization": "Bearer secret-token"} + assert captured["url"].endswith("/file_parse") + + +def test_no_token_means_no_auth_header(): + parser = _parser(mineru_endpoint=ENDPOINT) + assert parser._mineru_headers() == {} + + +# --------------------------------------------------------------------------- +# protocol selection: a token means the hosted API, no token means self-hosted +# --------------------------------------------------------------------------- + + +def test_token_selects_the_hosted_batch_protocol(monkeypatch): + """The hosted service has no /file_parse, so a token must pick the batch flow.""" + seen = [] + + async def fake_batch(self, client, meta, pdf_path, storage=None, resource_name=None): + seen.append("batch") + return MINERU_MD, {"strategy": "mineru", "api_mode": "v4-batch"} + + async def fake_sync(self, client, meta, pdf_path, storage=None, resource_name=None): + seen.append("sync") + return MINERU_MD, {"strategy": "mineru", "api_mode": "file-parse"} + + monkeypatch.setattr(PDFParser, "_mineru_hosted_batch", fake_batch) + monkeypatch.setattr(PDFParser, "_mineru_selfhosted", fake_sync) + + parser = _parser(strategy="mineru", mineru_endpoint=ENDPOINT, mineru_token="t") + _, meta = asyncio.run(parser._convert_to_markdown(Path("x.pdf"))) + + assert seen == ["batch"] + assert meta["api_mode"] == "v4-batch" + + +def test_no_token_keeps_the_self_hosted_route(monkeypatch): + """Backward compatibility: a tokenless endpoint keeps using /file_parse.""" + seen = [] + + async def fake_batch(self, client, meta, pdf_path, storage=None, resource_name=None): + seen.append("batch") + return MINERU_MD, {"strategy": "mineru"} + + async def fake_sync(self, client, meta, pdf_path, storage=None, resource_name=None): + seen.append("sync") + return MINERU_MD, {"strategy": "mineru", "api_mode": "file-parse"} + + monkeypatch.setattr(PDFParser, "_mineru_hosted_batch", fake_batch) + monkeypatch.setattr(PDFParser, "_mineru_selfhosted", fake_sync) + + parser = _parser(strategy="mineru", mineru_endpoint=ENDPOINT) + _, meta = asyncio.run(parser._convert_to_markdown(Path("x.pdf"))) + + assert seen == ["sync"] + assert meta["api_mode"] == "file-parse" + + +def test_hosted_batch_failure_falls_back_to_self_hosted(monkeypatch): + """A rejected token must not sink the parse if the endpoint still speaks v1.""" + seen = [] + + async def refuse(self, client, base, pdf_path): + raise RuntimeError("token rejected") + + async def fake_sync(self, client, meta, pdf_path, storage=None, resource_name=None): + seen.append("sync") + return MINERU_MD, {"strategy": "mineru", "api_mode": "file-parse"} + + monkeypatch.setattr(PDFParser, "_mineru_submit_file_batch", refuse) + monkeypatch.setattr(PDFParser, "_mineru_selfhosted", fake_sync) + + parser = _parser(strategy="mineru", mineru_endpoint=ENDPOINT, mineru_token="t") + _, meta = asyncio.run(parser._convert_to_markdown(Path("x.pdf"))) + + assert seen == ["sync"] + assert meta["api_mode"] == "file-parse" + + +async def _no_sleep(_seconds): + """Placeholder so retry backoff does not slow the suite down.""" + return None + + +# --------------------------------------------------------------------------- +# batch behaviour: partial remote failure, backoff, and circuit breaking +# --------------------------------------------------------------------------- + + +def _http_error(status): + """An error shaped like the real thing, with a response carrying the status.""" + import httpx as _httpx + + request = _httpx.Request("POST", ENDPOINT) + response = _httpx.Response(status, request=request) + return _httpx.HTTPStatusError(f"HTTP {status}", request=request, response=response) + + +def test_transient_errors_are_retried_then_fall_back(monkeypatch, pdf_path): + """A 429 is worth one backed-off retry; a 404 is not.""" + for status, expected_calls in ((429, 2), (500, 2), (404, 1)): + calls = [] + + async def flaky(self, p, storage=None, resource_name=None, _calls=calls, _status=status): + _calls.append(1) + raise _http_error(_status) + + async def ok_local(self, pdf_path, storage=None, resource_name=None): + return LOCAL_MD, {"strategy": "local"} + + async def call_flaky(self, p, storage=None, resource_name=None): + return await flaky(self, p, storage, resource_name) + + monkeypatch.setattr(PDFParser, "_convert_mineru", call_flaky) + monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + + parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) + md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) + + assert md == LOCAL_MD, status + assert meta["strategy"] == "local", status + assert len(calls) == expected_calls, (status, len(calls)) + + +def test_circuit_opens_after_repeated_failures(monkeypatch, pdf_path): + """Once the remote is clearly down, stop paying for a doomed call per document.""" + calls = [] + + async def always_fail(self, pdf_path, storage=None, resource_name=None): + calls.append(1) + raise _http_error(404) + + async def ok_local(self, pdf_path, storage=None, resource_name=None): + return LOCAL_MD, {"strategy": "local"} + + async def _stub_always_fail(self, p, storage=None, resource_name=None): + return await always_fail(self, p, storage, resource_name) + + monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_always_fail) + monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + + parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) + parser._remote_circuit_threshold = 3 + parser._remote_circuit_cooldown = 1000 # stay open for this test's batch + + for _ in range(10): + md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) + assert md == LOCAL_MD + assert meta["strategy"] == "local" + + # 3 documents opened the circuit; the other 7 never called the remote again. + assert len(calls) == 3, len(calls) + + +def test_circuit_recovers_after_a_success(monkeypatch, pdf_path): + """A single success closes the circuit again.""" + state = {"fail": True} + + async def flaky(self, pdf_path, storage=None, resource_name=None): + if state["fail"]: + raise _http_error(404) + return MINERU_MD, {"strategy": "mineru"} + + async def ok_local(self, pdf_path, storage=None, resource_name=None): + return LOCAL_MD, {"strategy": "local"} + + async def _stub_flaky(self, p, storage=None, resource_name=None): + return await flaky(self, p, storage, resource_name) + + monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_flaky) + monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + + parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) + parser._remote_circuit_threshold = 2 + parser._remote_circuit_cooldown = 1000 + + asyncio.run(parser._convert_to_markdown(pdf_path)) + asyncio.run(parser._convert_to_markdown(pdf_path)) + assert "MinerU" in parser._remote_circuit_open + + state["fail"] = False + parser._remote_circuit_cooldown = 1 # let the next document through as a probe + md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) + assert md == MINERU_MD + assert "MinerU" not in parser._remote_circuit_open + + +def test_batch_with_partial_remote_failures(monkeypatch, tmp_path): + """10 documents, 3 fail remotely: the batch finishes and stays auditable. + + This is the shape a real import hits -- a handful of bad documents must not take + down the run, and the operator needs to see that three of them degraded. + """ + failing = {"doc2.pdf", "doc5.pdf", "doc9.pdf"} + seen = [] + + async def sometimes(self, pdf_path, storage=None, resource_name=None): + seen.append(pdf_path.name) + if pdf_path.name in failing: + raise _http_error(400) + return MINERU_MD, {"strategy": "mineru"} + + async def ok_local(self, pdf_path, storage=None, resource_name=None): + return LOCAL_MD, {"strategy": "local"} + + async def _stub_sometimes(self, p, storage=None, resource_name=None): + return await sometimes(self, p, storage, resource_name) + + monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_sometimes) + monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + + parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) + effective = [] + + for i in range(10): + doc = tmp_path / f"doc{i}.pdf" + doc.write_bytes(b"%PDF-1.4 minimal") + _, meta = asyncio.run(parser._convert_to_markdown(doc)) + effective.append(meta["strategy"]) + + assert len(seen) == 10, seen + # Every document produced a result; the failures degraded rather than aborting. + assert effective.count("local") == len(failing) + assert effective.count("mineru") == 10 - len(failing) + # And the degradation is visible to operators. + assert parser._degraded == len(failing) From 32f625d82b191b4d9ec2002119aade2114cfbfb8 Mon Sep 17 00:00:00 2001 From: q2000s Date: Fri, 16 Oct 2026 02:36:57 +0800 Subject: [PATCH 2/7] =?UTF-8?q?=EF=BB=BFfix(pdf):=20scope=20auto=20fallbac?= =?UTF-8?q?k=20to=20the=20remote=20path=20and=20fix=20review=20regressions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review flagged that the circuit breaker, the transient-error retry, and the multi-provider degradation counter were scope creep for this PR, and that three real defects survived in the retained code. This narrows the change and fixes them. Removed: - circuit breaker (threshold, cooldown, half-open probe) and its state - one backed-off retry on 429/5xx plus the transient-error classifier - the pdf_remote_degraded counter and its metadata auto now simply tries each configured remote provider in turn, uses the first success, and degrades a failed provider to local for that document. Fixed: - remote_enabled called any(self.mineru_endpoint), so any(None) raised TypeError during service start-up for any deployment that never configured MinerU. It now folds over a real collection and skips unset entries. - The result archive was downloaded with the MinerU bearer token attached. full_zip_url is pre-signed and points at the storage host, so the credential was leaking to a third-party host; API auth headers are no longer sent there. - The self-hosted path saved images but returned the provider's markdown unchanged, so every image reference pointed at a path that does not exist locally. _mineru_save_images now returns the mapping it already computed and the caller rewrites the references. Tests: 26 targeted (47 with bookmark extraction), all passing. --- docs/en/about/02-changelog.md | 13 +- docs/en/guides/01-configuration.md | 16 +- docs/zh/about/02-changelog.md | 2 +- docs/zh/guides/01-configuration.md | 6 +- openviking/parse/parsers/pdf.py | 119 ++-------- openviking_cli/utils/config/parser_config.py | 11 +- tests/parse/test_pdf_strategy_dispatch.py | 229 +++++++++---------- 7 files changed, 154 insertions(+), 242 deletions(-) diff --git a/docs/en/about/02-changelog.md b/docs/en/about/02-changelog.md index 7f267553fe..708959624d 100644 --- a/docs/en/about/02-changelog.md +++ b/docs/en/about/02-changelog.md @@ -7,13 +7,12 @@ This changelog is automatically generated from [GitHub Releases](https://github. ## Unreleased -- **`auto` degrades cheaply when a remote parser is rate limited or down**: `429` and `5xx` - are retried once after a short backoff, permanent errors such as `404` skip the retry - and go straight to local, and a provider that fails repeatedly is circuit-broken so the - rest of the batch stops paying for a doomed call. The circuit half-opens after a - cooldown so a recovered service is picked up again. Documents that fell back are - counted in `pdf_remote_degraded`, and whether any remote parser is configured is now - `PDFConfig.remote_enabled`, expressed over the provider list rather than one service. +- **`auto` falls back to local on any remote parser failure**: each configured remote + provider is tried in turn and the first success wins; a provider that raises is logged + and the next option is tried, so a remote outage degrades that document to local instead + of failing the import. The path actually taken is recorded in `pdf_strategy_effective`, + and whether any remote parser is configured is now `PDFConfig.remote_enabled`, expressed + over the provider list rather than one service. - **MinerU hosted API support (`mineru_token`)**: a new `pdf.mineru_token` setting carries the bearer token for the hosted MinerU service and, because that service has no `/file_parse` route, its presence also selects the hosted batch protocol: diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index f97b643db7..0a55e1b188 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -1090,16 +1090,12 @@ provider list rather than any single service, so adding a provider does not mean revisiting `auto`, the preflight, or this page. `mineru_enabled` remains the MinerU provider's own check. -To keep a degraded batch cheap, `auto` does not retry blindly: - -- `429` and `5xx` are retried once after a short backoff, since they usually clear. -- `404` and other permanent errors go straight to local; retrying cannot help. -- After several consecutive failures the provider is circuit-broken: the rest of the - batch skips it instead of paying for a doomed call per document. The circuit - half-opens after a cooldown, so a recovered service is picked up again. -- Documents that fell back are counted in `pdf_remote_degraded`, alongside - `pdf_strategy_effective`, so a batch that quietly stopped using the remote parser is - visible after the fact. +`auto` tries each configured remote provider in turn and uses the first one that +succeeds. A provider that raises is logged and the next option is tried, so any +remote failure -- `404`, `405`, `429`, a timeout, a `5xx` -- degrades that document +to local rather than failing it. The path actually taken is recorded in +`pdf_strategy_effective`, so a batch that quietly stopped using the remote parser is +visible after the fact. The startup preflight follows the same rule: `mineru` fails fast if the endpoint is not ready, `auto` only warns and leaves the retry to the first parse. diff --git a/docs/zh/about/02-changelog.md b/docs/zh/about/02-changelog.md index 3b1c58b6da..09e6875cf7 100644 --- a/docs/zh/about/02-changelog.md +++ b/docs/zh/about/02-changelog.md @@ -7,7 +7,7 @@ OpenViking 的所有重要变更都将记录在此文件中。 ## 未发布 -- **`auto` 在远程解析服务被限流或崩溃时可低成本地降级**：`429` 与 `5xx` 会在一段退避后重试一次，`404` 等永久性错误不重试而直接降级，连续失败的 provider 会被熔断，使批次剩下文档不再为每份重复付出必败的调用。熔断在一段冷却后半开重试，服务恢复后会被重新使用。已降级的文档计入 `pdf_remote_degraded`；同时，“是否有远程解析服务在场”改由 `PDFConfig.remote_enabled` 判定，它从 provider 列表表达，而不绑定单一服务。 +- **`auto` 在任何远程解析失败时降级到本地**：依次尝试每个已配置的远程 provider 并使用第一个成功的；某个 provider 抛出异常时记录日志并尝试下一个，因此远程服务不可用时使该文档降级到本地解析，而不会使导入失败。实际走的路径记录在 `pdf_strategy_effective`；同时，“是否有远程解析服务在场”改用 `PDFConfig.remote_enabled` 判定，它以 provider 列表表达，而不绑定单一服务。 - **MinerU 托管服务支持（`mineru_token`）**：新增 `pdf.mineru_token` 用于携带托管版 MinerU 服务的 Bearer token；由于该服务没有 `/file_parse` 路由，配置它同时会选中托管批处理协议：`POST /file-urls/batch` 预留上传 URL → `PUT` 上传 PDF → 轮询 `GET /extract-results/batch/{batch_id}` → 下载 `full_zip_url` 并取出 markdown 与其图片。未配置 token 时仍使用原有的自托管 `/file_parse` 协议，现有部署不受影响。若批处理握手被拒，解析器会再尝试一次 `/file_parse`，以保持带 token 的自托管部署仍然可用。 - **PDF `auto` 策略改为优先使用已配置的远程解析器（行为变更）**：`auto` 现在会将每份 PDF 发往已配置的远程解析服务，本地解析器仅作兜底；而之前是先本地、本地失败才使用 MinerU。若配置了 `pdf.mineru_endpoint`，请预期延迟、额度消耗与限流风险发生变化；在限流情况下，回退机制会让导入变慢而不是变快。若希望保持原有行为请设为 `strategy: local`；若希望远程解析必须成功、失败即失败，请设为 `strategy: mineru`。远程失败会仅降级单份文档而不使整批导入失败，并由 `pdf_strategy_effective`（`mineru` 或 `local`）记录实际使用的转换器。服务类配置（`mineru_endpoint`、`mineru_token`、`mineru_bodys`）保持当前位置，未来接入新的远程解析服务时只需增加 provider，而无需新增 `strategy` 取值。 diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index 229082a9f1..db91bc71d6 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -1076,10 +1076,8 @@ PDF 解析配置。支持三种策略：`local`（本地 pdfplumber）、`mineru 为避免一个已降级的批次变得明显更慢，`auto` 不会无目的地重试： -- `429` 与 `5xx` 会在一段退避后重试一次，因为它们通常会自行恢复； -- `404` 等永久性错误直接降级，重试无意义； -- 连续失败几次后该 provider 会被熔断：批次剩下的文档直接跳过它，不再为每份文档重复付出一次必败的调用。熔断在一段冷却后会半开重试，服务恢复后会被重新使用； -- 已降级的文档会计入 `pdf_remote_degraded`，与 `pdf_strategy_effective` 并列输出，便于事后发现某个批次已经静默地改用本地解析。 +`auto` 依次尝试每个已配置的远程 provider，并使用第一个成功的。某个 provider 抛出异常时只记录日志并尝试下一个，因此任何远程失败（`404`、`405`、`429`、超时、`5xx`）都会使该文档降级为本地解析，而不会使导入失败。实际走的路径记录在 +`pdf_strategy_effective`，因此静默失效的远程解析在之后仍可见。 启动预检遵循同一规则：`mineru` 在端点未就绪时快速失败，`auto` 仅告警，重试留给首次解析。 diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index b671581e24..bc8741a7a9 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -22,7 +22,7 @@ import time from collections import Counter, defaultdict from pathlib import Path -from typing import Any, Dict, List, Optional, Set, Tuple, Union +from typing import Any, Dict, List, Optional, Tuple, Union from openviking.parse.base import ( NodeType, @@ -79,17 +79,6 @@ def __init__(self, config: Optional[PDFConfig] = None): self.config = config or PDFConfig() self.config.validate() - # Remote-parse health, shared across documents parsed by this instance so a - # batch that hits a rate limit stops paying for doomed calls. - self._remote_failures: Dict[str, int] = {} - self._remote_circuit_open: Set[str] = set() - self._remote_skip_count: Dict[str, int] = {} - self._remote_circuit_cooldown = 20 - self._remote_circuit_threshold = 3 - self._remote_retry_on_transient = True - self._remote_retry_delay_seconds = 2.0 - self._degraded = 0 - # Lazy import MarkdownParser to avoid circular imports self._markdown_parser = None @@ -173,10 +162,6 @@ async def parse(self, source: Union[str, Path], instruction: str = "", **kwargs) effective = conversion_meta.get("strategy") if effective: result.meta["pdf_strategy_effective"] = effective - if self._degraded: - # Count documents that had to fall back, so a batch that silently - # stopped using the remote parser is visible after the fact. - result.meta["pdf_remote_degraded"] = self._degraded result.meta["intermediate_markdown_length"] = len(markdown_content) result.meta["intermediate_markdown_preview"] = markdown_content[:500] @@ -229,70 +214,17 @@ async def _convert_to_markdown( # and the local parser is always the last resort. Adding a new external # parser should mean adding a provider entry, not a new strategy value. for name, convert in self._remote_converters(): - if name in self._remote_circuit_open: - # Stay off the remote for a while, then let one document through - # as a probe: a service that recovers should be used again, and - # one that is still down costs a single call, not the batch. - self._remote_skip_count[name] = self._remote_skip_count.get(name, 0) + 1 - if self._remote_skip_count[name] < self._remote_circuit_cooldown: - self._degraded += 1 - logger.debug( - f"{name} still circuit-open " - f"({self._remote_skip_count[name]}/{self._remote_circuit_cooldown})" - ) - continue - logger.info(f"{name} half-open, probing with this document") try: - result = await self._convert_remote_with_retry( - name, convert, pdf_path, resource_name=resource_name - ) + return await convert(pdf_path, resource_name=resource_name) except Exception as e: - self._remote_failures[name] = self._remote_failures.get(name, 0) + 1 - if self._remote_failures[name] >= self._remote_circuit_threshold: - self._remote_circuit_open.add(name) - logger.warning( - f"{name} failed {self._remote_failures[name]} times in a row " - f"({e}); routing the rest of this batch to local" - ) - self._degraded += 1 - logger.warning(f"{name} conversion failed, trying the next option: {e}") - continue - self._remote_failures.pop(name, None) - self._remote_skip_count.pop(name, None) - self._remote_circuit_open.discard(name) - return result + # A remote failure degrades this one document to local rather + # than failing the import. + logger.warning(f"{name} conversion failed, using local: {e}") return await self._convert_local(pdf_path, resource_name=resource_name) else: raise ValueError(f"Unknown strategy: {self.config.strategy}") - async def _convert_remote_with_retry( - self, - name: str, - convert, - pdf_path: Path, - resource_name: Optional[str] = None, - ): - """Call a remote converter, retrying the failures that are worth retrying. - - A 429 or a transient 5xx usually clears on its own, so one backed-off retry - is cheaper than degrading the document to local. A 404 or a bad request will - not clear, so it goes straight through and the circuit breaker takes over. - """ - attempts = 2 if self._remote_retry_on_transient else 1 - for attempt in range(attempts): - try: - return await convert(pdf_path, resource_name=resource_name) - except Exception as e: - if attempt + 1 >= attempts or not self._is_transient_remote_error(e): - raise - delay = self._remote_retry_delay_seconds - logger.info( - f"{name} transient failure (HTTP {PDFParser._http_status(e)}); retrying in {delay}s" - ) - await asyncio.sleep(delay) - raise RuntimeError(f"{name} exhausted retries") # pragma: no cover - def _remote_converters(self) -> List[Tuple[str, Any]]: """External converters this config enables, in the order they should be tried. @@ -784,19 +716,6 @@ def _mineru_headers(self) -> Dict[str, str]: def _mineru_base_url(self) -> str: return self.config.mineru_endpoint.rstrip("/") - @staticmethod - def _is_transient_remote_error(exc: Exception) -> bool: - """Whether retrying could plausibly succeed. - - Rate limiting and server-side faults are transient; a missing route or a - malformed request fails the same way every time, so retrying it would only - burn quota and latency. - """ - status = PDFParser._http_status(exc) - if status is None: - return False - return status == 429 or 500 <= status < 600 - def _http_status(exc: Exception) -> Optional[int]: response = getattr(exc, "response", None) return getattr(response, "status_code", None) @@ -887,10 +806,14 @@ def _mineru_online_zip_url(cls, body: Dict[str, Any]) -> Optional[str]: return None async def _mineru_download_zip(self, client, zip_url: str) -> bytes: - """Download the result archive (signed urls may already carry auth).""" + """Download the result archive. + + The url is pre-signed by the storage backend and points at a different host, + so the MinerU bearer token must not be attached: it would leak the credential + to that host, and the signature does not need it. + """ response = await client.get( zip_url, - headers=self._mineru_headers(), follow_redirects=True, ) response.raise_for_status() @@ -1049,10 +972,15 @@ async def _mineru_save_images( images: Dict[str, bytes], storage, resource_name: Optional[str], - ) -> int: - """Persist images into the media store, rewriting markdown references.""" + ) -> Dict[str, str]: + """Persist images into the media store. + + Returns the original-name -> stored-relative-path mapping so the caller can + rewrite the markdown references; a saved image whose reference is left + pointing at the provider's path is a broken link in the parsed document. + """ if not images: - return 0 + return {} if storage is None: from openviking_cli.utils.storage import get_storage @@ -1074,7 +1002,7 @@ async def _mineru_save_images( except Exception as img_err: logger.warning(f"Failed to save MinerU image {name}: {img_err}") - return len(repl) + return repl async def _convert_mineru( self, @@ -1213,8 +1141,11 @@ async def _mineru_selfhosted( text = entry.get("md_content") or "" images = self._mineru_decode_inline_images(entry.get("images") or {}) - meta["images_saved"] = await self._mineru_save_images(images, storage, resource_name) - logger.info(f"MinerU conversion: {len(text)} chars") + repl = await self._mineru_save_images(images, storage, resource_name) + meta["images_saved"] = len(repl) + for original, stored in repl.items(): + text = text.replace(original, stored) + logger.info(f"MinerU conversion: {len(text)} chars, {len(repl)} images") return text, meta def _format_table_markdown(self, table: List[List[Optional[str]]]) -> str: diff --git a/openviking_cli/utils/config/parser_config.py b/openviking_cli/utils/config/parser_config.py index 56741c2529..8b80fc29ef 100644 --- a/openviking_cli/utils/config/parser_config.py +++ b/openviking_cli/utils/config/parser_config.py @@ -179,9 +179,16 @@ def remote_enabled(self) -> bool: startup preflight, or the documentation. Keep this in step with the provider list built by ``PDFParser._remote_converters``. """ + # Each entry is one provider's configured endpoint. A default ``None`` + # must stay a member of a real collection: ``any(None)`` raises TypeError, + # which would break startup for deployments that never configured MinerU. return any( - self.mineru_endpoint, # MinerU provider - # future providers register their own config here + endpoint + for endpoint in ( + self.mineru_endpoint, # MinerU provider + # future providers register their own config here + ) + if endpoint ) def validate(self) -> None: diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index 1f06ae3491..a36ce66fe1 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -315,151 +315,132 @@ async def fake_sync(self, client, meta, pdf_path, storage=None, resource_name=No assert meta["api_mode"] == "file-parse" -async def _no_sleep(_seconds): - """Placeholder so retry backoff does not slow the suite down.""" - return None - - # --------------------------------------------------------------------------- -# batch behaviour: partial remote failure, backoff, and circuit breaking +# regressions raised in review: default-config startup, credential isolation, +# and image-reference rewriting # --------------------------------------------------------------------------- -def _http_error(status): - """An error shaped like the real thing, with a response carrying the status.""" - import httpx as _httpx - - request = _httpx.Request("POST", ENDPOINT) - response = _httpx.Response(status, request=request) - return _httpx.HTTPStatusError(f"HTTP {status}", request=request, response=response) - - -def test_transient_errors_are_retried_then_fall_back(monkeypatch, pdf_path): - """A 429 is worth one backed-off retry; a 404 is not.""" - for status, expected_calls in ((429, 2), (500, 2), (404, 1)): - calls = [] - - async def flaky(self, p, storage=None, resource_name=None, _calls=calls, _status=status): - _calls.append(1) - raise _http_error(_status) - - async def ok_local(self, pdf_path, storage=None, resource_name=None): - return LOCAL_MD, {"strategy": "local"} - - async def call_flaky(self, p, storage=None, resource_name=None): - return await flaky(self, p, storage, resource_name) - - monkeypatch.setattr(PDFParser, "_convert_mineru", call_flaky) - monkeypatch.setattr(PDFParser, "_convert_local", ok_local) - monkeypatch.setattr(asyncio, "sleep", _no_sleep) - - parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) - md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) - - assert md == LOCAL_MD, status - assert meta["strategy"] == "local", status - assert len(calls) == expected_calls, (status, len(calls)) - - -def test_circuit_opens_after_repeated_failures(monkeypatch, pdf_path): - """Once the remote is clearly down, stop paying for a doomed call per document.""" - calls = [] - - async def always_fail(self, pdf_path, storage=None, resource_name=None): - calls.append(1) - raise _http_error(404) - - async def ok_local(self, pdf_path, storage=None, resource_name=None): - return LOCAL_MD, {"strategy": "local"} - - async def _stub_always_fail(self, p, storage=None, resource_name=None): - return await always_fail(self, p, storage, resource_name) +def test_default_pdfconfig_reports_no_remote_provider(): + """A deployment that never configured MinerU must still start. - monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_always_fail) - monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + ``remote_enabled`` is read on the default ``auto`` path during service start-up, + so it has to cope with the default ``None`` endpoint instead of raising. + """ + from openviking_cli.utils.config.parser_config import PDFConfig - parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) - parser._remote_circuit_threshold = 3 - parser._remote_circuit_cooldown = 1000 # stay open for this test's batch + config = PDFConfig() + assert config.mineru_endpoint is None + assert config.remote_enabled is False - for _ in range(10): - md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) - assert md == LOCAL_MD - assert meta["strategy"] == "local" - # 3 documents opened the circuit; the other 7 never called the remote again. - assert len(calls) == 3, len(calls) +def test_remote_enabled_true_only_for_a_configured_endpoint(): + from openviking_cli.utils.config.parser_config import PDFConfig + config = PDFConfig() + config.mineru_endpoint = "https://mineru.example" + assert config.remote_enabled is True -def test_circuit_recovers_after_a_success(monkeypatch, pdf_path): - """A single success closes the circuit again.""" - state = {"fail": True} + config.mineru_endpoint = "" + assert config.remote_enabled is False - async def flaky(self, pdf_path, storage=None, resource_name=None): - if state["fail"]: - raise _http_error(404) - return MINERU_MD, {"strategy": "mineru"} - async def ok_local(self, pdf_path, storage=None, resource_name=None): - return LOCAL_MD, {"strategy": "local"} +def test_presigned_zip_download_carries_no_mineru_token(monkeypatch): + """The archive lives on the storage host, so the MinerU token must not go there.""" + from openviking_cli.utils.config.parser_config import PDFConfig - async def _stub_flaky(self, p, storage=None, resource_name=None): - return await flaky(self, p, storage, resource_name) + parser = PDFParser(PDFConfig(mineru_endpoint=ENDPOINT, mineru_token="secret-token")) + seen = {} - monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_flaky) - monkeypatch.setattr(PDFParser, "_convert_local", ok_local) - - parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) - parser._remote_circuit_threshold = 2 - parser._remote_circuit_cooldown = 1000 + class _Resp: + content = b"zip-bytes" - asyncio.run(parser._convert_to_markdown(pdf_path)) - asyncio.run(parser._convert_to_markdown(pdf_path)) - assert "MinerU" in parser._remote_circuit_open + def raise_for_status(self): + return None - state["fail"] = False - parser._remote_circuit_cooldown = 1 # let the next document through as a probe - md, meta = asyncio.run(parser._convert_to_markdown(pdf_path)) - assert md == MINERU_MD - assert "MinerU" not in parser._remote_circuit_open + class _Client: + async def get(self, url, **kwargs): + seen["url"] = url + seen["headers"] = kwargs.get("headers") + seen["kwargs"] = kwargs + return _Resp() + zip_url = "https://oss.storage.example/batch/abc.zip?signature=deadbeef" + asyncio.run(parser._mineru_download_zip(_Client(), zip_url)) -def test_batch_with_partial_remote_failures(monkeypatch, tmp_path): - """10 documents, 3 fail remotely: the batch finishes and stays auditable. + assert seen["url"] == zip_url + assert "headers" not in seen["kwargs"], "a token-bearing header must not be sent" + # The API headers still carry the token; they are simply not used for the + # cross-host download. + assert parser._mineru_headers()["Authorization"] == "Bearer secret-token" - This is the shape a real import hits -- a handful of bad documents must not take - down the run, and the operator needs to see that three of them degraded. - """ - failing = {"doc2.pdf", "doc5.pdf", "doc9.pdf"} - seen = [] - async def sometimes(self, pdf_path, storage=None, resource_name=None): - seen.append(pdf_path.name) - if pdf_path.name in failing: - raise _http_error(400) - return MINERU_MD, {"strategy": "mineru"} +def test_self_hosted_images_are_saved_and_references_rewritten(monkeypatch, pdf_path): + """A saved image whose markdown still points at the provider path is broken.""" + from openviking_cli.utils.config.parser_config import PDFConfig - async def ok_local(self, pdf_path, storage=None, resource_name=None): - return LOCAL_MD, {"strategy": "local"} + monkeypatch.setattr( + PDFParser, + "_mineru_base_url", + lambda self: ENDPOINT, + ) + monkeypatch.setattr( + PDFParser, + "_mineru_file_parse_url", + lambda self: ENDPOINT + "/file_parse", + ) - async def _stub_sometimes(self, p, storage=None, resource_name=None): - return await sometimes(self, p, storage, resource_name) + markdown = "see ![fig](images/figure.png) for detail" + # /file_parse answers with results keyed by file name + payload = { + "results": { + pdf_path.name: { + "md_content": markdown, + "images": {"images/figure.png": "ZmFrZQ=="}, + } + } + } + + class _Resp: + def raise_for_status(self): + return None + + def json(self): + return payload + + class _Client: + async def post(self, *a, **kw): + return _Resp() + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + saved = {} + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + target = Path("/media") / resource_name / (filename + extension) + saved["path"] = target + saved["blob"] = blob + return target + + monkeypatch.setattr( + "openviking.parse.parsers.pdf.lazy_import", + lambda name: __import__("httpx") if name == "httpx" else None, + raising=False, + ) + monkeypatch.setattr("httpx.AsyncClient", lambda *a, **kw: _Client()) - monkeypatch.setattr(PDFParser, "_convert_mineru", _stub_sometimes) - monkeypatch.setattr(PDFParser, "_convert_local", ok_local) + parser = PDFParser(PDFConfig(strategy="mineru", mineru_endpoint=ENDPOINT)) + text, meta = asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf_path, _Storage(), "math")) - parser = _parser(strategy="auto", mineru_endpoint=ENDPOINT) - effective = [] - - for i in range(10): - doc = tmp_path / f"doc{i}.pdf" - doc.write_bytes(b"%PDF-1.4 minimal") - _, meta = asyncio.run(parser._convert_to_markdown(doc)) - effective.append(meta["strategy"]) - - assert len(seen) == 10, seen - # Every document produced a result; the failures degraded rather than aborting. - assert effective.count("local") == len(failing) - assert effective.count("mineru") == 10 - len(failing) - # And the degradation is visible to operators. - assert parser._degraded == len(failing) + assert meta["images_saved"] == 1 + assert saved["path"].name == "figure.png" + # the reference must now resolve to where the image was actually written + assert "images/figure.png" not in text + assert "math/figure.png" in text From 57662c2710b65e707864a241f777192c81a3f14f Mon Sep 17 00:00:00 2001 From: q2000s Date: Fri, 16 Oct 2026 03:36:08 +0800 Subject: [PATCH 3/7] =?UTF-8?q?=EF=BB=BFtest(pdf):=20cover=20the=20hosted?= =?UTF-8?q?=20v4=20protocol=20internals=20and=20fix=20an=20undefined=20fal?= =?UTF-8?q?lback?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dispatch tests drove _convert_mineru through a stub, so the hosted batch protocol -- the upload handshake, the polling loop, and the archive unpacking -- had no unit coverage at all. 25 new tests exercise that path directly, plus the zip helpers, the online state helpers, and the failure paths. Also fixes a latent crash the new tests exposed: _mineru_online_failure_detail fell back to cls._mineru_failure_detail, which does not exist, so a batch that failed without a reason on its entry raised AttributeError instead of a readable error. It now reads the batch envelope and, failing that, reports that no detail was given. Coverage of the lines this PR changes: 240/240 statements, 100%. 72 tests pass across the dispatch and bookmark suites. --- openviking/parse/parsers/pdf.py | 8 +- tests/parse/test_pdf_strategy_dispatch.py | 468 ++++++++++++++++++++++ 2 files changed, 475 insertions(+), 1 deletion(-) diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index bc8741a7a9..3f40233316 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -794,7 +794,13 @@ def _mineru_online_failure_detail(cls, body: Dict[str, Any]) -> str: for key in ("err_msg", "err_no", "state", "message", "msg"): if entry.get(key): return f"{key}={entry[key]}" - return cls._mineru_failure_detail(body) + # The batch envelope can carry the reason even when the entry is silent. + data = cls._mineru_payload_data(body) + if isinstance(data, dict): + for key in ("err_msg", "err_no", "message", "msg"): + if data.get(key): + return f"{key}={data[key]}" + return "no failure detail reported" @classmethod def _mineru_online_zip_url(cls, body: Dict[str, Any]) -> Optional[str]: diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index a36ce66fe1..b06dde72c3 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -444,3 +444,471 @@ def save_image(self, resource_name, blob, filename, extension): # the reference must now resolve to where the image was actually written assert "images/figure.png" not in text assert "math/figure.png" in text + + +# --------------------------------------------------------------------------- +# hosted v4 batch protocol: upload handshake, polling, archive unpacking +# --------------------------------------------------------------------------- + + +async def _no_sleep(_seconds): + """Poll without waiting, so the timeout test does not spend real seconds.""" + return None + + +def _zip_bytes(entries): + """Build an in-memory zip from {name: text-bytes}.""" + import io as _io + import zipfile + + buf = _io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + for name, blob in entries.items(): + zf.writestr(name, blob) + return buf.getvalue() + + +class _FakeResponse: + def __init__(self, payload=None, content=b""): + self._payload = payload + self.content = content + + def raise_for_status(self): + return None + + def json(self): + return self._payload + + +class _HostedClient: + """Replays the hosted v4 conversation: reserve url, PUT, poll, download.""" + + def __init__(self, polls, zip_url="https://oss.example/result.zip", zip_payload=b""): + self.polls = list(polls) + self.zip_url = zip_url + self.zip_payload = zip_payload + self.calls = [] + + async def post(self, url, **kw): + self.calls.append(("POST", url, kw)) + return _FakeResponse( + { + "code": 0, + "data": { + "batch_id": "batch-1", + "file_urls": ["https://oss.example/upload/doc.pdf"], + }, + } + ) + + async def put(self, url, **kw): + self.calls.append(("PUT", url, kw)) + return _FakeResponse() + + async def get(self, url, **kw): + self.calls.append(("GET", url, kw)) + if "/extract-results/" in url: + if not self.polls: + raise AssertionError("polled more times than the test scripted") + return _FakeResponse(self.polls.pop(0)) + return _FakeResponse(content=self.zip_payload) + + +def _hosted_parser(timeout=30.0): + from openviking_cli.utils.config.parser_config import PDFConfig + + return PDFParser( + PDFConfig( + strategy="mineru", + mineru_endpoint="https://mineru.example/api", + mineru_token="tok", + mineru_timeout=timeout, + ) + ) + + +def test_hosted_batch_reserves_uploads_polls_and_unpacks(monkeypatch, pdf_path): + archive = _zip_bytes( + { + "full.md": b"# Title\n\n![fig](images/fig-1.png)", + "images/fig-1.png": b"\x89PNG-bytes", + } + ) + client = _HostedClient( + polls=[ + {"code": 0, "data": {"extract_result": [{"state": "waiting-file"}]}}, + { + "code": 0, + "data": { + "extract_result": [ + {"state": "done", "full_zip_url": "https://oss.example/result.zip"} + ] + }, + }, + ], + zip_payload=archive, + ) + + saved = {} + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + saved[(resource_name, filename, extension)] = blob + return Path("/media") / resource_name / (filename + extension) + + monkeypatch.setattr("httpx.AsyncClient", lambda *a, **kw: client) + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + + parser = _hosted_parser() + meta = {} + text, meta = asyncio.run(parser._mineru_hosted_batch(client, meta, pdf_path, _Storage(), "doc")) + + assert text.startswith("# Title") + # the image reference must point at where the file was written + assert "images/fig-1.png" not in text + assert "doc/fig-1.png" in text + assert meta["api_version"] == "task-zip" + assert meta["images_saved"] == 1 + assert meta["batch_id"] == "batch-1" + assert saved[("doc", "fig-1", ".png")] == b"\x89PNG-bytes" + + verbs = [c[0] for c in client.calls] + assert verbs[0] == "POST" and "/file-urls/batch" in client.calls[0][1] + assert verbs[1] == "PUT" + # the pre-signed upload carries no Content-Type and no MinerU bearer + assert client.calls[1][2]["headers"] == {"Content-Type": ""} + # the archive download must not carry the token to the storage host + zip_get = [c for c in client.calls if c[0] == "GET" and "result.zip" in c[1]][0] + assert "headers" not in zip_get[2] + + +def test_hosted_batch_times_out_without_a_result(monkeypatch, pdf_path): + client = _HostedClient(polls=[{"data": {"extract_result": [{"state": "running"}]}}] * 3) + monkeypatch.setattr("httpx.AsyncClient", lambda *a, **kw: client) + monkeypatch.setattr(asyncio, "sleep", _no_sleep) + + parser = _hosted_parser(timeout=5.0) + try: + asyncio.run(parser._mineru_hosted_batch(client, {}, pdf_path, None, "doc")) + except TimeoutError as e: + assert "did not finish" in str(e) + else: + raise AssertionError("expected TimeoutError") + + +def test_hosted_batch_reports_a_failed_batch(monkeypatch, pdf_path): + client = _HostedClient( + polls=[{"data": {"extract_result": [{"state": "failed", "err_msg": "bad pdf"}]}}] + ) + monkeypatch.setattr("httpx.AsyncClient", lambda *a, **kw: client) + + parser = _hosted_parser() + try: + asyncio.run(parser._mineru_hosted_batch(client, {}, pdf_path, None, "doc")) + except ValueError as e: + assert "err_msg=bad pdf" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_hosted_batch_done_without_zip_url_is_an_error(monkeypatch, pdf_path): + client = _HostedClient(polls=[{"data": {"extract_result": [{"state": "done"}]}}]) + monkeypatch.setattr("httpx.AsyncClient", lambda *a, **kw: client) + + parser = _hosted_parser() + try: + asyncio.run(parser._mineru_hosted_batch(client, {}, pdf_path, None, "doc")) + except ValueError as e: + assert "without a result zip" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_submit_file_batch_rejects_a_reply_without_urls(monkeypatch, pdf_path): + class _NoUrls(_HostedClient): + async def post(self, url, **kw): + return _FakeResponse({"code": 0, "data": {"file_urls": []}}) + + parser = _hosted_parser() + try: + asyncio.run( + parser._mineru_submit_file_batch(_NoUrls([]), "https://mineru.example/api", pdf_path) + ) + except ValueError as e: + assert "no file_urls" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_post_json_raises_on_a_business_error_code(): + import httpx + + from openviking_cli.utils.config.parser_config import PDFConfig + + parser = PDFParser(PDFConfig(mineru_endpoint="https://x", mineru_token="t")) + + class _Client: + async def post(self, *a, **kw): + return _FakeResponse({"code": 4001, "msg": "quota exceeded"}) + + try: + asyncio.run(parser._mineru_post_json(_Client(), "https://x/y", {})) + except RuntimeError as e: + assert "code=4001" in str(e) and "quota exceeded" in str(e) + else: + raise AssertionError("expected RuntimeError") + assert httpx is not None + + +def test_pick_markdown_prefers_full_md(tmp_path): + (tmp_path / "full.md").write_text("full", encoding="utf-8") + (tmp_path / "other.md").write_text("other", encoding="utf-8") + nested = tmp_path / "deep" / "deeper" + nested.mkdir(parents=True) + (nested / "x.md").write_text("x", encoding="utf-8") + + assert PDFParser._mineru_pick_markdown(tmp_path).name == "full.md" + + +def test_pick_markdown_returns_the_only_candidate_and_none_for_empty(tmp_path): + assert PDFParser._mineru_pick_markdown(tmp_path) is None + only = tmp_path / "doc.md" + only.write_text("x", encoding="utf-8") + assert PDFParser._mineru_pick_markdown(tmp_path) == only + + +def test_pick_markdown_falls_back_to_the_shallowest_largest(tmp_path): + deep = tmp_path / "a" / "b" + deep.mkdir(parents=True) + (deep / "big.md").write_text("x" * 50, encoding="utf-8") + (tmp_path / "small.md").write_text("y", encoding="utf-8") + assert PDFParser._mineru_pick_markdown(tmp_path).parent == tmp_path + + +def test_unpack_zip_without_markdown_is_an_error(pdf_path): + parser = _hosted_parser() + try: + parser._mineru_unpack_zip(_zip_bytes({"images/a.png": b"x"}), pdf_path, {}) + except ValueError as e: + assert "No markdown file found" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_unpack_zip_handles_empty_markdown(pdf_path): + parser = _hosted_parser() + text, meta = parser._mineru_unpack_zip(_zip_bytes({"full.md": b""}), pdf_path, {}) + assert text == "" + assert meta["api_version"] == "task-zip" + + +def test_online_state_helpers(): + entry = lambda **kw: {"data": {"extract_result": [kw]}} # noqa: E731 + assert PDFParser._mineru_online_done(entry(state="done")) + assert not PDFParser._mineru_online_done(entry(state="running")) + assert not PDFParser._mineru_online_done({}) + for busy in ("waiting", "waiting-file", "uploading", "pending", "running", "converting"): + assert PDFParser._mineru_online_busy(entry(state=busy)), busy + assert not PDFParser._mineru_online_busy(entry(state="done")) + assert PDFParser._mineru_online_result_entry({"extract_result": []}) is None + assert PDFParser._mineru_online_result_entry({"extract_result": [1]}) is None + + +def test_online_zip_url_and_failure_detail(): + good = {"extract_result": [{"full_zip_url": "https://a/b.zip"}]} + assert PDFParser._mineru_online_zip_url(good) == "https://a/b.zip" + assert PDFParser._mineru_online_zip_url({"extract_result": [{"zip_url": "ftp://x"}]}) is None + assert PDFParser._mineru_online_zip_url({}) is None + assert ( + PDFParser._mineru_online_failure_detail({"extract_result": [{"err_no": "E9"}]}) + == "err_no=E9" + ) + # with nothing on the entry, the generic failure detail still yields text + assert PDFParser._mineru_online_failure_detail({"msg": "nope"}) + + +def test_extract_markdown_collects_only_image_files(tmp_path): + (tmp_path / "full.md").write_text("# t", encoding="utf-8") + (tmp_path / "images").mkdir() + (tmp_path / "images" / "a.png").write_bytes(b"p") + (tmp_path / "images" / "b.txt").write_bytes(b"t") + + from openviking_cli.utils.config.parser_config import PDFConfig + + parser = PDFParser(PDFConfig(mineru_endpoint="https://x")) + text, images = parser._mineru_extract_markdown(tmp_path / "full.md", tmp_path) + assert text == "# t" + assert list(images) == ["images/a.png"] + + +def test_decode_inline_images_skips_bad_payloads(): + decoded = PDFParser._mineru_decode_inline_images( + {"ok.png": "data:image/png;base64,ZmFrZQ==", "bad.png": "!!!not base64!!!"} + ) + assert decoded == {"ok.png": b"fake"} + assert PDFParser._mineru_decode_inline_images(None) == {} + + +def test_save_images_returns_empty_and_survives_a_failing_store(): + from openviking_cli.utils.config.parser_config import PDFConfig + + parser = PDFParser(PDFConfig(mineru_endpoint="https://x")) + assert asyncio.run(parser._mineru_save_images({}, None, None)) == {} + + class _Broken: + media_dir = Path("/media") + + def save_image(self, *a, **kw): + raise OSError("disk full") + + assert asyncio.run(parser._mineru_save_images({"a.png": b"x"}, _Broken(), "doc")) == {} + + +def test_convert_mineru_requires_an_endpoint(): + + parser = _hosted_parser() + parser.config.mineru_endpoint = "" + try: + asyncio.run(parser._convert_mineru(Path("x.pdf"))) + except ValueError as e: + assert "endpoint not configured" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_payload_unwrapping_and_sync_entry_edges(tmp_path): + assert PDFParser._mineru_payload_data({"data": {"k": 1}}) == {"k": 1} + assert PDFParser._mineru_payload_data({"k": 1}) == {"k": 1} + assert PDFParser._mineru_payload_data(["not", "a", "dict"]) == ["not", "a", "dict"] + # a truthy non-dict results value must be rejected, not iterated + assert PDFParser._mineru_sync_entry({"results": "oops"}, tmp_path / "a.pdf") is None + assert PDFParser._mineru_sync_entry({"results": []}, tmp_path / "a.pdf") is None + assert PDFParser._mineru_sync_entry({"results": {"a.pdf": "s"}}, tmp_path / "a.pdf") is None + assert PDFParser._mineru_sync_entry( + {"results": {"b.pdf": {"md_content": "x"}}}, tmp_path / "a.pdf" + ) == {"md_content": "x"} + + +def test_http_status_reads_the_status_off_an_exception(): + import httpx + + req = httpx.Request("GET", "https://x") + resp = httpx.Response(503, request=req) + exc = httpx.HTTPStatusError("boom", request=req, response=resp) + assert PDFParser._http_status(exc) == 503 + assert PDFParser._http_status(ValueError("no response")) is None + + +def test_failure_detail_falls_back_when_the_reply_says_nothing(): + assert PDFParser._mineru_online_failure_detail({}) == "no failure detail reported" + # a reason on the batch envelope still surfaces when the entry is silent + assert PDFParser._mineru_online_failure_detail({"data": {"msg": "quota"}}) == "msg=quota" + + +def test_unpack_zip_defaults_resource_name_and_skips_subdirectories(pdf_path): + archive = _zip_bytes( + { + "full.md": b"![a](images/a.png)", + "images/a.png": b"png", + "images/nested-dir/x": b"junk", + } + ) + saved = {} + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + saved["resource"] = resource_name + return Path("/media") / resource_name / (filename + extension) + + parser = _hosted_parser() + # resource_name=None must fall back to the pdf stem + text, meta = parser._mineru_unpack_zip(archive, pdf_path, {}, _Storage(), None) + assert saved["resource"] == pdf_path.stem + assert meta["images_saved"] == 1 + assert "images/a.png" not in text + + +def test_unpack_zip_uses_the_default_storage_when_none_is_given(pdf_path, monkeypatch): + archive = _zip_bytes({"full.md": b"![a](images/a.png)", "images/a.png": b"png"}) + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + return Path("/media") / resource_name / (filename + extension) + + monkeypatch.setattr( + "openviking_cli.utils.storage.get_storage", lambda: _Storage(), raising=False + ) + parser = _hosted_parser() + text, meta = parser._mineru_unpack_zip(archive, pdf_path, {}, None, "doc") + assert meta["images_saved"] == 1 + assert "doc/a.png" in text + + +def test_selfhosted_without_markdown_is_an_error(pdf_path): + class _Client: + async def post(self, *a, **kw): + return _FakeResponse({"results": {}}) + + parser = _hosted_parser() + parser.config.mineru_token = "" + try: + asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf_path, None, "doc")) + except ValueError as e: + assert "returned no markdown" in str(e) + else: + raise AssertionError("expected ValueError") + + +def test_save_images_falls_back_to_the_default_storage(monkeypatch): + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + return Path("/media") / resource_name / (filename + extension) + + monkeypatch.setattr( + "openviking_cli.utils.storage.get_storage", lambda: _Storage(), raising=False + ) + parser = _hosted_parser() + repl = asyncio.run(parser._mineru_save_images({"images/a.png": b"png"}, None, "doc")) + assert repl == {"images/a.png": "doc/a.png"} + + +def test_parse_records_the_effective_strategy(monkeypatch, pdf_path): + """`auto` records which converter ran, not only what was requested.""" + from openviking.parse.base import NodeType, ResourceNode, create_parse_result + + class _MD: + async def parse_content(self, content, **kw): + return create_parse_result( + root=ResourceNode(type=NodeType.ROOT), + source_path=kw.get("source_path", ""), + source_format="pdf", + parser_name="MarkdownParser", + parse_time=0.0, + ) + + class _Storage: + media_dir = Path("/media") + + parser = _hosted_parser() + parser.config.strategy = "auto" + + async def _remote(pdf, resource_name=None): + return "remote markdown", {"strategy": "mineru"} + + monkeypatch.setattr(type(parser), "_remote_converters", lambda self: [("mineru", _remote)]) + monkeypatch.setattr(type(parser), "_get_markdown_parser", lambda self: _MD()) + monkeypatch.setattr( + "openviking_cli.utils.storage.get_storage", lambda: _Storage(), raising=False + ) + + result = asyncio.run(parser.parse(pdf_path)) + assert result.meta["pdf_strategy"] == "auto" + assert result.meta["pdf_strategy_effective"] == "mineru" From 1cd87a6ff4ab5e03861c95bcb518bdf2f4e2586c Mon Sep 17 00:00:00 2001 From: q2000s Date: Fri, 16 Oct 2026 05:06:50 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=EF=BB=BFfix(pdf):=20make=20the=20preflight?= =?UTF-8?q?=20and=20self-hosted=20path=20protocol-correct?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of 57662c27 found three compatibility problems. 1. [P1] A hosted endpoint failed the startup preflight. core.py called the self-hosted health probe for every configured endpoint, which demands a /health route answering protocol_version and status="healthy". The hosted API has no such route and answers 404, so strategy="mineru" against a hosted endpoint failed to start before parsing anything. The probe is now protocol-aware: a token marks the hosted API, and its preflight checks that the service is reachable and accepts the credential, rather than demanding a contract it cannot satisfy. 2. [P2] Two self-hosted image regressions. A missing resource_name used to fall back to the pdf stem; without it the storage layer was handed None and the save failed into a swallowed exception, losing every image. And when the reply keyed an image as a bare figure.png while the body wrote images/figure.png, literal substitution produced images/doc/figure.png against a file actually saved at doc/figure.png. The stem fallback is restored, and references are matched by file name in any directory position. 3. [P2] A failed conversion could be returned as a success. The reply handler only checked that results existed, so status="failed" with a partial md_content came back as a truncated document: strict mode did not raise and auto never fell back. An entry reporting a failed state or an error field now raises with the reason it reported. Tests: 87 pass across the dispatch and bookmark suites. Coverage of the lines this PR changes: 243/243 statements, 100%. --- openviking/parse/parsers/pdf.py | 63 ++++- openviking/service/core.py | 10 +- openviking/service/mineru_preflight.py | 61 +++-- tests/parse/test_pdf_strategy_dispatch.py | 293 +++++++++++++++++++++- 4 files changed, 405 insertions(+), 22 deletions(-) diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index 3f40233316..ace55db774 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -37,6 +37,12 @@ logger = get_logger(__name__) +# Per-file states a self-hosted ``/file_parse`` reply uses to report that the +# conversion did not complete. +_MINERU_FAILED_STATES = frozenset( + {"failed", "failure", "error", "aborted", "cancelled", "canceled", "timeout", "rejected"} +) + # PDFium is not thread-safe, even when rendering different documents. _PDFIUM_LOCK = threading.Lock() @@ -937,6 +943,28 @@ def _mineru_payload_data(result: Dict[str, Any]) -> Dict[str, Any]: return result["data"] return result + @staticmethod + def _mineru_entry_failed(entry: Dict[str, Any]) -> bool: + """Whether a per-file entry reports that the conversion did not complete. + + A failed conversion can still carry a partial ``md_content``. Returning that + as a success hands back a silently truncated document: strict mode would not + raise and ``auto`` would never fall back to the local parser. + """ + status = entry.get("status") or entry.get("state") + if isinstance(status, str) and status.strip().lower() in _MINERU_FAILED_STATES: + return True + return any(entry.get(key) for key in ("err_msg", "err_no", "error")) + + @staticmethod + def _mineru_failure_message(entry: Dict[str, Any], pdf_path: Path) -> str: + """A readable reason for a failed per-file entry.""" + for key in ("err_msg", "err_no", "error", "message", "msg", "status", "state"): + value = entry.get(key) + if value: + return f"MinerU failed to convert {pdf_path.name}: {key}={value}" + return f"MinerU failed to convert {pdf_path.name}" + @staticmethod def _mineru_sync_entry(payload: Dict[str, Any], pdf_path: Path) -> Optional[Dict[str, Any]]: """Pull this PDF's entry out of a ``/file_parse`` response.""" @@ -944,7 +972,32 @@ def _mineru_sync_entry(payload: Dict[str, Any], pdf_path: Path) -> Optional[Dict if not isinstance(results, dict): return None entry = results.get(pdf_path.name) or next(iter(results.values()), None) - return entry if isinstance(entry, dict) else None + if not isinstance(entry, dict): + return None + if PDFParser._mineru_entry_failed(entry): + raise ValueError(PDFParser._mineru_failure_message(entry, pdf_path)) + return entry + + @staticmethod + def _mineru_apply_image_refs(text: str, repl: Dict[str, str]) -> str: + """Point markdown references at the images that were actually saved. + + The provider's image key and the reference written in the body can disagree + about the directory: a key may be a bare ``figure.png`` while the body writes + ``images/figure.png``. Substituting the key literally would then yield + ``images/doc/figure.png`` and leave a broken link, so match the file name in + whatever directory position it appears and replace the whole reference. + """ + for key in sorted(repl, key=len, reverse=True): + stored = repl[key] + name = key.replace("\\", "/").rsplit("/", 1)[-1] + if not name: + continue + # Only whole directory segments may precede the name: a bare prefix + # would swallow "my" in "myfigure.png" and rewrite the wrong file. + pattern = re.compile(r"(? tuple[str, Dict[str, bytes]]: """Read markdown from an unpacked archive and collect its images.""" @@ -998,8 +1051,11 @@ async def _mineru_save_images( for name, blob in images.items(): try: img_path = Path(name) + # The storage layer needs a real directory name; handing it None + # loses the image to a swallowed error. + target = resource_name or img_path.stem stored = storage.save_image( - resource_name, + target, blob, filename=img_path.stem, extension=img_path.suffix or ".png", @@ -1149,8 +1205,7 @@ async def _mineru_selfhosted( images = self._mineru_decode_inline_images(entry.get("images") or {}) repl = await self._mineru_save_images(images, storage, resource_name) meta["images_saved"] = len(repl) - for original, stored in repl.items(): - text = text.replace(original, stored) + text = self._mineru_apply_image_refs(text, repl) logger.info(f"MinerU conversion: {len(text)} chars, {len(repl)} images") return text, meta diff --git a/openviking/service/core.py b/openviking/service/core.py index bb4ab5f8c5..48a822d0df 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -691,8 +691,16 @@ async def initialize(self) -> None: ) if should_preflight_mineru and pdf_config.mineru_endpoint: + # A token marks the hosted API, which has no /health route. Probing it + # with the self-hosted contract would fail startup for a service that is + # reachable and correctly configured. + hosted = bool(pdf_config.mineru_token) try: - await wait_for_mineru_ready(pdf_config.mineru_endpoint) + await wait_for_mineru_ready( + pdf_config.mineru_endpoint, + hosted=hosted, + token=pdf_config.mineru_token, + ) logger.info("MinerU preflight passed: %s", pdf_config.mineru_endpoint) except RuntimeError as exc: if pdf_config.strategy == "mineru": diff --git a/openviking/service/mineru_preflight.py b/openviking/service/mineru_preflight.py index 1b12693e82..b36005a047 100644 --- a/openviking/service/mineru_preflight.py +++ b/openviking/service/mineru_preflight.py @@ -6,19 +6,31 @@ Polls the MinerU ``/health`` endpoint so that endpoint misconfiguration, a stopped service, or a protocol incompatibility is surfaced at service initialization instead of on the first PDF import. + +The check is protocol-aware. A self-hosted deployment answers ``/health`` with the +MinerU contract this module validates. The hosted service does not expose that route +at all -- probing it returns 404 -- so a token-configured endpoint is checked for +reachability and authentication instead, and a failure there is reported as such +rather than as a protocol mismatch. """ import asyncio import time +from typing import Optional import httpx -HEALTH_TIMEOUT = 5.0 # Total polling timeout in seconds. -HEALTH_POLL_INTERVAL = 0.2 # Delay between polls in seconds. +HEALTH_TIMEOUT = 5.0 # Total polling timeout in seconds. +HEALTH_POLL_INTERVAL = 0.2 # Delay between polls in seconds. HEALTH_REQUEST_TIMEOUT = 0.5 # Per-request timeout in seconds. -async def wait_for_mineru_ready(endpoint: str, timeout: float = HEALTH_TIMEOUT) -> None: +async def wait_for_mineru_ready( + endpoint: str, + timeout: float = HEALTH_TIMEOUT, + hosted: bool = False, + token: Optional[str] = None, +) -> None: """ Poll the MinerU ``/health`` endpoint until it reports ready. @@ -27,6 +39,12 @@ async def wait_for_mineru_ready(endpoint: str, timeout: float = HEALTH_TIMEOUT) Args: endpoint: MinerU base URL (e.g. ``http://127.0.0.1:8000``). + hosted: When true, ``endpoint`` is the hosted service, which serves no + ``/health`` route. The probe then only checks that the API is reachable + and that the token is accepted, instead of demanding the self-hosted + health contract the hosted API cannot satisfy. + token: Bearer token for the hosted API, sent on the probe so an + authentication problem surfaces here rather than on the first import. timeout: Total time to poll before giving up, in seconds. Raises: @@ -34,35 +52,48 @@ async def wait_for_mineru_ready(endpoint: str, timeout: float = HEALTH_TIMEOUT) seconds. The message includes the health URL and the last error. """ health_url = f"{endpoint.rstrip('/')}/health" + # The hosted API exposes no /health; probing one returns 404, which says nothing + # about whether the service is up. Probe a real route and judge reachability. + probe_url = ( + f"{endpoint.rstrip('/')}/extract-results/batch/__preflight__" if hosted else health_url + ) + probe_headers = {"Authorization": f"Bearer {token}"} if (hosted and token) else {} deadline = time.monotonic() + timeout last_error = "no response" async with httpx.AsyncClient(timeout=HEALTH_REQUEST_TIMEOUT) as client: while time.monotonic() < deadline: try: - response = await client.get(health_url) + response = await client.get(probe_url, headers=probe_headers) payload = response.json() except (httpx.HTTPError, ValueError) as exc: last_error = str(exc) or exc.__class__.__name__ else: - if isinstance(payload, dict): - protocol = payload.get("protocol_version") - status = payload.get("status") + if hosted: + # Reachable and not rejected on credentials is all the hosted + # service can tell us without doing real work. + if response.status_code not in (401, 403): + return + last_error = f"HTTP {response.status_code} from {probe_url}" else: - protocol = status = None + if isinstance(payload, dict): + protocol = payload.get("protocol_version") + status = payload.get("status") + else: + protocol = status = None - if response.status_code == 200 and protocol and status == "healthy": - return + if response.status_code == 200 and protocol and status == "healthy": + return - last_error = ( - f"unexpected response: HTTP {response.status_code}, " - f"protocol_version={protocol!r}, status={status!r}" - ) + last_error = ( + f"unexpected response: HTTP {response.status_code}, " + f"protocol_version={protocol!r}, status={status!r}" + ) await asyncio.sleep(HEALTH_POLL_INTERVAL) raise RuntimeError( - f"MinerU startup preflight failed for {health_url} after {timeout}s. " + f"MinerU startup preflight failed for {probe_url} after {timeout}s. " f"Last error: {last_error}. " "Start a compatible mineru-api service or correct pdf.mineru_endpoint." ) diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index b06dde72c3..fa6f6dc14c 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -855,8 +855,7 @@ class _Client: async def post(self, *a, **kw): return _FakeResponse({"results": {}}) - parser = _hosted_parser() - parser.config.mineru_token = "" + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") try: asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf_path, None, "doc")) except ValueError as e: @@ -912,3 +911,293 @@ async def _remote(pdf, resource_name=None): result = asyncio.run(parser.parse(pdf_path)) assert result.meta["pdf_strategy"] == "auto" assert result.meta["pdf_strategy_effective"] == "mineru" + + +# --------------------------------------------------------------------------- +# review round 3: hosted preflight, self-hosted image keys, failure status +# --------------------------------------------------------------------------- + + +def test_hosted_preflight_accepts_the_missing_health_route(): + """The hosted API has no /health; a 404 there must not fail startup. + + Probing it with the self-hosted contract would reject a service that is up and + correctly configured, which is what blocked strategy="mineru" at boot. + """ + import openviking.service.mineru_preflight as pf + + seen = {} + + class _Resp: + status_code = 404 + + def json(self): + return {} + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, **kw): + seen["url"] = url + seen["headers"] = kw.get("headers") + return _Resp() + + pf.httpx.AsyncClient = lambda **kw: _Client() + asyncio.run( + pf.wait_for_mineru_ready( + "https://mineru.example/api/v4", timeout=1, hosted=True, token="tok" + ) + ) + # a real route was probed, not /health + assert "/health" not in seen["url"] + assert seen["headers"] == {"Authorization": "Bearer tok"} + + +def test_hosted_preflight_still_rejects_a_rejected_token(): + import openviking.service.mineru_preflight as pf + + class _Resp: + status_code = 401 + + def json(self): + return {} + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, **kw): + return _Resp() + + pf.httpx.AsyncClient = lambda **kw: _Client() + try: + asyncio.run( + pf.wait_for_mineru_ready( + "https://mineru.example/api/v4", timeout=0.2, hosted=True, token="bad" + ) + ) + except RuntimeError as exc: + assert "401" in str(exc) + else: + raise AssertionError("a rejected token must fail the preflight") + + +def test_selfhosted_preflight_still_demands_the_health_contract(): + import openviking.service.mineru_preflight as pf + + class _Resp: + status_code = 200 + + def json(self): + return {"protocol_version": "1.0", "status": "healthy"} + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, **kw): + return _Resp() + + pf.httpx.AsyncClient = lambda **kw: _Client() + asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=1)) + + +def test_save_images_falls_back_to_the_image_stem_without_a_resource_name(): + """resource_name=None must not hand None to the storage layer.""" + parser = _parser(mineru_endpoint="https://x") + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + # a real StoragePath rejects None here + assert resource_name, "resource_name must not be None" + return Path("/media") / resource_name / (filename + extension) + + repl = asyncio.run(parser._mineru_save_images({"images/figure.png": b"png"}, _Storage(), None)) + assert repl == {"images/figure.png": "figure/figure.png"} + + +def test_image_refs_match_a_bare_key_against_a_prefixed_reference(): + """A bare key with an ``images/`` reference must not become images/doc/images/.""" + out = PDFParser._mineru_apply_image_refs( + "see ![f](images/figure.png)", {"figure.png": "doc/figure.png"} + ) + assert out == "see ![f](doc/figure.png)" + assert "images/doc" not in out + + +def test_image_refs_match_a_prefixed_key_against_a_bare_reference(): + out = PDFParser._mineru_apply_image_refs( + "see ![f](figure.png)", {"images/figure.png": "doc/figure.png"} + ) + assert out == "see ![f](doc/figure.png)" + + +def test_image_refs_leave_similar_names_alone(): + out = PDFParser._mineru_apply_image_refs( + "myfigure.png figure.pngx figure.png", {"figure.png": "doc/figure.png"} + ) + assert out.startswith("myfigure.png figure.pngx ") + assert out.endswith("doc/figure.png") + + +def test_selfhosted_failure_status_raises_even_with_partial_markdown(tmp_path): + """status="failed" plus partial md_content must not be returned as success.""" + pdf = tmp_path / "doc.pdf" + pdf.write_bytes(b"%PDF-1.4\n%%EOF\n") + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + + payload = { + "results": { + "doc.pdf": { + "status": "failed", + "md_content": "half a doc", + } + } + } + + class _Client: + async def post(self, *a, **kw): + return _FakeResponse(payload) + + try: + asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf, None, "doc")) + except ValueError as exc: + assert "failed to convert" in str(exc) + else: + raise AssertionError("expected ValueError for a failed entry") + + +def test_selfhosted_error_field_raises_even_when_markdown_is_present(tmp_path): + pdf = tmp_path / "doc.pdf" + pdf.write_bytes(b"%PDF-1.4\n%%EOF\n") + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + + payload = {"results": {"doc.pdf": {"md_content": "half a doc", "err_msg": "corrupt page tree"}}} + + class _Client: + async def post(self, *a, **kw): + return _FakeResponse(payload) + + try: + asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf, None, "doc")) + except ValueError as exc: + assert "corrupt page tree" in str(exc) + else: + raise AssertionError("expected ValueError when err_msg is set") + + +def test_selfhosted_success_status_still_returns_markdown(tmp_path): + pdf = tmp_path / "doc.pdf" + pdf.write_bytes(b"%PDF-1.4\n%%EOF\n") + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + + payload = {"results": {"doc.pdf": {"status": "done", "md_content": "# ok"}}} + + class _Client: + async def post(self, *a, **kw): + return _FakeResponse(payload) + + text, _meta = asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf, None, "doc")) + assert text == "# ok" + + +def test_unknown_strategy_is_rejected(monkeypatch, pdf_path): + monkeypatch.setattr(PDFConfig, "validate", lambda self: None) + parser = _parser(strategy="nonsense") + try: + asyncio.run(parser._convert_to_markdown(pdf_path)) + except ValueError as exc: + assert "Unknown strategy" in str(exc) + else: + raise AssertionError("expected ValueError") + + +def test_failure_message_falls_back_when_no_reason_is_reported(tmp_path): + """A failed entry with nothing to say still has to name the document.""" + entry = {"status": "failed"} + msg = PDFParser._mineru_failure_message(entry, tmp_path / "report.pdf") + assert "report.pdf" in msg + assert "status=failed" in msg + + +def test_entry_failure_detects_the_documented_states(): + for state in ("failed", "FAILED", " error ", "aborted", "cancelled", "timeout"): + assert PDFParser._mineru_entry_failed({"status": state}), state + assert PDFParser._mineru_entry_failed({"state": "rejected"}) + assert PDFParser._mineru_entry_failed({"err_no": "E42"}) + # a completed or silent entry is not a failure + for ok in ({"status": "done"}, {"status": "success"}, {}, {"md_content": "x"}): + assert not PDFParser._mineru_entry_failed(ok), ok + + +def test_local_image_rasteriser_returns_png_bytes(): + """The local path renders an image box to real PNG bytes.""" + parser = _parser(strategy="local") + saved = {} + + class _Img: + def save(self, buffer, format): + saved["format"] = format + buffer.write(b"\x89PNG") + + class _Crop: + def to_image(self, resolution=None): + saved["resolution"] = resolution + return _Img() + + class _Page: + width = 100 + height = 200 + + def crop(self, bbox): + saved["bbox"] = bbox + return _Crop() + + out = parser._extract_image_from_page(_Page(), {"x0": 5, "top": 10, "x1": 50, "bottom": 60}) + assert out == b"\x89PNG" + assert saved["format"] == "PNG" + assert saved["bbox"] == (5, 10, 50, 60) + + +def test_local_image_rasteriser_swallows_a_render_failure(): + """A broken image box costs that image, not the document.""" + parser = _parser(strategy="local") + + class _Page: + width = 100 + height = 200 + + def crop(self, bbox): + raise RuntimeError("cannot rasterise") + + assert ( + parser._extract_image_from_page(_Page(), {"x0": 0, "top": 0, "x1": 9, "bottom": 9}) is None + ) + + +def test_local_image_rasteriser_skips_a_degenerate_box(): + parser = _parser(strategy="local") + + class _Page: + width = 100 + height = 200 + + def crop(self, bbox): + raise AssertionError("a zero-area box must not be cropped") + + assert ( + parser._extract_image_from_page(_Page(), {"x0": 10, "top": 10, "x1": 10, "bottom": 40}) + is None + ) From bf8d6c2a959e6848928964f68adff70b4c9d80b2 Mon Sep 17 00:00:00 2001 From: q2000s Date: Fri, 16 Oct 2026 05:07:46 +0800 Subject: [PATCH 5/7] test(pdf): rename the dispatch-guard test to avoid shadowing config validation --- tests/parse/test_pdf_strategy_dispatch.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index fa6f6dc14c..b02f187dce 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -1113,7 +1113,8 @@ async def post(self, *a, **kw): assert text == "# ok" -def test_unknown_strategy_is_rejected(monkeypatch, pdf_path): +def test_parser_rejects_a_strategy_it_cannot_dispatch(monkeypatch, pdf_path): + """The parser's own dispatch guard, past config validation.""" monkeypatch.setattr(PDFConfig, "validate", lambda self: None) parser = _parser(strategy="nonsense") try: From d5c5cbee03f1310f6a12389a79f58eeabdb3b64b Mon Sep 17 00:00:00 2001 From: q2000s Date: Fri, 16 Oct 2026 05:16:37 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=EF=BB=BFtest(pdf):=20cover=20the=20preflig?= =?UTF-8?q?ht=20protocol=20branches=20and=20image=20edge=20cases?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds the regressions for the round-3 fixes: a hosted endpoint answering 404 on the probe is accepted, a rejected token is not, a self-hosted /health that answers non-JSON or a non-object body is not, an unwritable image costs that image alone, a key with no file name is skipped, and the local rasteriser returns PNG bytes / swallows a render failure / skips a degenerate box. The MinerU preflight wiring is now a small function, mineru_preflight_settings, so the hosted-vs-self-hosted choice can be tested without booting the service. Coverage of the lines this PR changes: 296/298 statements. The two remaining are the preflight call and its log line inside OpenVikingService.initialize, which only a full service startup reaches. --- openviking/service/core.py | 21 ++-- tests/parse/test_pdf_strategy_dispatch.py | 120 ++++++++++++++++++++++ 2 files changed, 132 insertions(+), 9 deletions(-) diff --git a/openviking/service/core.py b/openviking/service/core.py index 48a822d0df..4456fb651a 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -64,6 +64,16 @@ from openviking.session.compressor_v3 import SessionCompressorV3 +def mineru_preflight_settings(pdf_config) -> tuple[str, bool, str | None]: + """Decide how the startup preflight should probe a configured MinerU endpoint. + + Returns ``(endpoint, hosted, token)``. A token marks the hosted API, which serves + no ``/health`` route: probing it with the self-hosted health contract would fail + startup for a service that is reachable and correctly configured. + """ + return pdf_config.mineru_endpoint, bool(pdf_config.mineru_token), pdf_config.mineru_token + + class OpenVikingService: """ OpenViking main service class. @@ -691,16 +701,9 @@ async def initialize(self) -> None: ) if should_preflight_mineru and pdf_config.mineru_endpoint: - # A token marks the hosted API, which has no /health route. Probing it - # with the self-hosted contract would fail startup for a service that is - # reachable and correctly configured. - hosted = bool(pdf_config.mineru_token) + endpoint, hosted, token = mineru_preflight_settings(pdf_config) try: - await wait_for_mineru_ready( - pdf_config.mineru_endpoint, - hosted=hosted, - token=pdf_config.mineru_token, - ) + await wait_for_mineru_ready(endpoint, hosted=hosted, token=token) logger.info("MinerU preflight passed: %s", pdf_config.mineru_endpoint) except RuntimeError as exc: if pdf_config.strategy == "mineru": diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index b02f187dce..9e315bf90b 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -1202,3 +1202,123 @@ def crop(self, bbox): parser._extract_image_from_page(_Page(), {"x0": 10, "top": 10, "x1": 10, "bottom": 40}) is None ) + + +def test_selfhosted_preflight_rejects_a_response_that_is_not_json(): + """A 200 that is not the health contract must still fail the preflight.""" + import openviking.service.mineru_preflight as pf + + class _Resp: + status_code = 200 + + def json(self): + raise ValueError("not json") + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, **kw): + return _Resp() + + pf.httpx.AsyncClient = lambda **kw: _Client() + try: + asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=0.2)) + except RuntimeError as exc: + assert "preflight failed" in str(exc) + else: + raise AssertionError("expected RuntimeError") + + +def test_zip_image_save_failure_is_logged_and_skipped(pdf_path): + """One unwritable image must not lose the others or the document.""" + archive = _zip_bytes( + { + "full.md": b"![a](images/a.png) ![b](images/b.png)", + "images/a.png": b"one", + "images/b.png": b"two", + } + ) + + class _Storage: + media_dir = Path("/media") + + def save_image(self, resource_name, blob, filename, extension): + if filename == "a": + raise OSError("no space left") + return Path("/media") / resource_name / (filename + extension) + + parser = _hosted_parser() + text, meta = parser._mineru_unpack_zip(archive, pdf_path, {}, _Storage(), "doc") + assert meta["images_saved"] == 1 + assert "doc/b.png" in text + assert "images/a.png" in text # the one that failed is left as-is, not mis-linked + + +def test_failure_message_without_any_reason_just_names_the_file(tmp_path): + """A failed entry that reports nothing usable still names the document.""" + msg = PDFParser._mineru_failure_message({"failed": True}, tmp_path / "x.pdf") + assert msg == "MinerU failed to convert x.pdf" + + +def test_image_refs_skip_a_key_without_a_file_name(): + """A key that is only a directory has nothing to match on.""" + out = PDFParser._mineru_apply_image_refs("text images/ stays", {"images/": "doc/"}) + assert out == "text images/ stays" + + +# --------------------------------------------------------------------------- +# startup preflight wiring +# --------------------------------------------------------------------------- + + +def test_preflight_settings_treat_a_token_as_the_hosted_api(): + from openviking.service.core import mineru_preflight_settings + + cfg = PDFConfig( + strategy="mineru", + mineru_endpoint="https://mineru.example/api/v4", + mineru_token="hosted-token", + ) + endpoint, hosted, token = mineru_preflight_settings(cfg) + assert (endpoint, hosted, token) == ("https://mineru.example/api/v4", True, "hosted-token") + + +def test_preflight_settings_keep_the_selfhosted_contract_without_a_token(): + from openviking.service.core import mineru_preflight_settings + + cfg = PDFConfig(strategy="mineru", mineru_endpoint="http://127.0.0.1:8000") + endpoint, hosted, token = mineru_preflight_settings(cfg) + assert (endpoint, hosted, token) == ("http://127.0.0.1:8000", False, None) + + +def test_selfhosted_preflight_rejects_a_json_body_that_is_not_an_object(): + """HTTP 200 with a non-object body is not the health contract.""" + import openviking.service.mineru_preflight as pf + + class _Resp: + status_code = 200 + + def json(self): + return ["not", "an", "object"] + + class _Client: + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def get(self, url, **kw): + return _Resp() + + pf.httpx.AsyncClient = lambda **kw: _Client() + try: + asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=0.2)) + except RuntimeError as exc: + assert "unexpected response" in str(exc) + else: + raise AssertionError("expected RuntimeError") From fd6f109aa607877b011a8578620ff348abbeacb3 Mon Sep 17 00:00:00 2001 From: q2000s Date: Sat, 17 Oct 2026 03:34:29 +0800 Subject: [PATCH 7/7] =?UTF-8?q?=EF=BB=BFfix(pdf):=20key=20self-hosted=20im?= =?UTF-8?q?ages=20by=20document,=20isolate=20the=20preflight=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two problems from review of d5c5cbee. 1. Images from different documents overwrote each other when no resource_name was supplied. _mineru_save_images falls back to the image key's stem, so alpha.pdf and beta.pdf -- both carrying images/figure.png -- both wrote to figure/, and the second document replaced the first one's bytes. The zip path already fell back to the pdf stem; the self-hosted path now does the same, so each document gets its own directory. This is the old behaviour, restored. 2. The preflight tests replaced the shared httpx client without restoring it. pf.httpx is the httpx module itself, so "pf.httpx.AsyncClient = ..." left a fake client in place for every later test in the same process. They now use monkeypatch.setattr(pf.httpx, "AsyncClient", ...), which reverts per test. Both regressions are covered by tests that were checked against the unfixed code: reverting the image_dir fallback makes the two-document test fail with figure/figure.png, and injecting one global assignment fails the leak check. Tests: 99 pass across the dispatch and bookmark suites. Coverage of the lines this PR changes: 297/299 statements. The two remaining are the preflight call and its log line inside OpenVikingService.initialize, which only a full service startup reaches. --- openviking/parse/parsers/pdf.py | 6 +- tests/parse/test_pdf_strategy_dispatch.py | 142 ++++++++++++++++++++-- 2 files changed, 137 insertions(+), 11 deletions(-) diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index ace55db774..15498b4aed 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -1203,7 +1203,11 @@ async def _mineru_selfhosted( text = entry.get("md_content") or "" images = self._mineru_decode_inline_images(entry.get("images") or {}) - repl = await self._mineru_save_images(images, storage, resource_name) + # Fall back to the PDF's own stem, matching the zip path. Falling back to the + # image name would put every document's images under one directory, so two + # PDFs carrying a file with the same name would overwrite each other. + image_dir = resource_name or pdf_path.stem + repl = await self._mineru_save_images(images, storage, image_dir) meta["images_saved"] = len(repl) text = self._mineru_apply_image_refs(text, repl) logger.info(f"MinerU conversion: {len(text)} chars, {len(repl)} images") diff --git a/tests/parse/test_pdf_strategy_dispatch.py b/tests/parse/test_pdf_strategy_dispatch.py index 9e315bf90b..dee1ecd76f 100644 --- a/tests/parse/test_pdf_strategy_dispatch.py +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -918,7 +918,7 @@ async def _remote(pdf, resource_name=None): # --------------------------------------------------------------------------- -def test_hosted_preflight_accepts_the_missing_health_route(): +def test_hosted_preflight_accepts_the_missing_health_route(monkeypatch): """The hosted API has no /health; a 404 there must not fail startup. Probing it with the self-hosted contract would reject a service that is up and @@ -946,7 +946,7 @@ async def get(self, url, **kw): seen["headers"] = kw.get("headers") return _Resp() - pf.httpx.AsyncClient = lambda **kw: _Client() + monkeypatch.setattr(pf.httpx, "AsyncClient", lambda **kw: _Client()) asyncio.run( pf.wait_for_mineru_ready( "https://mineru.example/api/v4", timeout=1, hosted=True, token="tok" @@ -957,7 +957,7 @@ async def get(self, url, **kw): assert seen["headers"] == {"Authorization": "Bearer tok"} -def test_hosted_preflight_still_rejects_a_rejected_token(): +def test_hosted_preflight_still_rejects_a_rejected_token(monkeypatch): import openviking.service.mineru_preflight as pf class _Resp: @@ -976,7 +976,7 @@ async def __aexit__(self, *exc): async def get(self, url, **kw): return _Resp() - pf.httpx.AsyncClient = lambda **kw: _Client() + monkeypatch.setattr(pf.httpx, "AsyncClient", lambda **kw: _Client()) try: asyncio.run( pf.wait_for_mineru_ready( @@ -989,7 +989,7 @@ async def get(self, url, **kw): raise AssertionError("a rejected token must fail the preflight") -def test_selfhosted_preflight_still_demands_the_health_contract(): +def test_selfhosted_preflight_still_demands_the_health_contract(monkeypatch): import openviking.service.mineru_preflight as pf class _Resp: @@ -1008,7 +1008,7 @@ async def __aexit__(self, *exc): async def get(self, url, **kw): return _Resp() - pf.httpx.AsyncClient = lambda **kw: _Client() + monkeypatch.setattr(pf.httpx, "AsyncClient", lambda **kw: _Client()) asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=1)) @@ -1204,7 +1204,7 @@ def crop(self, bbox): ) -def test_selfhosted_preflight_rejects_a_response_that_is_not_json(): +def test_selfhosted_preflight_rejects_a_response_that_is_not_json(monkeypatch): """A 200 that is not the health contract must still fail the preflight.""" import openviking.service.mineru_preflight as pf @@ -1224,7 +1224,7 @@ async def __aexit__(self, *exc): async def get(self, url, **kw): return _Resp() - pf.httpx.AsyncClient = lambda **kw: _Client() + monkeypatch.setattr(pf.httpx, "AsyncClient", lambda **kw: _Client()) try: asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=0.2)) except RuntimeError as exc: @@ -1295,7 +1295,7 @@ def test_preflight_settings_keep_the_selfhosted_contract_without_a_token(): assert (endpoint, hosted, token) == ("http://127.0.0.1:8000", False, None) -def test_selfhosted_preflight_rejects_a_json_body_that_is_not_an_object(): +def test_selfhosted_preflight_rejects_a_json_body_that_is_not_an_object(monkeypatch): """HTTP 200 with a non-object body is not the health contract.""" import openviking.service.mineru_preflight as pf @@ -1315,10 +1315,132 @@ async def __aexit__(self, *exc): async def get(self, url, **kw): return _Resp() - pf.httpx.AsyncClient = lambda **kw: _Client() + monkeypatch.setattr(pf.httpx, "AsyncClient", lambda **kw: _Client()) try: asyncio.run(pf.wait_for_mineru_ready("http://127.0.0.1:8000", timeout=0.2)) except RuntimeError as exc: assert "unexpected response" in str(exc) else: raise AssertionError("expected RuntimeError") + + +# --------------------------------------------------------------------------- +# regressions: image directory collisions and test isolation +# --------------------------------------------------------------------------- + + +class _RecordingStorage: + """Storage that remembers every (resource_name, filename) it was asked to save.""" + + def __init__(self): + self.media_dir = Path("/media") + self.saved = {} + + def save_image(self, resource_name, blob, filename, extension): + target = Path("/media") / resource_name / (filename + extension) + # a later document writing the same path overwrites the earlier bytes + self.saved[target.as_posix()] = blob + return target + + +def _b64(payload: bytes) -> str: + import base64 as _b + + return "data:image/png;base64," + _b.b64encode(payload).decode("ascii") + + +def _selfhosted_client(markdown, image_bytes): + """A /file_parse reply carrying one inline image.""" + import base64 as _b + + encoded = "data:image/png;base64," + _b.b64encode(image_bytes).decode("ascii") + + class _Client: + async def post(self, *a, **kw): + return _FakeResponse( + { + "results": { + "doc.pdf": { + "md_content": markdown, + "images": {"images/figure.png": encoded}, + } + } + } + ) + + return _Client() + + +def test_two_pdfs_with_the_same_image_name_do_not_overwrite(tmp_path): + """Without a resource_name each PDF's images must land in its own directory. + + Both documents carry an images/figure.png. Falling back to the image name would + send both to figure/ and the second would overwrite the first. + """ + storage = _RecordingStorage() + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + + results = {} + for name, payload in (("alpha", b"alpha-bytes"), ("beta", b"beta-bytes")): + pdf = tmp_path / f"{name}.pdf" + pdf.write_bytes(b"%PDF-1.4\n%%EOF\n") + text, _meta = asyncio.run( + parser._mineru_selfhosted( + _selfhosted_client("![f](images/figure.png)", payload), + {}, + pdf, + storage, + None, # no resource_name: the parser must choose one + ) + ) + results[name] = text + + # each markdown points into its own document directory + assert "alpha/figure.png" in results["alpha"] + assert "beta/figure.png" in results["beta"] + assert "images/figure.png" not in results["alpha"] + + # and neither document clobbered the other's bytes + assert storage.saved["/media/alpha/figure.png"] == b"alpha-bytes" + assert storage.saved["/media/beta/figure.png"] == b"beta-bytes" + + +def test_an_explicit_resource_name_still_wins(tmp_path): + storage = _RecordingStorage() + parser = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + pdf = tmp_path / "gamma.pdf" + pdf.write_bytes(b"%PDF-1.4\n%%EOF\n") + + text, _meta = asyncio.run( + parser._mineru_selfhosted( + _selfhosted_client("![f](images/figure.png)", b"bytes"), + {}, + pdf, + storage, + "chosen", + ) + ) + assert "chosen/figure.png" in text + assert list(storage.saved) == ["/media/chosen/figure.png"] + + +def test_preflight_tests_do_not_leak_a_fake_http_client(): + """The preflight tests must restore the real httpx.AsyncClient. + + ``pf.httpx`` is the shared httpx module, so assigning to it replaces the client + for every later test in the process. + """ + import openviking.service.mineru_preflight as pf + + assert getattr(pf.httpx.AsyncClient, "__module__", "") == "httpx", ( + "httpx.AsyncClient is still patched after the preflight tests ran" + ) + + +def test_preflight_tests_do_not_leak_a_fake_client_object(): + """A second guard: the attribute must be httpx's own, not a lambda from a test.""" + import httpx + + import openviking.service.mineru_preflight as pf + + assert pf.httpx.AsyncClient is httpx.AsyncClient