import base64
import math
import re
import sys
from io import BytesIO
from pathlib import Path
from typing import BinaryIO, Any
sys.path.insert(0, str(Path(__file__).parent))
from _vlm_base import VLMBase, sanitize_alt_text
from _slides import render_pdf_as_slides
import pypdfium2 as pdfium
import requests
import openai
from markitdown import MarkItDown
from markitdown._base_converter import DocumentConverterResult
from markitdown._stream_info import StreamInfo
MINERU_URL = "http://localhost:8000"
_SLIDE_PROMPT = "Describe the content of this slide accurately (all text must be verbatim) and concisely."
_FLOWCHART_PROMPT = (
"Describe this flowchart/block diagram accurately and concisely: list the labeled "
"blocks and the connections between them (all text must be verbatim)."
)
_DETAILS_BLOCK_RE = re.compile(
r'!\[[^\]]*\]\((images/[^)]+)\)\n+'
r'\n([\w -]+)
\n+'
r'(.*?)\n ',
re.DOTALL
)
_KNOWN_TAGS = re.compile(
r'^]*)?>$',
re.IGNORECASE
)
def _sanitize_html_tags(text: str) -> str:
"""Escape < > that are not part of known HTML tags (e.g. , )."""
def replace(m: re.Match) -> str:
tag = m.group(0)
if _KNOWN_TAGS.match(tag):
return tag
return tag.replace("<", "<").replace(">", ">")
return re.sub(r"<[^>]+>", replace, text)
def _is_landscape(pdf_bytes: bytes) -> bool:
"""Return True only for non-document landscape pages (slides: 16:9, 4:3, etc.).
A4/Letter landscape (ratio ≈ √2 ≈ 1.414) is treated as a normal document."""
try:
doc = pdfium.PdfDocument(pdf_bytes)
if not doc:
return False
page = doc[0]
w, h = page.get_width(), page.get_height()
if page.get_rotation() in (90, 270):
w, h = h, w
if w <= h:
return False
ratio = w / h
return abs(ratio - math.sqrt(2)) > 0.05
except Exception:
return False
class PDFMinerUConverter(VLMBase):
"""Converts PDF via MinerU (portrait) or page-by-page slides (landscape)."""
def accepts(self, _file_stream: BinaryIO, stream_info: StreamInfo, **_kwargs: Any) -> bool:
ext = (stream_info.extension or "").lower()
mime = (stream_info.mimetype or "").lower()
return ext == ".pdf" or "application/pdf" in mime
def convert(self, file_stream: BinaryIO, stream_info: StreamInfo, **kwargs: Any) -> DocumentConverterResult:
llm_client = kwargs.get("llm_client") or self._llm_client
llm_model = kwargs.get("llm_model") or self._llm_model
source_name = Path(stream_info.filename or "document").stem.replace(" ", "-")
local_path = stream_info.local_path or stream_info.filename
output_dir = Path(local_path).parent if local_path else Path.cwd()
pdf_bytes = file_stream.read()
if _is_landscape(pdf_bytes):
describe = (lambda p: self._describe(p, llm_client, llm_model, _SLIDE_PROMPT)) if llm_client and llm_model else None
markdown = render_pdf_as_slides(pdf_bytes, output_dir, source_name, describe)
return DocumentConverterResult(markdown=markdown)
return self._convert_via_mineru(pdf_bytes, stream_info, output_dir, source_name, llm_client, llm_model,
kwargs.get("mineru_url", MINERU_URL))
def _convert_via_mineru(self, pdf_bytes, stream_info, output_dir, source_name,
llm_client, llm_model, mineru_url) -> DocumentConverterResult:
filename = Path(stream_info.filename or "document.pdf").name
resp = requests.post(
f"{mineru_url}/file_parse",
files=[("files", (filename, BytesIO(pdf_bytes), "application/pdf"))],
data={"return_images": "true", "return_md": "true"},
)
resp.raise_for_status()
data = resp.json()
results = data.get("results", data)
result = next(iter(results.values())) if isinstance(results, dict) else results[0]
markdown: str = result.get("md_content", result.get("md", ""))
images: dict[str, str] = result.get("images", {})
images_dir = output_dir / f".{source_name}"
if images:
images_dir.mkdir(exist_ok=True)
path_remap: dict[str, Path] = {}
for img_name, data_val in images.items():
if isinstance(data_val, str) and "," in data_val:
data_val = data_val.split(",", 1)[1]
dest = images_dir / img_name
dest.write_bytes(base64.b64decode(data_val))
path_remap[f"images/{img_name}"] = dest
def replace_details(m: re.Match) -> str:
path, block_type, content = m.group(1), m.group(2), m.group(3)
dest = path_remap.get(path)
if dest is None:
return m.group(0)
if block_type == "flowchart":
if not (llm_client and llm_model):
return m.group(0)
alt = self._describe(dest, llm_client, llm_model, _FLOWCHART_PROMPT) or "Flowchart"
else:
alt = sanitize_alt_text(content.strip()) or block_type
return f"})"
markdown = _DETAILS_BLOCK_RE.sub(replace_details, markdown)
def replace_image(m: re.Match) -> str:
alt, path = m.group(1), m.group(2)
dest = path_remap.get(path)
if dest is None:
return m.group(0)
if not alt and llm_client and llm_model:
alt = self._describe(dest, llm_client, llm_model) or ""
return f"})"
markdown = re.sub(r"!\[([^\]]*)\]\(([^)]+)\)", replace_image, markdown)
markdown = _sanitize_html_tags(markdown)
return DocumentConverterResult(markdown=markdown)
def register_converters(markitdown_instance, **_kwargs):
markitdown_instance.register_converter(PDFMinerUConverter())
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("pdf", help="Path to the .pdf file")
parser.add_argument("--output", "-o", help="Output markdown file (default: .md)")
args = parser.parse_args()
pdf_path = Path(args.pdf)
out_path = Path(args.output) if args.output else pdf_path.with_stem(pdf_path.stem.replace(" ", "-")).with_suffix(".md")
client = openai.OpenAI(base_url="http://localhost:8500/v1", api_key="none")
md = MarkItDown(llm_client=client, llm_model="Qwen3.6-35B-A3B-UD-IQ2_XXS.gguf")
md.register_converter(PDFMinerUConverter())
result = md.convert(str(pdf_path))
out_path.write_text(result.text_content)
print(f"Written to {out_path}")