83 lines
2.8 KiB
Python
83 lines
2.8 KiB
Python
"""PDF document parser for Mosaic — PyMuPDF text extraction."""
|
|
import logging
|
|
from typing import Optional
|
|
|
|
from .base import ParseResult
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class PDFDocParser:
|
|
"""Parse PDF documents using PyMuPDF (fitz) for text extraction."""
|
|
|
|
def __init__(self, max_pages: int = 200, min_chars_per_page: int = 50):
|
|
self.max_pages = max_pages
|
|
self.min_chars_per_page = min_chars_per_page
|
|
|
|
def parse(self, file_path: str) -> ParseResult:
|
|
"""Parse a PDF file.
|
|
|
|
Extracts text page by page. Large docs (>max_pages) are split.
|
|
After extraction, checks chars/page ratio for OCR detection.
|
|
"""
|
|
try:
|
|
import fitz # PyMuPDF
|
|
except ImportError:
|
|
return ParseResult(error="PyMuPDF (fitz) not installed")
|
|
|
|
try:
|
|
doc = fitz.open(file_path)
|
|
total_pages = len(doc)
|
|
metadata = {
|
|
'page_count': total_pages,
|
|
'format': doc.metadata.get('format', ''),
|
|
'title': doc.metadata.get('title', ''),
|
|
'author': doc.metadata.get('author', ''),
|
|
'subject': doc.metadata.get('subject', ''),
|
|
'creator': doc.metadata.get('creator', ''),
|
|
}
|
|
|
|
title = metadata['title'] or ''
|
|
|
|
# Limit pages
|
|
pages_to_process = min(total_pages, self.max_pages)
|
|
if total_pages > self.max_pages:
|
|
logger.warning("PDF has %d pages, processing first %d",
|
|
total_pages, self.max_pages)
|
|
metadata['truncated'] = True
|
|
metadata['truncated_at_page'] = self.max_pages
|
|
|
|
# Extract text page by page with form feed separators
|
|
text_parts = []
|
|
for page_num in range(pages_to_process):
|
|
page = doc[page_num]
|
|
page_text = page.get_text("text")
|
|
if page_text.strip():
|
|
text_parts.append(page_text.strip())
|
|
|
|
doc.close()
|
|
|
|
text = '\f'.join(text_parts) # Form feed between pages
|
|
char_count = len(text)
|
|
|
|
# Check for OCR need
|
|
needs_ocr = False
|
|
if pages_to_process > 0:
|
|
chars_per_page = char_count / pages_to_process
|
|
if chars_per_page < self.min_chars_per_page:
|
|
needs_ocr = True
|
|
metadata['chars_per_page'] = chars_per_page
|
|
|
|
return ParseResult(
|
|
success=True,
|
|
text=text,
|
|
metadata=metadata,
|
|
doc_type='pdf',
|
|
title=title,
|
|
char_count=char_count,
|
|
needs_ocr=needs_ocr,
|
|
)
|
|
|
|
except Exception as e:
|
|
return ParseResult(error=f"PDF parse error: {str(e)}")
|