# type: ignore
"""
Generic Document Parser Utility
This module provides functionality for parsing documents using the built-in
MinerU, Docling, and PaddleOCR parsers, and exposes a small registry for
**in-process** custom parsers (see :func:`register_parser`).
Important notes:
- The custom parser registry is primarily intended for Python usage, where your
application imports a parser implementation and calls :func:`register_parser`
before invoking RAGAnything APIs.
- The standalone CLI (``python -m raganything.parser`` or the installed console
script) does **not** perform automatic plugin discovery; it will only see
custom parsers that have already been registered in the current process
(for example via a wrapper script or :mod:`sitecustomize`).
MinerU 2.0 no longer includes LibreOffice document conversion module.
For Office documents (.doc, .docx, .ppt, .pptx), please convert them to PDF
format first.
"""
from __future__ import annotations
import os
import platform
import hashlib
import json
import argparse
import base64
import subprocess
import tempfile
import threading
import logging
import time
import urllib.parse
import urllib.request
import shutil
from pathlib import Path
from typing import (
Dict,
List,
Optional,
Union,
Tuple,
Any,
Iterator,
)
from raganything.asset_urls import attach_public_media_urls
_IS_WINDOWS: bool = platform.system() == "Windows"
class MineruExecutionError(Exception):
"""catch mineru error"""
def __init__(self, return_code, error_msg):
self.return_code = return_code
self.error_msg = error_msg
super().__init__(
f"Mineru command failed with return code {return_code}: {error_msg}"
)
class Parser:
"""
Base class for document parsing utilities.
Defines common functionality and constants for parsing different document types.
"""
# Define common file formats
OFFICE_FORMATS = {".doc", ".docx", ".ppt", ".pptx", ".xls", ".xlsx"}
IMAGE_FORMATS = {".png", ".jpeg", ".jpg", ".bmp", ".tiff", ".tif", ".gif", ".webp"}
TEXT_FORMATS = {".txt", ".md"}
# Class-level logger
logger = logging.getLogger(__name__)
@staticmethod
def _is_url(path: str) -> bool:
"""Check if the path is a URL."""
try:
result = urllib.parse.urlparse(str(path))
return all([result.scheme, result.netloc])
except ValueError:
return False
def _download_file(self, url: str) -> Path:
"""
Download a file from a URL to a temporary file.
Attempts to preserve the file extension from the URL or Content-Type header.
"""
tmp_path = None
response = None
try:
self.logger.info(f"Downloading file from URL: {url}")
# Parse URL to get path and extension
parsed_url = urllib.parse.urlparse(url)
path = Path(parsed_url.path)
suffix = path.suffix if path.suffix else ""
# Create request with User-Agent to avoid 403 Forbidden from some sites
req = urllib.request.Request(
url,
data=None,
headers={
"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36"
},
)
# Open connection to get headers (with an explicit timeout to prevent hanging)
response = urllib.request.urlopen(req, timeout=30)
# If no extension in URL, try Content-Type header
if not suffix:
content_type = (
response.headers.get("Content-Type", "").split(";")[0].strip()
)
if content_type:
import mimetypes
guessed_ext = mimetypes.guess_extension(content_type)
if guessed_ext:
suffix = guessed_ext
self.logger.info(
f"Inferred file extension '{suffix}' from Content-Type: {content_type}"
)
# Create a temporary file with the correct extension
fd, tmp_path = tempfile.mkstemp(suffix=suffix)
os.close(fd)
tmp_path = Path(tmp_path)
# Download the file content
with open(tmp_path, "wb") as out_file:
shutil.copyfileobj(response, out_file)
self.logger.info(
f"Downloaded to temporary file: {tmp_path} ({tmp_path.stat().st_size} bytes)"
)
return tmp_path
except Exception as e:
# Clean up temp file if it was created
if tmp_path and tmp_path.exists():
try:
tmp_path.unlink()
self.logger.debug(
f"Cleaned up temporary file after failed download: {tmp_path}"
)
except Exception as cleanup_error:
self.logger.warning(
f"Failed to clean up temp file {tmp_path}: {cleanup_error}"
)
self.logger.error(f"Failed to download file from {url}: {e}")
raise RuntimeError(f"Failed to download file from {url}: {e}")
finally:
if response:
response.close()
def __init__(self) -> None:
"""Initialize the base parser."""
pass
@staticmethod
def _unique_output_dir(
base_dir: Union[str, Path], file_path: Union[str, Path]
) -> Path:
"""Create a unique output subdirectory for a file to prevent same-name collisions.
When multiple files share the same name (e.g. dir1/paper.pdf and dir2/paper.pdf),
their parser output would collide in the same output directory. This creates a
unique subdirectory by appending a short hash of the file's absolute path. (Fixes #51)
Args:
base_dir: The base output directory
file_path: Path to the input file
Returns:
Path like base_dir/paper_a1b2c3d4/ unique per absolute file path.
"""
file_path = Path(file_path).resolve()
stem = file_path.stem
path_hash = hashlib.md5(str(file_path).encode()).hexdigest()[:8]
return Path(base_dir) / f"{stem}_{path_hash}"
@classmethod
def _libreoffice_command_candidates(cls) -> List[str]:
"""Return LibreOffice executable candidates for office conversion.
On Windows the ``libreoffice``/``soffice`` commands are frequently not
on PATH even when LibreOffice is installed, so we also probe the
``.exe`` names and the standard ``Program Files`` install locations.
"""
command_names = ["libreoffice", "soffice"]
candidates: List[str] = []
for command_name in command_names:
resolved = shutil.which(command_name)
if resolved:
candidates.append(resolved)
candidates.append(command_name)
if _IS_WINDOWS:
for command_name in ["soffice.exe", "libreoffice.exe"]:
resolved = shutil.which(command_name)
if resolved:
candidates.append(resolved)
candidates.append(command_name)
for env_name in ("PROGRAMFILES", "PROGRAMFILES(X86)"):
program_files = os.environ.get(env_name)
if not program_files:
continue
libreoffice_program = Path(program_files) / "LibreOffice" / "program"
for exe_name in ("soffice.exe", "libreoffice.exe"):
exe_path = libreoffice_program / exe_name
if exe_path.exists():
candidates.append(str(exe_path))
deduped: List[str] = []
seen = set()
for candidate in candidates:
normalized = os.path.normcase(candidate)
if normalized not in seen:
seen.add(normalized)
deduped.append(candidate)
return deduped
@classmethod
def convert_office_to_pdf(
cls, doc_path: Union[str, Path], output_dir: Optional[str] = None
) -> Path:
"""
Convert Office document (.doc, .docx, .ppt, .pptx, .xls, .xlsx) to PDF.
Requires LibreOffice to be installed.
Args:
doc_path: Path to the Office document file
output_dir: Output directory for the PDF file
Returns:
Path to the generated PDF file
"""
try:
# Convert to Path object for easier handling
doc_path = Path(doc_path)
if not doc_path.exists():
raise FileNotFoundError(f"Office document does not exist: {doc_path}")
name_without_suff = doc_path.stem
# Prepare output directory
if output_dir:
base_output_dir = Path(output_dir)
else:
base_output_dir = doc_path.parent / "libreoffice_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
# Create temporary directory for PDF conversion
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
# Convert to PDF using LibreOffice
cls.logger.info(
f"Converting {doc_path.name} to PDF using LibreOffice..."
)
# Try LibreOffice commands in order of preference, including
# Windows .exe names and standard install locations.
commands_to_try = cls._libreoffice_command_candidates()
conversion_successful = False
last_cmd = commands_to_try[-1]
for cmd in commands_to_try:
is_last = cmd == last_cmd
try:
convert_cmd = [
cmd,
"--headless",
"--convert-to",
"pdf",
"--outdir",
str(temp_path),
str(doc_path),
]
# Prepare conversion subprocess parameters
convert_subprocess_kwargs = {
"capture_output": True,
"text": True,
"timeout": 60, # 60 second timeout
"encoding": "utf-8",
"errors": "ignore",
}
# Hide console window on Windows
if _IS_WINDOWS:
convert_subprocess_kwargs["creationflags"] = (
subprocess.CREATE_NO_WINDOW
)
result = subprocess.run(
convert_cmd, **convert_subprocess_kwargs
)
if result.returncode == 0:
conversion_successful = True
cls.logger.info(
f"Successfully converted {doc_path.name} to PDF using {cmd}"
)
break
else:
cls.logger.warning(
f"LibreOffice command '{cmd}' failed: {result.stderr}"
)
except FileNotFoundError:
# Only warn when all candidates are exhausted; otherwise
# log at debug level so that the normal fallback from
# 'libreoffice' → 'soffice' does not surface a spurious
# WARNING to users whose system only has 'soffice'.
if is_last:
cls.logger.warning(f"LibreOffice command '{cmd}' not found")
else:
cls.logger.debug(
f"LibreOffice command '{cmd}' not found, "
f"trying next candidate"
)
except subprocess.TimeoutExpired:
cls.logger.warning(f"LibreOffice command '{cmd}' timed out")
except Exception as e:
cls.logger.error(
f"LibreOffice command '{cmd}' failed with exception: {e}"
)
if not conversion_successful:
raise RuntimeError(
f"LibreOffice conversion failed for {doc_path.name}. "
f"Please ensure LibreOffice is installed:\n"
"- Windows: Download from https://www.libreoffice.org/download/download/\n"
"- macOS: brew install --cask libreoffice\n"
"- Ubuntu/Debian: sudo apt-get install libreoffice\n"
"- CentOS/RHEL: sudo yum install libreoffice\n"
"Alternatively, convert the document to PDF manually."
)
# Find the generated PDF
pdf_files = list(temp_path.glob("*.pdf"))
if not pdf_files:
raise RuntimeError(
f"PDF conversion failed for {doc_path.name} - no PDF file generated. "
f"Please check LibreOffice installation or try manual conversion."
)
pdf_path = pdf_files[0]
cls.logger.info(
f"Generated PDF: {pdf_path.name} ({pdf_path.stat().st_size} bytes)"
)
# Validate the generated PDF
if pdf_path.stat().st_size < 100: # Very small file, likely empty
raise RuntimeError(
"Generated PDF appears to be empty or corrupted. "
"Original file may have issues or LibreOffice conversion failed."
)
# Copy PDF to final output directory
final_pdf_path = base_output_dir / f"{name_without_suff}.pdf"
import shutil
shutil.copy2(pdf_path, final_pdf_path)
return final_pdf_path
except Exception as e:
cls.logger.error(f"Error in convert_office_to_pdf: {str(e)}")
raise
@classmethod
def convert_text_to_pdf(
cls, text_path: Union[str, Path], output_dir: Optional[str] = None
) -> Path:
"""
Convert text file (.txt, .md) to PDF using ReportLab with full markdown support.
Args:
text_path: Path to the text file
output_dir: Output directory for the PDF file
Returns:
Path to the generated PDF file
"""
try:
text_path = Path(text_path)
if not text_path.exists():
raise FileNotFoundError(f"Text file does not exist: {text_path}")
# Supported text formats
supported_text_formats = {".txt", ".md"}
if text_path.suffix.lower() not in supported_text_formats:
raise ValueError(f"Unsupported text format: {text_path.suffix}")
# Read the text content
try:
with open(text_path, "r", encoding="utf-8") as f:
text_content = f.read()
except UnicodeDecodeError:
# Try with different encodings
for encoding in ["gbk", "latin-1", "cp1252"]:
try:
with open(text_path, "r", encoding=encoding) as f:
text_content = f.read()
cls.logger.info(
f"Successfully read file with {encoding} encoding"
)
break
except UnicodeDecodeError:
continue
else:
raise RuntimeError(
f"Could not decode text file {text_path.name} with any supported encoding"
)
# Prepare output directory
if output_dir:
base_output_dir = Path(output_dir)
else:
base_output_dir = text_path.parent / "reportlab_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
pdf_path = base_output_dir / f"{text_path.stem}.pdf"
# Convert text to PDF
cls.logger.info(f"Converting {text_path.name} to PDF...")
try:
from reportlab.lib.pagesizes import A4
from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer
from reportlab.lib.styles import getSampleStyleSheet, ParagraphStyle
from reportlab.lib.units import inch
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
support_chinese = True
try:
if "WenQuanYi" not in pdfmetrics.getRegisteredFontNames():
if not Path(
"/usr/share/fonts/wqy-microhei/wqy-microhei.ttc"
).exists():
support_chinese = False
cls.logger.warning(
"WenQuanYi font not found at /usr/share/fonts/wqy-microhei/wqy-microhei.ttc. Chinese characters may not render correctly."
)
else:
pdfmetrics.registerFont(
TTFont(
"WenQuanYi",
"/usr/share/fonts/wqy-microhei/wqy-microhei.ttc",
)
)
except Exception as e:
support_chinese = False
cls.logger.warning(
f"Failed to register WenQuanYi font: {e}. Chinese characters may not render correctly."
)
# Create PDF document
doc = SimpleDocTemplate(
str(pdf_path),
pagesize=A4,
leftMargin=inch,
rightMargin=inch,
topMargin=inch,
bottomMargin=inch,
)
# Get styles
styles = getSampleStyleSheet()
normal_style = styles["Normal"]
heading_style = styles["Heading1"]
if support_chinese:
normal_style.fontName = "WenQuanYi"
heading_style.fontName = "WenQuanYi"
# Try to register a font that supports Chinese characters
# UnicodeCIDFont only supports specific CID font names:
# STSong-Light (Chinese), MSung-Light (Chinese Traditional),
# HeiseiMin-W3 / HeiseiKakuGo-W5 (Japanese),
# HYSMyeongJo-Medium (Korean)
# System font names like "SimSun", "SimHei", "STHeiti" are
# NOT valid CID names and silently fail (#24).
try:
from reportlab.pdfbase.cidfonts import UnicodeCIDFont
# STSong-Light is the standard CID font for Simplified
# Chinese and works cross-platform (reportlab ships the
# required CID resources internally).
pdfmetrics.registerFont(UnicodeCIDFont("STSong-Light"))
if not support_chinese:
normal_style.fontName = "STSong-Light"
heading_style.fontName = "STSong-Light"
except Exception:
pass # Use default fonts if Chinese font setup fails
# Build content
story = []
# Handle markdown or plain text
if text_path.suffix.lower() == ".md":
# Handle markdown content - simplified implementation
lines = text_content.split("\n")
for line in lines:
line = line.strip()
if not line:
story.append(Spacer(1, 12))
continue
# Headers
if line.startswith("#"):
level = len(line) - len(line.lstrip("#"))
header_text = line.lstrip("#").strip()
if header_text:
header_style = ParagraphStyle(
name=f"Heading{level}",
parent=heading_style,
fontSize=max(16 - level, 10),
spaceAfter=8,
spaceBefore=16 if level <= 2 else 12,
)
story.append(Paragraph(header_text, header_style))
else:
# Regular text
story.append(Paragraph(line, normal_style))
story.append(Spacer(1, 6))
else:
# Handle plain text files (.txt)
cls.logger.info(
f"Processing plain text file with {len(text_content)} characters..."
)
# Split text into lines and process each line
lines = text_content.split("\n")
line_count = 0
for line in lines:
line = line.rstrip()
line_count += 1
# Empty lines
if not line.strip():
story.append(Spacer(1, 6))
continue
# Regular text lines
# Escape special characters for ReportLab
safe_line = (
line.replace("&", "&")
.replace("<", "<")
.replace(">", ">")
)
# Create paragraph
story.append(Paragraph(safe_line, normal_style))
story.append(Spacer(1, 3))
cls.logger.info(f"Added {line_count} lines to PDF")
# If no content was added, add a placeholder
if not story:
story.append(Paragraph("(Empty text file)", normal_style))
# Build PDF
doc.build(story)
cls.logger.info(
f"Successfully converted {text_path.name} to PDF ({pdf_path.stat().st_size / 1024:.1f} KB)"
)
except ImportError:
raise RuntimeError(
"reportlab is required for text-to-PDF conversion. "
"Please install it using: pip install reportlab"
)
except Exception as e:
raise RuntimeError(
f"Failed to convert text file {text_path.name} to PDF: {str(e)}"
)
# Validate the generated PDF
if not pdf_path.exists() or pdf_path.stat().st_size < 100:
raise RuntimeError(
f"PDF conversion failed for {text_path.name} - generated PDF is empty or corrupted."
)
return pdf_path
except Exception as e:
cls.logger.error(f"Error in convert_text_to_pdf: {str(e)}")
raise
@classmethod
def _process_inline_markdown(cls, text: str) -> str:
"""
Process inline markdown formatting (bold, italic, code, links)
Args:
text: Raw text with markdown formatting
Returns:
Text with ReportLab markup
"""
import re
# Escape special characters for ReportLab
text = text.replace("&", "&").replace("<", "<").replace(">", ">")
# Bold text: **text** or __text__
text = re.sub(r"\*\*(.*?)\*\*", r"\1", text)
text = re.sub(r"__(.*?)__", r"\1", text)
# Italic text: *text* or _text_ (but not in the middle of words)
text = re.sub(r"(?\1", text)
text = re.sub(r"(?\1", text)
# Inline code: `code`
text = re.sub(
r"`([^`]+?)`",
r'\1',
text,
)
# Links: [text](url) - convert to text with URL annotation
def link_replacer(match):
link_text = match.group(1)
url = match.group(2)
return f'{link_text}'
text = re.sub(r"\[([^\]]+?)\]\(([^)]+?)\)", link_replacer, text)
# Strikethrough: ~~text~~
text = re.sub(r"~~(.*?)~~", r"\1", text)
return text
def parse_pdf(
self,
pdf_path: Union[str, Path],
output_dir: Optional[str] = None,
method: str = "auto",
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Abstract method to parse PDF document.
Must be implemented by subclasses.
Args:
pdf_path: Path to the PDF file
output_dir: Output directory path
method: Parsing method (auto, txt, ocr)
lang: Document language for OCR optimization
**kwargs: Additional parameters for parser-specific command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
raise NotImplementedError("parse_pdf must be implemented by subclasses")
def parse_image(
self,
image_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Abstract method to parse image document.
Must be implemented by subclasses.
Note: Different parsers may support different image formats.
Check the specific parser's documentation for supported formats.
Args:
image_path: Path to the image file
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for parser-specific command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
raise NotImplementedError("parse_image must be implemented by subclasses")
def parse_document(
self,
file_path: Union[str, Path],
method: str = "auto",
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Abstract method to parse a document.
Must be implemented by subclasses.
Args:
file_path: Path to the file to be parsed
method: Parsing method (auto, txt, ocr)
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for parser-specific command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
raise NotImplementedError("parse_document must be implemented by subclasses")
def check_installation(self) -> bool:
"""
Abstract method to check if the parser is properly installed.
Must be implemented by subclasses.
Returns:
bool: True if installation is valid, False otherwise
"""
raise NotImplementedError(
"check_installation must be implemented by subclasses"
)
class MineruParser(Parser):
"""
MinerU 2.0 document parsing utility class
Supports parsing PDF and image documents, converting the content into structured data
and generating markdown and JSON output.
Note: Office documents are no longer directly supported. Please convert them to PDF first.
"""
__slots__ = ()
# Class-level logger
logger = logging.getLogger(__name__)
def __init__(self) -> None:
"""Initialize MineruParser"""
super().__init__()
@classmethod
def _is_mineru_unsafe_windows_path(cls, path: Union[str, Path]) -> bool:
if not _IS_WINDOWS:
return False
path = Path(path)
path_text = str(path)
try:
path_text.encode("ascii")
except UnicodeEncodeError:
return True
return any(
part.endswith((" ", ".")) for part in path.parts
) or path.stem.endswith((" ", "."))
@classmethod
def _mineru_safe_path_hash(cls, path: Union[str, Path]) -> str:
path_text = str(Path(path).resolve())
return hashlib.md5(path_text.encode("utf-8")).hexdigest()[:10]
@classmethod
def _prepare_mineru_paths(
cls,
input_path: Union[str, Path],
output_dir: Union[str, Path],
hash_path: Optional[Union[str, Path]] = None,
) -> Tuple[Path, Path, str, Optional[Path]]:
input_path = Path(input_path)
output_dir = Path(output_dir)
hash_source = Path(hash_path) if hash_path is not None else input_path
input_is_unsafe = cls._is_mineru_unsafe_windows_path(input_path)
output_is_unsafe = cls._is_mineru_unsafe_windows_path(output_dir)
if not input_is_unsafe and not output_is_unsafe:
return input_path, output_dir, input_path.stem, None
path_hash = cls._mineru_safe_path_hash(hash_source)
temp_dir = Path(tempfile.mkdtemp(prefix="raganything_mineru_"))
mineru_input_path = input_path
if input_is_unsafe:
suffix = input_path.suffix.lower()
mineru_input_path = temp_dir / f"input_{path_hash}{suffix}"
shutil.copy2(input_path, mineru_input_path)
mineru_output_dir = output_dir
if output_is_unsafe:
mineru_output_dir = temp_dir / f"mineru_{path_hash}"
mineru_output_dir.mkdir(parents=True, exist_ok=True)
return mineru_input_path, mineru_output_dir, mineru_input_path.stem, temp_dir
@classmethod
def _copy_mineru_output_tree(cls, source_dir: Path, target_dir: Path) -> None:
if source_dir == target_dir:
return
target_dir.mkdir(parents=True, exist_ok=True)
if not source_dir.exists():
return
for item in source_dir.iterdir():
target = target_dir / item.name
if item.is_dir():
shutil.copytree(item, target, dirs_exist_ok=True)
else:
shutil.copy2(item, target)
@classmethod
def _cleanup_mineru_temp_dir(cls, temp_dir: Optional[Path]) -> None:
if temp_dir is not None and temp_dir.exists():
shutil.rmtree(temp_dir, ignore_errors=True)
@classmethod
def _run_mineru_command(
cls,
input_path: Union[str, Path],
output_dir: Union[str, Path],
method: str = "auto",
lang: Optional[str] = None,
backend: Optional[str] = None,
start_page: Optional[int] = None,
end_page: Optional[int] = None,
formula: bool = True,
table: bool = True,
device: Optional[str] = None,
source: Optional[str] = None,
vlm_url: Optional[str] = None,
timeout: Optional[int] = None,
**kwargs,
) -> None:
"""
Run mineru command line tool
Args:
input_path: Path to input file or directory
output_dir: Output directory path
method: Parsing method (auto, txt, ocr)
lang: Document language for OCR optimization
backend: Parsing backend
start_page: Starting page number (0-based)
end_page: Ending page number (0-based)
formula: Enable formula parsing
table: Enable table parsing
device: Inference device
source: Model source
vlm_url: When the backend is `vlm-http-client`, you need to specify the server_url
timeout: Maximum seconds to wait for MinerU to complete. None means no limit.
Raises TimeoutError if the process does not finish within this duration.
**kwargs: Additional parameters for subprocess (e.g., env)
"""
cmd = [
"mineru",
"-p",
str(input_path),
"-o",
str(output_dir),
"-m",
method,
]
if backend:
cmd.extend(["-b", backend])
if source:
cmd.extend(["--source", source])
if lang:
cmd.extend(["-l", lang])
if start_page is not None:
cmd.extend(["-s", str(start_page)])
if end_page is not None:
cmd.extend(["-e", str(end_page)])
if not formula:
cmd.extend(["-f", "false"])
if not table:
cmd.extend(["-t", "false"])
if device:
cmd.extend(["-d", device])
if vlm_url:
cmd.extend(["-u", vlm_url])
output_lines = []
error_lines = []
# Handle and validate environment variables
custom_env = kwargs.pop("env", None)
# Validate env if provided
if custom_env is not None:
if not isinstance(custom_env, dict):
raise TypeError(
f"env must be a dictionary, got {type(custom_env).__name__}"
)
for k, v in custom_env.items():
if not isinstance(k, str) or not isinstance(v, str):
raise TypeError("env keys and values must be strings")
# Check for unsupported arguments to fail fast
if kwargs:
unsupported = ", ".join(kwargs.keys())
raise TypeError(
f"MineruParser._run_mineru_command received unexpected keyword argument(s): {unsupported}"
)
try:
# Prepare subprocess parameters to hide console window on Windows
import threading
from queue import Queue, Empty
# Log the command being executed
cls.logger.info(f"Executing mineru command: {' '.join(cmd)}")
env = None
if custom_env:
env = os.environ.copy()
env.update(custom_env)
subprocess_kwargs = {
"stdout": subprocess.PIPE,
"stderr": subprocess.PIPE,
"text": True,
"encoding": "utf-8",
"errors": "ignore",
"bufsize": 1, # Line buffered
"env": env,
}
# Hide console window on Windows
if _IS_WINDOWS:
subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
# Function to read output from subprocess and add to queue
def enqueue_output(pipe, queue, prefix):
try:
for line in iter(pipe.readline, ""):
if line.strip(): # Only add non-empty lines
queue.put((prefix, line.strip()))
pipe.close()
except Exception as e:
queue.put((prefix, f"Error reading {prefix}: {e}"))
# Start subprocess
process = subprocess.Popen(cmd, **subprocess_kwargs)
# Create queues for stdout and stderr
stdout_queue = Queue()
stderr_queue = Queue()
# Start threads to read output
stdout_thread = threading.Thread(
target=enqueue_output, args=(process.stdout, stdout_queue, "STDOUT")
)
stderr_thread = threading.Thread(
target=enqueue_output, args=(process.stderr, stderr_queue, "STDERR")
)
stdout_thread.daemon = True
stderr_thread.daemon = True
stdout_thread.start()
stderr_thread.start()
# Process output in real time
start_time = time.monotonic()
while process.poll() is None:
# Check stdout queue
try:
while True:
prefix, line = stdout_queue.get_nowait()
output_lines.append(line)
# Log mineru output with INFO level, prefixed with [MinerU]
cls.logger.info(f"[MinerU] {line}")
except Empty:
pass
# Check stderr queue
try:
while True:
prefix, line = stderr_queue.get_nowait()
# Log mineru errors with WARNING level
if "warning" in line.lower():
cls.logger.warning(f"[MinerU] {line}")
elif "error" in line.lower():
cls.logger.error(f"[MinerU] {line}")
error_message = line.split("\n")[0]
error_lines.append(error_message)
else:
cls.logger.info(f"[MinerU] {line}")
except Empty:
pass
# Enforce timeout — kill the process and raise if exceeded
if timeout is not None and (time.monotonic() - start_time) > timeout:
process.kill()
process.wait()
# Give reader threads a moment to drain before raising
stdout_thread.join(timeout=1)
stderr_thread.join(timeout=1)
raise TimeoutError(
f"MinerU did not finish within {timeout}s. "
"This often means a model download is stuck due to network issues. "
"Check your internet connection or pre-download the required models."
)
# Small delay to prevent busy waiting
time.sleep(0.1)
# Process any remaining output after process completion
try:
while True:
prefix, line = stdout_queue.get_nowait()
output_lines.append(line)
cls.logger.info(f"[MinerU] {line}")
except Empty:
pass
try:
while True:
prefix, line = stderr_queue.get_nowait()
if "warning" in line.lower():
cls.logger.warning(f"[MinerU] {line}")
elif "error" in line.lower():
cls.logger.error(f"[MinerU] {line}")
error_message = line.split("\n")[0]
error_lines.append(error_message)
else:
cls.logger.info(f"[MinerU] {line}")
except Empty:
pass
# Wait for process to complete and get return code
return_code = process.wait()
# Wait for threads to finish
stdout_thread.join(timeout=5)
stderr_thread.join(timeout=5)
if return_code != 0 or error_lines:
cls.logger.info("[MinerU] Command executed failed")
raise MineruExecutionError(return_code, error_lines)
else:
cls.logger.info("[MinerU] Command executed successfully")
except MineruExecutionError:
raise
except subprocess.CalledProcessError as e:
cls.logger.error(f"Error running mineru subprocess command: {e}")
cls.logger.error(f"Command: {' '.join(cmd)}")
cls.logger.error(f"Return code: {e.returncode}")
raise
except FileNotFoundError:
raise RuntimeError(
"mineru command not found. Please ensure MinerU 2.0 is properly installed:\n"
"pip install -U 'mineru[core]' or uv pip install -U 'mineru[core]'"
)
except Exception as e:
error_message = f"Unexpected error running mineru command: {e}"
cls.logger.error(error_message)
raise RuntimeError(error_message) from e
@classmethod
def _read_output_files(
cls, output_dir: Path, file_stem: str, method: str = "auto"
) -> Tuple[List[Dict[str, Any]], str]:
"""
Read the output files generated by mineru
Args:
output_dir: Output directory
file_stem: File name without extension
method: Parsing method (used as fallback if subdirectory scan fails)
Returns:
Tuple containing (content list JSON, Markdown text)
"""
# Look for the generated files
md_file = output_dir / f"{file_stem}.md"
json_file = output_dir / f"{file_stem}_content_list.json"
images_base_dir = output_dir # Base directory for images
file_stem_subdir = output_dir / file_stem
if file_stem_subdir.is_dir():
# Scan for actual output subdirectory instead of assuming method name
found = False
for subdir in file_stem_subdir.iterdir():
if not subdir.is_dir():
continue
# Check if this subdirectory contains the expected JSON output file
candidate_json = subdir / f"{file_stem}_content_list.json"
if candidate_json.exists():
# Found the actual output directory
md_file = subdir / f"{file_stem}.md"
json_file = candidate_json
images_base_dir = subdir
found = True
cls.logger.info(
f"Found MinerU output in subdirectory: {subdir.name}"
)
break
# Fallback to method-based path if scanning didn't find output
if not found:
cls.logger.debug(
f"No output found by scanning, falling back to method-based path: {method}"
)
md_file = file_stem_subdir / method / f"{file_stem}.md"
json_file = file_stem_subdir / method / f"{file_stem}_content_list.json"
images_base_dir = file_stem_subdir / method
# Read markdown content
md_content = ""
if md_file.exists():
try:
with open(md_file, "r", encoding="utf-8") as f:
md_content = f.read()
except Exception as e:
cls.logger.warning(f"Could not read markdown file {md_file}: {e}")
# Read JSON content list
content_list = []
if json_file.exists():
try:
with open(json_file, "r", encoding="utf-8") as f:
content_list = json.load(f)
# Normalize MinerU 2.0 field names to expected names for backward compatibility.
# MinerU 2.0 renamed: img_caption -> image_caption, img_footnote -> image_footnote
# The codebase primarily uses image_caption/image_footnote with img_caption/img_footnote
# as fallback, but we ensure both fields exist so downstream code works regardless.
_FIELD_ALIASES = {
# MinerU 1.x name -> MinerU 2.0 name (canonical)
"img_caption": "image_caption",
"img_footnote": "image_footnote",
}
for item in content_list:
if isinstance(item, dict):
for old_name, new_name in _FIELD_ALIASES.items():
# If only the old field exists, copy it to the new field name
if old_name in item and new_name not in item:
item[new_name] = item[old_name]
# If only the new field exists, copy it to the old field name (for any legacy code)
elif new_name in item and old_name not in item:
item[old_name] = item[new_name]
# Always fix relative paths in content_list to absolute paths
cls.logger.info(
f"Fixing image paths in {json_file} with base directory: {images_base_dir}"
)
for item in content_list:
if isinstance(item, dict):
for field_name in [
"img_path",
"table_img_path",
"equation_img_path",
]:
if field_name in item and item[field_name]:
img_path = item[field_name]
absolute_img_path = (
images_base_dir / img_path
).resolve()
# Security check: ensure the image path is within the base directory
resolved_base = images_base_dir.resolve()
if not absolute_img_path.is_relative_to(resolved_base):
cls.logger.warning(
f"Potential path traversal detected in {field_name}: {img_path}. Skipping."
)
item[field_name] = "" # Clear unsafe path
continue
item[field_name] = str(absolute_img_path)
attach_public_media_urls(item)
except Exception as e:
cls.logger.warning(f"Could not read JSON file {json_file}: {e}")
return content_list, md_content
def parse_pdf(
self,
pdf_path: Union[str, Path],
output_dir: Optional[str] = None,
method: str = "auto",
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse PDF document using MinerU 2.0
Args:
pdf_path: Path to the PDF file
output_dir: Output directory path
method: Parsing method (auto, txt, ocr)
lang: Document language for OCR optimization
**kwargs: Additional parameters for mineru command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert to Path object for easier handling
pdf_path = Path(pdf_path)
if not pdf_path.exists():
raise FileNotFoundError(f"PDF file does not exist: {pdf_path}")
# Prepare output directory — use unique subdirectory to prevent
# same-name file collisions when output_dir is shared (#51)
if output_dir:
base_output_dir = self._unique_output_dir(output_dir, pdf_path)
else:
base_output_dir = pdf_path.parent / "mineru_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
mineru_input_path, mineru_output_dir, file_stem, temp_dir = (
self._prepare_mineru_paths(pdf_path, base_output_dir)
)
try:
# Run mineru command
self._run_mineru_command(
input_path=mineru_input_path,
output_dir=mineru_output_dir,
method=method,
lang=lang,
**kwargs,
)
self._copy_mineru_output_tree(mineru_output_dir, base_output_dir)
# Read the generated output files
# Map backend to expected output directory name for better compatibility
# MinerU 2.7.0+ uses different directory names based on backend:
# - pipeline -> auto/
# - vlm-* -> vlm/
# - hybrid-* -> hybrid_auto/
# Note: _read_output_files() will scan subdirectories automatically,
# so this mapping is just for optimization and fallback
# Use `or ""` to handle both missing keys and explicit None values
backend = kwargs.get("backend") or ""
if backend.startswith("vlm-"):
method = "vlm"
elif backend.startswith("hybrid-"):
method = "hybrid_auto"
content_list, _ = self._read_output_files(
base_output_dir, file_stem, method=method
)
return content_list
finally:
self._cleanup_mineru_temp_dir(temp_dir)
except MineruExecutionError:
raise
except Exception as e:
self.logger.error(f"Error in parse_pdf: {str(e)}")
raise
def parse_image(
self,
image_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse image document using MinerU 2.0
Note: MinerU 2.0 natively supports .png, .jpeg, .jpg formats.
Other formats (.bmp, .tiff, .tif, etc.) will be automatically converted to .png.
Args:
image_path: Path to the image file
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for mineru command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert to Path object for easier handling
image_path = Path(image_path)
if not image_path.exists():
raise FileNotFoundError(f"Image file does not exist: {image_path}")
# Supported image formats by MinerU 2.0
mineru_supported_formats = {".png", ".jpeg", ".jpg"}
# All supported image formats (including those we can convert)
all_supported_formats = {
".png",
".jpeg",
".jpg",
".bmp",
".tiff",
".tif",
".gif",
".webp",
}
ext = image_path.suffix.lower()
if ext not in all_supported_formats:
raise ValueError(
f"Unsupported image format: {ext}. Supported formats: {', '.join(all_supported_formats)}"
)
# Determine the actual image file to process
actual_image_path = image_path
temp_converted_file = None
# If format is not natively supported by MinerU, convert it
if ext not in mineru_supported_formats:
self.logger.info(
f"Converting {ext} image to PNG for MinerU compatibility..."
)
try:
from PIL import Image
except ImportError:
raise RuntimeError(
"PIL/Pillow is required for image format conversion. "
"Please install it using: pip install Pillow"
)
# Create temporary directory for conversion
temp_dir = Path(tempfile.mkdtemp())
temp_converted_file = temp_dir / f"{image_path.stem}_converted.png"
try:
# Open and convert image
with Image.open(image_path) as img:
# Handle different image modes
if img.mode in ("RGBA", "LA", "P"):
# For images with transparency or palette, convert to RGB first
if img.mode == "P":
img = img.convert("RGBA")
# Create white background for transparent images
background = Image.new("RGB", img.size, (255, 255, 255))
if img.mode == "RGBA":
background.paste(
img, mask=img.split()[-1]
) # Use alpha channel as mask
else:
background.paste(img)
img = background
elif img.mode not in ("RGB", "L"):
# Convert other modes to RGB
img = img.convert("RGB")
# Save as PNG
img.save(temp_converted_file, "PNG", optimize=True)
self.logger.info(
f"Successfully converted {image_path.name} to PNG ({temp_converted_file.stat().st_size / 1024:.1f} KB)"
)
actual_image_path = temp_converted_file
except Exception as e:
if temp_converted_file and temp_converted_file.exists():
temp_converted_file.unlink()
raise RuntimeError(
f"Failed to convert image {image_path.name}: {str(e)}"
)
# Prepare output directory — use unique subdirectory to prevent
# same-name file collisions when output_dir is shared (#51)
if output_dir:
base_output_dir = self._unique_output_dir(output_dir, image_path)
else:
base_output_dir = image_path.parent / "mineru_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
mineru_input_path, mineru_output_dir, file_stem, mineru_temp_dir = (
self._prepare_mineru_paths(
actual_image_path, base_output_dir, hash_path=image_path
)
)
try:
# Run mineru command (images are processed with OCR method)
self._run_mineru_command(
input_path=mineru_input_path,
output_dir=mineru_output_dir,
method="ocr", # Images require OCR method
lang=lang,
**kwargs,
)
self._copy_mineru_output_tree(mineru_output_dir, base_output_dir)
# Read the generated output files
content_list, _ = self._read_output_files(
base_output_dir, file_stem, method="ocr"
)
return content_list
except MineruExecutionError:
raise
finally:
self._cleanup_mineru_temp_dir(mineru_temp_dir)
# Clean up temporary converted file if it was created
if temp_converted_file and temp_converted_file.exists():
try:
temp_converted_file.unlink()
temp_converted_file.parent.rmdir() # Remove temp directory if empty
except Exception:
pass # Ignore cleanup errors
except Exception as e:
self.logger.error(f"Error in parse_image: {str(e)}")
raise
def parse_office_doc(
self,
doc_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse office document by first converting to PDF, then parsing with MinerU 2.0
Note: This method requires LibreOffice to be installed separately for PDF conversion.
MinerU 2.0 no longer includes built-in Office document conversion.
Supported formats: .doc, .docx, .ppt, .pptx, .xls, .xlsx
Args:
doc_path: Path to the document file (.doc, .docx, .ppt, .pptx, .xls, .xlsx)
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for mineru command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert Office document to PDF using base class method
pdf_path = self.convert_office_to_pdf(doc_path, output_dir)
# Parse the converted PDF
return self.parse_pdf(
pdf_path=pdf_path, output_dir=output_dir, lang=lang, **kwargs
)
except Exception as e:
self.logger.error(f"Error in parse_office_doc: {str(e)}")
raise
def parse_text_file(
self,
text_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse text file by first converting to PDF, then parsing with MinerU 2.0
Supported formats: .txt, .md
Args:
text_path: Path to the text file (.txt, .md)
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for mineru command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert text file to PDF using base class method
pdf_path = self.convert_text_to_pdf(text_path, output_dir)
# Parse the converted PDF
return self.parse_pdf(
pdf_path=pdf_path, output_dir=output_dir, lang=lang, **kwargs
)
except Exception as e:
self.logger.error(f"Error in parse_text_file: {str(e)}")
raise
def parse_document(
self,
file_path: Union[str, Path],
method: str = "auto",
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse document using MinerU 2.0 based on file extension
Args:
file_path: Path to the file to be parsed
method: Parsing method (auto, txt, ocr)
output_dir: Output directory path
lang: Document language for OCR optimization
**kwargs: Additional parameters for mineru command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
# Convert to Path object
file_path = Path(file_path)
if not file_path.exists():
raise FileNotFoundError(f"File does not exist: {file_path}")
# Get file extension
ext = file_path.suffix.lower()
# Choose appropriate parser based on file type
if ext == ".pdf":
return self.parse_pdf(file_path, output_dir, method, lang, **kwargs)
elif ext in self.IMAGE_FORMATS:
return self.parse_image(file_path, output_dir, lang, **kwargs)
elif ext in self.OFFICE_FORMATS:
self.logger.warning(
f"Warning: Office document detected ({ext}). "
f"MinerU 2.0 requires conversion to PDF first."
)
return self.parse_office_doc(file_path, output_dir, lang, **kwargs)
elif ext in self.TEXT_FORMATS:
return self.parse_text_file(file_path, output_dir, lang, **kwargs)
else:
# For unsupported file types, try as PDF
self.logger.warning(
f"Warning: Unsupported file extension '{ext}', "
f"attempting to parse as PDF"
)
return self.parse_pdf(file_path, output_dir, method, lang, **kwargs)
def check_installation(self) -> bool:
"""
Check if MinerU 2.0 is properly installed
Returns:
bool: True if installation is valid, False otherwise
"""
try:
# Prepare subprocess parameters to hide console window on Windows
subprocess_kwargs = {
"capture_output": True,
"text": True,
"check": True,
"encoding": "utf-8",
"errors": "ignore",
}
# Hide console window on Windows
if _IS_WINDOWS:
subprocess_kwargs["creationflags"] = subprocess.CREATE_NO_WINDOW
result = subprocess.run(["mineru", "--version"], **subprocess_kwargs)
self.logger.debug(f"MinerU version: {result.stdout.strip()}")
return True
except (subprocess.CalledProcessError, FileNotFoundError):
self.logger.debug(
"MinerU 2.0 is not properly installed. "
"Please install it using: pip install -U 'mineru[core]'"
)
return False
class DoclingParser(Parser):
"""
Docling document parsing utility class.
Specialized in parsing Office documents and HTML files, converting the content
into structured data and generating markdown and JSON output.
Backed by the Docling Python API (`docling.document_converter.DocumentConverter`)
to avoid subprocess overhead and re-initialization of Docling's deep-learning
models on every call. A `DocumentConverter` instance is built lazily on first
use and cached per pipeline-option combination so that subsequent parses
against the same configuration reuse already-loaded models.
Compatibility changes vs. earlier CLI-subprocess implementation
----------------------------------------------------------------
- `check_installation()` now returns True iff the Docling Python package
can be imported (`docling.document_converter.DocumentConverter`). The
previous behavior of probing the `docling` CLI executable on PATH is
gone; environments that ship the CLI without the importable package
(or vice versa) will see a different result than before.
- The legacy `env={...}` kwarg is still accepted for source-level
compatibility but is **ignored**: the Python API does not run a
subprocess, so per-call environment overrides no longer take effect.
Callers needing model-cache, proxy, or CUDA configuration should set
the corresponding environment variables in the parent process before
instantiating `DoclingParser`, or configure Docling directly via
`_get_converter` kwargs (`artifacts_path`, `table_mode`, ...).
- JSON and Markdown artifacts are still written to
`//docling/` for backward compatibility, but
they are produced by Docling's `export_to_dict()` /
`export_to_markdown()` rather than by the CLI's serializer; expect the
same logical content but not byte-identical files (key ordering,
whitespace, optional fields may differ).
Concurrency
-----------
The internal converter cache is guarded by a lock so that a single
`DoclingParser` instance can be safely shared across threads without
duplicating Docling model loads on first use.
"""
# Define Docling-specific formats
HTML_FORMATS = {".html", ".htm", ".xhtml"}
def __init__(self) -> None:
"""Initialize DoclingParser"""
super().__init__()
# Cache of DocumentConverter instances keyed by pipeline-option tuple,
# so that loaded layout/OCR/table models are reused across calls.
# The lock guards concurrent first-use from creating duplicate
# converters (and re-loading models) when the same DoclingParser
# instance is shared across threads.
self._converter_cache: Dict[Tuple, Any] = {}
self._converter_cache_lock = threading.Lock()
def parse_pdf(
self,
pdf_path: Union[str, Path],
output_dir: Optional[str] = None,
method: str = "auto",
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse PDF document using Docling
Args:
pdf_path: Path to the PDF file
output_dir: Output directory path
method: Parsing method (auto, txt, ocr)
lang: Document language for OCR optimization
**kwargs: Additional parameters for docling command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert to Path object for easier handling
pdf_path = Path(pdf_path)
if not pdf_path.exists():
raise FileNotFoundError(f"PDF file does not exist: {pdf_path}")
name_without_suff = pdf_path.stem
# Prepare output directory — use unique subdirectory to prevent
# same-name file collisions when output_dir is shared (#51)
if output_dir:
base_output_dir = self._unique_output_dir(output_dir, pdf_path)
else:
base_output_dir = pdf_path.parent / "docling_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
# Parse via the Docling Python API and convert directly from the
# in-memory dict, bypassing the JSON disk round-trip.
doc_dict = self._run_docling_python(
input_path=pdf_path,
output_dir=base_output_dir,
file_stem=name_without_suff,
**kwargs,
)
file_subdir = base_output_dir / name_without_suff / "docling"
content_list = self.read_from_block_recursive(
doc_dict["body"], "body", file_subdir, 0, "0", doc_dict
)
return content_list
except Exception as e:
self.logger.error(f"Error in parse_pdf: {str(e)}")
raise
def parse_document(
self,
file_path: Union[str, Path],
method: str = "auto",
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse document using Docling based on file extension
Args:
file_path: Path to the file to be parsed or URL
method: Parsing method
output_dir: Output directory path
lang: Document language for optimization
**kwargs: Additional parameters for docling command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
downloaded_temp_file = None
try:
# Check if input is a URL
if self._is_url(file_path):
file_path = self._download_file(file_path)
downloaded_temp_file = file_path
# Convert to Path object
file_path = Path(file_path)
if not file_path.exists():
raise FileNotFoundError(f"File does not exist: {file_path}")
# Get file extension
ext = file_path.suffix.lower()
# Choose appropriate parser based on file type
if ext == ".pdf":
return self.parse_pdf(file_path, output_dir, method, lang, **kwargs)
elif ext in self.OFFICE_FORMATS:
return self.parse_office_doc(file_path, output_dir, lang, **kwargs)
elif ext in self.HTML_FORMATS:
return self.parse_html(file_path, output_dir, lang, **kwargs)
else:
raise ValueError(
f"Unsupported file format: {ext}. "
f"Docling only supports PDF files, Office formats ({', '.join(self.OFFICE_FORMATS)}) "
f"and HTML formats ({', '.join(self.HTML_FORMATS)})"
)
finally:
# Clean up temporary file if we downloaded one
if downloaded_temp_file and downloaded_temp_file.exists():
try:
downloaded_temp_file.unlink()
self.logger.debug(f"Removed temporary file: {downloaded_temp_file}")
except Exception as e:
self.logger.warning(
f"Failed to remove temporary file {downloaded_temp_file}: {e}"
)
def _get_converter(self, **kwargs) -> Any:
"""
Lazily build and cache a `DocumentConverter` configured from kwargs.
Caches one converter per distinct pipeline-option tuple so that Docling's
layout, OCR, and TableFormer models are loaded only once per process for
a given configuration, drastically reducing per-document latency on
multi-document workloads.
Recognized kwargs (all optional):
table_mode (str): "fast" (default) or "accurate" – TableFormer mode.
tables (bool): Enable table structure recognition (default: True).
allow_ocr (bool): Enable OCR on scanned content (default: True).
artifacts_path (str): Path to a custom Docling models directory.
"""
from docling.document_converter import DocumentConverter, PdfFormatOption
from docling.datamodel.base_models import InputFormat
from docling.datamodel.pipeline_options import (
PdfPipelineOptions,
TableFormerMode,
)
table_mode = str(kwargs.get("table_mode", "fast")).lower()
do_tables = bool(kwargs.get("tables", True))
do_ocr = bool(kwargs.get("allow_ocr", True))
artifacts_path = kwargs.get("artifacts_path")
cache_key = (table_mode, do_tables, do_ocr, artifacts_path)
# Fast path: snapshot read outside the lock (dict reads are atomic in
# CPython for hashable keys) so the common cache-hit case stays
# contention-free.
cached = self._converter_cache.get(cache_key)
if cached is not None:
return cached
pipeline_options = PdfPipelineOptions()
if hasattr(pipeline_options, "do_ocr"):
pipeline_options.do_ocr = do_ocr
if hasattr(pipeline_options, "do_table_structure"):
pipeline_options.do_table_structure = do_tables
if hasattr(pipeline_options, "table_structure_options"):
try:
pipeline_options.table_structure_options.mode = (
TableFormerMode.ACCURATE
if table_mode == "accurate"
else TableFormerMode.FAST
)
except Exception as e: # pragma: no cover - defensive
self.logger.debug(f"Could not set TableFormer mode '{table_mode}': {e}")
if artifacts_path and hasattr(pipeline_options, "artifacts_path"):
pipeline_options.artifacts_path = artifacts_path
# Ask Docling to embed picture bytes in the exported dict so that
# `read_from_block` can extract them from `block["image"]["uri"]`
# without a second pass over the source document.
if hasattr(pipeline_options, "generate_picture_images"):
pipeline_options.generate_picture_images = True
if hasattr(pipeline_options, "images_scale"):
pipeline_options.images_scale = 2.0
# Slow path: serialize converter construction so that concurrent
# first-use against the same cache_key doesn't load Docling's models
# twice. We re-check the cache under the lock to avoid a double build
# when two threads race past the fast-path check above.
with self._converter_cache_lock:
cached = self._converter_cache.get(cache_key)
if cached is not None:
return cached
converter = DocumentConverter(
format_options={
InputFormat.PDF: PdfFormatOption(pipeline_options=pipeline_options),
}
)
self._converter_cache[cache_key] = converter
return converter
def _run_docling_python(
self,
input_path: Union[str, Path],
output_dir: Union[str, Path],
file_stem: str,
**kwargs,
) -> Dict[str, Any]:
"""
Parse `input_path` through the Docling Python API and return the
exported document dict.
Replaces the legacy `_run_docling_command` path that shelled out to the
`docling` CLI. JSON and Markdown artifacts are still written to
`//docling/` for backward compatibility, but the
document dict is also fed directly to `read_from_block_recursive`
without an intermediate disk round-trip.
Args:
input_path: Source document.
output_dir: Base output directory (a `/docling`
subdirectory will be created inside it).
file_stem: File name without extension, used for the subdirectory
and the output artifact filenames.
**kwargs: Forwarded to `_get_converter`. The legacy `env` kwarg is
still accepted for backward compatibility but has no effect
under the Python API.
Returns:
The Docling document exported via `export_to_dict()`.
"""
file_output_dir = Path(output_dir) / file_stem / "docling"
file_output_dir.mkdir(parents=True, exist_ok=True)
# The legacy CLI path accepted an `env` mapping. Validate it for type
# compatibility but otherwise drop it: the Python API does not need
# subprocess environment overrides.
custom_env = kwargs.pop("env", None)
if custom_env is not None:
if not isinstance(custom_env, dict):
raise TypeError(
f"env must be a dictionary, got {type(custom_env).__name__}"
)
for k, v in custom_env.items():
if not isinstance(k, str) or not isinstance(v, str):
raise TypeError("env keys and values must be strings")
self.logger.debug(
"DoclingParser: 'env' kwarg accepted for backward compatibility "
"but ignored by the Python API path."
)
try:
converter = self._get_converter(**kwargs)
except ImportError as e:
raise RuntimeError(
"Docling Python API is not available. Install it with: "
"pip install docling"
) from e
try:
result = converter.convert(str(input_path))
except Exception as e:
self.logger.error(f"Error running Docling Python API on {input_path}: {e}")
raise
doc = result.document
try:
doc_dict = doc.export_to_dict()
except Exception as e:
self.logger.error(f"Failed to export Docling document to dict: {e}")
raise
# Persist JSON + Markdown artifacts on disk to preserve the file layout
# produced by the previous CLI-based implementation. Failures here are
# logged but do not abort parsing, since callers only require the
# in-memory dict.
json_path = file_output_dir / f"{file_stem}.json"
try:
with open(json_path, "w", encoding="utf-8") as f:
json.dump(doc_dict, f, ensure_ascii=False, indent=2)
except Exception as e:
self.logger.warning(f"Could not write Docling JSON to {json_path}: {e}")
md_path = file_output_dir / f"{file_stem}.md"
try:
with open(md_path, "w", encoding="utf-8") as f:
f.write(doc.export_to_markdown())
except Exception as e:
self.logger.warning(f"Could not write Docling Markdown to {md_path}: {e}")
self.logger.info(
f"Docling Python API parse completed for {Path(input_path).name}"
)
return doc_dict
def read_from_block_recursive(
self,
block,
type: str,
output_dir: Path,
cnt: int,
num: str,
docling_content: Dict[str, Any],
) -> List[Dict[str, Any]]:
content_list = []
if not block.get("children"):
cnt += 1
content_list.append(self.read_from_block(block, type, output_dir, cnt, num))
else:
if type not in ["groups", "body"]:
cnt += 1
content_list.append(
self.read_from_block(block, type, output_dir, cnt, num)
)
members = block["children"]
for member in members:
cnt += 1
member_tag = member["$ref"]
# JSON References follow the form "#//" (e.g. "#/body/0")
ref_parts = member_tag.split("/")
if len(ref_parts) < 3:
self.logger.warning(
f"Unexpected $ref format (expected #//): {member_tag!r}"
)
continue
member_type = ref_parts[1]
member_num = ref_parts[2]
try:
member_block = docling_content[member_type][int(member_num)]
except (KeyError, ValueError, IndexError) as e:
self.logger.warning(f"Could not resolve $ref {member_tag!r}: {e}")
continue
content_list.extend(
self.read_from_block_recursive(
member_block,
member_type,
output_dir,
cnt,
member_num,
docling_content,
)
)
return content_list
def read_from_block(
self, block, type: str, output_dir: Path, cnt: int, num: str
) -> Dict[str, Any]:
if type == "texts":
if block["label"] == "formula":
return {
"type": "equation",
"img_path": "",
"text": block["orig"],
"text_format": "unknown",
"page_idx": cnt // 10,
}
else:
return {
"type": "text",
"text": block["orig"],
"page_idx": cnt // 10,
}
elif type == "pictures":
try:
base64_uri = block["image"]["uri"]
# base64 data URIs have the form "data:;base64,"
# but some exporters may omit the prefix
parts = base64_uri.split(",", 1)
base64_str = parts[1] if len(parts) == 2 else parts[0]
# Create images directory within the docling subdirectory
image_dir = output_dir / "images"
image_dir.mkdir(parents=True, exist_ok=True) # Ensure directory exists
image_path = image_dir / f"image_{num}.png"
with open(image_path, "wb") as f:
f.write(base64.b64decode(base64_str))
return {
"type": "image",
"img_path": str(image_path.resolve()), # Convert to absolute path
"image_caption": block.get("caption", ""),
"image_footnote": block.get("footnote", ""),
"page_idx": cnt // 10,
}
except Exception as e:
self.logger.warning(f"Failed to process image {num}: {e}")
return {
"type": "text",
"text": f"[Image processing failed: {block.get('caption', '')}]",
"page_idx": cnt // 10,
}
else:
try:
return {
"type": "table",
"img_path": "",
"table_caption": block.get("caption", ""),
"table_footnote": block.get("footnote", ""),
"table_body": block.get("data", []),
"page_idx": cnt // 10,
}
except Exception as e:
self.logger.warning(f"Failed to process table {num}: {e}")
return {
"type": "text",
"text": f"[Table processing failed: {block.get('caption', '')}]",
"page_idx": cnt // 10,
}
def parse_office_doc(
self,
doc_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse office document directly using Docling
Supported formats: .doc, .docx, .ppt, .pptx, .xls, .xlsx
Args:
doc_path: Path to the document file
output_dir: Output directory path
lang: Document language for optimization
**kwargs: Additional parameters for docling command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert to Path object
doc_path = Path(doc_path)
if not doc_path.exists():
raise FileNotFoundError(f"Document file does not exist: {doc_path}")
if doc_path.suffix.lower() not in self.OFFICE_FORMATS:
raise ValueError(f"Unsupported office format: {doc_path.suffix}")
name_without_suff = doc_path.stem
# Prepare output directory — use unique subdirectory to prevent
# same-name file collisions when output_dir is shared (#51)
if output_dir:
base_output_dir = self._unique_output_dir(output_dir, doc_path)
else:
base_output_dir = doc_path.parent / "docling_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
doc_dict = self._run_docling_python(
input_path=doc_path,
output_dir=base_output_dir,
file_stem=name_without_suff,
**kwargs,
)
file_subdir = base_output_dir / name_without_suff / "docling"
content_list = self.read_from_block_recursive(
doc_dict["body"], "body", file_subdir, 0, "0", doc_dict
)
return content_list
except Exception as e:
self.logger.error(f"Error in parse_office_doc: {str(e)}")
raise
def parse_html(
self,
html_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
"""
Parse HTML document using Docling
Supported formats: .html, .htm, .xhtml
Args:
html_path: Path to the HTML file
output_dir: Output directory path
lang: Document language for optimization
**kwargs: Additional parameters for docling command
Returns:
List[Dict[str, Any]]: List of content blocks
"""
try:
# Convert to Path object
html_path = Path(html_path)
if not html_path.exists():
raise FileNotFoundError(f"HTML file does not exist: {html_path}")
if html_path.suffix.lower() not in self.HTML_FORMATS:
raise ValueError(f"Unsupported HTML format: {html_path.suffix}")
name_without_suff = html_path.stem
# Prepare output directory — use unique subdirectory to prevent
# same-name file collisions when output_dir is shared (#51)
if output_dir:
base_output_dir = self._unique_output_dir(output_dir, html_path)
else:
base_output_dir = html_path.parent / "docling_output"
base_output_dir.mkdir(parents=True, exist_ok=True)
doc_dict = self._run_docling_python(
input_path=html_path,
output_dir=base_output_dir,
file_stem=name_without_suff,
**kwargs,
)
file_subdir = base_output_dir / name_without_suff / "docling"
content_list = self.read_from_block_recursive(
doc_dict["body"], "body", file_subdir, 0, "0", doc_dict
)
return content_list
except Exception as e:
self.logger.error(f"Error in parse_html: {str(e)}")
raise
def check_installation(self) -> bool:
"""
Check whether the Docling Python package is importable.
Returns:
bool: True if `docling.document_converter.DocumentConverter` can be
imported, False otherwise.
Note:
This is a behavior change from the previous CLI-subprocess
implementation, which probed the `docling` executable on PATH.
Some environments may have the CLI installed without the Python
package (or vice versa) and will therefore see a different
result. The Python-API path is what `parse_pdf`,
`parse_office_doc`, and `parse_html` actually exercise, so this
check is now a faithful pre-flight for those entry points.
"""
try:
from docling.document_converter import DocumentConverter # noqa: F401
return True
except ImportError:
self.logger.debug(
"Docling Python package is not installed. "
"Install it with: pip install docling"
)
return False
class PaddleOCRParser(Parser):
"""PaddleOCR document parser with optional PDF page rendering support."""
def __init__(self, default_lang: str = "en") -> None:
super().__init__()
self.default_lang = default_lang
self._ocr_instances: Dict[str, Any] = {}
def _require_paddleocr(self):
try:
from paddleocr import PaddleOCR
except ImportError as exc:
raise ImportError(
"PaddleOCR parser requires optional dependency `paddleocr`. "
"Install with `pip install -e '.[paddleocr]'` or "
"`uv sync --extra paddleocr`. "
"PaddleOCR also needs `paddlepaddle`; install it from "
"https://www.paddlepaddle.org.cn/install/quick."
) from exc
return PaddleOCR
def _get_ocr(self, lang: Optional[str] = None):
PaddleOCR = self._require_paddleocr()
language = (lang or self.default_lang).strip() or self.default_lang
cached = self._ocr_instances.get(language)
if cached is not None:
return cached
init_candidates = [
{"lang": language, "show_log": False},
{"lang": language},
{},
]
last_exception = None
for candidate_kwargs in init_candidates:
try:
ocr = PaddleOCR(**candidate_kwargs)
self._ocr_instances[language] = ocr
return ocr
except Exception as exc: # pragma: no cover - defensive fallback
last_exception = exc
continue
raise RuntimeError(
f"Unable to initialize PaddleOCR for language '{language}': {last_exception}"
)
def _extract_text_lines(self, result: Any) -> List[str]:
lines: List[str] = []
def append_text(text: str) -> None:
clean_text = text.strip()
if clean_text:
lines.append(clean_text)
if isinstance(result, str):
append_text(result)
return lines
def visit(node: Any) -> None:
if node is None:
return
if hasattr(node, "to_dict"):
try:
visit(node.to_dict())
return
except Exception:
pass
if isinstance(node, dict):
rec_texts = node.get("rec_texts")
if isinstance(rec_texts, list):
for item in rec_texts:
if isinstance(item, str):
append_text(item)
else:
visit(item)
text_value = node.get("text")
if isinstance(text_value, str):
append_text(text_value)
texts_value = node.get("texts")
if isinstance(texts_value, list):
for item in texts_value:
if isinstance(item, str):
append_text(item)
else:
visit(item)
# Avoid double-visiting keys we already handled above; this prevents
# accidental duplication without content-level deduplication.
for key, value in node.items():
if key in {"rec_texts", "text", "texts"}:
continue
visit(value)
return
if isinstance(node, (list, tuple)):
if node and all(isinstance(item, str) for item in node):
for item in node:
append_text(item)
return
if (
len(node) >= 2
and isinstance(node[1], (list, tuple))
and len(node[1]) >= 1
and isinstance(node[1][0], str)
):
append_text(node[1][0])
return
if (
len(node) >= 1
and isinstance(node[0], str)
and (len(node) == 1 or isinstance(node[1], (int, float)))
):
append_text(node[0])
return
for item in node:
visit(item)
return
if isinstance(node, str):
append_text(node)
return
visit(result)
return lines
def _ocr_input(
self, input_data: Any, lang: Optional[str] = None, cls_enabled: bool = True
) -> List[str]:
ocr = self._get_ocr(lang=lang)
if hasattr(ocr, "ocr"):
try:
result = ocr.ocr(input_data, cls=cls_enabled)
except TypeError:
result = ocr.ocr(input_data)
return self._extract_text_lines(result)
if hasattr(ocr, "predict"):
result = ocr.predict(input_data)
return self._extract_text_lines(result)
raise RuntimeError(
"Unsupported PaddleOCR API: expected `ocr` or `predict` method."
)
def _extract_pdf_page_inputs(self, pdf_path: Path) -> Iterator[Tuple[int, Any]]:
try:
import pypdfium2 as pdfium
except ImportError as exc:
raise ImportError(
"PDF parsing with parser='paddleocr' requires `pypdfium2`. "
"Install with `pip install -e '.[paddleocr]'` or "
"`uv sync --extra paddleocr`."
) from exc
pdf = pdfium.PdfDocument(str(pdf_path))
try:
total_pages = len(pdf)
for page_idx in range(total_pages):
page = pdf[page_idx]
try:
rendered = page.render(scale=2.0)
if hasattr(rendered, "to_pil"):
yield (page_idx, rendered.to_pil())
elif hasattr(rendered, "to_numpy"):
yield (page_idx, rendered.to_numpy())
else:
raise RuntimeError(
"Unsupported rendered page format from pypdfium2."
)
finally:
if hasattr(page, "close"):
page.close()
finally:
if hasattr(pdf, "close"):
pdf.close()
def _ocr_rendered_page(
self, rendered_page: Any, lang: Optional[str] = None, cls_enabled: bool = True
) -> List[str]:
if hasattr(rendered_page, "save"):
temp_image_path: Optional[Path] = None
try:
with tempfile.NamedTemporaryFile(suffix=".png", delete=False) as temp:
temp_image_path = Path(temp.name)
rendered_page.save(temp_image_path)
return self._ocr_input(
str(temp_image_path), lang=lang, cls_enabled=cls_enabled
)
finally:
if temp_image_path is not None and temp_image_path.exists():
try:
temp_image_path.unlink()
except Exception:
pass
return self._ocr_input(rendered_page, lang=lang, cls_enabled=cls_enabled)
def parse_pdf(
self,
pdf_path: Union[str, Path],
output_dir: Optional[str] = None,
method: str = "auto",
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
del output_dir, method
pdf_path = Path(pdf_path)
if not pdf_path.exists():
raise FileNotFoundError(f"PDF file does not exist: {pdf_path}")
cls_enabled = kwargs.get("cls", True)
content_list: List[Dict[str, Any]] = []
page_inputs = self._extract_pdf_page_inputs(pdf_path)
try:
for page_idx, rendered_page in page_inputs:
page_lines = self._ocr_rendered_page(
rendered_page, lang=lang, cls_enabled=cls_enabled
)
for text in page_lines:
content_list.append(
{"type": "text", "text": text, "page_idx": int(page_idx)}
)
finally:
# Ensure we promptly release PDF handles even if OCR fails mid-stream.
close = getattr(page_inputs, "close", None)
if callable(close):
close()
return content_list
def parse_image(
self,
image_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
del output_dir
image_path = Path(image_path)
if not image_path.exists():
raise FileNotFoundError(f"Image file does not exist: {image_path}")
ext = image_path.suffix.lower()
if ext not in self.IMAGE_FORMATS:
raise ValueError(
f"Unsupported image format: {ext}. Supported formats: {', '.join(sorted(self.IMAGE_FORMATS))}"
)
cls_enabled = kwargs.get("cls", True)
page_idx = int(kwargs.get("page_idx", 0))
text_lines = self._ocr_input(
str(image_path), lang=lang, cls_enabled=cls_enabled
)
return [
{"type": "text", "text": text, "page_idx": page_idx} for text in text_lines
]
def parse_office_doc(
self,
doc_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
pdf_path = self.convert_office_to_pdf(doc_path, output_dir)
return self.parse_pdf(
pdf_path=pdf_path, output_dir=output_dir, lang=lang, **kwargs
)
def parse_text_file(
self,
text_path: Union[str, Path],
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
pdf_path = self.convert_text_to_pdf(text_path, output_dir)
return self.parse_pdf(
pdf_path=pdf_path, output_dir=output_dir, lang=lang, **kwargs
)
def parse_document(
self,
file_path: Union[str, Path],
method: str = "auto",
output_dir: Optional[str] = None,
lang: Optional[str] = None,
**kwargs,
) -> List[Dict[str, Any]]:
del method
file_path = Path(file_path)
if not file_path.exists():
raise FileNotFoundError(f"File does not exist: {file_path}")
ext = file_path.suffix.lower()
if ext == ".pdf":
return self.parse_pdf(file_path, output_dir, lang=lang, **kwargs)
if ext in self.IMAGE_FORMATS:
return self.parse_image(file_path, output_dir, lang=lang, **kwargs)
if ext in self.OFFICE_FORMATS:
return self.parse_office_doc(file_path, output_dir, lang=lang, **kwargs)
if ext in self.TEXT_FORMATS:
return self.parse_text_file(file_path, output_dir, lang=lang, **kwargs)
raise ValueError(
f"Unsupported file format: {ext}. "
"PaddleOCR parser supports PDF, image, office, and text formats."
)
def check_installation(self) -> bool:
try:
self._require_paddleocr()
return True
except ImportError:
return False
def _normalize_parser_name(name: str) -> str:
"""Normalize and validate a parser name for registry APIs."""
if not isinstance(name, str):
raise TypeError(
f"parser name must be a non-empty string, got {type(name).__name__}"
)
normalized = name.strip().lower()
if not normalized:
raise ValueError("parser name must be a non-empty string")
return normalized
# Custom parser registry for Bring-Your-Own-Parser support (see #151)
_CUSTOM_PARSERS: Dict[str, type] = {}
def register_parser(name: str, parser_class: type) -> None:
"""Register a custom parser class for use with RAGAnything.
This enables the Bring-Your-Own-Parser pattern: users can integrate
any document parser (e.g., Marker, Unstructured, Surya) by subclassing
``Parser`` and registering it here.
Args:
name: Unique identifier for the parser (e.g., "marker", "surya").
Must not collide with built-in names ("mineru", "docling", "paddleocr").
parser_class: A subclass of ``Parser`` that implements at least
``parse_document``, ``check_installation``, and
optionally ``parse_pdf``, ``parse_image``, ``parse_office_doc``.
Raises:
TypeError: If *parser_class* is not a subclass of ``Parser``.
ValueError: If *name* collides with a built-in parser name.
Example::
from raganything.parser import Parser, register_parser
class MarkerParser(Parser):
def check_installation(self) -> bool:
try:
import marker
return True
except ImportError:
return False
def parse_pdf(self, pdf_path, output_dir="./output", method="auto", **kw):
import marker
# ... your implementation ...
return content_list
def parse_document(self, file_path, output_dir="./output", method="auto", **kw):
return self.parse_pdf(pdf_path=file_path, output_dir=output_dir, method=method, **kw)
register_parser("marker", MarkerParser)
"""
normalized_name = _normalize_parser_name(name)
if not isinstance(parser_class, type) or not issubclass(parser_class, Parser):
raise TypeError(
f"parser_class must be a subclass of Parser, got {parser_class!r}"
)
_BUILTIN_NAMES = {"mineru", "docling", "paddleocr"}
if normalized_name in _BUILTIN_NAMES:
raise ValueError(
f"Cannot override built-in parser '{normalized_name}'. "
f"Choose a different name for your custom parser."
)
_CUSTOM_PARSERS[normalized_name] = parser_class
Parser.logger.info(
"Registered custom parser: '%s' -> %s", normalized_name, parser_class.__name__
)
def unregister_parser(name: str) -> None:
"""Remove a previously registered custom parser.
Args:
name: The parser name to remove.
Raises:
TypeError: If *name* is not a string.
ValueError: If *name* is empty or only whitespace.
KeyError: If no custom parser with that name is registered.
"""
normalized_name = _normalize_parser_name(name)
if normalized_name not in _CUSTOM_PARSERS:
raise KeyError(f"No custom parser registered with name '{normalized_name}'")
del _CUSTOM_PARSERS[normalized_name]
Parser.logger.info("Unregistered custom parser: '%s'", normalized_name)
def list_parsers() -> Dict[str, str]:
"""Return a mapping of all available parser names to their class names.
Returns:
Dict mapping parser name to the fully-qualified class name.
Includes both built-in and custom parsers.
"""
result: Dict[str, str] = {
"mineru": "MineruParser",
"docling": "DoclingParser",
"paddleocr": "PaddleOCRParser",
}
for name, cls in _CUSTOM_PARSERS.items():
result[name] = cls.__name__
return result
SUPPORTED_PARSERS = ("mineru", "docling", "paddleocr")
def get_supported_parsers() -> tuple:
"""Return all supported parser names including custom registered parsers."""
return SUPPORTED_PARSERS + tuple(_CUSTOM_PARSERS.keys())
def get_parser(parser_type: str) -> Parser:
"""Get a parser instance by name.
Checks built-in parsers first, then falls back to the custom parser
registry populated via :func:`register_parser`.
Args:
parser_type: Parser name (e.g., "mineru", "docling", "paddleocr",
or any custom registered name).
Returns:
An instance of the requested parser.
Raises:
ValueError: If the parser name is not recognized.
"""
parser_name = (parser_type or "mineru").strip().lower()
if parser_name == "mineru":
return MineruParser()
if parser_name == "docling":
return DoclingParser()
if parser_name == "paddleocr":
return PaddleOCRParser()
# Check custom parser registry
if parser_name in _CUSTOM_PARSERS:
return _CUSTOM_PARSERS[parser_name]()
raise ValueError(
f"Unsupported parser type: {parser_type}. "
f"Supported parsers: {', '.join(get_supported_parsers())}"
)
def main():
"""
Main function to run the document parser from command line
"""
parser = argparse.ArgumentParser(
description="Parse documents using MinerU 2.0, Docling, or PaddleOCR"
)
parser.add_argument("file_path", help="Path to the document to parse")
parser.add_argument("--output", "-o", help="Output directory path")
parser.add_argument(
"--method",
"-m",
choices=["auto", "txt", "ocr"],
default="auto",
help="Parsing method (auto, txt, ocr)",
)
parser.add_argument(
"--lang",
"-l",
help="Document language for OCR optimization (e.g., ch, en, ja)",
)
parser.add_argument(
"--backend",
"-b",
choices=[
"pipeline",
"hybrid-auto-engine",
"hybrid-http-client",
"vlm-auto-engine",
"vlm-http-client",
],
default="pipeline",
help="Parsing backend",
)
parser.add_argument(
"--device",
"-d",
help="Inference device (e.g., cpu, cuda, cuda:0, npu, mps)",
)
parser.add_argument(
"--source",
choices=["huggingface", "modelscope", "local"],
default="huggingface",
help="Model source",
)
parser.add_argument(
"--no-formula",
action="store_true",
help="Disable formula parsing",
)
parser.add_argument(
"--no-table",
action="store_true",
help="Disable table parsing",
)
parser.add_argument(
"--stats", action="store_true", help="Display content statistics"
)
parser.add_argument(
"--check",
action="store_true",
help="Check parser installation",
)
parser.add_argument(
"--parser",
default="mineru",
help=(
"Parser selection. Built-ins: mineru, docling, paddleocr. "
"Custom parsers registered via register_parser() in the same "
"Python process are also accepted when you integrate RAGAnything "
"as a library. The standalone CLI itself only sees parsers that "
"have already been registered in this process."
),
)
parser.add_argument(
"--vlm_url",
help="When the backend is `vlm-http-client`, you need to specify the server_url, for example:`http://127.0.0.1:30000`",
)
args = parser.parse_args()
# Check installation if requested
if args.check:
doc_parser = get_parser(args.parser)
if doc_parser.check_installation():
print(f"✅ {args.parser.title()} is properly installed")
return 0
else:
print(f"❌ {args.parser.title()} installation check failed")
return 1
try:
# Parse the document
doc_parser = get_parser(args.parser)
content_list = doc_parser.parse_document(
file_path=args.file_path,
method=args.method,
output_dir=args.output,
lang=args.lang,
backend=args.backend,
device=args.device,
source=args.source,
formula=not args.no_formula,
table=not args.no_table,
vlm_url=args.vlm_url,
)
print(f"✅ Successfully parsed: {args.file_path}")
print(f"📊 Extracted {len(content_list)} content blocks")
# Display statistics if requested
if args.stats:
print("\n📈 Document Statistics:")
print(f"Total content blocks: {len(content_list)}")
# Count different types of content
content_types = {}
for item in content_list:
if isinstance(item, dict):
content_type = item.get("type", "unknown")
content_types[content_type] = content_types.get(content_type, 0) + 1
if content_types:
print("\n📋 Content Type Distribution:")
for content_type, count in sorted(content_types.items()):
print(f" • {content_type}: {count}")
except Exception as e:
print(f"❌ Error: {str(e)}")
return 1
return 0
if __name__ == "__main__":
exit(main())