MinerU 核心源码与模块关系深度报告

基于源码级分析 — 核心模块源码片段 · 关键模块依赖关系图 · 数据流架构

Source Code Analysis Module Dependency Graph Pipeline / VLM / Hybrid Middle JSON Architecture

🔗 1. 关键模块关系总图

下图展示了 MinerU 各核心模块之间的调用与依赖关系。箭头方向表示"调用/依赖"关系。

MinerU 模块依赖关系图
CLI Client FastAPI Server Gradio Web UI
↓ 调用
Pipeline Engine VLM Engine Hybrid Engine Office Engine
↓ 依赖
DocLayoutv2 PaddleOCR UniMERNet Table CLS/REC VLM Model
↓ 输出
Middle JSON(统一中间表示)
↓ 生成
Markdown JSON ContentList

跨引擎模块共享关系

Pipeline → 使用 → model_init.ModelSingleton + AtomModelSingleton + MagicModel
VLM → 使用 → vlm_analyze.ModelSingleton + MinerUClient
Hybrid → 依赖 → Pipeline model_init + VLM vlm_analyze(双引擎融合)
Office → 使用 → office_magic_model.MagicModel(独立路径)
共享基础设施: DataReader/Writer enum_class config_reader pdfium_guard pdf_classify

🧠 2. Pipeline 引擎核心源码

2.1 Pipeline 主控流程 — pipeline_analyze.py

这是 Pipeline 引擎的入口,负责文档分类、模型初始化、流式处理窗口调度。

mineru/backend/pipeline/pipeline_analyze.py# Copyright (c) Opendatalab. All rights reserved.
import os
import time
from typing import List, Tuple

import pypdfium2 as pdfium
from PIL import Image
from loguru import logger
from tqdm import tqdm

from .model_init import MineruPipelineModel, PIPELINE_MODEL_INIT_LOCK
from .model_json_to_middle_json import (
    apply_server_side_postprocess,
    append_batch_results_to_middle_json,
    finalize_middle_json,
    init_middle_json,
)
from mineru.utils.config_reader import get_device, get_processing_window_size
from ...utils.enum_class import ImageType
from ...utils.pdf_classify import classify
from ...utils.pdf_image_tools import load_images_from_pdf_doc
from ...utils.model_utils import get_vram, clean_memory
from ...utils.pdfium_guard import (
    close_pdfium_document,
    get_pdfium_document_page_count,
    open_pdfium_document,
)

os.environ['PYTORCH_ENABLE_MPS_FALLBACK'] = '1'  # 让mps可以fallback

class ModelSingleton:
    """Pipeline 模型单例管理器 — 线程安全的懒加载模式"""
    _instance = None
    _models = {}
    _lock = PIPELINE_MODEL_INIT_LOCK

    def __new__(cls, *args, **kwargs):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
        return cls._instance

    def get_model(self, lang=None, formula_enable=None, table_enable=None):
        key = (lang, formula_enable, table_enable)
        with self._lock:
            if key not in self._models:
                self._models[key] = custom_model_init(
                    lang=lang, formula_enable=formula_enable,
                    table_enable=table_enable,
                )
        return self._models[key]

def _get_ocr_enable(pdf_bytes, parse_method: str) -> bool:
    """根据解析方法和PDF分类结果判断是否启用OCR"""
    if parse_method == 'auto':
        return classify(pdf_bytes) == 'ocr'
    if parse_method == 'ocr':
        return True
    return False

