cc
329 lines
12 KiB
Python
329 lines
12 KiB
Python
import logging
|
|
import re
|
|
import time
|
|
from typing import Dict, Optional
|
|
|
|
import markdownify
|
|
import requests
|
|
|
|
from docreader.config import CONFIG
|
|
from docreader.models.document import Document
|
|
from docreader.parser.base_parser import BaseParser
|
|
from docreader.parser.chain_parser import PipelineParser
|
|
from docreader.parser.markdown_parser import MarkdownImageUtil, MarkdownTableFormatter
|
|
from docreader.utils import endecode
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StdMinerUParser(BaseParser):
|
|
"""
|
|
Standard MinerU Parser for document parsing.
|
|
|
|
This parser uses MinerU API to parse documents (especially PDFs) into markdown format,
|
|
with support for tables, formulas, and images extraction.
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
enable_markdownify: bool = True,
|
|
mineru_endpoint: Optional[str] = None, # Added: 支持传入自定义 endpoint
|
|
**kwargs,
|
|
):
|
|
"""
|
|
Initialize MinerU parser.
|
|
|
|
Args:
|
|
enable_markdownify: Whether to convert HTML tables to markdown format
|
|
mineru_endpoint: MinerU API endpoint URL
|
|
**kwargs: Additional arguments passed to BaseParser
|
|
"""
|
|
super().__init__(**kwargs)
|
|
# Get MinerU endpoint from environment variable or parameter
|
|
# Modified: 优先使用传入的参数，否则使用 Config
|
|
base_url = mineru_endpoint if mineru_endpoint else CONFIG.mineru_endpoint
|
|
self.minerU = base_url.rstrip("/") if base_url else ""
|
|
|
|
self.enable_markdownify = enable_markdownify
|
|
# Helper for processing markdown images
|
|
self.image_helper = MarkdownImageUtil()
|
|
# Pattern to match base64 encoded images
|
|
self.base64_pattern = re.compile(r"data:image/(\w+);base64,(.*)")
|
|
# Check if MinerU API is available
|
|
self.enable = self.ping()
|
|
|
|
def ping(self, timeout: int = 5) -> bool:
|
|
"""
|
|
Check if MinerU API is available.
|
|
|
|
Args:
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
True if API is available, False otherwise
|
|
"""
|
|
try:
|
|
response = requests.get(
|
|
self.minerU + "/docs", timeout=timeout, allow_redirects=True
|
|
)
|
|
response.raise_for_status()
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def parse_into_text(self, content: bytes) -> Document:
|
|
"""
|
|
Parse document content into text using MinerU API.
|
|
|
|
Args:
|
|
content: Raw document content in bytes
|
|
|
|
Returns:
|
|
Document object containing parsed text and images
|
|
"""
|
|
if not self.enable:
|
|
logger.debug("MinerU API is not enabled")
|
|
return Document()
|
|
|
|
logger.info(f"Parsing scanned PDF via MinerU API (size: {len(content)} bytes)")
|
|
md_content: str = ""
|
|
images_b64: Dict[str, str] = {}
|
|
try:
|
|
# Call MinerU API to parse document
|
|
response = requests.post(
|
|
url=self.minerU + "/file_parse",
|
|
data={
|
|
"return_md": True, # Return markdown content
|
|
"return_images": True, # Return extracted images
|
|
"lang_list": ["ch", "en"], # Support Chinese and English
|
|
"table_enable": True, # Enable table parsing
|
|
"formula_enable": True, # Enable formula parsing
|
|
"parse_method": "auto", # Auto detect parsing method
|
|
"start_page_id": 0, # Start from first page
|
|
"end_page_id": 99999, # Parse all pages
|
|
"backend": "pipeline", # Use pipeline backend
|
|
"response_format_zip": False, # Return JSON instead of ZIP
|
|
"return_middle_json": False, # Don't return intermediate JSON
|
|
"return_model_output": False, # Don't return model output
|
|
"return_content_list": False, # Don't return content list
|
|
},
|
|
files={"files": content},
|
|
timeout=1000,
|
|
)
|
|
response.raise_for_status()
|
|
result = response.json()["results"]["files"]
|
|
md_content = result["md_content"]
|
|
images_b64 = result.get("images", {})
|
|
except Exception as e:
|
|
logger.error(f"MinerU parsing failed: {e}", exc_info=True)
|
|
return Document()
|
|
|
|
# Convert HTML tables in markdown to markdown table format
|
|
if self.enable_markdownify:
|
|
logger.debug("Converting HTML to Markdown")
|
|
md_content = markdownify.markdownify(md_content)
|
|
|
|
images = {}
|
|
image_replace = {}
|
|
# Filter images that are actually used in markdown content
|
|
# Some images in images_b64 may not be referenced in md_content
|
|
# (e.g., images embedded in tables), so we need to filter them
|
|
for ipath, b64_str in images_b64.items():
|
|
# Skip images that are not referenced in markdown content
|
|
if f"images/{ipath}" not in md_content:
|
|
logger.debug(f"Image {ipath} not used in markdown")
|
|
continue
|
|
# Parse base64 image data
|
|
match = self.base64_pattern.match(b64_str)
|
|
if match:
|
|
# Extract image format (e.g., png, jpg)
|
|
file_ext = match.group(1)
|
|
# Extract base64 encoded data
|
|
b64_str = match.group(2)
|
|
|
|
# Decode base64 string to bytes
|
|
image_bytes = endecode.encode_image(b64_str, errors="ignore")
|
|
if not image_bytes:
|
|
logger.error("Failed to decode base64 image skip it")
|
|
continue
|
|
|
|
# Upload image to storage and get URL
|
|
image_url = self.storage.upload_bytes(
|
|
image_bytes, file_ext=f".{file_ext}"
|
|
)
|
|
|
|
# Store image mapping for later use
|
|
images[image_url] = b64_str
|
|
# Prepare replacement mapping for markdown content
|
|
image_replace[f"images/{ipath}"] = image_url
|
|
|
|
logger.info(f"Replaced {len(image_replace)} images in markdown")
|
|
# Replace image paths in markdown with uploaded URLs
|
|
text = self.image_helper.replace_path(md_content, image_replace)
|
|
|
|
logger.info(
|
|
f"Successfully parsed PDF, text: {len(text)}, images: {len(images)}"
|
|
)
|
|
return Document(content=text, images=images)
|
|
|
|
|
|
# Added: 新增 MinerUCloudParser 类，支持异步任务提交
|
|
class MinerUCloudParser(StdMinerUParser):
|
|
"""
|
|
MinerU Parser for REMOTE/CLOUD API (Asynchronous).
|
|
Uses the /submit -> /status -> /result workflow.
|
|
"""
|
|
|
|
SUBMIT_TIMEOUT = 30
|
|
POLL_INTERVAL = 2
|
|
MAX_WAIT_TIME = 600
|
|
|
|
def parse_into_text(self, content: bytes) -> Document:
|
|
"""
|
|
Parse document content using Cloud MinerU API (Async/Polling).
|
|
"""
|
|
if not self.enable:
|
|
return Document()
|
|
|
|
logger.info(f"Parsing PDF via Cloud MinerU API (size: {len(content)} bytes)")
|
|
|
|
try:
|
|
# --- Step 1: Submit Task ---
|
|
submit_url = f"{self.minerU}/submit"
|
|
logger.info(f"Submitting task to {submit_url}")
|
|
|
|
response = requests.post(
|
|
url=submit_url,
|
|
files={"files": content},
|
|
data={
|
|
"enable_formula": "true",
|
|
"enable_table": "true",
|
|
"layout_model": "doclayout_yolo",
|
|
"backend": "pipeline",
|
|
},
|
|
timeout=self.SUBMIT_TIMEOUT,
|
|
)
|
|
response.raise_for_status()
|
|
|
|
# Robust task_id extraction
|
|
resp_data = response.json()
|
|
task_id = resp_data.get("task_id") or resp_data.get("data", {}).get("task_id")
|
|
|
|
if not task_id:
|
|
raise ValueError(f"No task_id in response: {resp_data}")
|
|
|
|
logger.info(f"Task submitted, ID: {task_id}, waiting for completion...")
|
|
|
|
# --- Step 2: Poll Status ---
|
|
start_time = time.time()
|
|
|
|
while True:
|
|
if time.time() - start_time > self.MAX_WAIT_TIME:
|
|
raise TimeoutError(f"Task {task_id} timed out after {self.MAX_WAIT_TIME}s")
|
|
|
|
try:
|
|
status_resp = requests.get(
|
|
f"{self.minerU}/status/{task_id}",
|
|
timeout=10
|
|
)
|
|
status_resp.raise_for_status()
|
|
status_data = status_resp.json()
|
|
except requests.RequestException as e:
|
|
logger.warning(f"Status check failed for {task_id}: {e}. Retrying...")
|
|
time.sleep(self.POLL_INTERVAL)
|
|
continue
|
|
|
|
state = status_data.get("status") or status_data.get("state")
|
|
|
|
if state in ["done", "success"]:
|
|
break
|
|
elif state == "failed":
|
|
error_msg = status_data.get("error") or "Unknown error"
|
|
raise RuntimeError(f"Task {task_id} failed: {error_msg}")
|
|
else:
|
|
time.sleep(self.POLL_INTERVAL)
|
|
|
|
# --- Step 3: Get Result ---
|
|
result_resp = requests.get(
|
|
f"{self.minerU}/result/{task_id}",
|
|
timeout=30
|
|
)
|
|
result_resp.raise_for_status()
|
|
result_json = result_resp.json()
|
|
|
|
# Normalize result data
|
|
result_data = result_json.get("result", result_json)
|
|
|
|
md_content = result_data.get("md_content", "")
|
|
images_b64 = result_data.get("images", {})
|
|
|
|
# 使用父类的方法处理图片和Markdown转换 (复用现有逻辑)
|
|
|
|
# Convert HTML tables
|
|
if self.enable_markdownify:
|
|
md_content = markdownify.markdownify(md_content)
|
|
|
|
images = {}
|
|
image_replace = {}
|
|
|
|
for ipath, b64_str in images_b64.items():
|
|
if f"images/{ipath}" not in md_content:
|
|
continue
|
|
match = self.base64_pattern.match(b64_str)
|
|
if match:
|
|
file_ext = match.group(1)
|
|
b64_str_clean = match.group(2)
|
|
image_bytes = endecode.encode_image(b64_str_clean, errors="ignore")
|
|
if not image_bytes: continue
|
|
|
|
if self.storage:
|
|
image_url = self.storage.upload_bytes(image_bytes, file_ext=f".{file_ext}")
|
|
images[image_url] = b64_str_clean
|
|
image_replace[f"images/{ipath}"] = image_url
|
|
|
|
if image_replace:
|
|
md_content = self.image_helper.replace_path(md_content, image_replace)
|
|
|
|
return Document(content=md_content, images=images)
|
|
|
|
except Exception as e:
|
|
logger.error(f"Cloud MinerU parsing failed: {e}", exc_info=True)
|
|
return Document()
|
|
|
|
|
|
class MinerUParser(PipelineParser):
|
|
"""
|
|
MinerU Parser with pipeline processing.
|
|
|
|
This parser combines StdMinerUParser for document parsing and
|
|
MarkdownTableFormatter for table formatting in a pipeline.
|
|
"""
|
|
|
|
_parser_cls = (StdMinerUParser, MarkdownTableFormatter)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
import os
|
|
|
|
# Example usage for testing
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
|
|
# Configure your file path and MinerU endpoint
|
|
your_file = "/path/to/your/file.pdf"
|
|
|
|
# Added: 修改为 Localhost 方便测试
|
|
test_endpoint = "http://localhost:9987"
|
|
os.environ["MINERU_ENDPOINT"] = test_endpoint
|
|
|
|
# Create parser instance
|
|
# Modified: 传入 endpoint
|
|
parser = MinerUParser(mineru_endpoint=test_endpoint)
|
|
|
|
# Parse PDF file
|
|
if os.path.exists(your_file):
|
|
with open(your_file, "rb") as f:
|
|
content = f.read()
|
|
document = parser.parse_into_text(content)
|
|
logger.error(document.content)
|
|
else:
|
|
print(f"File not found: {your_file}") |