Extract Structured Data from Engineering Drawings with DocTR and LlamaIndex

                      
                     
"""
Production implementation for extracting structured data from engineering drawings.
Utilizes DocTR for document processing and LlamaIndex for data indexing.
"""
from typing import Dict, Any, List
import os
import logging
import time
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from doctr.models import Document
from llama_index import DocumentIndex

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

Base = declarative_base()

class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///drawings.db')
    retry_attempts: int = int(os.getenv('RETRY_ATTEMPTS', 3))

class Drawing(Base):
    __tablename__ = 'drawings'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(String)

engine = create_engine(Config.database_url)
Session = sessionmaker(bind=engine)
Base.metadata.create_all(engine)

def validate_input(data: Dict[str, Any]) -> bool:
    """Validate request data.
    
    Args:
        data: Input to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'file_path' not in data:
        raise ValueError('Missing file_path')
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields.
    
    Args:
        data: Input data to sanitize
    Returns:
        Sanitized data
    """
    return {k: v.strip() for k, v in data.items()}

def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform raw records into structured format.
    
    Args:
        records: List of raw records
    Returns:
        Transformed records
    """
    return [{'title': record['title'], 'content': record['content']} for record in records]

def fetch_data(file_path: str) -> Document:
    """Fetch and process document data.
    
    Args:
        file_path: Path to the document
    Returns:
        Document object from DocTR
    Raises:
        FileNotFoundError: If the file does not exist
    """
    if not os.path.exists(file_path):
        raise FileNotFoundError(f'File not found: {file_path}')
    return Document.from_file(file_path)

def save_to_db(session, data: Dict[str, Any]) -> None:
    """Save structured data to the database.
    
    Args:
        session: Database session
        data: Data to save
    """
    drawing = Drawing(**data)
    session.add(drawing)
    session.commit()

def call_api(data: Dict[str, Any]) -> Any:
    """Mock API call for processing.
    
    Args:
        data: Data to process
    Returns:
        Processed result
    """
    logger.info('Calling external API...')
    return {'status': 'success', 'data': data}

def process_batch(file_paths: List[str]) -> None:
    """Process a batch of files.
    
    Args:
        file_paths: List of file paths to process
    """
    for file_path in file_paths:
        try:
            logger.info(f'Processing file: {file_path}')
            data = fetch_data(file_path)
            # Extract and transform data
            structured_data = transform_records(data)
            # Persist data to DB
            with Session() as session:
                save_to_db(session, structured_data)
        except Exception as e:
            logger.error(f'Error processing {file_path}: {e}')

def aggregate_metrics(data: List[Dict[str, Any]]) -> Dict[str, Any]:
    """Aggregate metrics from processed data.
    
    Args:
        data: Processed data
    Returns:
        Aggregated metrics
    """
    return {'total_records': len(data)}

class DrawingExtractor:
    """Main orchestrator for extracting drawings data.
    """
    def __init__(self, file_paths: List[str]) -> None:
        self.file_paths = file_paths

    def run(self) -> None:
        """Run the extraction process.
        """
        for attempt in range(1, Config.retry_attempts + 1):
            try:
                logger.info(f'Starting extraction attempt {attempt}')
                process_batch(self.file_paths)
                logger.info('Extraction completed successfully.')
                break
            except Exception as e:
                logger.warning(f'Attempt {attempt} failed: {e}')
                time.sleep(2 ** attempt)  # Exponential backoff
                if attempt == Config.retry_attempts:
                    logger.error('Max retries reached. Process failed.')

if __name__ == '__main__':
    # Example usage
    file_paths = ['drawing1.pdf', 'drawing2.pdf']
    extractor = DrawingExtractor(file_paths)
    extractor.run()