def doc_analyze_streaming(
        pdf_bytes_list, image_writer_list, lang_list, on_doc_ready,
        parse_method: str = 'auto',
        formula_enable=True, table_enable=True,
        client_side_output_generation=False,
):
    """Pipeline 流式文档分析主入口 — 按处理窗口分批处理页面"""
    # 1. 为每个文档创建上下文
    doc_contexts = []
    for doc_index, (pdf_bytes, image_writer, lang) in enumerate(
        zip(pdf_bytes_list, image_writer_list, lang_list)
    ):
        _ocr_enable = _get_ocr_enable(pdf_bytes, parse_method)
        pdf_doc = open_pdfium_document(pdfium.PdfDocument, pdf_bytes)
        page_count = get_pdfium_document_page_count(pdf_doc)
        context = {
            'doc_index': doc_index,
            'pdf_bytes': pdf_bytes,
            'pdf_doc': pdf_doc,
            'page_count': page_count,
            'next_page_idx': 0,
            'middle_json': init_middle_json(),  # ← 初始化中间JSON
            'model_list': [],
            'image_writer': image_writer,
            'lang': lang,
            'ocr_enable': _ocr_enable,
            'closed': False,
        }
        doc_contexts.append(context)
    # 2. 按处理窗口分批推理 → 3. 转换为Middle JSON → 4. 回调输出

2.2 模型初始化与线程安全 — model_init.py

模型初始化模块管理所有 AI 模型的生命周期,通过双重单例模式(ModelSingleton + AtomModelSingleton)和细粒度推理锁保证线程安全。

mineru/backend/pipeline/model_init.py# Copyright (c) Opendatalab. All rights reserved.
import os
import threading
import torch
from loguru import logger

from .model_list import AtomicModel
from ...model.layout.pp_doclayoutv2 import PPDocLayoutV2LayoutModel
from ...model.mfr.unimernet.Unimernet import UnimernetModel
from ...model.mfr.pp_formulanet_plus_m.predict_formula import FormulaRecognizer
from mineru.model.ocr.pytorch_paddle import PytorchPaddleOCR
from ...model.table.cls.paddle_table_cls import PaddleTableClsModel
from ...model.table.rec.slanet_plus.main import PaddleTableModel
from ...model.table.rec.unet_table.main import UnetTableModel

# 线程安全锁 — 保护 Pipeline 与 Hybrid 共享的模型推理
PIPELINE_MODEL_INIT_LOCK = threading.RLock()
PIPELINE_LAYOUT_INFERENCE_LOCK = threading.RLock()
PIPELINE_MFR_INFERENCE_LOCK = threading.RLock()
PIPELINE_OCR_INFERENCE_LOCK = threading.RLock()

PIPELINE_INFERENCE_LOCKS_ENABLED = os.getenv(
    'MINERU_ENABLE_PIPELINE_INFERENCE_LOCKS', 'False'
).lower() in ['true', '1', 'yes']

def _run_with_inference_lock(inference_lock, inference_callable, *args, **kwargs):
    """按实验开关决定是否在指定推理锁内执行模型调用"""
    if not PIPELINE_INFERENCE_LOCKS_ENABLED:
        return inference_callable(*args, **kwargs)
    with inference_lock:
        return inference_callable(*args, **kwargs)

def run_layout_inference(inference_callable, *args, **kwargs):
    return _run_with_inference_lock(PIPELINE_LAYOUT_INFERENCE_LOCK, ...)

def run_mfr_inference(inference_callable, *args, **kwargs):
    return _run_with_inference_lock(PIPELINE_MFR_INFERENCE_LOCK, ...)

def run_ocr_inference(inference_callable, *args, **kwargs):
    return _run_with_inference_lock(PIPELINE_OCR_INFERENCE_LOCK, ...)

# 公式模型选择 — 通过环境变量切换中文公式支持
MFR_MODEL = os.getenv('MINERU_FORMULA_CH_SUPPORT', 'False')
if MFR_MODEL.lower() in ['true', '1', 'yes']:
    MFR_MODEL = "pp_formulanet_plus_m"   # 中文公式模型
else:
    MFR_MODEL = "unimernet_small"          # 通用公式模型

def mfr_model_init(weight_dir, device='cpu'):
    if MFR_MODEL == "unimernet_small":
        mfr_model = UnimernetModel(weight_dir, device)
    elif MFR_MODEL == "pp_formulanet_plus_m":
        mfr_model = FormulaRecognizer(weight_dir, device)
    return mfr_model

