Extract Structured Safety Data from Factory Incident and Hazard Reports with Surya and spaCy

"""
Production implementation for extracting structured safety data from factory incident and hazard reports.
This module provides a secure and scalable operation using Surya and spaCy.
"""

from typing import Dict, Any, List
import os
import logging
import spacy
import requests
from sqlalchemy import create_engine, text
from sqlalchemy.orm import sessionmaker

# Setup logger for monitoring
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration class to manage environment variables
class Config:
    database_url: str = os.getenv('DATABASE_URL')
    spacy_model: str = os.getenv('SPACY_MODEL', 'en_core_web_sm')

# Initialize spaCy model
nlp = spacy.load(Config.spacy_model)

# Create a SQLAlchemy engine and session factory for connection pooling
engine = create_engine(Config.database_url, pool_size=20, max_overflow=0)
Session = sessionmaker(bind=engine)

async 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 'reports' not in data:
        raise ValueError('Missing reports field')
    if not isinstance(data['reports'], list):
        raise ValueError('Reports must be a list')
    return True

async def sanitize_fields(record: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize fields in the report.
    
    Args:
        record: Report record to sanitize
    Returns:
        Sanitized report
    """
    return {key: str(value).strip() for key, value in record.items() if value is not None}

async def transform_records(records: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Transform raw records into structured format.
    
    Args:
        records: List of raw report records
    Returns:
        List of transformed records
    """
    structured_records = []
    for record in records:
        sanitized_record = await sanitize_fields(record)
        structured_records.append(sanitized_record)
    return structured_records

async def process_batch(batch: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    """Process a batch of records and extract safety data.
    
    Args:
        batch: List of records to process
    Returns:
        List of extracted safety data
    """  
    extracted_data = []
    for report in batch:
        doc = nlp(report['description'])  # Analyze text with spaCy
        safety_info = {'incidents': [], 'hazards': []}
        for ent in doc.ents:
            if ent.label_ in ('INJURY', 'HAZARD'):
                safety_info[ent.label_.lower() + 's'].append(ent.text)
        extracted_data.append(safety_info)
    return extracted_data

async def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch data from the provided API.
    
    Args:
        api_url: URL of the API to fetch data from
    Returns:
        List of report records
    Raises:
        RuntimeError: If API call fails
    """
    response = requests.get(api_url)
    if response.status_code != 200:
        raise RuntimeError('Failed to fetch data from API')
    return response.json()

async def save_to_db(records: List[Dict[str, Any]]) -> None:
    """Save extracted records to the database.
    
    Args:
        records: List of records to save
    """  
    with Session() as session:
        for record in records:
            session.execute(
                text('INSERT INTO safety_data (incidents, hazards) VALUES (:incidents, :hazards)'),
                {'incidents': record['incidents'], 'hazards': record['hazards']}
            )
        session.commit()  # Commit changes to the database

async def format_output(data: List[Dict[str, Any]]) -> str:
    """Format the output for display.
    
    Args:
        data: Data to format
    Returns:
        Formatted string output
    """  
    return '\n'.join([f'Incidents: {d['incidents']}, Hazards: {d['hazards']}' for d in data])

async def handle_errors(func):
    """Decorator to handle errors in async functions.
    
    Args:
        func: Function to wrap
    """  
    async def wrapper(*args, **kwargs):
        try:
            return await func(*args, **kwargs)
        except Exception as e:
            logger.error(f'Error in {func.__name__}: {e}')
    return wrapper

class SafetyDataExtractor:
    """Main orchestrator for extracting safety data.
    
    This class ties together the helper functions to form a complete workflow.
    """  

    async def run(self, api_url: str) -> None:
        """Run the extraction process.
        
        Args:
            api_url: URL of the API to fetch data from
        """  
        logger.info('Starting data extraction process.')
        raw_data = await fetch_data(api_url)  # Fetch raw data
        await validate_input({'reports': raw_data})  # Validate input data
        transformed_data = await transform_records(raw_data)  # Transform records
        processed_data = await process_batch(transformed_data)  # Process data
        await save_to_db(processed_data)  # Save to database
        logger.info('Data extraction process completed.')  

if __name__ == '__main__':
    # Example usage
    extractor = SafetyDataExtractor()
    import asyncio
    asyncio.run(extractor.run('http://example.com/api/reports'))