Extract Technical Drawings from PDF Specs with PyMuPDF and Supervision

                      
                     
"""
Production implementation for extracting technical drawings from PDF specifications.
Provides secure, scalable operations using PyMuPDF and supervision.
"""
from typing import Dict, Any, List
import os
import logging
import fitz  # PyMuPDF
import time
from contextlib import contextmanager

# Logging configuration
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class Config:
    """
    Configuration class for environment variables.
    """
    pdf_directory: str = os.getenv('PDF_DIRECTORY', '/path/to/pdfs')
    output_directory: str = os.getenv('OUTPUT_DIRECTORY', '/path/to/output')

@contextmanager
def pdf_context_manager(file_path: str):
    """
    Context manager for handling PDF file operations.
    
    Args:
        file_path: Path to the PDF file.
    """  
    try:
        pdf_document = fitz.open(file_path)  # Open the PDF file
        yield pdf_document  # Yield control to the caller
    except Exception as e:
        logger.error(f"Failed to open PDF: {e}")
        raise
    finally:
        pdf_document.close()  # Ensure the document is closed

async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for extraction.
    
    Args:
        data: Input data containing 'pdf_file'.
    Returns:
        True if valid.
    Raises:
        ValueError: If validation fails.
    """  
    if 'pdf_file' not in data:
        raise ValueError('Missing pdf_file in input data')
    if not os.path.isfile(data['pdf_file']):
        raise ValueError(f"File not found: {data['pdf_file']}")
    return True

async def extract_drawings_from_pdf(file_path: str) -> List[str]:
    """Extract technical drawings from a PDF file.
    
    Args:
        file_path: Path to the PDF file.
    Returns:
        List of extracted drawing file paths.
    Raises:
        RuntimeError: If extraction fails.
    """  
    extracted_files = []
    with pdf_context_manager(file_path) as pdf:
        for page_number in range(len(pdf)):
            page = pdf.load_page(page_number)
            image_list = page.get_images(full=True)
            for img_index, img in enumerate(image_list):
                xref = img[0]  # Image reference
                base_image = pdf.extract_image(xref)
                image_bytes = base_image["image"]
                image_filename = os.path.join(Config.output_directory, f"drawing_{page_number}_{img_index}.png")
                with open(image_filename, "wb") as img_file:
                    img_file.write(image_bytes)
                extracted_files.append(image_filename)
                logger.info(f"Extracted image saved to: {image_filename}")
    if not extracted_files:
        raise RuntimeError("No drawings were extracted from the PDF.")
    return extracted_files

async def save_to_db(data: List[str]) -> None:
    """Simulate saving extracted data to a database.
    
    Args:
        data: List of extracted drawing file paths.
    """  
    for file_path in data:
        logger.info(f"Saving {file_path} to the database...")
        # Simulate database save with sleep
        time.sleep(1)  # Simulate time delay for DB operation
    logger.info("All drawings saved to the database.")

async def main_extraction_workflow(input_data: Dict[str, Any]) -> None:
    """Main workflow for the extraction process.
    
    Args:
        input_data: Input data with PDF file information.
    """  
    try:
        await validate_input(input_data)  # Validate the input
        extracted_files = await extract_drawings_from_pdf(input_data['pdf_file'])  # Extract drawings
        await save_to_db(extracted_files)  # Save the extracted files to DB
    except ValueError as ve:
        logger.warning(f"Validation error: {ve}")
    except RuntimeError as re:
        logger.error(f"Runtime error during extraction: {re}")
    except Exception as ex:
        logger.exception(f"An unexpected error occurred: {ex}")

if __name__ == '__main__':
    input_data = {"pdf_file": os.path.join(Config.pdf_directory, "specifications.pdf")}
    import asyncio
    asyncio.run(main_extraction_workflow(input_data))