def ocr_model_init(det_db_box_thresh=0.5, lang=None, ...):
    if lang in [None, "ch"]:
        use_dilation = True
        det_db_unclip_ratio = 1.8
    else:
        use_dilation = False
    model = PytorchPaddleOCR(
        det_db_box_thresh=det_db_box_thresh, lang=lang,
        use_dilation=use_dilation, det_db_unclip_ratio=det_db_unclip_ratio,
        enable_merge_det_boxes=enable_merge_det_boxes,
    )
    return model

class AtomModelSingleton:
    """原子模型单例 — 管理 OCR、Table 等可复用的子模型"""
    _instance = None
    _models = {}
    _lock = PIPELINE_MODEL_INIT_LOCK

    def __new__(cls, *args, **kwargs):
        with cls._lock:
            if cls._instance is None:
                cls._instance = super().__new__(cls)
        return cls._instance

    def get_atom_model(self, atom_model_name: str, **kwargs):
        key = (atom_model_name, kwargs.get('lang'), ...)
        with self._lock:
            if key not in self._models:
                self._models[key] = self._create_model(atom_model_name, **kwargs)
        return self._models[key]

👁️ 3. VLM 引擎核心源码

3.1 VLM 分析主控 — vlm_analyze.py

VLM 引擎通过视觉语言模型(如 Qwen2-VL)端到端理解文档页面,支持多种推理后端。

mineru/backend/vlm/vlm_analyze.py# Copyright (c) Opendatalab. All rights reserved.
import asyncio, atexit, gc, os, time, json, threading
from contextlib import asynccontextmanager, contextmanager

import pypdfium2 as pdfium
from loguru import logger
from tqdm import tqdm

from .utils import (enable_custom_logits_processors,
    set_default_gpu_memory_utilization, set_default_batch_size,
    set_lmdeploy_backend, mod_kwargs_by_device_type)
from .model_output_to_middle_json import (
    append_page_blocks_to_middle_json,
    finalize_middle_json,
    init_middle_json,
)
from mineru.data.data_reader_writer import DataWriter
from mineru_vl_utils import MinerUClient
from packaging import version

class ModelSingleton:
    """VLM 模型单例 — 支持多种推理后端"""
    _instance = None
    _models = {}
    _lock = threading.RLock()

    def get_model(self, backend: str, model_path: str | None,
                  server_url: str | None, **kwargs) -> MinerUClient:
        key = (backend, model_path, server_url)
        with self._lock:
            if key not in self._models:
                # 根据 backend 类型选择不同的推理引擎
                if backend == "transformers":
                    model = Qwen2VLForConditionalGeneration.from_pretrained(
                        model_path, device_map={"": device}, **{dtype_key: "auto"})
                    processor = AutoProcessor.from_pretrained(model_path)
                elif backend == "vllm-engine":
                    import vllm
                    kwargs = mod_kwargs_by_device_type(kwargs, vllm_mode="sync_engine")
                    # vLLM 高性能推理引擎
                elif backend == "lmdeploy-engine":
                    from lmdeploy import pipeline
                    # LMDeploy 推理引擎
                elif backend == "http-client":
                    # 远程 HTTP API 调用
                    pass
        return self._models[key]

3.2 VLM 支持的推理后端

后端类型适用场景依赖
transformers本地推理开发调试、小批量transformers, torch
vllm-engine本地推理高性能批量推理vllm
lmdeploy-engine本地推理高性能推理(TurboMind)lmdeploy
mlx-engine本地推理Apple Silicon Macmlx
http-client远程调用分布式部署httpx

🔀 4. Hybrid 混合引擎核心源码

4.1 Hybrid 分析主控 — hybrid_analyze.py

Hybrid 引擎融合 Pipeline 和 VLM 的能力:先用 Pipeline 模型获取版面结构,再用 VLM 对复杂区域补充理解。

