mirror of
https://github.com/infiniflow/ragflow.git
synced 2026-11-11 05:40:52 +08:00
1293 lines
59 KiB
Python
1293 lines
59 KiB
Python
#
|
|
# Copyright 2025 The InfiniFlow Authors. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
import json
|
|
import html
|
|
import logging
|
|
import os
|
|
import re
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import zipfile
|
|
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
from os import PathLike
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Optional
|
|
|
|
import numpy as np
|
|
import pdfplumber
|
|
import requests
|
|
from PIL import Image
|
|
from enum import StrEnum
|
|
|
|
from deepdoc.parser.pdf_parser import RAGFlowPdfParser
|
|
from deepdoc.parser.utils import extract_pdf_outlines
|
|
|
|
from common.constants import MAXIMUM_PAGE_NUMBER
|
|
|
|
LOCK_KEY_pdfplumber = "global_shared_lock_pdfplumber"
|
|
if LOCK_KEY_pdfplumber not in sys.modules:
|
|
sys.modules[LOCK_KEY_pdfplumber] = threading.Lock()
|
|
|
|
|
|
class MinerUContentType(StrEnum):
|
|
IMAGE = "image"
|
|
TABLE = "table"
|
|
TEXT = "text"
|
|
EQUATION = "equation"
|
|
CODE = "code"
|
|
LIST = "list"
|
|
HEADER = "header"
|
|
FOOTER = "footer"
|
|
PAGE_NUMBER = "page_number"
|
|
DISCARDED = "discarded"
|
|
# MinerU 3.4.x VLM backend emits chart blocks for figures whose visual is a
|
|
# chart rather than a plain image. They carry chart_caption / chart_footnote
|
|
# / sub_type / img_path / bbox / page_idx — the same shape as IMAGE blocks —
|
|
# so they are routed through the image pipeline instead of being dropped.
|
|
CHART = "chart"
|
|
|
|
|
|
# Mapping from language names to MinerU language codes
|
|
LANGUAGE_TO_MINERU_MAP = {
|
|
"English": "en",
|
|
"Chinese": "ch",
|
|
"Traditional Chinese": "chinese_cht",
|
|
"Russian": "east_slavic",
|
|
"Ukrainian": "east_slavic",
|
|
"Indonesian": "latin",
|
|
"Spanish": "latin",
|
|
"Vietnamese": "latin",
|
|
"Japanese": "japan",
|
|
"Korean": "korean",
|
|
"Portuguese BR": "latin",
|
|
"German": "latin",
|
|
"French": "latin",
|
|
"Italian": "latin",
|
|
"Tamil": "ta",
|
|
"Telugu": "te",
|
|
"Kannada": "ka",
|
|
"Thai": "th",
|
|
"Greek": "el",
|
|
"Hindi": "devanagari",
|
|
"Bulgarian": "cyrillic",
|
|
"Turkish": "latin",
|
|
}
|
|
|
|
|
|
class MinerUBackend(StrEnum):
|
|
"""MinerU processing backend options (current public API names)."""
|
|
|
|
PIPELINE = "pipeline"
|
|
VLM_ENGINE = "vlm-engine"
|
|
HYBRID_ENGINE = "hybrid-engine"
|
|
VLM_HTTP_CLIENT = "vlm-http-client"
|
|
HYBRID_HTTP_CLIENT = "hybrid-http-client"
|
|
|
|
|
|
class MinerULanguage(StrEnum):
|
|
"""MinerU supported languages for OCR (pipeline backend only)."""
|
|
|
|
CH = "ch" # Chinese
|
|
CH_SERVER = "ch_server" # Chinese (server)
|
|
CH_LITE = "ch_lite" # Chinese (lite)
|
|
EN = "en" # English
|
|
KOREAN = "korean" # Korean
|
|
JAPAN = "japan" # Japanese
|
|
CHINESE_CHT = "chinese_cht" # Chinese Traditional
|
|
TA = "ta" # Tamil
|
|
TE = "te" # Telugu
|
|
KA = "ka" # Kannada
|
|
TH = "th" # Thai
|
|
EL = "el" # Greek
|
|
LATIN = "latin" # Latin
|
|
ARABIC = "arabic" # Arabic
|
|
EAST_SLAVIC = "east_slavic" # East Slavic
|
|
CYRILLIC = "cyrillic" # Cyrillic
|
|
DEVANAGARI = "devanagari" # Devanagari
|
|
|
|
|
|
class MinerUParseMethod(StrEnum):
|
|
"""MinerU PDF parsing methods (pipeline backend only)."""
|
|
|
|
AUTO = "auto" # Automatically determine the method based on the file type
|
|
TXT = "txt" # Use text extraction method
|
|
OCR = "ocr" # Use OCR method for image-based PDFs
|
|
|
|
|
|
@dataclass
|
|
class MinerUParseOptions:
|
|
"""Options for MinerU PDF parsing."""
|
|
|
|
backend: MinerUBackend = MinerUBackend.PIPELINE
|
|
lang: Optional[MinerULanguage] = None # language for OCR (pipeline backend only)
|
|
method: MinerUParseMethod = MinerUParseMethod.AUTO
|
|
server_url: Optional[str] = None
|
|
delete_output: bool = True
|
|
parse_method: str = "raw"
|
|
formula_enable: bool = True
|
|
table_enable: bool = True
|
|
|
|
|
|
class MinerUParser(RAGFlowPdfParser):
|
|
def __init__(self, mineru_path: str = "mineru", mineru_api: str = "", mineru_server_url: str = ""):
|
|
self.mineru_api = mineru_api.rstrip("/")
|
|
self.mineru_server_url = mineru_server_url.rstrip("/")
|
|
self.outlines = []
|
|
self.page_from = 0
|
|
self.page_to = MAXIMUM_PAGE_NUMBER
|
|
self.logger = logging.getLogger(self.__class__.__name__)
|
|
# Initialize the coverage stamp so the attribute exists from
|
|
# construction regardless of whether parse_pdf has run yet.
|
|
# ``vlm_configured`` is pre-seeded with ``False``; parse_pdf
|
|
# overwrites it with the actual vision-model status once it has
|
|
# the constructor wiring available. Pre-seeding means any
|
|
# caller that reads ``last_image_coverage[\"vlm_configured\"]``
|
|
# after _transfer_to_sections (but before parse_pdf fills it in)
|
|
# doesn't KeyError.
|
|
self.last_image_coverage: dict = {
|
|
"images_detected": 0,
|
|
"images_chunked": 0,
|
|
"images_described": 0,
|
|
"images_dropped_no_text": 0,
|
|
"images_unreadable_resource": 0,
|
|
"vlm_configured": False,
|
|
}
|
|
|
|
@staticmethod
|
|
def _is_zipinfo_symlink(member: zipfile.ZipInfo) -> bool:
|
|
return (member.external_attr >> 16) & 0o170000 == 0o120000
|
|
|
|
def _extract_zip_no_root(self, zip_path, extract_to, root_dir):
|
|
self.logger.info(f"[MinerU] Extract zip: zip_path={zip_path}, extract_to={extract_to}, root_hint={root_dir}")
|
|
base_dir = Path(extract_to).resolve()
|
|
with zipfile.ZipFile(zip_path, "r") as zip_ref:
|
|
members = zip_ref.infolist()
|
|
if not root_dir:
|
|
if members and members[0].filename.endswith("/"):
|
|
root_dir = members[0].filename
|
|
else:
|
|
root_dir = None
|
|
if root_dir:
|
|
root_dir = root_dir.replace("\\", "/")
|
|
if not root_dir.endswith("/"):
|
|
root_dir += "/"
|
|
|
|
for member in members:
|
|
if member.flag_bits & 0x1:
|
|
raise RuntimeError(f"[MinerU] Encrypted zip entry not supported: {member.filename}")
|
|
if self._is_zipinfo_symlink(member):
|
|
raise RuntimeError(f"[MinerU] Symlink zip entry not supported: {member.filename}")
|
|
|
|
name = member.filename.replace("\\", "/")
|
|
if root_dir and name == root_dir:
|
|
self.logger.info("[MinerU] Ignore root folder...")
|
|
continue
|
|
if root_dir and name.startswith(root_dir):
|
|
name = name[len(root_dir) :]
|
|
if not name:
|
|
continue
|
|
if name.startswith("/") or name.startswith("//") or re.match(r"^[A-Za-z]:", name):
|
|
raise RuntimeError(f"[MinerU] Unsafe zip path (absolute): {member.filename}")
|
|
|
|
parts = [p for p in name.split("/") if p not in ("", ".")]
|
|
if any(p == ".." for p in parts):
|
|
raise RuntimeError(f"[MinerU] Unsafe zip path (traversal): {member.filename}")
|
|
|
|
rel_path = os.path.join(*parts) if parts else ""
|
|
dest_path = (Path(extract_to) / rel_path).resolve(strict=False)
|
|
if dest_path != base_dir and base_dir not in dest_path.parents:
|
|
raise RuntimeError(f"[MinerU] Unsafe zip path (escape): {member.filename}")
|
|
|
|
if member.is_dir():
|
|
os.makedirs(dest_path, exist_ok=True)
|
|
continue
|
|
|
|
os.makedirs(dest_path.parent, exist_ok=True)
|
|
with zip_ref.open(member) as src, open(dest_path, "wb") as dst:
|
|
shutil.copyfileobj(src, dst)
|
|
|
|
@staticmethod
|
|
def _is_http_endpoint_valid(url, timeout=5):
|
|
try:
|
|
response = requests.head(url, timeout=timeout, allow_redirects=True)
|
|
return response.status_code in [200, 301, 302, 307, 308]
|
|
except Exception:
|
|
return False
|
|
|
|
@staticmethod
|
|
def _sanitize_section_text(section: str) -> str:
|
|
"""Normalize MinerU text blocks before chunking.
|
|
|
|
MinerU may return HTML fragments (e.g. table_body with <tr>/<td>/<br>).
|
|
Keep human-readable text while removing tag noise that hurts chunking.
|
|
"""
|
|
if not section:
|
|
return ""
|
|
section = html.unescape(section)
|
|
# Preserve rough structure before dropping tags.
|
|
section = re.sub(r"(?is)<\s*br\s*/?\s*>", "\n", section)
|
|
section = re.sub(r"(?is)</\s*(p|div|li|tr|h[1-6]|table|caption)\s*>", "\n", section)
|
|
section = re.sub(r"(?is)<[^>]+>", "", section)
|
|
# Collapse whitespace while preserving line boundaries.
|
|
section = re.sub(r"[ \t]+\n", "\n", section)
|
|
section = re.sub(r"\n{3,}", "\n\n", section)
|
|
section = re.sub(r"[ \t]{2,}", " ", section)
|
|
return section.strip()
|
|
|
|
def check_installation(self, backend: str = "pipeline", server_url: Optional[str] = None) -> tuple[bool, str]:
|
|
reason = ""
|
|
|
|
valid_backends = [b.value for b in MinerUBackend]
|
|
if backend not in valid_backends:
|
|
reason = f"[MinerU] Invalid backend '{backend}'. Valid backends are: {valid_backends}"
|
|
self.logger.warning(reason)
|
|
return False, reason
|
|
|
|
if not self.mineru_api:
|
|
reason = "[MinerU] MINERU_APISERVER not configured."
|
|
self.logger.warning(reason)
|
|
return False, reason
|
|
|
|
api_openapi = f"{self.mineru_api}/openapi.json"
|
|
try:
|
|
api_ok = self._is_http_endpoint_valid(api_openapi)
|
|
self.logger.info(f"[MinerU] API openapi.json reachable={api_ok} url={api_openapi}")
|
|
if not api_ok:
|
|
reason = f"[MinerU] MinerU API not accessible: {api_openapi}"
|
|
return False, reason
|
|
except Exception as exc:
|
|
reason = f"[MinerU] MinerU API check failed: {exc}"
|
|
self.logger.warning(reason)
|
|
return False, reason
|
|
|
|
if backend in (MinerUBackend.VLM_HTTP_CLIENT, MinerUBackend.HYBRID_HTTP_CLIENT):
|
|
resolved_server = server_url or self.mineru_server_url
|
|
if not resolved_server:
|
|
reason = f"[MinerU] MINERU_SERVER_URL required for {backend} backend."
|
|
self.logger.warning(reason)
|
|
return False, reason
|
|
try:
|
|
server_ok = self._is_http_endpoint_valid(resolved_server)
|
|
self.logger.info(f"[MinerU] {backend} server check reachable={server_ok} url={resolved_server}")
|
|
except Exception as exc:
|
|
self.logger.warning(f"[MinerU] {backend} server probe failed: {resolved_server}: {exc}")
|
|
|
|
return True, reason
|
|
|
|
def _run_mineru(self, input_path: Path, output_dir: Path, options: MinerUParseOptions, callback: Optional[Callable] = None, *, page_from: int = 0, page_to: int = MAXIMUM_PAGE_NUMBER) -> Path:
|
|
return self._run_mineru_api(input_path, output_dir, options, callback, page_from=page_from, page_to=page_to)
|
|
|
|
def _run_mineru_api(self, input_path: Path, output_dir: Path, options: MinerUParseOptions, callback: Optional[Callable] = None, *, page_from: int = 0, page_to: int = MAXIMUM_PAGE_NUMBER) -> Path:
|
|
pdf_file_path = str(input_path)
|
|
|
|
if not os.path.exists(pdf_file_path):
|
|
raise RuntimeError(f"[MinerU] PDF file not exists: {pdf_file_path}")
|
|
|
|
pdf_file_name = Path(pdf_file_path).stem.strip()
|
|
output_path = tempfile.mkdtemp(prefix=f"{pdf_file_name}_{options.method}_", dir=str(output_dir))
|
|
output_zip_path = os.path.join(str(output_dir), f"{Path(output_path).name}.zip")
|
|
|
|
# RAGFlow's page_to is exclusive (Python slice stop index), MinerU's
|
|
# end_page_id is 0-based inclusive. Translate at the API boundary so
|
|
# the dispatcher can pass its native range through unchanged.
|
|
if page_to == MAXIMUM_PAGE_NUMBER:
|
|
end_page_id = 99999
|
|
else:
|
|
end_page_id = page_to - 1
|
|
|
|
data = {
|
|
"output_dir": "./output",
|
|
"lang_list": options.lang.value if isinstance(options.lang, MinerULanguage) else options.lang,
|
|
"backend": options.backend.value if isinstance(options.backend, MinerUBackend) else options.backend,
|
|
"parse_method": options.method.value if isinstance(options.method, MinerUParseMethod) else options.method,
|
|
"formula_enable": options.formula_enable,
|
|
"table_enable": options.table_enable,
|
|
"return_md": True,
|
|
"return_middle_json": True,
|
|
"return_model_output": True,
|
|
"return_content_list": True,
|
|
"return_images": True,
|
|
"response_format_zip": True,
|
|
"start_page_id": page_from,
|
|
"end_page_id": end_page_id,
|
|
}
|
|
|
|
if options.server_url:
|
|
data["server_url"] = options.server_url
|
|
elif self.mineru_server_url:
|
|
data["server_url"] = self.mineru_server_url
|
|
|
|
self.logger.info(f"[MinerU] resolved page range start_page_id={page_from} end_page_id={end_page_id} (from page_from={page_from} page_to={page_to})")
|
|
self.logger.info(f"[MinerU] request {data=}")
|
|
self.logger.info(f"[MinerU] request {options=}")
|
|
|
|
headers = {"Accept": "application/json"}
|
|
try:
|
|
self.logger.info(f"[MinerU] invoke api: {self.mineru_api}/file_parse backend={data['backend']} server_url={data.get('server_url')}")
|
|
if callback:
|
|
callback(0.20, f"[MinerU] invoke api: {self.mineru_api}/file_parse")
|
|
with open(pdf_file_path, "rb") as pdf_file:
|
|
files = {"files": (pdf_file_name + ".pdf", pdf_file, "application/pdf")}
|
|
with requests.post(
|
|
url=f"{self.mineru_api}/file_parse",
|
|
files=files,
|
|
data=data,
|
|
headers=headers,
|
|
timeout=1800,
|
|
stream=True,
|
|
) as response:
|
|
if not response.ok:
|
|
body = (response.text or "")[:2000]
|
|
raise RuntimeError(f"[MinerU] api failed status={response.status_code} body={body}")
|
|
content_type = response.headers.get("Content-Type", "")
|
|
if not content_type.startswith("application/zip"):
|
|
raise RuntimeError(f"[MinerU] not zip returned from api: {content_type}")
|
|
self.logger.info(f"[MinerU] zip file returned, saving to {output_zip_path}...")
|
|
if callback:
|
|
callback(0.30, f"[MinerU] zip file returned, saving to {output_zip_path}...")
|
|
with open(output_zip_path, "wb") as f:
|
|
response.raw.decode_content = True
|
|
shutil.copyfileobj(response.raw, f)
|
|
self.logger.info(f"[MinerU] Unzip to {output_path}...")
|
|
self._extract_zip_no_root(output_zip_path, output_path, pdf_file_name + "/")
|
|
if callback:
|
|
callback(0.40, f"[MinerU] Unzip to {output_path}...")
|
|
self.logger.info("[MinerU] Api completed successfully.")
|
|
return Path(output_path)
|
|
except requests.RequestException as e:
|
|
detail = ""
|
|
resp = getattr(e, "response", None)
|
|
if resp is not None:
|
|
try:
|
|
detail = f" body={(resp.text or '')[:2000]}"
|
|
except Exception:
|
|
pass
|
|
raise RuntimeError(f"[MinerU] api failed with exception {e}{detail}") from e
|
|
|
|
def __images__(self, fnm, zoomin: int = 1, page_from=0, page_to=MAXIMUM_PAGE_NUMBER, callback=None):
|
|
self.page_from = page_from
|
|
self.page_to = page_to
|
|
if callback:
|
|
callback(0.16, "[MinerU] Rendering PDF pages...")
|
|
try:
|
|
with sys.modules[LOCK_KEY_pdfplumber]:
|
|
with pdfplumber.open(fnm) if isinstance(fnm, (str, PathLike)) else pdfplumber.open(BytesIO(fnm)) as pdf:
|
|
self.pdf = pdf
|
|
self.page_images = [p.to_image(resolution=72 * zoomin, antialias=True).original for _, p in enumerate(self.pdf.pages[page_from:page_to])]
|
|
except Exception as e:
|
|
self.page_images = None
|
|
self.total_page = 0
|
|
self.logger.exception("[MinerU] PDF page rendering failed for pages %s:%s: %s", page_from, page_to, e)
|
|
if callback:
|
|
callback(0.16, f"[MinerU] PDF page rendering failed for pages {page_from}:{page_to}: {e}")
|
|
else:
|
|
# Report success only after every selected page rendered successfully.
|
|
self.logger.info("[MinerU] Rendered %d PDF page images.", len(self.page_images))
|
|
if callback:
|
|
callback(0.19, f"[MinerU] Rendered {len(self.page_images)} PDF page images.")
|
|
|
|
@staticmethod
|
|
def _normalize_bbox(bbox):
|
|
x0, top, x1, bott = [float(v) for v in bbox[:4]]
|
|
# Normalize flipped coordinates (MinerU may report inverted bbox for flipped images)
|
|
if x0 > x1:
|
|
x0, x1 = x1, x0
|
|
if top > bott:
|
|
top, bott = bott, top
|
|
return x0, top, x1, bott
|
|
|
|
def _content_bbox_to_page_space(self, page_idx, bbox):
|
|
x0, top, x1, bott = self._normalize_bbox(bbox)
|
|
if hasattr(self, "page_images") and self.page_images and len(self.page_images) > page_idx:
|
|
page_width, page_height = self.page_images[page_idx].size
|
|
x0 = (x0 / 1000.0) * page_width
|
|
x1 = (x1 / 1000.0) * page_width
|
|
top = (top / 1000.0) * page_height
|
|
bott = (bott / 1000.0) * page_height
|
|
return x0, top, x1, bott
|
|
|
|
def _middle_bbox_to_page_space(self, page_idx, bbox, page_size=None):
|
|
x0, top, x1, bott = self._normalize_bbox(bbox)
|
|
if hasattr(self, "page_images") and self.page_images and len(self.page_images) > page_idx:
|
|
page_width, page_height = self.page_images[page_idx].size
|
|
if max(abs(x0), abs(x1), abs(top), abs(bott)) <= 1:
|
|
x0, x1 = x0 * page_width, x1 * page_width
|
|
top, bott = top * page_height, bott * page_height
|
|
elif page_size and len(page_size) >= 2:
|
|
source_width, source_height = float(page_size[0]), float(page_size[1])
|
|
if source_width > 0 and source_height > 0:
|
|
x0, x1 = x0 / source_width * page_width, x1 / source_width * page_width
|
|
top, bott = top / source_height * page_height, bott / source_height * page_height
|
|
else:
|
|
self.logger.warning("[MinerU] Invalid middle-json page_size=%s for page_idx=%s; using raw bbox=%s", page_size, page_idx, bbox)
|
|
else:
|
|
self.logger.warning("[MinerU] Missing middle-json page_size for page_idx=%s; using raw bbox=%s", page_idx, bbox)
|
|
return x0, top, x1, bott
|
|
|
|
@staticmethod
|
|
def _format_line_tag(page_idx, bbox):
|
|
x0, top, x1, bott = bbox
|
|
return "@@{}\t{:.1f}\t{:.1f}\t{:.1f}\t{:.1f}##".format(page_idx + 1, x0, x1, top, bott)
|
|
|
|
def _line_tag(self, bx):
|
|
middle_positions = bx.get("_mineru_positions")
|
|
if middle_positions:
|
|
return "".join(self._format_line_tag(pos["page_idx"], pos["bbox"]) for pos in middle_positions)
|
|
|
|
return self._format_line_tag(bx["page_idx"], self._content_bbox_to_page_space(bx["page_idx"], bx.get("bbox", (0, 0, 0, 0))))
|
|
|
|
def crop(self, text, ZM=1, need_position=False):
|
|
imgs = []
|
|
poss = self.extract_positions(text)
|
|
if not poss:
|
|
if need_position:
|
|
return None, None
|
|
return
|
|
|
|
if not getattr(self, "page_images", None):
|
|
self.logger.warning("[MinerU] crop called without page images; skipping image generation.")
|
|
if need_position:
|
|
return None, None
|
|
return
|
|
|
|
page_count = len(self.page_images)
|
|
|
|
filtered_poss = []
|
|
for pns, left, right, top, bottom in poss:
|
|
if not pns:
|
|
self.logger.warning("[MinerU] Empty page index list in crop; skipping this position.")
|
|
continue
|
|
valid_pns = [p for p in pns if 0 <= p < page_count]
|
|
if not valid_pns:
|
|
self.logger.warning(f"[MinerU] All page indices {pns} out of range for {page_count} pages; skipping.")
|
|
continue
|
|
filtered_poss.append((valid_pns, left, right, top, bottom))
|
|
|
|
poss = filtered_poss
|
|
if not poss:
|
|
self.logger.warning("[MinerU] No valid positions after filtering; skip cropping.")
|
|
if need_position:
|
|
return None, None
|
|
return
|
|
|
|
max_width = max(np.max([right - left for (_, left, right, _, _) in poss]), 6)
|
|
GAP = 6
|
|
pos = poss[0]
|
|
first_page_idx = pos[0][0]
|
|
poss.insert(0, ([first_page_idx], pos[1], pos[2], max(0, pos[3] - 120), max(pos[3] - GAP, 0)))
|
|
pos = poss[-1]
|
|
last_page_idx = pos[0][-1]
|
|
if not (0 <= last_page_idx < page_count):
|
|
self.logger.warning(f"[MinerU] Last page index {last_page_idx} out of range for {page_count} pages; skipping crop.")
|
|
if need_position:
|
|
return None, None
|
|
return
|
|
last_page_height = self.page_images[last_page_idx].size[1]
|
|
poss.append(
|
|
(
|
|
[last_page_idx],
|
|
pos[1],
|
|
pos[2],
|
|
min(last_page_height, pos[4] + GAP),
|
|
min(last_page_height, pos[4] + 120),
|
|
)
|
|
)
|
|
|
|
positions = []
|
|
for ii, (pns, left, right, top, bottom) in enumerate(poss):
|
|
right = left + max_width
|
|
|
|
if bottom <= top:
|
|
bottom = top + 2
|
|
|
|
for pn in pns[1:]:
|
|
if 0 <= pn - 1 < page_count:
|
|
bottom += self.page_images[pn - 1].size[1]
|
|
else:
|
|
self.logger.warning(f"[MinerU] Page index {pn}-1 out of range for {page_count} pages during crop; skipping height accumulation.")
|
|
|
|
if not (0 <= pns[0] < page_count):
|
|
self.logger.warning(f"[MinerU] Base page index {pns[0]} out of range for {page_count} pages during crop; skipping this segment.")
|
|
continue
|
|
|
|
img0 = self.page_images[pns[0]]
|
|
x0, y0, x1, y1 = int(left), int(top), int(right), int(min(bottom, img0.size[1]))
|
|
if x0 > x1:
|
|
x0, x1 = x1, x0
|
|
if y0 > y1:
|
|
y0, y1 = y1, y0
|
|
if x1 <= x0 or y1 <= y0:
|
|
continue
|
|
crop0 = img0.crop((x0, y0, x1, y1))
|
|
imgs.append(crop0)
|
|
if 0 < ii < len(poss) - 1:
|
|
positions.append((pns[0] + self.page_from, x0, x1, y0, y1))
|
|
|
|
bottom -= img0.size[1]
|
|
for pn in pns[1:]:
|
|
if not (0 <= pn < page_count):
|
|
self.logger.warning(f"[MinerU] Page index {pn} out of range for {page_count} pages during crop; skipping this page.")
|
|
continue
|
|
page = self.page_images[pn]
|
|
x0, y0, x1, y1 = int(left), 0, int(right), int(min(bottom, page.size[1]))
|
|
if x0 > x1:
|
|
x0, x1 = x1, x0
|
|
if y0 > y1:
|
|
y0, y1 = y1, y0
|
|
if x1 <= x0 or y1 <= y0:
|
|
bottom -= page.size[1]
|
|
continue
|
|
cimgp = page.crop((x0, y0, x1, y1))
|
|
imgs.append(cimgp)
|
|
if 0 < ii < len(poss) - 1:
|
|
positions.append((pn + self.page_from, x0, x1, y0, y1))
|
|
bottom -= page.size[1]
|
|
|
|
if not imgs:
|
|
if need_position:
|
|
return None, None
|
|
return
|
|
|
|
height = 0
|
|
for img in imgs:
|
|
height += img.size[1] + GAP
|
|
height = int(height)
|
|
width = int(np.max([i.size[0] for i in imgs]))
|
|
pic = Image.new("RGB", (width, height), (245, 245, 245))
|
|
height = 0
|
|
for ii, img in enumerate(imgs):
|
|
if ii == 0 or ii + 1 == len(imgs):
|
|
img = img.convert("RGBA")
|
|
overlay = Image.new("RGBA", img.size, (0, 0, 0, 0))
|
|
overlay.putalpha(128)
|
|
img = Image.alpha_composite(img, overlay).convert("RGB")
|
|
pic.paste(img, (0, int(height)))
|
|
height += img.size[1] + GAP
|
|
|
|
if need_position:
|
|
return pic, positions
|
|
return pic
|
|
|
|
@staticmethod
|
|
def extract_positions(txt: str):
|
|
poss = []
|
|
for tag in re.findall(r"@@[0-9-]+\t[0-9.\t]+##", txt):
|
|
pn, left, right, top, bottom = tag.strip("#").strip("@").split("\t")
|
|
left, right, top, bottom = float(left), float(right), float(top), float(bottom)
|
|
poss.append(([int(p) - 1 for p in pn.split("-")], left, right, top, bottom))
|
|
return poss
|
|
|
|
def _find_middle_json(self, output_dir: Path, subdir: Path, file_stem: str, safe_stem: str) -> Path | None:
|
|
middle_names = tuple(dict.fromkeys((f"{file_stem}_middle.json", f"{safe_stem}_middle.json", "middle.json")))
|
|
for base in (subdir, output_dir):
|
|
for name in middle_names:
|
|
candidate = base / name
|
|
if candidate.exists():
|
|
return candidate
|
|
|
|
stem_dirs = tuple(dict.fromkeys((file_stem, safe_stem)))
|
|
for pattern in ("**/*_middle.json", "**/middle.json"):
|
|
for candidate in sorted(output_dir.glob(pattern)):
|
|
rel_parts = candidate.relative_to(output_dir).parts
|
|
if candidate.name in middle_names or any(stem_dir in rel_parts for stem_dir in stem_dirs):
|
|
return candidate
|
|
return None
|
|
|
|
@staticmethod
|
|
def _normalize_match_text(text: str) -> str:
|
|
text = html.unescape(str(text or ""))
|
|
text = re.sub(r"(?is)<[^>]+>", "", text)
|
|
return re.sub(r"\s+", "", text).lower()
|
|
|
|
@classmethod
|
|
def _table_match_text(cls, output: dict[str, Any]) -> str:
|
|
text = output.get("table_body", "") + "\n".join(output.get("table_caption", [])) + "\n".join(output.get("table_footnote", []))
|
|
return cls._normalize_match_text(text)
|
|
|
|
@classmethod
|
|
def _middle_block_text(cls, value: Any) -> str:
|
|
parts = []
|
|
|
|
def collect(node):
|
|
if isinstance(node, str):
|
|
parts.append(node)
|
|
elif isinstance(node, list):
|
|
for item in node:
|
|
collect(item)
|
|
elif isinstance(node, dict):
|
|
for key, child in node.items():
|
|
if key in {"type", "bbox", "page_idx", "page_size"}:
|
|
continue
|
|
collect(child)
|
|
|
|
collect(value)
|
|
return cls._normalize_match_text(" ".join(parts))
|
|
|
|
def _iter_middle_blocks(self, middle_data: dict[str, Any]):
|
|
for default_page_idx, page in enumerate(middle_data.get("pdf_info", []) or []):
|
|
if not isinstance(page, dict):
|
|
continue
|
|
page_idx = int(page.get("page_idx", default_page_idx))
|
|
page_size = page.get("page_size")
|
|
for key in ("para_blocks", "preproc_blocks"):
|
|
for block in page.get(key, []) or []:
|
|
if not isinstance(block, dict) or not isinstance(block.get("bbox"), (list, tuple)) or len(block["bbox"]) < 4:
|
|
continue
|
|
yield {
|
|
"type": block.get("type"),
|
|
"page_idx": page_idx,
|
|
"bbox": self._middle_bbox_to_page_space(page_idx, block["bbox"], page_size),
|
|
"text": self._middle_block_text(block),
|
|
}
|
|
|
|
@staticmethod
|
|
def _overlap_ratio(a, b):
|
|
ax0, atop, ax1, abott = a
|
|
bx0, btop, bx1, bbott = b
|
|
x0, x1 = max(ax0, bx0), min(ax1, bx1)
|
|
top, bott = max(atop, btop), min(abott, bbott)
|
|
if x1 <= x0 or bott <= top:
|
|
return 0
|
|
area = (x1 - x0) * (bott - top)
|
|
base = max((ax1 - ax0) * (abott - atop), 1)
|
|
return area / base
|
|
|
|
def _middle_positions_for_output(self, output: dict[str, Any], middle_blocks: list[dict[str, Any]]):
|
|
if output.get("type") != MinerUContentType.TABLE:
|
|
return []
|
|
if not isinstance(output.get("bbox"), (list, tuple)) or len(output["bbox"]) < 4 or output.get("page_idx") is None:
|
|
return []
|
|
|
|
target_text = self._table_match_text(output)
|
|
page_idx = int(output["page_idx"])
|
|
anchor_bbox = self._content_bbox_to_page_space(page_idx, output["bbox"])
|
|
matches = []
|
|
|
|
for idx, block in enumerate(middle_blocks):
|
|
if block["type"] != "table":
|
|
continue
|
|
|
|
block_text = block.get("text", "")
|
|
is_anchor = block["page_idx"] == page_idx and self._overlap_ratio(anchor_bbox, block["bbox"]) >= 0.5
|
|
is_text_match = len(block_text) >= 4 and target_text and (block_text in target_text or target_text in block_text)
|
|
if is_anchor or is_text_match:
|
|
matches.append((idx, block, is_anchor, is_text_match))
|
|
|
|
anchor_indices = [idx for idx, _, is_anchor, _ in matches if is_anchor]
|
|
if not anchor_indices:
|
|
return []
|
|
|
|
first_anchor_idx = min(anchor_indices)
|
|
positions = []
|
|
seen = set()
|
|
|
|
for idx, block, is_anchor, is_text_match in matches:
|
|
if not (is_anchor or (idx >= first_anchor_idx and is_text_match)):
|
|
continue
|
|
|
|
key = (block["page_idx"], *[round(v, 3) for v in block["bbox"]])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
positions.append({"page_idx": block["page_idx"], "bbox": block["bbox"]})
|
|
|
|
positions.sort(key=lambda item: (item["page_idx"], item["bbox"][1], item["bbox"][0]))
|
|
return positions
|
|
|
|
def _enrich_outputs_with_middle_positions(self, outputs: list[dict[str, Any]], middle_json: Path):
|
|
try:
|
|
with open(middle_json, "r", encoding="utf-8") as f:
|
|
middle_data = json.load(f)
|
|
except Exception as exc:
|
|
self.logger.warning("[MinerU] Failed to read middle json %s: %s", middle_json, exc)
|
|
return
|
|
|
|
middle_blocks = list(self._iter_middle_blocks(middle_data))
|
|
if not middle_blocks:
|
|
self.logger.info("[MinerU] No middle-json blocks found in %s; skipping cross-page position enrichment.", middle_json)
|
|
return
|
|
|
|
enriched_count = 0
|
|
for output in outputs:
|
|
positions = self._middle_positions_for_output(output, middle_blocks)
|
|
if len(positions) > 1:
|
|
output["_mineru_positions"] = positions
|
|
enriched_count += 1
|
|
self.logger.info("[MinerU] Enriched %s outputs with cross-page table positions from %s.", enriched_count, middle_json)
|
|
|
|
def _read_output(self, output_dir: Path, file_stem: str, method: str = "auto", backend: str = "pipeline") -> list[dict[str, Any]]:
|
|
json_file = None
|
|
subdir = None
|
|
attempted = []
|
|
|
|
# mirror MinerU's sanitize_filename to align ZIP naming
|
|
def _sanitize_filename(name: str) -> str:
|
|
sanitized = re.sub(r"[/\\\.]{2,}|[/\\]", "", name)
|
|
sanitized = re.sub(r"[^\w.-]", "_", sanitized, flags=re.UNICODE)
|
|
if sanitized.startswith("."):
|
|
sanitized = "_" + sanitized[1:]
|
|
return sanitized or "unnamed"
|
|
|
|
safe_stem = _sanitize_filename(file_stem)
|
|
content_names = tuple(dict.fromkeys((f"{file_stem}_content_list.json", f"{safe_stem}_content_list.json")))
|
|
allowed_names = set(content_names)
|
|
self.logger.info(f"[MinerU] Expected output files: {', '.join(sorted(allowed_names))}")
|
|
self.logger.info(f"[MinerU] Searching output in: {output_dir}")
|
|
|
|
jf = output_dir / f"{file_stem}_content_list.json"
|
|
self.logger.info(f"[MinerU] Trying original path: {jf}")
|
|
attempted.append(jf)
|
|
if jf.exists():
|
|
subdir = output_dir
|
|
json_file = jf
|
|
else:
|
|
alt = output_dir / f"{safe_stem}_content_list.json"
|
|
self.logger.info(f"[MinerU] Trying sanitized filename: {alt}")
|
|
attempted.append(alt)
|
|
if alt.exists():
|
|
subdir = output_dir
|
|
json_file = alt
|
|
else:
|
|
nested_alt = output_dir / safe_stem / f"{safe_stem}_content_list.json"
|
|
self.logger.info(f"[MinerU] Trying sanitized nested path: {nested_alt}")
|
|
attempted.append(nested_alt)
|
|
if nested_alt.exists():
|
|
subdir = nested_alt.parent
|
|
json_file = nested_alt
|
|
else:
|
|
# Try vlm subdirectory (for vlm-http-client backend)
|
|
vlm_path = output_dir / "vlm" / f"{file_stem}_content_list.json"
|
|
self.logger.info(f"[MinerU] Trying vlm subdirectory: {vlm_path}")
|
|
attempted.append(vlm_path)
|
|
if vlm_path.exists():
|
|
subdir = vlm_path.parent
|
|
json_file = vlm_path
|
|
else:
|
|
vlm_safe = output_dir / "vlm" / f"{safe_stem}_content_list.json"
|
|
self.logger.info(f"[MinerU] Trying vlm subdirectory with sanitized name: {vlm_safe}")
|
|
attempted.append(vlm_safe)
|
|
if vlm_safe.exists():
|
|
subdir = vlm_safe.parent
|
|
json_file = vlm_safe
|
|
|
|
if not json_file:
|
|
parse_subdir = None
|
|
if backend.startswith("pipeline"):
|
|
parse_subdir = method
|
|
elif backend.startswith("hybrid"):
|
|
parse_subdir = f"hybrid_{method}"
|
|
elif backend.startswith("vlm"):
|
|
parse_subdir = "vlm"
|
|
|
|
if parse_subdir:
|
|
for content_name in content_names:
|
|
for candidate in output_dir.glob(f"**/{parse_subdir}/{content_name}"):
|
|
self.logger.info(f"[MinerU] Trying parse-method path: {candidate}")
|
|
attempted.append(candidate)
|
|
subdir = candidate.parent
|
|
json_file = candidate
|
|
break
|
|
if json_file:
|
|
break
|
|
|
|
if not json_file:
|
|
stem_dirs = tuple(dict.fromkeys((file_stem, safe_stem)))
|
|
patterns = []
|
|
if parse_subdir:
|
|
for stem_dir in stem_dirs:
|
|
patterns.extend(
|
|
[
|
|
f"**/{stem_dir}/{parse_subdir}/content_list.json",
|
|
f"**/{stem_dir}/{parse_subdir}/*_content_list.json",
|
|
]
|
|
)
|
|
patterns.extend(
|
|
[
|
|
f"**/{parse_subdir}/content_list.json",
|
|
f"**/{parse_subdir}/*_content_list.json",
|
|
]
|
|
)
|
|
for stem_dir in stem_dirs:
|
|
patterns.extend(
|
|
[
|
|
f"**/{stem_dir}/content_list.json",
|
|
f"**/{stem_dir}/*_content_list.json",
|
|
]
|
|
)
|
|
patterns.extend(["**/content_list.json", "**/*_content_list.json"])
|
|
|
|
for pattern in patterns:
|
|
for candidate in sorted(output_dir.glob(pattern)):
|
|
self.logger.info(f"[MinerU] Trying fallback path: {candidate}")
|
|
if candidate.name.endswith("_content_list.json"):
|
|
rel_parts = candidate.relative_to(output_dir).parts
|
|
in_stem_dir = any(stem_dir in rel_parts for stem_dir in stem_dirs)
|
|
stem_match = candidate.stem.startswith(file_stem) or candidate.stem.startswith(safe_stem)
|
|
if not (stem_match or in_stem_dir):
|
|
self.logger.info(f"[MinerU] Skip unrelated fallback candidate: {candidate}")
|
|
continue
|
|
attempted.append(candidate)
|
|
subdir = candidate.parent
|
|
json_file = candidate
|
|
break
|
|
if json_file:
|
|
break
|
|
|
|
if not json_file:
|
|
raise FileNotFoundError(f"[MinerU] Missing output file, tried: {', '.join(str(p) for p in attempted)}")
|
|
|
|
with open(json_file, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
|
|
middle_json = self._find_middle_json(output_dir, subdir, file_stem, safe_stem)
|
|
if middle_json:
|
|
self.logger.info(f"[MinerU] Found middle json: {middle_json}")
|
|
self._enrich_outputs_with_middle_positions(data, middle_json)
|
|
else:
|
|
self.logger.info("[MinerU] No middle json found; skipping cross-page position enrichment.")
|
|
|
|
for item in data:
|
|
for key in ("img_path", "table_img_path", "equation_img_path"):
|
|
if key in item and item[key]:
|
|
item[key] = str((subdir / item[key]).resolve())
|
|
return data
|
|
|
|
def _select_configured_pages(self, outputs: list[dict[str, Any]], page_ranges, page_from: int) -> list[dict[str, Any]]:
|
|
"""Keep only the blocks whose page falls inside ``page_ranges``.
|
|
|
|
The dispatcher covers every configured range with one request, so the PDF is
|
|
uploaded once. MinerU then also returns the pages between the ranges.
|
|
|
|
``page_idx`` counts from ``page_from``. A configured range ``(s, e)`` selects
|
|
the 1-based pages ``s`` to ``e - 1``, because ``queue_tasks`` builds each task
|
|
stop with ``e = min(e - 1, pages)``. The end stays exclusive here so the merged
|
|
span selects the same pages that one task per range selected.
|
|
|
|
A cross-page table is one block whose ``page_idx`` is its first page and whose
|
|
``_mineru_positions`` hold every page it covers. A table that starts on a gap
|
|
page and continues into a configured range must stay, so the block is kept when
|
|
any of its pages is in a range, not only the first one.
|
|
"""
|
|
if not page_ranges:
|
|
return outputs
|
|
|
|
spans = [(int(s), int(e)) for s, e in page_ranges]
|
|
kept = []
|
|
for output in outputs:
|
|
pages = [output.get("page_idx")]
|
|
pages += [pos.get("page_idx") for pos in output.get("_mineru_positions") or []]
|
|
pages = [int(page_idx) for page_idx in pages if page_idx is not None]
|
|
if not pages:
|
|
# A block with no page cannot be placed on a gap page either. The rest of
|
|
# the parser already treats it as unplaced: _line_tag gives it no position
|
|
# and _middle_positions_for_output returns nothing for it. One task per
|
|
# range kept it too, so dropping it here would lose content.
|
|
kept.append(output)
|
|
continue
|
|
if any(s <= page_from + page_idx + 1 < e for page_idx in pages for s, e in spans):
|
|
kept.append(output)
|
|
|
|
dropped = len(outputs) - len(kept)
|
|
if dropped:
|
|
self.logger.info("[MinerU] Dropped %d block(s) outside the configured page ranges %s.", dropped, spans)
|
|
return kept
|
|
|
|
def _transfer_to_sections(self, outputs: list[dict[str, Any]], parse_method: str = None, table_enable: bool = False):
|
|
sections = []
|
|
parse_method = (parse_method or "raw").lower()
|
|
# Track image coverage so silent loss of embedded PDF/DOCX images is
|
|
# queryable at the serialization boundary (issue #16978). When an
|
|
# IMAGE block yields nothing — no caption, no footnote, no VLM
|
|
# description — the image body is dropped from the chunk stream, so
|
|
# we surface that gap explicitly instead of letting it disappear.
|
|
image_coverage = {
|
|
"images_detected": 0,
|
|
"images_chunked": 0,
|
|
"images_dropped_no_text": 0,
|
|
"images_described": 0,
|
|
"images_unreadable_resource": 0,
|
|
# Pre-seeded so callers can read the key after this method
|
|
# returns without going through parse_pdf first. parse_pdf
|
|
# overwrites it with the real vision_model status.
|
|
"vlm_configured": False,
|
|
}
|
|
for output in outputs:
|
|
output_type = output.get("type")
|
|
# Count every visual block we see (IMAGE and CHART — MinerU emits
|
|
# both as visual content) so the image_coverage stamp reports
|
|
# "detected" symmetrically with "chunked" / "described" (which
|
|
# both already include CHART via _transfer_to_tables).
|
|
if output_type in {MinerUContentType.IMAGE, MinerUContentType.CHART}:
|
|
image_coverage["images_detected"] += 1
|
|
# These chunkers consume tables and images separately, so exclude
|
|
# media from their text sections. Raw consumers keep legacy sections.
|
|
# Charts are routed the same way as images: MinerU emits them as
|
|
# visual blocks that _transfer_to_tables turns into image chunks.
|
|
if parse_method in {"naive", "manual", "paper"} and (output_type in {MinerUContentType.IMAGE, MinerUContentType.CHART} or (output_type == MinerUContentType.TABLE and table_enable)):
|
|
continue
|
|
|
|
match output_type:
|
|
case MinerUContentType.TEXT:
|
|
section = output.get("text", "")
|
|
case MinerUContentType.TABLE:
|
|
section = output.get("table_body", "") + "\n".join(output.get("table_caption", [])) + "\n".join(output.get("table_footnote", []))
|
|
if not section.strip():
|
|
self.logger.warning("[MinerU] Empty table content at page_idx=%s; using fallback text.", output.get("page_idx"))
|
|
section = "FAILED TO PARSE TABLE"
|
|
case MinerUContentType.IMAGE:
|
|
section = "".join(output.get("image_caption", [])) + "\n" + "".join(output.get("image_footnote", []))
|
|
# If a vision model enriched this image with a semantic
|
|
# description (see _enhance_images_with_vlm), embed it in
|
|
# the chunk so it becomes searchable / retrievable.
|
|
vlm_description = (output.get("vlm_description") or "").strip()
|
|
if vlm_description:
|
|
image_coverage["images_described"] += 1
|
|
section = (section.strip("\n") + "\n" + vlm_description).strip("\n") if section.strip() else vlm_description
|
|
# Strip any residual newlines so the empty-after-strip case
|
|
# ("\n" from empty caption+footnote joined by "\n") is
|
|
# correctly counted as dropped at the boundary check below
|
|
# (issue #16978).
|
|
if not section.strip():
|
|
section = ""
|
|
case MinerUContentType.CHART:
|
|
section = "".join(output.get("chart_caption", [])) + "\n" + "".join(output.get("chart_footnote", []))
|
|
vlm_description = (output.get("vlm_description") or "").strip()
|
|
if vlm_description:
|
|
image_coverage["images_described"] += 1
|
|
section = (section.strip("\n") + "\n" + vlm_description).strip("\n") if section.strip() else vlm_description
|
|
case MinerUContentType.EQUATION:
|
|
section = output.get("text", "")
|
|
case MinerUContentType.CODE:
|
|
code_body = output.get("code_body", "")
|
|
code_caption = "\n".join(output.get("code_caption", []))
|
|
if code_caption:
|
|
# MinerU VLM returns a fenced body while the pipeline backend
|
|
# may return plain code. Replace only one complete outer fence.
|
|
outer_fence = re.fullmatch(r"```[^`\r\n]*\r?\n(?P<body>.*)\r?\n```", code_body, flags=re.DOTALL)
|
|
if outer_fence:
|
|
code_body = outer_fence.group("body")
|
|
section = f"```{code_caption}\n{code_body}\n```"
|
|
else:
|
|
section = code_body
|
|
case MinerUContentType.LIST:
|
|
section = "\n".join(output.get("list_items", []))
|
|
case MinerUContentType.HEADER | MinerUContentType.FOOTER | MinerUContentType.PAGE_NUMBER | MinerUContentType.DISCARDED:
|
|
continue
|
|
case _:
|
|
self.logger.warning("[MinerU] Skip unsupported section type=%s", output.get("type"))
|
|
continue
|
|
|
|
# Only flatten table HTML when table extraction is disabled; the
|
|
# same sanitization would corrupt valid text, code, and equations.
|
|
if output_type == MinerUContentType.TABLE and not table_enable:
|
|
section = self._sanitize_section_text(section)
|
|
if not section:
|
|
if output.get("type") in {MinerUContentType.IMAGE, MinerUContentType.CHART}:
|
|
image_coverage["images_dropped_no_text"] += 1
|
|
# DEBUG (not WARNING) because PDFs with many decorative
|
|
# images would flood the log; the parse_pdf summary
|
|
# already reports the aggregate dropped_no_text count.
|
|
self.logger.debug(
|
|
"[MinerU] Dropped embedded %s at page_idx=%s with no caption, footnote, or VLM description "
|
|
"(img_path=%s). Configure an IMAGE2TEXT vision model to make embedded visuals retrievable.",
|
|
"chart" if output.get("type") == MinerUContentType.CHART else "image",
|
|
output.get("page_idx"),
|
|
output.get("img_path"),
|
|
)
|
|
else:
|
|
self.logger.debug("[MinerU] Skip section after sanitization: type=%s", output.get("type"))
|
|
continue
|
|
|
|
# Mirror the same IMAGE/CHART scope as images_detected at the
|
|
# top of this loop and as images_chunked in
|
|
# _transfer_to_tables: a CHART block whose caption/footnote/
|
|
# vlm_description was emitted into a raw-mode text section is
|
|
# just as \"chunked\" as a caption-only IMAGE block — not
|
|
# silently dropped.
|
|
if output.get("type") in {MinerUContentType.IMAGE, MinerUContentType.CHART}:
|
|
image_coverage["images_chunked"] += 1
|
|
|
|
if section and parse_method in {"manual", "pipeline"}:
|
|
sections.append((section, output["type"], self._line_tag(output)))
|
|
elif section and parse_method == "paper":
|
|
sections.append((section + self._line_tag(output), output["type"]))
|
|
else:
|
|
sections.append((section, self._line_tag(output)))
|
|
# Stash on the instance so callers (and parse_pdf) can inspect the
|
|
# coverage stamp for the most recent parse.
|
|
self.last_image_coverage = image_coverage
|
|
return sections
|
|
|
|
def _transfer_to_tables(self, outputs: list[dict[str, Any]], table_enable: bool = True):
|
|
"""Convert MinerU media blocks to RAGFlow table/image tuples.
|
|
|
|
Args:
|
|
outputs: MinerU content-list blocks in source order.
|
|
table_enable: Whether table blocks should be emitted.
|
|
"""
|
|
tables = []
|
|
for output in outputs:
|
|
output_type = output.get("type")
|
|
if output_type not in {MinerUContentType.TABLE, MinerUContentType.IMAGE, MinerUContentType.CHART}:
|
|
continue
|
|
if output_type == MinerUContentType.TABLE and not table_enable:
|
|
continue
|
|
|
|
position_tag = self._line_tag(output) if "page_idx" in output and "bbox" in output else ""
|
|
# MinerU numbers pages from zero within the selected PDF slice.
|
|
# Media positions leave the parser in the document-global domain.
|
|
positions = [(page + self.page_from, left, right, top, bottom) for pages, left, right, top, bottom in self.extract_positions(position_tag) for page in pages]
|
|
|
|
if output_type == MinerUContentType.TABLE:
|
|
text = output.get("table_body", "") + "\n".join(output.get("table_caption", [])) + "\n".join(output.get("table_footnote", []))
|
|
if not text.strip():
|
|
self.logger.warning("[MinerU] Empty table content at page_idx=%s; using fallback text.", output.get("page_idx"))
|
|
text = "FAILED TO PARSE TABLE"
|
|
tables.append(((None, text), positions))
|
|
continue
|
|
|
|
# IMAGE and CHART share the same visual pipeline; only the caption
|
|
# field names differ (image_caption/image_footnote vs chart_*).
|
|
if output_type == MinerUContentType.CHART:
|
|
texts = [*output.get("chart_caption", []), *output.get("chart_footnote", [])]
|
|
else:
|
|
texts = [*output.get("image_caption", []), *output.get("image_footnote", [])]
|
|
vlm_description = (output.get("vlm_description") or "").strip()
|
|
if vlm_description:
|
|
texts.append(vlm_description)
|
|
# The text payload was non-empty if any of caption / footnote /
|
|
# vlm_description contributed a non-whitespace string. Use this
|
|
# to disambiguate "text was there but binary was unreadable"
|
|
# (images_unreadable_resource) from "text was empty" (text
|
|
# never made it into the chunk stream — images_dropped_no_text).
|
|
has_meaningful_text = any(t.strip() for t in texts)
|
|
|
|
image = None
|
|
image_path = output.get("img_path")
|
|
if image_path:
|
|
try:
|
|
with Image.open(image_path) as source:
|
|
source.load()
|
|
image = source.copy()
|
|
except Exception as e:
|
|
self.logger.debug(f"[MinerU] Failed to load image '{image_path}': {e}")
|
|
if image is None:
|
|
self.logger.debug("[MinerU] Skip image without a readable resource: %s", image_path)
|
|
# Image.open() couldn't read the resource. Route the
|
|
# observation to whichever counter matches the underlying
|
|
# failure mode:
|
|
# - has_meaningful_text → binary was unreadable; route to
|
|
# images_unreadable_resource.
|
|
# - text was empty → the image never made it into the
|
|
# chunk stream at all; route to
|
|
# images_dropped_no_text.
|
|
# The IMAGE block was detected (in _transfer_to_sections)
|
|
# but never produced an image chunk, so we do NOT count
|
|
# it as chunked/described here.
|
|
if has_meaningful_text:
|
|
self.last_image_coverage["images_unreadable_resource"] += 1
|
|
else:
|
|
self.last_image_coverage["images_dropped_no_text"] += 1
|
|
continue
|
|
|
|
# Count chunked/described only on confirmed emit (image actually
|
|
# reached the tables list). This avoids the brittle
|
|
# increment-then-decrement pattern that depends on every branch
|
|
# taking the right counter.
|
|
self.last_image_coverage["images_chunked"] += 1
|
|
if vlm_description:
|
|
self.last_image_coverage["images_described"] += 1
|
|
|
|
tables.append(((image, texts or [""]), positions))
|
|
return tables
|
|
|
|
def _enhance_images_with_vlm(self, outputs: list[dict[str, Any]], vision_model, callback: Optional[Callable] = None, language: str = "English"):
|
|
"""Generate semantic descriptions for image blocks via the tenant's
|
|
VISION model, mirroring deepdoc's VisionFigureParser. Each
|
|
IMAGE or CHART block with a readable img_path gets a
|
|
``vlm_description`` field that ``_transfer_to_sections`` then folds
|
|
into the chunk text — closing issue #14869.
|
|
"""
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from rag.app.picture import vision_llm_chunk
|
|
from rag.prompts.generator import vision_llm_figure_describe_prompt
|
|
|
|
image_jobs = [
|
|
(idx, item) for idx, item in enumerate(outputs) if item.get("type") in {MinerUContentType.IMAGE, MinerUContentType.CHART} and item.get("img_path") and os.path.exists(item["img_path"])
|
|
]
|
|
if not image_jobs:
|
|
return
|
|
|
|
if callback:
|
|
callback(0.78, f"[MinerU] Generating VLM descriptions for {len(image_jobs)} images...")
|
|
|
|
prompt = vision_llm_figure_describe_prompt(language=language or "English")
|
|
|
|
def worker(idx, item):
|
|
try:
|
|
with Image.open(item["img_path"]) as img:
|
|
img.load()
|
|
desc = vision_llm_chunk(binary=img, vision_model=vision_model, prompt=prompt)
|
|
return idx, (desc or "").strip()
|
|
except Exception as e:
|
|
logging.warning(f"[MinerU] VLM description failed for image #{idx}: {e}")
|
|
return idx, ""
|
|
|
|
with ThreadPoolExecutor(max_workers=10) as executor:
|
|
futures = [executor.submit(worker, idx, item) for idx, item in image_jobs]
|
|
for fut in as_completed(futures):
|
|
idx, desc = fut.result()
|
|
if desc:
|
|
outputs[idx]["vlm_description"] = desc
|
|
|
|
def parse_pdf(
|
|
self,
|
|
filepath: str | PathLike[str],
|
|
binary: BytesIO | bytes,
|
|
callback: Optional[Callable] = None,
|
|
*,
|
|
output_dir: Optional[str] = None,
|
|
backend: str = "pipeline",
|
|
server_url: Optional[str] = None,
|
|
delete_output: bool = True,
|
|
parse_method: str = "raw",
|
|
page_from: int = 0,
|
|
page_to: int = MAXIMUM_PAGE_NUMBER,
|
|
**kwargs,
|
|
) -> tuple:
|
|
import shutil
|
|
|
|
self.outlines = extract_pdf_outlines(binary if binary is not None else filepath)
|
|
temp_pdf = None
|
|
created_tmp_dir = False
|
|
|
|
parser_cfg = kwargs.get("parser_config", {})
|
|
lang = parser_cfg.get("mineru_lang") or kwargs.get("lang") or "English"
|
|
mineru_lang_code = LANGUAGE_TO_MINERU_MAP.get(lang, "ch") # Defaults to Chinese if not matched
|
|
mineru_method_raw_str = parser_cfg.get("mineru_parse_method", "auto")
|
|
enable_formula = parser_cfg.get("mineru_formula_enable", True)
|
|
enable_table = parser_cfg.get("mineru_table_enable", True)
|
|
|
|
# remove spaces, or mineru crash, and _read_output fail too
|
|
file_path = Path(filepath)
|
|
pdf_file_name = file_path.stem.replace(" ", "") + ".pdf"
|
|
pdf_file_path_valid = os.path.join(file_path.parent, pdf_file_name)
|
|
|
|
if binary:
|
|
temp_dir = Path(tempfile.mkdtemp(prefix="mineru_bin_pdf_"))
|
|
temp_pdf = temp_dir / pdf_file_name
|
|
with open(temp_pdf, "wb") as f:
|
|
f.write(binary)
|
|
pdf = temp_pdf
|
|
self.logger.info(f"[MinerU] Received binary PDF -> {temp_pdf}")
|
|
if callback:
|
|
callback(0.15, f"[MinerU] Received binary PDF -> {temp_pdf}")
|
|
else:
|
|
if pdf_file_path_valid != filepath:
|
|
self.logger.info(f"[MinerU] Remove all space in file name: {pdf_file_path_valid}")
|
|
shutil.move(filepath, pdf_file_path_valid)
|
|
pdf = Path(pdf_file_path_valid)
|
|
if not pdf.exists():
|
|
if callback:
|
|
callback(-1, f"[MinerU] PDF not found: {pdf}")
|
|
raise FileNotFoundError(f"[MinerU] PDF not found: {pdf}")
|
|
|
|
if output_dir:
|
|
out_dir = Path(output_dir)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
else:
|
|
out_dir = Path(tempfile.mkdtemp(prefix="mineru_pdf_"))
|
|
created_tmp_dir = True
|
|
|
|
self.logger.info(f"[MinerU] Output directory: {out_dir} backend={backend} api={self.mineru_api} server_url={server_url or self.mineru_server_url}")
|
|
if callback:
|
|
callback(0.15, f"[MinerU] Output directory: {out_dir}")
|
|
|
|
self.__images__(pdf, zoomin=1, page_from=page_from, page_to=page_to, callback=callback)
|
|
|
|
try:
|
|
options = MinerUParseOptions(
|
|
backend=MinerUBackend(backend),
|
|
lang=MinerULanguage(mineru_lang_code),
|
|
method=MinerUParseMethod(mineru_method_raw_str),
|
|
server_url=server_url,
|
|
delete_output=delete_output,
|
|
parse_method=parse_method,
|
|
formula_enable=enable_formula,
|
|
table_enable=enable_table,
|
|
)
|
|
final_out_dir = self._run_mineru(pdf, out_dir, options, callback=callback, page_from=page_from, page_to=page_to)
|
|
outputs = self._read_output(final_out_dir, pdf.stem, method=mineru_method_raw_str, backend=backend)
|
|
outputs = self._select_configured_pages(outputs, parser_cfg.get("pages"), page_from)
|
|
self.logger.info(f"[MinerU] Parsed {len(outputs)} blocks from PDF.")
|
|
if callback:
|
|
callback(0.75, f"[MinerU] Parsed {len(outputs)} blocks from PDF.")
|
|
|
|
vision_model = kwargs.get("vision_model")
|
|
if vision_model is not None:
|
|
try:
|
|
self._enhance_images_with_vlm(outputs, vision_model, callback=callback, language=lang)
|
|
except Exception as e:
|
|
self.logger.warning(f"[MinerU] VLM image enhancement failed: {e}. Continuing without descriptions.")
|
|
|
|
sections = self._transfer_to_sections(outputs, parse_method, enable_table)
|
|
tables = self._transfer_to_tables(outputs, enable_table) if (parse_method or "raw").lower() in {"naive", "manual", "paper"} else []
|
|
# Emit an image_coverage stamp so silent loss of embedded PDF/DOCX
|
|
# images is visible at the parser boundary (issue #16978). The
|
|
# vlm_configured flag tells operators whether downstream chunks
|
|
# could have been enriched by a vision model at all. The
|
|
# coverage dict is initialized in __init__ so this access is
|
|
# unconditional — see issue #16978.
|
|
coverage = self.last_image_coverage
|
|
coverage["vlm_configured"] = vision_model is not None
|
|
self.logger.info(
|
|
"[MinerU] image_coverage final detected=%s chunked=%s described=%s dropped_no_text=%s unreadable=%s vlm_configured=%s",
|
|
coverage["images_detected"],
|
|
coverage["images_chunked"],
|
|
coverage["images_described"],
|
|
coverage["images_dropped_no_text"],
|
|
coverage["images_unreadable_resource"],
|
|
coverage["vlm_configured"],
|
|
)
|
|
return sections, tables
|
|
finally:
|
|
if temp_pdf and temp_pdf.exists():
|
|
try:
|
|
temp_pdf.unlink()
|
|
temp_pdf.parent.rmdir()
|
|
except Exception:
|
|
pass
|
|
if delete_output and created_tmp_dir and out_dir.exists():
|
|
try:
|
|
shutil.rmtree(out_dir)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = MinerUParser("mineru")
|
|
ok, reason = parser.check_installation()
|
|
print("MinerU available:", ok)
|
|
|
|
filepath = ""
|
|
with open(filepath, "rb") as file:
|
|
outputs = parser.parse_pdf(filepath=filepath, binary=file.read())
|
|
for output in outputs:
|
|
print(output)
|