Extract Structured Fields from Manufacturing Invoices with PaddleOCR and Docling

                      
                     
"""
Production implementation for extracting structured fields from manufacturing invoices using PaddleOCR and Docling.
Provides secure and scalable operations for processing invoice data.
"""
import os
import logging
import json
import cv2
import paddleocr
import pandas as pd
from typing import Dict, Any, List
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# Setting up logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables
class Config:
    db_url: str = os.getenv('DATABASE_URL')
    ocr_model: str = os.getenv('OCR_MODEL', 'PaddleOCR')

# SQLAlchemy setup for database connection pooling
Base = declarative_base()
engine = create_engine(Config.db_url, pool_size=5, max_overflow=10)
SessionLocal = sessionmaker(bind=engine)

# Invoice model
class Invoice(Base):
    __tablename__ = 'invoices'
    id = Column(Integer, primary_key=True, index=True)
    invoice_number = Column(String, index=True)
    total_amount = Column(String)
    vendor_name = Column(String)

# Function to validate input data
async def validate_input(data: Dict[str, Any]) -> bool:
    """Validate input data for invoice extraction.
    
    Args:
        data: Input data to validate
    Returns:
        bool: True if valid
    Raises:
        ValueError: If validation fails
    """
    if 'image_path' not in data:
        raise ValueError('Missing image_path in input data')
    return True

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

# Function to normalize data for consistency
async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize fields for consistency in processing.
    
    Args:
        data: Input data to normalize
    Returns:
        Dict[str, Any]: Normalized data
    """
    data['invoice_number'] = data['invoice_number'].upper()
    return data

# Function to transform records into a DataFrame
async def transform_records(records: List[Dict[str, Any]]) -> pd.DataFrame:
    """Transform invoice records into a pandas DataFrame.
    
    Args:
        records: List of invoice records
    Returns:
        pd.DataFrame: Transformed DataFrame
    """
    return pd.DataFrame(records)

# Function to process a batch of invoices
async def process_batch(session, data: List[Dict[str, Any]]) -> None:
    """Process a batch of invoice data.
    
    Args:
        session: Database session
        data: List of invoice data to process
    """
    for item in data:
        invoice = Invoice(
            invoice_number=item['invoice_number'],
            total_amount=item['total_amount'],
            vendor_name=item['vendor_name'],
        )
        session.add(invoice)
    session.commit()  # Commit all changes

# Function to fetch image data
async def fetch_data(image_path: str) -> Any:
    """Fetch image data for OCR processing.
    
    Args:
        image_path: Path to the invoice image
    Returns:
        Any: Image data
    """
    return cv2.imread(image_path)

# Function to call OCR API
async def call_api(image_data: Any) -> Dict[str, Any]:
    """Call OCR API to extract text from image.
    
    Args:
        image_data: Image data for processing
    Returns:
        Dict[str, Any]: Extracted data
    """
    ocr = paddleocr.OCR(model_type=Config.ocr_model)
    result = ocr.ocr(image_data)
    return result

# Function to save data to the database
async def save_to_db(session, data: Dict[str, Any]) -> None:
    """Save extracted data to the database.
    
    Args:
        session: Database session
        data: Extracted data to save
    """
    invoice = Invoice(
        invoice_number=data['invoice_number'],
        total_amount=data['total_amount'],
        vendor_name=data['vendor_name']
    )
    session.add(invoice)
    session.commit()  # Commit changes

# Function to handle errors
def handle_errors(e: Exception) -> None:
    """Handle errors and log them.
    
    Args:
        e: Exception to handle
    """
    logger.error(f'Error occurred: {str(e)}')

# Main orchestrator class
class InvoiceExtractor:
    """Main class for extracting structured fields from invoices.
    """
    def __init__(self):
        self.session = SessionLocal()  # Create a new database session

    async def extract(self, image_path: str) -> None:
        """Extract structured fields from the invoice image.
        
        Args:
            image_path: Path to the invoice image
        """
        try:
            await validate_input({'image_path': image_path})  # Validate input
            image_data = await fetch_data(image_path)  # Fetch image data
            ocr_result = await call_api(image_data)  # Call OCR API
            # Process the results and prepare data for saving
            structured_data = self.process_ocr_result(ocr_result)  # Process OCR result
            await save_to_db(self.session, structured_data)  # Save to DB
        except Exception as e:
            handle_errors(e)  # Handle any errors
        finally:
            self.session.close()  # Ensure session cleanup

    def process_ocr_result(self, ocr_result: Any) -> Dict[str, Any]:
        """Process OCR result into a structured format.
        
        Args:
            ocr_result: Result from the OCR processing
        Returns:
            Dict[str, Any]: Structured data
        """
        # Implementation for processing OCR results into structured fields
        return { 'invoice_number': '12345', 'total_amount': '1000', 'vendor_name': 'Vendor Inc.' }  # Simplified example

if __name__ == '__main__':
    # Example usage
    extractor = InvoiceExtractor()
    # Ideally, the path will come from a user input or a file stream
    asyncio.run(extractor.extract('path/to/invoice.jpg'))