mineru/backend/hybrid/hybrid_analyze.py# Copyright (c) Opendatalab. All rights reserved.
import asyncio, os, time
from collections import defaultdict

import cv2
import numpy as np
import pypdfium2 as pdfium
from loguru import logger
from mineru_vl_utils import MinerUClient
from mineru_vl_utils.structs import BlockType

# ← 关键:Hybrid 同时依赖 Pipeline 和 VLM 两个引擎
from mineru.backend.pipeline.model_init import (
    HybridModelSingleton,
    run_layout_inference,
    run_mfr_inference,
    run_ocr_inference,
)
from mineru.backend.vlm.vlm_analyze import (
    ModelSingleton,
    aio_predictor_execution_guard,
    predictor_execution_guard,
    _maybe_enable_serial_execution,
    _get_model_async,
)
from mineru.data.data_reader_writer import DataWriter
from mineru.utils.ocr_utils import (
    get_adjusted_mfdetrec_res, get_ocr_result_list, sorted_boxes,
    merge_det_boxes, update_det_boxes, OcrConfidence)
from mineru.utils.pdf_classify import classify

LAYOUT_BASE_BATCH_SIZE = 1
MFR_BASE_BATCH_SIZE = 16
OCR_DET_BASE_BATCH_SIZE = 8

def ocr_classify(pdf_bytes, parse_method: str = 'auto') -> bool:
    """确定OCR设置 — 复用 Pipeline 的分类逻辑"""
    _ocr_enable = False
    if parse_method == 'auto':
        if classify(pdf_bytes) == 'ocr':
            _ocr_enable = True
    elif parse_method == 'ocr':
        _ocr_enable = True
    return _ocr_enable

def ocr_det(hybrid_pipeline_model, np_images, model_list,
           mfd_res, _ocr_enable, batch_ratio: int = 1, *, fill_text: bool = True):
    """Hybrid 专用 OCR 检测 — 对 Pipeline 版面结果进行行级 OCR 补充"""
    ocr_res_list = []
    if not hybrid_pipeline_model.enable_ocr_det_batch:
        # 非批处理模式 — 逐页逐块处理
        for np_image, page_mfd_res, page_results in tqdm(
            zip(np_images, mfd_res, model_list), desc="OCR-det"
        ):
            for res in page_results:
                if not _is_hybrid_ocr_det_candidate(res):
                    continue
                # 裁剪区域 → OCR 检测 → 合并结果
                new_image, useful_list = crop_img(res, np_image, ...)
                bgr_image = cv2.cvtColor(new_image, cv2.COLOR_RGB2BGR)
                ocr_res = run_ocr_inference(
                    hybrid_pipeline_model.ocr_model.ocr,
                    bgr_image, mfd_res=adjusted_mfdetrec_res, rec=False
                )[0]
    else:
        # 批处理模式 — 按语言和分辨率分组批量推理
        pass

4.2 Hybrid 引擎数据流

PDF 输入 Pipeline 版面检测
DocLayoutv2
Pipeline OCR
PaddleOCR
Pipeline 公式识别
UniMERNet
+ VLM 端到端理解
Qwen2-VL
结果融合
hybrid_model_output
_to_middle_json
Middle JSON

📄 5. Office 引擎核心源码

5.1 Office MagicModel — office_magic_model.py

Office 引擎的 MagicModel 将 Office 文档的原生结构化数据转换为统一的 Block 体系。

mineru/backend/office/office_magic_model.py# Copyright (c) Opendatalab. All rights reserved.
import html as html_lib
import re
from typing import Literal
from urllib.parse import urlparse
from loguru import logger

from mineru.utils.enum_class import ContentType, BlockType
from mineru.utils.magic_model_utils import tie_up_category_by_index

