MinerU 4.0.12 实战笔记：从PDF到可用的Markdown – weiwei22844

#!/usr/bin/env python3
"""MinerU 4.0.12 V1 API 的三种 PDF 解析调用方式。

用法:
    python mineru_api_examples.py uploads  doc.pdf   # 标准链路: /v1/uploads -> /v1/parse/jobs -> /v1/files
    python mineru_api_examples.py tasks    doc.pdf   # 便捷异步: /v1/tasks -> /v1/tasks/{id}/result
    python mineru_api_examples.py sync     doc.pdf   # 同步包装: /v1/file_parse

环境变量:
    MINERU_API_URL  服务地址，默认 
    MINERU_API_KEY  API key；未设置时读取 /root/zww/minerU/.mineru_api_key

依赖: requests (pip install requests)
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sys
import time
from pathlib import Path
from urllib.parse import urljoin, urlparse

import requests

DEFAULT_BASE_URL = ""
DEFAULT_KEY_FILE = Path("/root/zww/minerU/.mineru_api_key")
TERMINAL_STATUSES = {"completed", "partial", "failed", "canceled"}
OUTPUT_SUFFIX = {
    "markdown": ".md",
    "middle_json": ".middle.json",
    "structured_content": ".structured_content.json",
    "zip": ".zip",
}


def load_api_key() -> str:
    key = os.environ.get("MINERU_API_KEY", "").strip()
    if key:
        return key
    if DEFAULT_KEY_FILE.is_file():
        return DEFAULT_KEY_FILE.read_text(encoding="utf-8").strip()
    return ""


class Client:
    """所有请求都显式带上 Bearer Token；上传/下载的跨源场景单独处理。"""
    def __init__(self, base_url: str, api_key: str) -> None:
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.session = requests.Session()

    def headers(self, extra: dict[str, str] | None = None) -> dict[str, str]:
        headers = {"Authorization": f"Bearer {self.api_key}"} if self.api_key else {}
        if extra:
            headers.update(extra)
        return headers

    def call(self, method: str, path: str, *, headers: dict[str, str] | None = None, **kwargs) -> requests.Response:
        response = self.session.request(method, f"{self.base_url}{path}", headers=self.headers(headers), **kwargs)
        if response.status_code >= 400:
            raise RuntimeError(f"{method} {path} -> {response.status_code}: {response.text[:500]}")
        return response

    def poll(self, path: str, *, timeout: float = 600.0, interval: float = 1.0) -> dict:
        deadline = time.monotonic() + timeout
        while True:
            payload = self.call("GET", path).json()
            status = payload.get("status")
            print(f"  状态: {status}")
            if status in TERMINAL_STATUSES:
                return payload
            if time.monotonic() >= deadline:
                raise TimeoutError(f"{path} 在 {timeout:.0f}s 内没有进入终态 (最后状态 {status})")
            time.sleep(interval)

    def download(self, file_id: str) -> bytes:
        """下载产物；遇到 302 时不把 API key 带给另一个源。"""
        response = self.session.get(
            f"{self.base_url}/v1/files/{file_id}/content", headers=self.headers(), allow_redirects=False
        )
        if response.is_redirect:
            response = requests.get(response.headers["Location"], timeout=300)
        if response.status_code >= 400:
            raise RuntimeError(f"GET /v1/files/{file_id}/content -> {response.status_code}: {response.text[:300]}")
        return response.content


def same_origin(url: str, base_url: str) -> bool:
    """方案 + 主机 + 有效端口都一致才算同源，同主机不同端口也算跨源。"""
    def origin(value: str) -> tuple[str, str | None, int]:
        parsed = urlparse(value)
        return parsed.scheme, parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80)

    return origin(url) == origin(base_url)


def save_job_outputs(client: Client, files: list[dict], out_dir: Path) -> list[Path]:
    """Job / task 响应里的 files[] 形状一致：name + status + output_files。"""
    out_dir.mkdir(parents=True, exist_ok=True)
    saved: list[Path] = []
    for entry in files:
        if entry.get("status") != "completed":
            print(f"  跳过 {entry.get('name')}: {entry.get('status')} {entry.get('error')}")
            continue
        stem = Path(entry["name"]).stem
        for fmt, ref in (entry.get("output_files") or {}).items():
            if not ref:
                continue
            target = out_dir / f"{stem}{OUTPUT_SUFFIX.get(fmt, '.' + fmt)}"
            target.write_bytes(client.download(ref["file_id"]))
            saved.append(target)
            print(f"  保存 {target} ({target.stat().st_size} 字节)")
    return saved


def save_task_content(payload: dict, out_dir: Path) -> list[Path]:
    """/v1/tasks/{id}/result 与 /v1/file_parse 返回的内联结果。"""
    out_dir.mkdir(parents=True, exist_ok=True)
    saved: list[Path] = []
    for entry in payload.get("files", []):
        if entry.get("status") != "completed":
            print(f"  跳过 {entry.get('name')}: {entry.get('status')} {entry.get('error')}")
            continue
        stem = Path(entry["name"]).stem
        for fmt, content in (entry.get("content") or {}).items():
            target = out_dir / f"{stem}{OUTPUT_SUFFIX.get(fmt, '.' + fmt)}"
            if isinstance(content, str):
                target.write_text(content, encoding="utf-8")
            else:
                target.write_text(json.dumps(content, ensure_ascii=False, indent=2), encoding="utf-8")
            saved.append(target)
            print(f"  保存 {target} ({target.stat().st_size} 字节)")
    return saved


# ── 示例一：/v1/uploads + /v1/parse/jobs（标准链路）────────────────────
def example_uploads_then_jobs(
    client: Client, pdf: Path, out_dir: Path, *, tier: str, ocr_mode: str, formats: list[str]
) -> None:
    """上传字节 -> 拿到 file_id -> 提交 Job -> 轮询 -> 从 /v1/files 下载产物。"""
    data = pdf.read_bytes()
    sha256 = hashlib.sha256(data).hexdigest()

    # 1) 创建上传会话。带 sha256sum 时，服务端已有相同字节会直接返回 completed（秒传）。
    upload = client.call(
        "POST",
        "/v1/uploads",
        json={
            "filename": pdf.name,
            "bytes": len(data),
            "mime_type": "application/pdf",
            "purpose": "parse",
            "sha256sum": sha256,
        },
    ).json()
    print(f"1. upload {upload['id']} status={upload['status']}")

    if upload["status"] == "pending":
        upload_url = urljoin(f"{client.base_url}/", upload["upload_url"])
        upload_headers = dict(upload.get("upload_headers") or {})
        if same_origin(upload_url, client.base_url):
            # 本地服务的上传地址同源，需要带 API key；云端的预签名 URL 是跨源，不能带。
            upload_headers = client.headers(upload_headers)
        response = client.session.put(upload_url, data=data, headers=upload_headers, timeout=300)
        if response.status_code >= 400:
            raise RuntimeError(f"PUT {upload_url} -> {response.status_code}: {response.text[:300]}")
        print(f"2. PUT bytes {len(data)}")

        # 3) complete 生成 File 对象。
        upload = client.call(
            "POST", f"/v1/uploads/{upload['id']}/complete", json={"sha256sum": sha256}
        ).json()
        print(f"3. complete status={upload['status']}")

    file_id = upload["file"]["id"]
    print(f"   file_id={file_id}")

    # 4) 提交解析任务：POST /v1/parse/jobs 永远返回 202，不会同步等结果。
    job = client.call(
        "POST",
        "/v1/parse/jobs",
        json={
            "files": [{"source": {"type": "file_id", "file_id": file_id}}],
            "tier": tier,
            "ocr_mode": ocr_mode,
            "output_formats": formats,
        },
    ).json()
    job_id = job["job_id"]
    print(f"4. job {job_id} status={job['status']}")

    # 5) 轮询到终态。completed / partial / failed / canceled 都是终态。
    job = client.poll(f"/v1/parse/jobs/{job_id}")
    print(f"5. job 结束: {job['status']}")

    # 6) 产物只给 file_id，内容必须从 /v1/files/{file_id}/content 取。
    save_job_outputs(client, job.get("files", []), out_dir)


# ── 示例二：/v1/tasks（自部署便捷异步接口）───────────────────────────
def example_tasks(
    client: Client, pdf: Path, out_dir: Path, *, tier: str, ocr_mode: str, formats: list[str]
) -> None:
    """multipart 直接提交，主进程不用先走 uploads；结果可从 /result 内联取。"""
    params: list[tuple[str, str]] = [("tier", tier), ("ocr_mode", ocr_mode)]
    params += [("output_formats", fmt) for fmt in formats]
    form: list[tuple[str, tuple[str, bytes, str]]] = [("files", (pdf.name, pdf.read_bytes(), "application/pdf"))]

    task = client.call("POST", "/v1/tasks", params=params, files=form).json()
    task_id = task["task_id"]
    print(f"1. task {task_id} status={task['status']}")

    task = client.poll(f"/v1/tasks/{task_id}")
    print(f"2. task 结束: {task['status']} (task_id 就是 job_id，可用 /v1/parse/jobs/{task_id} 查)")

    # 3) /result: 未终态 202，completed/partial 200，failed/canceled 409。
    if "zip" in formats:
        response = client.session.get(
            f"{client.base_url}/v1/tasks/{task_id}/result",
            params={"response_format": "zip"},
            headers=client.headers(),
            timeout=600,
        )
        if response.status_code == 409:
            print(f"   任务失败: {response.text[:500]}")
            return
        response.raise_for_status()
        out_dir.mkdir(parents=True, exist_ok=True)
        target = out_dir / f"{pdf.stem}{OUTPUT_SUFFIX['zip']}"
        target.write_bytes(response.content)
        print(f"3. 保存 {target} ({target.stat().st_size} 字节)")
        return

    response = client.session.get(
        f"{client.base_url}/v1/tasks/{task_id}/result",
        params={"response_format": "json"},
        headers=client.headers(),
        timeout=600,
    )
    if response.status_code == 409:
        print(f"   任务失败: {response.text[:500]}")
        return
    response.raise_for_status()
    save_task_content(response.json(), out_dir)


# ── 示例三：/v1/file_parse（自部署同步包装）──────────────────────────
def example_file_parse(
    client: Client, pdf: Path, out_dir: Path, *, tier: str, ocr_mode: str, formats: list[str], wait_timeout: int = 600
) -> None:
    """一次请求等结果；超过 wait_timeout 会返回 202 和查询地址，后台任务继续跑。"""
    response_format = "zip" if "zip" in formats else "json"
    params: list[tuple[str, str]] = [
        ("tier", tier),
        ("ocr_mode", ocr_mode),
        ("response_format", response_format),
        ("wait_timeout", str(wait_timeout)),
    ]
    params += [("output_formats", fmt) for fmt in formats]
    form: list[tuple[str, tuple[str, bytes, str]]] = [("files", (pdf.name, pdf.read_bytes(), "application/pdf"))]

    response = client.session.post(
        f"{client.base_url}/v1/file_parse", params=params, files=form, headers=client.headers(), timeout=wait_timeout + 60
    )
    if response.status_code == 202:
        task = response.json()
        task_id = task["task_id"]
        print(f"1. 等待超时，任务仍在跑: {task_id}，转为轮询")
        client.poll(f"/v1/tasks/{task_id}")
        response = client.session.get(
            f"{client.base_url}/v1/tasks/{task_id}/result",
            params={"response_format": response_format},
            headers=client.headers(),
            timeout=600,
        )
    if response.status_code == 409:
        print(f"任务失败: {response.text[:500]}")
        return
    if response.status_code >= 400:
        raise RuntimeError(f"POST /v1/file_parse -> {response.status_code}: {response.text[:500]}")

    print(f"1. 同步返回: {response.status_code}")
    if response_format == "zip":
        out_dir.mkdir(parents=True, exist_ok=True)
        target = out_dir / f"{pdf.stem}{OUTPUT_SUFFIX['zip']}"
        target.write_bytes(response.content)
        print(f"2. 保存 {target} ({target.stat().st_size} 字节)")
        return
    save_task_content(response.json(), out_dir)


def main() -> int:
    parser = argparse.ArgumentParser(description="MinerU 4.0.12 V1 API PDF 解析示例")
    parser.add_argument("mode", choices=["uploads", "tasks", "sync"], help="要演示的接口组合")
    parser.add_argument("pdf", type=Path, help="待解析的 PDF")
    parser.add_argument("--out", type=Path, default=None, help="产物目录，默认 ./_")
    parser.add_argument("--tier", default="flash", choices=["flash", "basic", "standard", "advanced"])
    parser.add_argument("--ocr-mode", default="auto", choices=["auto", "txt", "ocr"])
    parser.add_argument("--formats", default="markdown", help="逗号分隔: markdown,middle_json,structured_content,zip")
    parser.add_argument("--wait-timeout", type=int, default=600, help="sync 模式等待秒数 (1-3600)")
    args = parser.parse_args()

    if not args.pdf.is_file():
        print(f"找不到文件: {args.pdf}", file=sys.stderr)
        return 1
    if args.pdf.suffix.lower() != ".pdf":
        print(f"这个示例只处理 PDF: {args.pdf}", file=sys.stderr)
        return 1

    formats = [item.strip() for item in args.formats.split(",") if item.strip()]
    out_dir = args.out or Path.cwd() / f"{args.pdf.stem}_{args.mode}"
    client = Client(os.environ.get("MINERU_API_URL", DEFAULT_BASE_URL), load_api_key())
    if not client.api_key:
        print("提示: 未找到 API key，将按匿名请求发送", file=sys.stderr)

    print(f"服务 {client.base_url} | 文件 {args.pdf} | tier={args.tier} ocr_mode={args.ocr_mode} formats={formats}")
    if args.mode == "uploads":
        example_uploads_then_jobs(client, args.pdf, out_dir, tier=args.tier, ocr_mode=args.ocr_mode, formats=formats)
    elif args.mode == "tasks":
        example_tasks(client, args.pdf, out_dir, tier=args.tier, ocr_mode=args.ocr_mode, formats=formats)
    else:
        example_file_parse(
            client,
            args.pdf,
            out_dir,
            tier=args.tier,
            ocr_mode=args.ocr_mode,
            formats=formats,
            wait_timeout=args.wait_timeout,
        )
    print(f"完成，产物目录: {out_dir}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())