diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md index 8bdfb2c981..d85090e4be 100644 --- a/docs/en/guides/01-configuration.md +++ b/docs/en/guides/01-configuration.md @@ -892,11 +892,32 @@ PDF parsing configuration. Three strategies are supported: `local` (local pdfplu | Parameter | Type | Description | |-----------|------|-------------| | `strategy` | str | Parsing strategy: `local` / `mineru` / `auto` (default `auto`) | -| `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_endpoint` | str | MinerU API **base URL** (e.g. `http://127.0.0.1:8000`, or `https://mineru.net/api/v4` for the online API) | +| `mineru_timeout` | float | Request/polling timeout in seconds (default `300.0`) | +| `mineru_bodys` | dict | MinerU API multipart form fields (passed through) | +| `mineru_api_mode` | str | Protocol flavor: `auto` (detect from the response, default) / `sync` (legacy inline `/file_parse`) / `async` (task APIs) | +| `mineru_token` | str | Optional bearer token for the online MinerU API (sent as `Authorization: Bearer ...`) | -**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 protocols** — three flavors are supported: + +- **Legacy self-hosted (`sync`)**: a single `POST {mineru_endpoint}/file_parse` answering inline with the markdown; form parameters are passed through from `mineru_bodys`. +- **Current self-hosted (`async`)**: `POST {mineru_endpoint}/tasks` answers `202` with `task_id`/`status_url`/`result_url`; the task is polled until `completed` and the markdown is downloaded from the result zip. +- **Online batch API (`async`)**: `POST {mineru_endpoint}/extract/task/batch` (Bearer-token auth via `mineru_token`), poll `{mineru_endpoint}/extract-results/batch/{batch_id}` until `state=done`, download `full_zip_url` and unpack. + +Under the default `mineru_api_mode=auto` the flavor is detected automatically: the inline endpoint is tried first and a `404` (or a task-shaped response) switches to the task flow, so existing self-hosted v1 deployments keep working without config changes. The startup `/health` preflight only applies to the sync flavor (the online service has no such endpoint). + +**Online example:** + +```json +{ + "pdf": { + "strategy": "mineru", + "mineru_endpoint": "https://mineru.net/api/v4", + "mineru_api_mode": "auto", + "mineru_token": "your-mineru-api-token" + } +} +``` ### rerank diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md index 2bd31b0f19..8270e3b40d 100644 --- a/docs/zh/guides/01-configuration.md +++ b/docs/zh/guides/01-configuration.md @@ -860,11 +860,32 @@ PDF 解析配置。支持三种策略：`local`（本地 pdfplumber）、`mineru | 参数 | 类型 | 说明 | |------|------|------| | `strategy` | str | 解析策略：`local` / `mineru` / `auto`（默认 `auto`） | -| `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_endpoint` | str | MinerU API **base URL**（如自托管 `http://127.0.0.1:8000`，在线服务 `https://mineru.net/api/v4`） | +| `mineru_timeout` | float | 请求/轮询超时秒数（默认 `300.0`） | +| `mineru_bodys` | dict | MinerU API multipart form 参数（透传） | +| `mineru_api_mode` | str | 协议形态：`auto`（按响应自动识别，默认）/ `sync`（旧版内联 `/file_parse`）/ `async`（任务式 API） | +| `mineru_token` | str | 在线 MinerU API 的 bearer token（以 `Authorization: Bearer ...` 发送） | -**MinerU 协议**：同步调用 `POST {mineru_endpoint}/file_parse`，multipart 文件字段为 `files`，form 参数由 `mineru_bodys` 透传。 +**MinerU 协议**——支持三种形态： + +- **旧版自托管（`sync`）**：单次 `POST {mineru_endpoint}/file_parse` 内联返回 markdown；form 参数由 `mineru_bodys` 透传。 +- **新版自托管（`async`）**：`POST {mineru_endpoint}/tasks` 返回 `202` 与 `task_id`/`status_url`/`result_url`；轮询任务至 `completed` 后从结果 zip 下载 markdown。 +- **在线批量 API（`async`）**：`POST {mineru_endpoint}/extract/task/batch`（通过 `mineru_token` 做 Bearer 认证），轮询 `{mineru_endpoint}/extract-results/batch/{batch_id}` 至 `state=done`，下载 `full_zip_url` 并解包。 + +默认 `mineru_api_mode=auto` 时自动识别协议形态：先尝试内联端点，返回 `404`（或任务式响应）即切换到任务流，存量自托管 v1 部署无需改配置。启动时的 `/health` 预检仅适用于 sync 形态（在线服务没有该端点）。 + +**在线服务示例：** + +```json +{ + "pdf": { + "strategy": "mineru", + "mineru_endpoint": "https://mineru.net/api/v4", + "mineru_api_mode": "auto", + "mineru_token": "your-mineru-api-token" + } +} +``` ### rerank diff --git a/openviking/parse/parsers/pdf.py b/openviking/parse/parsers/pdf.py index f299a0680d..734bb6d9cf 100644 --- a/openviking/parse/parsers/pdf.py +++ b/openviking/parse/parsers/pdf.py @@ -17,6 +17,7 @@ import hashlib import io import re +import tempfile import time from collections import Counter, defaultdict from pathlib import Path @@ -30,6 +31,7 @@ lazy_import, ) from openviking.parse.parsers.base_parser import BaseParser +from openviking.utils.zip_safe import normalize_zip_filenames 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 @@ -680,6 +682,66 @@ def _extract_image_from_page(self, page, img_info: dict) -> Optional[bytes]: logger.debug(f"Image extraction error: {e}") return None + def _mineru_auth_headers(self) -> Dict[str, str]: + """Authorization headers for the online MinerU API (empty when unset).""" + 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 _mineru_file_parse_url(self) -> str: + """Upload URL: ``/file_parse`` unless the endpoint already ends with it.""" + 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 parsed JSON body.""" + with open(pdf_path, "rb") as f: + files = {"files": (pdf_path.name, f, "application/pdf")} + logger.info(f"Calling MinerU API: {url}") + response = await client.post( + url, + files=files, + data=data, + headers=self._mineru_auth_headers(), + ) + response.raise_for_status() + return response.json() + + @staticmethod + def _is_http_404(exc: Exception) -> bool: + status = getattr(exc, "response", None) + return getattr(status, "status_code", None) == 404 + + @staticmethod + def _mineru_payload_data(result: Dict[str, Any]) -> Dict[str, Any]: + data = result.get("data") + return data if isinstance(data, dict) else {} + + @staticmethod + def _mineru_response_flavor(result: Dict[str, Any]) -> Optional[str]: + """Classify a create/submit response as one of the three protocols. + + Returns ``"v1-sync"``, ``"v2-tasks"``, ``"online-batch"`` or None when + the shape is unrecognized. + """ + data = PDFParser._mineru_payload_data(result) + if isinstance(result.get("task_id"), str) and isinstance( + result.get("status_url") or result.get("result_url"), str + ): + return "v2-tasks" + if isinstance(data.get("batch_id"), str): + return "online-batch" + if result.get("status") == "completed" and isinstance( + result.get("results"), dict + ): + return "v1-sync" + return None + async def _convert_mineru( self, pdf_path: Path, @@ -687,7 +749,23 @@ async def _convert_mineru( resource_name: Optional[str] = None, ) -> tuple[str, Dict[str, Any]]: """ - Convert PDF to Markdown using the MinerU /file_parse API. + Convert PDF to Markdown via the MinerU API. + + Three protocol flavors are supported, selected by + ``PDFConfig.mineru_api_mode``: + + - ``"sync"``: the legacy self-hosted single-shot contract — one + ``POST /file_parse`` answering inline with + ``{"status": "completed", "results": ...}``. + - ``"async"``: the task-based contracts — the current self-hosted + ``POST /tasks`` API (202 + ``status_url``/``result_url``) + and the online batch API (``/extract/task/batch`` + polling + ``/extract-results/batch/{id}``, Bearer-token auth). Both deliver + the markdown inside a zip archive. The self-hosted endpoint is + probed first; a 404 falls through to the online endpoint. + - ``"auto"`` (default): the flavor is detected — the inline POST is + attempted first and a 404 (or a task-shaped response) switches to + the task flow, so existing v1 deployments keep working unchanged. Args: pdf_path: Path to PDF file @@ -698,8 +776,8 @@ async def _convert_mineru( Returns: Tuple of (markdown_content, metadata) where metadata includes - strategy, endpoint, api_version, backend, task_id, processing_time - (seconds) and images_saved + strategy, endpoint, api_mode, api_version, backend, task_id, + processing_time (seconds) and images_saved Raises: ValueError: If MinerU endpoint is not configured, or the task does @@ -711,11 +789,7 @@ async def _convert_mineru( if not self.config.mineru_endpoint: raise ValueError("MinerU endpoint not configured") - # 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" - - meta = { + meta: Dict[str, Any] = { "strategy": "mineru", "endpoint": self.config.mineru_endpoint, "api_version": None, @@ -723,43 +797,261 @@ async def _convert_mineru( 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 + mode = self.config.mineru_api_mode + if mode != "async": data: Dict[str, Any] = dict(self.config.mineru_bodys or {}) - # MinerU must return extracted images for the markdown refs. data["return_images"] = True + try: + result = await self._mineru_post( + client, self._mineru_file_parse_url(), pdf_path, data + ) + except Exception as exc: + if mode == "sync" or not self._is_http_404(exc): + raise + # auto: /file_parse is gone -> task-based protocol + result = None + if result is not None: + flavor = self._mineru_response_flavor(result) + if mode == "sync" or flavor in (None, "v1-sync"): + meta["api_mode"] = "v1-sync" + return await self._convert_mineru_v1( + result, pdf_path, meta, + storage=storage, resource_name=resource_name, + ) + return await self._mineru_continue_task( + client, result, meta, pdf_path, + storage=storage, resource_name=resource_name, + ) + return await self._mineru_run_task_flow( + client, meta, pdf_path, + storage=storage, resource_name=resource_name, + ) + except Exception as e: + logger.error(f"MinerU API call failed: {e}") + raise - # Make API request - logger.info(f"Calling MinerU API: {url}") - response = await client.post( - url, - files=files, - data=data, - ) - response.raise_for_status() - - # Parse response - result = response.json() - if result.get("status") != "completed": - raise ValueError(f"MinerU task not completed: {result.get('status')}") - - results = result.get("results") or {} - file_result = results.get(pdf_path.name) or next(iter(results.values()), {}) - markdown_content = file_result.get("md_content") or "" - - # Extract metadata from response - meta["api_version"] = result.get("version") - meta["backend"] = result.get("backend") - meta["task_id"] = result.get("task_id") - started_at, completed_at = result.get("started_at"), result.get("completed_at") - if started_at and completed_at: - meta["processing_time"] = ( - parse_iso_datetime(completed_at) - parse_iso_datetime(started_at) - ).total_seconds() + async def _mineru_run_task_flow( + self, client, meta: Dict[str, Any], pdf_path: Path, **kwargs + ) -> tuple[str, Dict[str, Any]]: + """Explicit async mode: probe the task endpoints until one accepts.""" + errors = [] + base = self._mineru_base_url() + for url, kind, data in ( + (f"{base}/tasks", "v2-tasks", self._mineru_v2_form()), + (f"{base}/extract/task/batch", "online-batch", self._mineru_online_form()), + ): + try: + result = await self._mineru_post(client, url, pdf_path, data) + except Exception as exc: + if self._is_http_404(exc): + errors.append(f"{url}: 404") + continue + raise + flavor = self._mineru_response_flavor(result) + if flavor in (None, kind): + return await self._mineru_continue_task( + client, result, meta, pdf_path, **kwargs + ) + errors.append(f"{url}: unexpected response shape") + raise ValueError( + "No MinerU task endpoint responded on " + f"{base} (tried /tasks, /extract/task/batch; {errors}). " + "Check pdf.mineru_endpoint and, for the online API, pdf.mineru_token." + ) + + def _mineru_v2_form(self) -> Dict[str, Any]: + """Form fields for the current self-hosted /tasks API.""" + data: Dict[str, Any] = { + "return_md": "true", + "return_images": "true", + "response_format_zip": "true", + } + data.update(self.config.mineru_bodys or {}) + return data + + def _mineru_online_form(self) -> Dict[str, Any]: + """Form fields for the online batch API (pass-through of mineru_bodys).""" + return dict(self.config.mineru_bodys or {}) + + async def _mineru_continue_task( + self, + client, + result: Dict[str, Any], + meta: Dict[str, Any], + pdf_path: Path, + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Drive an already-submitted task to completion and unpack the zip.""" + flavor = self._mineru_response_flavor(result) or "v2-tasks" + meta["api_mode"] = flavor + base = self._mineru_base_url() + + if flavor == "v2-tasks": + task_id = result["task_id"] + meta["task_id"] = task_id + status_url = self._mineru_absolute(result.get("status_url"), base) + result_url = self._mineru_absolute(result.get("result_url"), base) + logger.info(f"MinerU task {task_id} submitted; polling {status_url}") + deadline = time.monotonic() + self.config.mineru_timeout + await self._mineru_poll( + client, status_url, deadline, + done_when=lambda body: body.get("status") == "completed", + busy_when=lambda body: body.get("status") in ("pending", "processing"), + failure_detail=self._mineru_failure_detail, + ) + zip_url = result_url + else: # online-batch + batch_id = self._mineru_payload_data(result)["batch_id"] + meta["task_id"] = batch_id + poll_url = f"{base}/extract-results/batch/{batch_id}" + logger.info(f"MinerU online batch {batch_id} submitted; polling {poll_url}") + deadline = time.monotonic() + self.config.mineru_timeout + await self._mineru_poll( + client, poll_url, deadline, + done_when=self._mineru_online_done, + busy_when=self._mineru_online_busy, + failure_detail=self._mineru_online_failure_detail, + ) + zip_url = self._mineru_online_zip_url( + await self._mineru_get_json(client, poll_url) + ) + if not zip_url: + raise ValueError( + "MinerU online task completed but no zip url found in " + f"{poll_url} response" + ) + + zip_bytes = await self._mineru_download_zip(client, zip_url) + markdown_content, meta = self._mineru_unpack_zip( + zip_bytes, pdf_path, meta, + storage=storage, resource_name=resource_name, + ) + logger.info(f"MinerU {flavor} conversion: {len(markdown_content)} chars") + return markdown_content, meta + + @staticmethod + def _mineru_absolute(url: Optional[str], base: str) -> str: + if url and url.startswith(("http://", "https://")): + return url + return base + (url or "") + + async def _mineru_poll( + self, client, url: str, deadline: float, + done_when, busy_when, failure_detail, + ) -> None: + """Poll until ``done_when`` holds; raise on failure or timeout.""" + last_state = None + while time.monotonic() < deadline: + body = await self._mineru_get_json(client, url) + if done_when(body): + return + if busy_when(body): + state = repr(body)[:120] + if state != last_state: + logger.info(f"MinerU task still busy: {state}") + last_state = state + await asyncio.sleep(3.0) + continue + raise ValueError( + f"MinerU task failed: {failure_detail(body)}" + ) + raise TimeoutError( + f"MinerU task did not finish within {self.config.mineru_timeout}s: {url}" + ) + + async def _mineru_get_json(self, client, url: str) -> Dict[str, Any]: + response = await client.get(url, headers=self._mineru_auth_headers()) + response.raise_for_status() + return response.json() + + @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 + + @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) + return bool(entry) and entry.get("state") in ("waiting", "pending", "running") + + @classmethod + def _mineru_online_failure_detail(cls, body: Dict[str, Any]) -> str: + entry = cls._mineru_online_result_entry(body) or {} + for key in ("err_msg", "err_no", "state", "message", "msg"): + if entry.get(key): + return f"{key}={entry[key]}" + return cls._mineru_failure_detail(body) + + @staticmethod + def _mineru_failure_detail(body: Dict[str, Any]) -> str: + for key in ("msg", "message", "error", "err_msg"): + value = body.get(key) + if isinstance(value, str) and value: + return value + return repr(body)[:200] + + @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 + + async def _mineru_download_zip(self, client, zip_url: str) -> bytes: + """Download the result archive (signed urls may already carry auth).""" + response = await client.get( + zip_url, + headers=self._mineru_auth_headers(), + follow_redirects=True, + ) + response.raise_for_status() + return response.content + + def _mineru_unpack_zip( + self, + zip_bytes: bytes, + pdf_path: Path, + meta: Dict[str, Any], + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Unpack the zip result: locate the markdown and persist images. + + The archive layout follows the standard MinerU export: a markdown file + (``full.md`` preferred) plus an ``images/`` directory it references. + """ + from openviking.utils.zip_safe import safe_extract_zip + + import zipfile + + with tempfile.TemporaryDirectory(prefix="mineru-") as tmp: + tmp_dir = Path(tmp) + zf = zipfile.ZipFile(io.BytesIO(zip_bytes)) + normalize_zip_filenames(zf) + safe_extract_zip(zf, tmp_dir) + zf.close() + + md_path = self._mineru_pick_markdown(tmp_dir) + if md_path is None: + raise ValueError( + f"No markdown file found in MinerU zip result extracted to {tmp_dir}" + ) + markdown_content = md_path.read_text(encoding="utf-8", errors="replace") + root_dir = md_path.parent + + meta["api_version"] = "task-zip" if not markdown_content: logger.warning(f"MinerU returned empty content for {pdf_path}") @@ -773,89 +1065,128 @@ 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 + # (``images/``) to the stored relative paths. 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)) + "|".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 ) - 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 + + @staticmethod + def _mineru_pick_markdown(root_dir: Path) -> Optional[Path]: + """Pick the document markdown from an extracted MinerU zip. + Preference order: ``full.md`` next to the images dir, any single + ``*.md`` in a flat layout, else the shallowest ``*.md``. + """ + 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)) + + async def _convert_mineru_v1( + self, + result: Dict[str, Any], + pdf_path: Path, + meta: Dict[str, Any], + storage=None, + resource_name: Optional[str] = None, + ) -> tuple[str, Dict[str, Any]]: + """Handle the legacy inline response (``md_content`` + base64 images).""" + 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() + + if not markdown_content: + logger.warning(f"MinerU returned empty content for {pdf_path}") return markdown_content, meta - except Exception as e: - logger.error(f"MinerU API call failed: {e}") - raise + if storage is None: + from openviking_cli.utils.storage import get_storage - def _format_table_markdown(self, table: List[List[Optional[str]]]) -> str: - """ - Convert table data to Markdown table format. + storage = get_storage() - Args: - table: 2D array of table cells + if resource_name is None: + resource_name = pdf_path.stem - Returns: - Markdown table string - - Examples: - >>> table = [["Name", "Age"], ["Alice", "30"], ["Bob", "25"]] - >>> print(parser._format_table_markdown(table)) - | Name | Age | - | --- | --- | - | Alice | 30 | - | Bob | 25 | - """ - if not table or not table[0]: - return "" - - # Clean cells and handle None values - def clean_cell(cell): - if cell is None: - return "" - return str(cell).strip().replace("|", "\\|") # Escape pipe characters - - lines = [] - - # Header row - header = table[0] - header_cells = [clean_cell(cell) for cell in header] - lines.append("| " + " | ".join(header_cells) + " |") - - # Separator row - separator = ["---"] * len(header) - lines.append("| " + " | ".join(separator) + " |") - - # Data rows - for row in table[1:]: - # Pad row to match header length - padded_row = row + [None] * (len(header) - len(row)) - cells = [clean_cell(cell) for cell in padded_row[: len(header)]] - lines.append("| " + " | ".join(cells) + " |") - - return "\n".join(lines) + # 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. + 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}") + + 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 async def parse_content( self, content: str, source_path: Optional[str] = None, instruction: str = "", **kwargs diff --git a/openviking/service/core.py b/openviking/service/core.py index 51bc2ed06f..02b61bcff8 100644 --- a/openviking/service/core.py +++ b/openviking/service/core.py @@ -513,9 +513,10 @@ async def initialize(self) -> None: # misconfiguration or a stopped service surfaces now instead of on the # 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 - ) + should_preflight_mineru = ( + pdf_config.strategy == "mineru" + or (pdf_config.strategy == "auto" and pdf_config.mineru_endpoint is not None) + ) and pdf_config.mineru_api_mode != "async" if should_preflight_mineru and pdf_config.mineru_endpoint: try: diff --git a/openviking_cli/utils/config/parser_config.py b/openviking_cli/utils/config/parser_config.py index 2423637dc3..4e9e24c485 100644 --- a/openviking_cli/utils/config/parser_config.py +++ b/openviking_cli/utils/config/parser_config.py @@ -149,6 +149,10 @@ class PDFConfig(ParserConfig): mineru_endpoint: MinerU API endpoint URL mineru_timeout: MinerU request timeout in seconds mineru_bodys: Additional MinerU API multipart form fields + mineru_api_mode: MinerU protocol flavor — "auto" (detect from the + response shape), "sync" (self-hosted single-shot /file_parse) or + "async" (online task API: create task, poll, download zip) + mineru_token: Optional bearer token for the online MinerU API """ strategy: str = "auto" # "local" | "mineru" | "auto" @@ -157,6 +161,12 @@ 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 + # "auto" (default) detects the protocol from the response shape; "sync" + # forces the self-hosted single-shot /file_parse contract; "async" forces + # the online task API (create task -> poll -> download zip result). + mineru_api_mode: str = "auto" # "auto" | "sync" | "async" + # Bearer token for the online MinerU API (sent as Authorization header). + mineru_token: Optional[str] = None # Heading detection configuration heading_detection: str = "auto" # "bookmarks" | "font" | "auto" | "none" @@ -189,6 +199,12 @@ def validate(self) -> None: if self.mineru_timeout <= 0: raise ValueError("mineru_timeout must be positive") + if self.mineru_api_mode not in ("auto", "sync", "async"): + raise ValueError( + f"Invalid mineru_api_mode '{self.mineru_api_mode}'. " + "Must be 'auto', 'sync' or 'async'" + ) + if self.heading_detection not in ("bookmarks", "font", "auto", "none"): raise ValueError(f"Invalid heading_detection: {self.heading_detection}") diff --git a/tests/parse/test_pdf_mineru.py b/tests/parse/test_pdf_mineru.py new file mode 100644 index 0000000000..2947dac9f7 --- /dev/null +++ b/tests/parse/test_pdf_mineru.py @@ -0,0 +1,431 @@ +# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd. +# SPDX-License-Identifier: AGPL-3.0 +"""Unit tests for the MinerU protocol flavors in PDFParser. + +Covers the three supported protocols: +- v1-sync: legacy self-hosted inline contract (md_content + base64 images), + unchanged under ``mineru_api_mode="sync"`` and auto-detected under "auto" +- v2-tasks: current self-hosted task API (POST /tasks -> status_url/result_url) +- online-batch: the online batch API (POST /extract/task/batch -> poll + extract-results -> full_zip_url, Bearer-token auth) + +plus the failure surfaces and config validation. +""" + +import base64 +import io +import zipfile +from pathlib import Path + +import httpx +import pytest + +from openviking.parse.parsers.pdf import PDFParser +from openviking_cli.utils.config.parser_config import PDFConfig + +BASE = "https://mineru.example" +ENDPOINT = f"{BASE}/api/v4" +FILE_PARSE_URL = f"{ENDPOINT}/file_parse" +TASKS_URL = f"{ENDPOINT}/tasks" +BATCH_URL = f"{ENDPOINT}/extract/task/batch" + +FAKE_PDF = b"%PDF-1.4 minimal" + + +class _FakeStorage: + """Media storage double: records saved images under a fake media dir.""" + + def __init__(self, tmp_path: Path): + self.media_dir = tmp_path / "media" + self.media_dir.mkdir() + self.saved = [] + + def save_image(self, resource_name, image_bytes, filename="", extension=".png"): + target = self.media_dir / f"{resource_name}-{filename}{extension}" + target.write_bytes(image_bytes) + self.saved.append(target.name) + return target + + +class _FakeResponse: + def __init__(self, payload=None, content=b"", status_code=200): + self._payload = payload + self.content = content + self.status_code = status_code + + def json(self): + if self._payload is None: + raise ValueError("no json") + return self._payload + + def raise_for_status(self): + if self.status_code >= 400: + raise httpx.HTTPStatusError( + f"HTTP {self.status_code}", request=None, response=self + ) + + +class _RoutedClient: + """Routes POST/GET by url substring to scripted response queues. + + One dict for POSTs (endpoint keys) and one for GETs (poll/result keys) so + a poll url like ``/tasks//status`` never collides with the ``/tasks`` + POST key. POSTs always replay the first entry; GETs consume entries in + order and the last one repeats. + """ + + def __init__(self, routes, get_routes=None): + self._routes = routes + self._get_routes = get_routes if get_routes is not None else routes + self.posts = [] + self.gets = [] + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def post(self, url, **kwargs): + self.posts.append((url, kwargs)) + for key, queue in self._routes.items(): + if key in url: + return queue[0] + raise AssertionError(f"unexpected POST {url}") + + async def get(self, url, **kwargs): + self.gets.append((url, kwargs)) + for key, queue in self._get_routes.items(): + if key in url: + if len(queue) > 1: + return queue.pop(0) + return queue[0] + raise AssertionError(f"unexpected GET {url}") + + def __call__(self, **kwargs): + return self + + +def _split(routes): + """POST endpoint keys vs GET url keys (keys only reachable by GET).""" + post_keys = ("file_parse", "/tasks", "/extract/task/batch") + posts = {k: v for k, v in routes.items() if any(p in k for p in post_keys)} + gets = {k: v for k, v in routes.items() if k not in posts} + return _RoutedClient(posts, gets) + + +def _sync_payload(md="# Doc\n\n![img](images/a.jpg)"): + data_url = "data:image/jpeg;base64," + base64.b64encode(b"jpegdata").decode() + return { + "status": "completed", + "version": 2, + "backend": "pipeline", + "task_id": "t-1", + "results": {"doc.pdf": {"md_content": md, "images": {"a.jpg": data_url}}}, + } + + +def _v2_create_payload(): + return { + "task_id": "task-42", + "status_url": f"{ENDPOINT}/tasks/task-42/status", + "result_url": f"{ENDPOINT}/tasks/task-42/result", + "file_names": ["doc.pdf"], + "queued_ahead": 0, + } + + +def _online_create_payload(): + return {"code": 0, "msg": "success", "data": {"batch_id": "batch-77"}} + + +def _result_zip(md="# Online\n\n![pic](images/pic.jpg)"): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("full.md", md) + zf.writestr("images/pic.jpg", b"fakejpg") + zf.writestr("layout.pdf", b"fakepdf") + return buf.getvalue() + + +@pytest.fixture +def pdf_path(tmp_path): + p = tmp_path / "doc.pdf" + p.write_bytes(FAKE_PDF) + return p + + +def _patch_client(monkeypatch, client): + monkeypatch.setattr(httpx, "AsyncClient", lambda **kwargs: client) + + +def _parser(**overrides): + defaults = dict(strategy="mineru", mineru_endpoint=ENDPOINT) + defaults.update(overrides) + return PDFParser(PDFConfig(**defaults)) + + +# --------------------------------------------------------------------------- +# v1 sync flavor + + +@pytest.mark.asyncio +async def test_v1_sync_mode_unchanged(monkeypatch, pdf_path, tmp_path): + """Explicit sync mode: identical contract as before the task support.""" + client = _RoutedClient({FILE_PARSE_URL: [_FakeResponse(_sync_payload())]}) + parser = _parser(mineru_api_mode="sync") + _patch_client(monkeypatch, client) + storage = _FakeStorage(tmp_path) + + md, meta = await parser._convert_mineru(pdf_path, storage=storage, resource_name="doc") + + assert md.startswith("# Doc") + assert meta["api_mode"] == "v1-sync" + assert meta["backend"] == "pipeline" + assert meta["images_saved"] == 1 + assert "doc-a.jpg" in md # reference rewritten to the stored relative path + assert "images/a.jpg" not in md + assert client.gets == [] # no polling on the sync path + + +@pytest.mark.asyncio +async def test_auto_mode_picks_v1_for_inline_response(monkeypatch, pdf_path, tmp_path): + client = _RoutedClient({FILE_PARSE_URL: [_FakeResponse(_sync_payload())]}) + parser = _parser() + _patch_client(monkeypatch, client) + + md, meta = await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + assert meta["api_mode"] == "v1-sync" + assert client.gets == [] + + +@pytest.mark.asyncio +async def test_auth_header_sent_when_token_configured(monkeypatch, pdf_path, tmp_path): + client = _RoutedClient({FILE_PARSE_URL: [_FakeResponse(_sync_payload())]}) + parser = _parser(mineru_token="tok123") + _patch_client(monkeypatch, client) + + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + url, kwargs = client.posts[0] + assert url == FILE_PARSE_URL + assert kwargs["headers"]["Authorization"] == "Bearer tok123" + + +# --------------------------------------------------------------------------- +# v2-tasks flavor (self-hosted task API) + + +def _v2_routes(zip_content=None): + return { + TASKS_URL: [_FakeResponse(_v2_create_payload(), status_code=202)], + "/status": [ + _FakeResponse({"status": "processing"}), + _FakeResponse({"status": "completed"}), + ], + "/result": [_FakeResponse(content=zip_content or _result_zip())], + } + + +@pytest.mark.asyncio +async def test_async_mode_uses_v2_tasks_flow(monkeypatch, pdf_path, tmp_path): + """Explicit async mode: /tasks -> 202 -> poll status -> download result zip.""" + client = _split(_v2_routes()) + parser = _parser(mineru_api_mode="async", mineru_timeout=30) + _patch_client(monkeypatch, client) + storage = _FakeStorage(tmp_path) + + md, meta = await parser._convert_mineru(pdf_path, storage=storage, resource_name="doc") + + assert meta["api_mode"] == "v2-tasks" + assert meta["task_id"] == "task-42" + assert meta["images_saved"] == 1 + assert "doc-pic.jpg" in md + assert "images/pic.jpg" not in md + # upload went to /tasks with the v2 form defaults + url, kwargs = client.posts[0] + assert url == TASKS_URL + assert kwargs["data"]["response_format_zip"] == "true" + assert kwargs["data"]["return_md"] == "true" + + +@pytest.mark.asyncio +async def test_auto_mode_falls_back_to_tasks_on_404(monkeypatch, pdf_path, tmp_path): + """auto: /file_parse 404 -> probe /tasks -> v2 flow.""" + routes = _v2_routes() + routes[FILE_PARSE_URL] = [_FakeResponse(status_code=404)] + client = _split(routes) + parser = _parser() + _patch_client(monkeypatch, client) + + md, meta = await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + assert meta["api_mode"] == "v2-tasks" + assert md.startswith("# Online") + + +@pytest.mark.asyncio +async def test_auto_mode_accepts_task_shaped_file_parse_response(monkeypatch, pdf_path, tmp_path): + """auto: /file_parse answering 200 with a task payload -> continue the task.""" + routes = { + FILE_PARSE_URL: [_FakeResponse(_v2_create_payload())], + "/status": [_FakeResponse({"status": "completed"})], + "/result": [_FakeResponse(content=_result_zip())], + } + client = _split(routes) + parser = _parser() + _patch_client(monkeypatch, client) + + md, meta = await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + assert meta["api_mode"] == "v2-tasks" + assert md.startswith("# Online") + assert len(client.posts) == 1 # no re-upload on the fallback + + +@pytest.mark.asyncio +async def test_v2_task_failed_raises_with_detail(monkeypatch, pdf_path, tmp_path): + routes = { + TASKS_URL: [_FakeResponse(_v2_create_payload(), status_code=202)], + "/status": [_FakeResponse({"status": "failed", "msg": "bad pdf"})], + } + client = _split(routes) + parser = _parser(mineru_api_mode="async", mineru_timeout=5) + _patch_client(monkeypatch, client) + + with pytest.raises(ValueError, match="bad pdf"): + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + +# --------------------------------------------------------------------------- +# online-batch flavor + + +def _online_routes(zip_content=None): + poll = f"{ENDPOINT}/extract-results/batch/batch-77" + done = { + "code": 0, + "data": { + "extract_result": [ + {"file_name": "doc.pdf", "state": "done", "full_zip_url": "https://cdn.example/result.zip"} + ] + }, + } + return { + BATCH_URL: [_FakeResponse(_online_create_payload())], + poll: [ + _FakeResponse({"code": 0, "data": {"extract_result": [{"state": "running"}]}}), + _FakeResponse(done), + ], + "cdn.example/result.zip": [_FakeResponse(content=zip_content or _result_zip())], + } + + +@pytest.mark.asyncio +async def test_async_mode_falls_through_to_online_batch(monkeypatch, pdf_path, tmp_path): + """Explicit async mode: /tasks 404 -> online batch endpoint.""" + routes = _online_routes() + routes[TASKS_URL] = [_FakeResponse(status_code=404)] + client = _split(routes) + parser = _parser(mineru_api_mode="async", mineru_timeout=30, mineru_token="tok") + _patch_client(monkeypatch, client) + storage = _FakeStorage(tmp_path) + + md, meta = await parser._convert_mineru(pdf_path, storage=storage, resource_name="doc") + + assert meta["api_mode"] == "online-batch" + assert meta["task_id"] == "batch-77" + assert meta["images_saved"] == 1 + assert "doc-pic.jpg" in md + poll_urls = [u for u, _ in client.gets] + assert any("/extract-results/batch/batch-77" in u for u in poll_urls) + assert poll_urls[-1] == "https://cdn.example/result.zip" + + +@pytest.mark.asyncio +async def test_online_task_failed_raises_with_detail(monkeypatch, pdf_path, tmp_path): + poll = f"{ENDPOINT}/extract-results/batch/batch-77" + routes = { + TASKS_URL: [_FakeResponse(status_code=404)], + BATCH_URL: [_FakeResponse(_online_create_payload())], + poll: [ + _FakeResponse( + {"code": 0, "data": {"extract_result": [{"state": "failed", "err_msg": "quota exhausted"}]}} + ) + ], + } + client = _split(routes) + parser = _parser(mineru_api_mode="async", mineru_timeout=5) + _patch_client(monkeypatch, client) + + with pytest.raises(ValueError, match="quota exhausted"): + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + +# --------------------------------------------------------------------------- +# failure surfaces + + +@pytest.mark.asyncio +async def test_async_timeout_raises(monkeypatch, pdf_path, tmp_path): + routes = { + TASKS_URL: [_FakeResponse(_v2_create_payload(), status_code=202)], + "/status": [_FakeResponse({"status": "processing"})], + } + client = _split(routes) + parser = _parser(mineru_api_mode="async", mineru_timeout=0.5) + _patch_client(monkeypatch, client) + + real_sleep = __import__("asyncio").sleep + + async def fast_sleep(seconds): + await real_sleep(0) + + monkeypatch.setattr("openviking.parse.parsers.pdf.asyncio.sleep", fast_sleep) + + with pytest.raises(TimeoutError): + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + +@pytest.mark.asyncio +async def test_async_no_task_endpoint_raises_helpful_error(monkeypatch, pdf_path, tmp_path): + """Both task endpoints 404 -> actionable message mentioning the token.""" + routes = { + TASKS_URL: [_FakeResponse(status_code=404)], + BATCH_URL: [_FakeResponse(status_code=404)], + } + client = _split(routes) + parser = _parser(mineru_api_mode="async") + _patch_client(monkeypatch, client) + + with pytest.raises(ValueError, match="mineru_token"): + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + +@pytest.mark.asyncio +async def test_async_zip_without_markdown_raises(monkeypatch, pdf_path, tmp_path): + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + zf.writestr("layout.pdf", b"only") + + client = _split(_v2_routes(zip_content=buf.getvalue())) + parser = _parser(mineru_api_mode="async", mineru_timeout=10) + _patch_client(monkeypatch, client) + + with pytest.raises(ValueError, match="[Nn]o markdown"): + await parser._convert_mineru(pdf_path, storage=_FakeStorage(tmp_path)) + + +# --------------------------------------------------------------------------- +# config validation + + +def test_api_mode_validation_rejects_unknown_value(): + with pytest.raises(ValueError, match="mineru_api_mode"): + PDFConfig(strategy="mineru", mineru_endpoint=ENDPOINT, mineru_api_mode="fast").validate() + + +def test_api_mode_accepts_all_documented_values(): + for mode in ("auto", "sync", "async"): + PDFConfig(strategy="mineru", mineru_endpoint=ENDPOINT, mineru_api_mode=mode).validate()