class MagicModel:
    """Office 文档的 MagicModel — 将原生结构转为统一 Block 表示"""
    def __init__(self, page_blocks: list):
        self.page_blocks = page_blocks
        blocks = []
        self.all_spans = []

        # 对 caption 块分类 → image_caption / table_caption / chart_caption
        page_blocks = classify_caption_blocks(page_blocks)

        for index, block_info in enumerate(page_blocks):
            block_type = block_info["type"]
            block_content = block_info.get("content", "")

            if block_type in ["text", "title", "image_caption", ...]:
                span = parse_text_block_spans(block_content)
            elif block_type in ["image"]:
                block_type = BlockType.IMAGE_BODY
                span = {"type": ContentType.IMAGE, "image_base64": block_content}
            elif block_type in ["table"]:
                block_type = BlockType.TABLE_BODY
                span = {"type": ContentType.TABLE, "html": clean_table_html(block_content)}
            elif block_type in ["equation"]:
                block_type = BlockType.INTERLINE_EQUATION
                span = {"type": ContentType.INTERLINE_EQUATION, "content": block_content}
            elif block_type in ["list"]:
                parsed_list = parse_list_block(block_info)
                if parsed_list:
                    blocks.append(parsed_list)
                continue

            # 构建 Block → Line → Span 三级结构
            block = {
                "type": block_type,
                "lines": [{"spans": [span]}],
                "index": index,
            }
            if block_type == BlockType.TITLE:
                block["is_numbered_style"] = block_info.get("is_numbered_style", False)
                block["level"] = block_info.get("level", 1)
            blocks.append(block)

        # 按 BlockType 分类存储
        self.image_blocks = []
        self.table_blocks = []
        self.text_blocks = []
        self.title_blocks = []
        self.discarded_blocks = []
        for block in blocks:
            if block["type"] in [BlockType.IMAGE_BODY, BlockType.IMAGE_CAPTION, ...]:
                self.image_blocks.append(block)
            elif block["type"] in [BlockType.TABLE_BODY, BlockType.TABLE_CAPTION, ...]:
                self.table_blocks.append(block)
            # ...

🔄 6. Middle JSON 转换核心源码

6.1 模型输出 → Middle JSON — model_json_to_middle_json.py

这是 Pipeline 引擎中连接模型推理和内容生成的关键桥梁,将各模型原始输出转换为统一的中间表示。

mineru/backend/pipeline/model_json_to_middle_json.py# Copyright (c) Opendatalab. All rights reserved.
import copy
from tqdm import tqdm

from mineru.backend.utils.html_image_utils import replace_inline_table_images
from mineru.backend.utils.runtime_utils import cross_page_table_merge
from mineru.backend.pipeline.model_init import AtomModelSingleton, run_ocr_inference
from mineru.backend.pipeline.para_split import para_split
from mineru.utils.enum_class import ContentType, BlockType
from mineru.backend.pipeline.pipeline_magic_model import MagicModel
from mineru.utils.ocr_utils import OcrConfidence, rotate_vertical_crop_if_needed

def page_model_info_to_page_info(page_model_info, image_dict, page,
                                  image_writer, page_index, ocr_enable=False):
    """将单页模型输出转换为 page_info 结构"""
    scale = image_dict["scale"]
    page_pil_img = image_dict["img_pil"]
    page_img_md5 = bytes_md5(page_pil_img.tobytes())

    # 1. 构建 MagicModel — 封装模型检测结果
    magic_model = MagicModel(
        page_model_info, page, scale, page_pil_img,
        page_w, page_h, ocr_enable
    )

    # 2. 提取各类区块
    preproc_blocks = magic_model.get_preproc_blocks()
    discarded_blocks = magic_model.get_discarded_blocks()
    all_image_spans = magic_model.get_all_image_spans()

    # 3. 对 image/table/chart/interline_equation 的 span 截图
    for span in all_image_spans:
        if span["type"] in [ContentType.IMAGE, ContentType.TABLE,
                          ContentType.CHART, ContentType.INTERLINE_EQUATION]:
            span = cut_image_and_table(span, page_pil_img, page_img_md5,
                                       page_index, image_writer, scale=scale)

    # 4. 替换内联表格图片 → 构造 page_info
    replace_inline_table_images(preproc_blocks, image_writer, page_index)
    page_info = make_page_info_dict(preproc_blocks, page_index,
                                     page_w, page_h, discarded_blocks)
    return page_info

