# pyright: ignore import argparse import asyncio import tempfile from concurrent.futures import ThreadPoolExecutor from pathlib import Path import sys from typing import Optional, Tuple, List import fitz # PyMuPDF from loguru import logger from docling.document_converter import DocumentConverter, PdfFormatOption from docling.datamodel.base_models import InputFormat from docling.datamodel.pipeline_options import ( PdfPipelineOptions, VlmPipelineOptions ) from docling.datamodel.vlm_model_specs import GRANITEDOCLING_TRANSFORMERS from docling.pipeline.vlm_pipeline import VlmPipeline class TruePage2PageConverter: def __init__(self, output_dir: str = "output", max_concurrent: int = 2, debug_dir: str = "debug_pages"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) self.debug_dir = Path(debug_dir) self.debug_dir.mkdir(exist_ok=True) self.max_concurrent = max_concurrent self.semaphore = asyncio.Semaphore(max_concurrent) # Phase 1: Regular Docling pipeline - using EXACT config from working script regular_pipeline_options = PdfPipelineOptions() regular_pipeline_options.do_table_structure = True regular_pipeline_options.table_structure_options.do_cell_matching = True regular_pipeline_options.do_ocr = True regular_pipeline_options.images_scale = 2.0 regular_format_options = PdfFormatOption( pipeline_options=regular_pipeline_options ) self.regular_converter = DocumentConverter( format_options={ InputFormat.PDF: regular_format_options } ) # Phase 2: VLM pipeline with granite_docling model vlm_pipeline_options = VlmPipelineOptions( vlm_options=GRANITEDOCLING_TRANSFORMERS ) vlm_format_options = PdfFormatOption( pipeline_cls=VlmPipeline, pipeline_options=vlm_pipeline_options ) self.vlm_converter = DocumentConverter( format_options={ InputFormat.PDF: vlm_format_options } ) logger.info("Initialized page-by-page PDF converter") logger.debug(f"Output: {self.output_dir.absolute()}, Debug: {self.debug_dir.absolute()}, Concurrency: {max_concurrent}") def extract_pages_to_temp(self, pdf_path: Path, temp_dir: Path) -> List[Path]: """Extract each page of the PDF to individual PDF files in temp directory""" logger.debug(f"Extracting {pdf_path.name} pages") doc = fitz.open(str(pdf_path)) total_pages = len(doc) page_files = [] for page_num in range(total_pages): # Create a new document for this single page new_doc = fitz.open() # Create empty document new_doc.insert_pdf(doc, from_page=page_num, to_page=page_num) # Save to temp directory page_filename = f"page_{page_num + 1:03d}.pdf" page_path = temp_dir / page_filename new_doc.save(str(page_path)) # ALSO save to debug directory for examination (only in verbose mode) debug_filename = f"{pdf_path.stem}_page_{page_num + 1:03d}.pdf" debug_path = self.debug_dir / debug_filename new_doc.save(str(debug_path)) logger.debug(f"Debug: {debug_path}") new_doc.close() page_files.append(page_path) doc.close() logger.debug(f"Extracted {total_pages} pages, debug artifacts in {self.debug_dir}") return page_files def analyze_page_for_tables(self, result, page_num: int) -> Tuple[bool, int, List[str]]: """Analyze a page using DoclingDocument structure to detect tables - using the user's proven approach""" table_count = 0 table_types = [] # Get document from result document = result.document # Count tables on this page using the user's proven approach page_tables = [ table for table in document.tables if table.prov and len(table.prov) > 0 and table.prov[0].page_no == page_num ] table_count = len(page_tables) table_types = [type(table).__name__ for table in page_tables] # Additional debugging for single page documents (only in DEBUG mode) if len(document.pages) == 1: logger.debug(f"Document has {len(document.tables)} total tables") logger.debug("Single page document - checking if page numbering starts at 1") # For single page documents, try checking page 1 if we were looking for page_num != 1 if len(document.pages) == 1 and page_num != 1: page_1_tables = [ table for table in document.tables if table.prov and len(table.prov) > 0 and table.prov[0].page_no == 1 ] if page_1_tables: logger.debug(f"Found {len(page_1_tables)} tables on page 1 instead") table_count = len(page_1_tables) table_types = [type(table).__name__ for table in page_1_tables] has_tables = table_count > 0 return has_tables, table_count, table_types def generate_page_markdown(self, result, page_num: int) -> str: """Generate markdown from a single page result""" try: if len(result.document.pages) > 0: # Should only have one page since we're processing individual page files page_content = result.document.export_to_markdown() header = f"# Page {page_num} " return header + page_content.strip() else: return f"# Page {page_num} No content found for this page." except Exception as e: logger.warning(f"Error generating markdown for page {page_num}: {str(e)}") return f"# Page {page_num} Error processing this page: {str(e)}" async def process_single_page_file(self, page_file: Path, page_num: int, total_pages: int, original_pdf_name: str) -> Optional[str]: """Process a single extracted page file through both phases""" async with self.semaphore: try: loop = asyncio.get_event_loop() # Phase 1: Regular Docling processing with ThreadPoolExecutor() as executor: regular_result = await loop.run_in_executor( executor, self._process_with_regular_pipeline, page_file ) if regular_result is None: logger.debug(f"Phase 1 failed for page {page_num}") return None # Check if pages exist if not regular_result.document.pages: logger.debug(f"No pages found in regular result for page {page_num}") return None # Analyze for tables using the user's proven document-level approach has_tables, table_count, table_types = self.analyze_page_for_tables(regular_result, page_num) # Log once after Phase 1 completes if has_tables: logger.info(f"Page {page_num}/{total_pages}: Processing") else: logger.info(f"Page {page_num}/{total_pages}: Processing (no tables)") if has_tables: # Phase 2: VLM processing with ThreadPoolExecutor() as executor: vlm_result = await loop.run_in_executor( executor, self._process_with_vlm_pipeline, page_file ) if vlm_result is not None: final_result = vlm_result else: logger.debug(f"VLM processing failed for page {page_num}, using regular result") final_result = regular_result else: final_result = regular_result # Generate markdown immediately markdown_content = self.generate_page_markdown(final_result, page_num) # Save to output file base_name = Path(original_pdf_name).stem output_filename = f"{base_name}_page_{page_num:03d}.md" output_path = self.output_dir / output_filename with open(output_path, "w", encoding="utf-8") as f: f.write(markdown_content) # Don't log completion - it's inferred return str(output_path) except Exception as e: import traceback logger.error(f"Page {page_num}/{total_pages}: Failed - {str(e)}") logger.debug(f"Full traceback: {traceback.format_exc()}") return None def _process_with_regular_pipeline(self, page_file: Path): """Process single page file with regular pipeline""" try: result = self.regular_converter.convert(str(page_file)) return result except Exception as e: logger.error(f"Regular pipeline error: {str(e)}") return None def _process_with_vlm_pipeline(self, page_file: Path): """Process single page file with VLM pipeline""" try: result = self.vlm_converter.convert(str(page_file)) return result except Exception as e: logger.error(f"VLM pipeline error: {str(e)}") return None async def convert_document(self, pdf_path: Path) -> dict: """Convert PDF using true page-by-page processing with temporary files""" logger.info(f"Starting conversion of {pdf_path.name}") if not pdf_path.exists(): logger.error(f"File not found: {pdf_path}") return {"success": 0, "failed": 0, "files": []} # Use temporary directory for clean processing with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) try: # Extract pages to temporary files page_files = self.extract_pages_to_temp(pdf_path, temp_path) total_pages = len(page_files) logger.info(f"Processing {total_pages} pages with {self.max_concurrent} concurrent workers") # Create tasks for processing each page file tasks = [] for i, page_file in enumerate(page_files): page_num = i + 1 task = asyncio.create_task( self.process_single_page_file(page_file, page_num, total_pages, pdf_path.name) ) tasks.append(task) # Process all pages concurrently results = await asyncio.gather(*tasks, return_exceptions=True) # Count successes and failures successful_files = [r for r in results if r is not None and not isinstance(r, Exception)] failed_count = total_pages - len(successful_files) logger.info(f"Conversion complete: {len(successful_files)} successful, {failed_count} failed") return { "success": len(successful_files), "failed": failed_count, "files": successful_files } except Exception as e: logger.error(f"Critical error during document conversion: {str(e)}") return {"success": 0, "failed": 0, "files": []} # Temporary directory is automatically cleaned up here async def main(): parser = argparse.ArgumentParser( description="True page-by-page PDF converter with intelligent table detection and VLM processing" ) parser.add_argument("pdf_path", help="Path to the PDF file to convert") parser.add_argument( "--output-dir", "-o", default="output", help="Output directory for markdown files (default: output)" ) parser.add_argument( "--concurrent", "-c", type=int, default=2, help="Number of pages to process concurrently (default: 2)" ) parser.add_argument( "--debug-dir", "-d", default="debug_pages", help="Directory to save extracted page PDFs for debugging (default: debug_pages)" ) parser.add_argument( "--verbose", "-v", action="store_true", help="Enable verbose logging" ) args = parser.parse_args() # Configure logging logger.remove() # Remove default handler if args.verbose: logger.add(sys.stderr, level="DEBUG", format="{time:YYYY-MM-DD HH:mm:ss} | {level} | {message}") else: logger.add(sys.stderr, level="INFO", format="{time:HH:mm:ss} | {level} | {message}") # Suppress verbose logging from Docling and other libraries import logging if not args.verbose: # Set Docling and related libraries to WARNING level to suppress verbose INFO messages logging.getLogger("docling").setLevel(logging.WARNING) logging.getLogger("docling.document_converter").setLevel(logging.WARNING) logging.getLogger("docling.pipeline").setLevel(logging.WARNING) logging.getLogger("docling.datamodel").setLevel(logging.WARNING) logging.getLogger().setLevel(logging.WARNING) # Root logger pdf_path = Path(args.pdf_path) # Create converter and process document converter = TruePage2PageConverter( output_dir=args.output_dir, max_concurrent=args.concurrent, debug_dir=args.debug_dir ) results = await converter.convert_document(pdf_path) # Print final summary if results["success"] > 0: logger.success(f"Successfully converted {results['success']} pages") if args.verbose: for file_path in results["files"]: logger.info(f" {file_path}") if results["failed"] > 0: logger.warning(f"{results['failed']} pages failed to convert") return 0 if results["success"] > 0 else 1 if __name__ == "__main__": sys.exit(asyncio.run(main()))