diff --git a/docs/en/about/02-changelog.md b/docs/en/about/02-changelog.md index eee9c12ff1..857ab43390 100644 --- a/docs/en/about/02-changelog.md +++ b/docs/en/about/02-changelog.md @@ -7,6 +7,33 @@ This changelog is automatically generated from [GitHub Releases](https://github. ## Unreleased +- **`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: + `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. + - **LangChain import migration (breaking change)**: The server no longer ships `openviking.integrations.langchain` or its forwarding modules. Old imports fail immediately; there is no deprecation period or compatibility stub. Install diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index fc78e2fa39..462772acde 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -1038,7 +1038,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 { @@ -1061,8 +1061,67 @@ 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. + +`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. + +:::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 4b3c44461c..d3102a79c1 100644 --- a/docs/zh/about/02-changelog.md +++ b/docs/zh/about/02-changelog.md @@ -7,6 +7,10 @@ OpenViking 的所有重要变更都将记录在此文件中。 ## 未发布 +- **`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` 取值。 + - **LangChain 导入迁移（不兼容变更）**：服务端不再提供 `openviking.integrations.langchain` 及其转发模块，旧导入会立即失败， 不提供弃用过渡期或兼容占位模块。请安装 `langchain-openviking` diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index 7a219cc789..5e965e254a 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -1032,7 +1032,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 { @@ -1055,8 +1055,44 @@ 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` 不会无目的地重试： + +`auto` 依次尝试每个已配置的远程 provider，并使用第一个成功的。某个 provider 抛出异常时只记录日志并尝试下一个，因此任何远程失败（`404`、`405`、`429`、超时、`5xx`）都会使该文档降级为本地解析，而不会使导入失败。实际走的路径记录在 +`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..15498b4aed 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, Tuple, Union from openviking.parse.base import ( NodeType, @@ -31,12 +32,17 @@ 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 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() @@ -53,7 +59,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 @@ -156,6 +162,12 @@ 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 result.meta["intermediate_markdown_length"] = len(markdown_content) result.meta["intermediate_markdown_preview"] = markdown_content[:500] @@ -200,27 +212,38 @@ 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(): + try: + return await convert(pdf_path, resource_name=resource_name) + except Exception as e: + # 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}") + 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 +709,157 @@ 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. + """ + if self.config.mineru_token: + return {"Authorization": f"Bearer {self.config.mineru_token}"} + return {} + + def _mineru_base_url(self) -> str: + return self.config.mineru_endpoint.rstrip("/") + + 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. """ - Convert PDF to Markdown using the MinerU /file_parse API. + 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 - 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 + @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 - Returns: - Tuple of (markdown_content, metadata) where metadata includes - strategy, endpoint, api_version, backend, task_id, processing_time - (seconds) and images_saved + @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", + ) - Raises: - ValueError: If MinerU endpoint is not configured, or the task does - not complete - Exception: If the API call fails - """ - httpx = lazy_import("httpx") + @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]}" + # 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]: + 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 - if not self.config.mineru_endpoint: - raise ValueError("MinerU endpoint not configured") + async def _mineru_download_zip(self, client, zip_url: str) -> bytes: + """Download the result archive. - # 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" + 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, + follow_redirects=True, + ) + response.raise_for_status() + return response.content - meta = { - "strategy": "mineru", - "endpoint": self.config.mineru_endpoint, - "api_version": None, - } + 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 - 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() + meta["api_version"] = "task-zip" if not markdown_content: logger.warning(f"MinerU returned empty content for {pdf_path}") @@ -779,42 +873,345 @@ 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_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.""" + results = payload.get("results") or {} + if not isinstance(results, dict): + return None + entry = results.get(pdf_path.name) or next(iter(results.values()), 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.""" + 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], + ) -> 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 {} + + 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) + # 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( + target, + 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 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 {}) + # 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") + 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 593c4c2308..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. @@ -687,12 +697,13 @@ 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: + endpoint, hosted, token = mineru_preflight_settings(pdf_config) try: - await wait_for_mineru_ready(pdf_config.mineru_endpoint) + 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/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/openviking_cli/utils/config/parser_config.py b/openviking_cli/utils/config/parser_config.py index 26cfcd99ce..8b80fc29ef 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,36 @@ 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``. + """ + # 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( + endpoint + for endpoint in ( + self.mineru_endpoint, # MinerU provider + # future providers register their own config here + ) + if endpoint + ) + 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..dee1ecd76f --- /dev/null +++ b/tests/parse/test_pdf_strategy_dispatch.py @@ -0,0 +1,1446 @@ +"""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" + + +# --------------------------------------------------------------------------- +# regressions raised in review: default-config startup, credential isolation, +# and image-reference rewriting +# --------------------------------------------------------------------------- + + +def test_default_pdfconfig_reports_no_remote_provider(): + """A deployment that never configured MinerU must still start. + + ``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 + + config = PDFConfig() + assert config.mineru_endpoint is None + assert config.remote_enabled is False + + +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 + + config.mineru_endpoint = "" + assert config.remote_enabled is False + + +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 + + parser = PDFParser(PDFConfig(mineru_endpoint=ENDPOINT, mineru_token="secret-token")) + seen = {} + + class _Resp: + content = b"zip-bytes" + + def raise_for_status(self): + return None + + 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)) + + 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" + + +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 + + monkeypatch.setattr( + PDFParser, + "_mineru_base_url", + lambda self: ENDPOINT, + ) + monkeypatch.setattr( + PDFParser, + "_mineru_file_parse_url", + lambda self: ENDPOINT + "/file_parse", + ) + + 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()) + + parser = PDFParser(PDFConfig(strategy="mineru", mineru_endpoint=ENDPOINT)) + text, meta = asyncio.run(parser._mineru_selfhosted(_Client(), {}, pdf_path, _Storage(), "math")) + + 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 + + +# --------------------------------------------------------------------------- +# 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 = _parser(strategy="mineru", mineru_endpoint="https://mineru.example") + 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" + + +# --------------------------------------------------------------------------- +# review round 3: hosted preflight, self-hosted image keys, failure status +# --------------------------------------------------------------------------- + + +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 + 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() + + 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" + ) + ) + # 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(monkeypatch): + 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() + + monkeypatch.setattr(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(monkeypatch): + 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() + + monkeypatch.setattr(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_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: + 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 + ) + + +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 + + 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() + + 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 "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(monkeypatch): + """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() + + 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