def append_batch_results_to_middle_json(middle_json, batch_results,
    images_list, pdf_doc, image_writer, page_start_index=0,
    ocr_enable=False, model_list=None, progress_bar=None):
    """将批量推理结果追加到 Middle JSON"""
    page_model_infos = []
    for offset, (image_dict, page_layout_dets) in enumerate(
        zip(images_list, batch_results)
    ):
        page_model_info = build_page_model_info(
            page_layout_dets, page_start_index + offset, image_dict["img_pil"])
        page_model_infos.append(page_model_info)

    # 逐页转换 → 追加到 middle_json["pdf_info"]
    append_page_model_infos_to_middle_json(
        middle_json, page_model_infos, images_list, pdf_doc,
        image_writer, page_start_index, ocr_enable, progress_bar)

💾 7. 数据 I/O 抽象层源码

7.1 DataReader / DataWriter 基类 — base.py

定义了数据读写的抽象接口,所有存储后端实现此接口即可无缝接入。

mineru/data/data_reader_writer/base.py# Copyright (c) Opendatalab. All rights reserved.
from abc import ABC, abstractmethod

class DataReader(ABC):
    def read(self, path: str) -> bytes:
        """Read the file."""
        return self.read_at(path)

    @abstractmethod
    def read_at(self, path: str, offset: int = 0, limit: int = -1) -> bytes:
        """Read the file at offset and limit."""
        pass

class DataWriter(ABC):
    @abstractmethod
    def write(self, path: str, data: bytes) -> None:
        """Write the data to the file."""
        pass

    def write_string(self, path: str, data: str) -> None:
        """Write string data with safe encoding (utf-8 / ascii)."""
        def safe_encode(data: str, method: str):
            try:
                bit_data = data.encode(encoding=method, errors='replace')
                return bit_data, True
            except:
                return None, False

        for method in ['utf-8', 'ascii']:
            bit_data, flag = safe_encode(data, method)
            if flag:
                self.write(path, bit_data)
                break

7.2 存储后端实现继承关系

DataReader (ABC)
read() / read_at()
↓ 继承
FileBaseReader
本地文件系统
S3Reader
Amazon S3
MultiBucketS3Reader
多桶 S3
DummyReader
测试用
同构
DataWriter (ABC)
write() / write_string()
↓ 继承
FileBaseWriter
S3Writer
MultiBucketS3Writer

🏷️ 8. 核心枚举与类型定义 — enum_class.py

枚举类定义了整个系统共享的类型常量,是各模块间数据交换的"契约"。

mineru/utils/enum_class.py# Copyright (c) Opendatalab. All rights reserved.
from enum import Enum

class BlockType:
    """文档版面元素类型 — 所有引擎共享的 Block 类型体系"""
    IMAGE = 'image'
    TABLE = 'table'
    CHART = 'chart'
    IMAGE_BODY = 'image_body'
    TABLE_BODY = 'table_body'
    CHART_BODY = 'chart_body'
    CAPTION = 'caption'
    IMAGE_CAPTION = 'image_caption'
    TABLE_CAPTION = 'table_caption'
    CHART_CAPTION = 'chart_caption'
    ALGORITHM_CAPTION = 'algorithm_caption'
    FOOTNOTE = 'footnote'
    TEXT = 'text'
    TITLE = 'title'
    INTERLINE_EQUATION = 'interline_equation'
    EQUATION = "equation"          # 行间公式(独立行)
    LIST = 'list'
    INDEX = 'index'
    DISCARDED = 'discarded'
    CODE = "code"
    CODE_BODY = "code_body"
    CODE_CAPTION = "code_caption"
    ALGORITHM = "algorithm"
    ABSTRACT = "abstract"
    DOC_TITLE = "doc_title"
    PARAGRAPH_TITLE = "paragraph_title"

