Extract Structured Fields from Equipment Service Reports with DocTR and spaCy

"""
Production implementation for extracting structured fields from equipment service reports using DocTR and spaCy.
Provides secure, scalable operations for processing and extracting relevant information.
"""

from typing import Dict, Any, List
import os
import logging
import spacy
from doctr.io import DocumentFile
from doctr.models import ocr_predictor
from sqlalchemy import create_engine, Column, String, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from backoff import on_exception, expo

# Logger setup
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Database configuration
Base = declarative_base()
DATABASE_URL = os.getenv('DATABASE_URL', 'sqlite:///service_reports.db')
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(bind=engine)

class EquipmentReport(Base):
    """ORM model for equipment service reports."""
    __tablename__ = 'reports'
    id = Column(Integer, primary_key=True, index=True)
    service_id = Column(String, index=True)
    extracted_text = Column(String)
    structured_data = Column(String)

Base.metadata.create_all(bind=engine)

class Config:
    """Configuration class for environment variables."""
    nlp_model: str = os.getenv('NLP_MODEL', 'en_core_web_sm')

# Load spaCy model
nlp = spacy.load(Config.nlp_model)

@on_exception(expo, Exception, max_tries=5)
def fetch_data(report_file: str) -> str:
    """Fetch data from the provided report file.
    
    Args:
        report_file: Path to the service report file
    Returns:
        Extracted text from the report
    Raises:
        FileNotFoundError: If the report file does not exist
    """
    if not os.path.exists(report_file):
        raise FileNotFoundError(f'Report file {report_file} not found.')  
    doc = DocumentFile.from_images(report_file)
    return doc[0].pages[0].content

def sanitize_fields(fields: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize fields to ensure they are clean and usable.
    
    Args:
        fields: Dictionary of fields to sanitize
    Returns:
        Sanitized dictionary of fields
    """
    sanitized = {key: str(value).strip() for key, value in fields.items()}
    return sanitized

def validate_input_data(data: Dict[str, Any]) -> bool:
    """Validate input data for required fields.
    
    Args:
        data: Input data to validate
    Returns:
        True if valid
    Raises:
        ValueError: If validation fails
    """
    required_fields = ['service_id', 'extracted_text']
    for field in required_fields:
        if field not in data:
            raise ValueError(f'Missing required field: {field}')
    return True

def transform_records(text: str) -> Dict[str, Any]:
    """Transform raw text into structured data fields.
    
    Args:
        text: Raw extracted text
    Returns:
        Dictionary of structured data
    """
    doc = nlp(text)
    structured_data = {
        'service_id': '',
        'description': '',
    }
    for ent in doc.ents:
        if ent.label_ == 'SERVICE_ID':
            structured_data['service_id'] = ent.text
        elif ent.label_ == 'DESCRIPTION':
            structured_data['description'] = ent.text
    return structured_data

def save_to_db(data: Dict[str, Any]) -> None:
    """Save structured data to the database.
    
    Args:
        data: Structured data to save
    Raises:
        Exception: If database operation fails
    """
    with SessionLocal() as session:
        report = EquipmentReport(**data)
        session.add(report)
        session.commit()

def process_batch(report_files: List[str]) -> None:
    """Process a batch of report files.
    
    Args:
        report_files: List of file paths to process
    """
    for report_file in report_files:
        try:
            logger.info(f'Processing file: {report_file}')
            raw_text = fetch_data(report_file)
            structured_data = transform_records(raw_text)
            structured_data = sanitize_fields(structured_data)
            validate_input_data(structured_data)
            save_to_db(structured_data)
        except Exception as e:
            logger.error(f'Error processing {report_file}: {e}')  

if __name__ == '__main__':
    # Example usage
    report_files = ['report1.pdf', 'report2.pdf']  # List of report files
    process_batch(report_files)