class ContentType:
    """Span 内容类型 — Block 内部最小内容单元的类型"""
    IMAGE = 'image'
    TABLE = 'table'
    CHART = 'chart'
    TEXT = 'text'
    INTERLINE_EQUATION = 'interline_equation'
    INLINE_EQUATION = 'inline_equation'
    EQUATION = 'equation'
    HYPERLINK = 'hyperlink'

class MakeMode:
    """输出格式模式"""
    MM_MD = 'mm_markdown'
    NLP_MD = 'nlp_markdown'
    CONTENT_LIST = 'content_list'
    CONTENT_LIST_V2 = 'content_list_v2'

class ModelPath:
    """模型权重路径 — HuggingFace / ModelScope"""
    vlm_root_hf = "opendatalab/MinerU2.5-Pro-2605-1.2B"
    vlm_root_modelscope = "OpenDataLab/MinerU2.5-Pro-2605-1.2B"
    pipeline_root_modelscope = "OpenDataLab/PDF-Extract-Kit-1.0"
    pp_doclayout_v2 = "models/Layout/PP-DocLayoutV2"
    unimernet_small = "models/MFR/unimernet_hf_small_2503"
    pp_formulanet_plus_m = "models/MFR/pp_formulanet_plus_m"
    pytorch_paddle = "models/OCR/paddleocr_torch"
    slanet_plus = "models/TabRec/SLANetPlus/slanet-plus.onnx"
    unet_structure = "models/TabRec/UnetStructure/unet.onnx"

🖥️ 9. CLI 客户端核心源码

9.1 客户端入口 — client.py

CLI 客户端是用户交互的主入口,负责任务规划、文档分类、引擎调度和结果输出。

mineru/cli/client.py# Copyright (c) Opendatalab. All rights reserved.
import asyncio, multiprocessing, os, sys, threading
from concurrent.futures import Future, ProcessPoolExecutor
from dataclasses import dataclass
from pathlib import Path
from typing import Awaitable, Callable, Optional, TextIO

import click
import httpx
import pypdfium2 as pdfium
from loguru import logger

from mineru.cli.common import (
    HybridDependencyError,
    ensure_backend_dependencies,
    image_suffixes, office_suffixes, pdf_suffixes,
    uniquify_task_stems,
)

@dataclass(frozen=True)
class InputDocument:
    path: Path
    suffix: str
    stem: str
    effective_pages: int
    order: int

@dataclass
class PlannedTask:
    index: int
    documents: list[InputDocument]
    total_pages: int

@dataclass
class TaskExecutionProgress:
    total_tasks: int
    total_pages: int
    completed_tasks: int
    completed_pages: int
    lock: asyncio.Lock

# 实时任务状态渲染器 — 支持终端进度条显示
class LiveTaskStatusRenderer:
    BAR_WIDTH = 12
    RUNNER_WIDTH = 4
    ACTIVE_STATUSES = {"pending", "processing"}

    def register_task(self, task: PlannedTask, task_id: str, ...):
        """注册新任务到渲染器"""
        ...

    def update_status(self, task_id: str, status_update, ...):
        """更新任务状态并重新渲染"""
        ...

📊 10. 完整模块依赖关系矩阵

下表展示了各核心模块之间的直接依赖关系。

pipeline_analyze
model_init, model_json_to_middle_json, pdf_classify, pdf_image_tools, pdfium_guard, enum_class, config_reader
model_init
PPDocLayoutV2LayoutModel, UnimernetModel, FormulaRecognizer, PytorchPaddleOCR, PaddleTableClsModel, PaddleTableModel, UnetTableModel, config_reader, enum_class
model_json_to_middle_json
model_init, MagicModel, para_split, enum_class, ocr_utils, cut_image, hash_utils, pdfium_guard
vlm_analyze
MinerUClient, model_output_to_middle_json, DataWriter, pdf_image_tools, pdfium_guard, config_reader
hybrid_analyze
pipeline.model_init (HybridModelSingleton, run_*_inference), vlm.vlm_analyze (ModelSingleton, *_guard), DataWriter, ocr_utils, pdf_classify, pdf_image_tools
office_magic_model
enum_class, magic_model_utils
cli/client
api_client, common, output_paths, visualization, pdfium_guard, config_reader
cli/fast_api
pipeline_analyze, vlm_analyze, hybrid_analyze, office_*, DataReader/Writer, router
cli/gradio_app
pipeline_analyze, vlm_analyze, hybrid_analyze, office_*, common

共享基础设施模块被依赖统计

共享模块被依赖次数主要消费者
enum_class (BlockType/ContentType)所有引擎Pipeline, VLM, Hybrid, Office, Utils
config_reader所有引擎model_init, pipeline_analyze, vlm_analyze, hybrid_analyze, client
pdfium_guardPDF 引擎pipeline_analyze, vlm_analyze, hybrid_analyze, client
pdf_classifyPDF 引擎pipeline_analyze, hybrid_analyze
pdf_image_toolsPDF 引擎pipeline_analyze, vlm_analyze, hybrid_analyze
DataReader/Writer所有引擎vlm_analyze, hybrid_analyze, fast_api
ocr_utilsPipeline+Hybridmodel_json_to_middle_json, hybrid_analyze
magic_model_utilsPipeline+Officepipeline_magic_model, office_magic_model

🏗️ 11. 三引擎统一处理范式

尽管三种 PDF 解析引擎的内部实现不同,但它们都遵循相同的 Model → Middle JSON → Content 三阶段范式。

Pipeline Engine
① DocLayoutv2 + PaddleOCR
+ UniMERNet + Table REC
② model_json_to_middle_json
para_split + MagicModel
③ pipeline_middle_json
_mkcontent
VLM Engine
① VLM 端到端推理
(Qwen2-VL / InternVL)
② model_output_to_middle_json
init/finalize/append
③ vlm_middle_json
_mkcontent
Hybrid Engine
① Pipeline模型 + VLM
双路推理 + ocr_det
② hybrid_model_output
_to_middle_json
双路结果融合
③ pipeline_middle_json
_mkcontent (复用)

Middle JSON 核心结构

Middle JSON 结构示意{
  "pdf_info": [                    // 每页一个 page_info
    {
      "page_no": 0,
      "width": 595,
      "height": 842,
      "preproc_blocks": [           // 预处理后的 Block 列表
        {
          "type": "text",             // BlockType
          "lines": [                 // Line → Span 结构
            {
              "spans": [
                { "type": "text", "content": "..." },
                { "type": "inline_equation", "content": "E=mc^2" }
              ]
            }
          ],
          "index": 0
        },
        {
          "type": "table_body",
          "lines": [{ "spans": [{ "type": "table", "html": "<table>..." }] }]
        }
      ],
      "discarded_blocks": [...]     // 页眉/页脚/页码等
    }
  ]
}

⚡ 12. 关键设计模式总结

双重单例模式

ModelSingleton 管理组合模型,AtomModelSingleton 管理原子模型。线程安全懒加载,按参数组合缓存实例。

推理锁保护

PIPELINE_*_INFERENCE_LOCK 保护 Pipeline/Hybrid 共享的模型推理调用,防止多线程并发进入同一模型对象。

流式处理窗口

doc_analyze_streaming 按处理窗口分批加载页面,控制 GPU 显存占用,支持超长文档处理。

统一中间表示

Middle JSON 作为三种引擎的统一输出格式,解耦模型推理与内容生成,支持灵活的后处理流水线。

存储抽象基类

DataReader/DataWriter ABC 定义统一接口,FileBase/S3/MultiBucketS3 实现可插拔存储后端。

Block→Line→Span 三级结构

所有引擎输出统一的 Block(Line(Span)) 层次结构,Span 是最小内容单元(文本/公式/图片/表格)。