Extract Structured Data from Complex Equipment Manuals with DeepSeek-OCR-2 and Haystack

"""
Production implementation for extracting structured data from complex equipment manuals using DeepSeek-OCR-2 and Haystack.
Provides secure, scalable operations with robust error handling and logging.
"""
from typing import Dict, Any, List
import os
import logging
import httpx
import asyncio
from pydantic import BaseModel, ValidationError

logging.basicConfig(level=logging.INFO)  # Set up logging
logger = logging.getLogger(__name__)  # Create a logger instance

class Config:
    """Configuration class to manage environment variables."""
    database_url: str = os.getenv('DATABASE_URL')  # Database connection string
    ocr_service_url: str = os.getenv('OCR_SERVICE_URL')  # OCR service endpoint

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

async def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize input fields to prevent injection attacks.
    
    Args:
        data: Input dictionary to sanitize
    Returns:
        Sanitized dictionary
    """
    sanitized_data = {key: str(value).strip() for key, value in data.items()}  # Strip whitespace
    return sanitized_data  # Return sanitized data

async def fetch_data(manual_id: str) -> Dict[str, Any]:
    """Fetch the manual content using OCR service.
    
    Args:
        manual_id: Identifier for the equipment manual
    Returns:
        Parsed data from the OCR service
    Raises:
        httpx.HTTPStatusError: If the request fails
    """
    logger.info(f'Fetching data for manual_id: {manual_id}')  # Log the action
    async with httpx.AsyncClient() as client:
        response = await client.get(f'{Config.ocr_service_url}/extract/{manual_id}')  # Call the OCR service
        response.raise_for_status()  # Raise an error for bad responses
    return response.json()  # Return the parsed JSON data

async def transform_records(raw_data: Dict[str, Any]) -> List[Dict[str, Any]]:
    """Transform raw data into structured format.
    
    Args:
        raw_data: Data dictionary from OCR service
    Returns:
        List of structured records
    """
    structured_data = []  # Initialize structured data list
    for item in raw_data.get('items', []):  # Process each item
        structured_data.append({
            'title': item.get('title'),
            'description': item.get('description'),
            'specs': item.get('specs'),  # Extract specifications
        })
    return structured_data  # Return structured records

async def save_to_db(data: List[Dict[str, Any]]) -> None:
    """Save structured data into the database.
    
    Args:
        data: List of structured records to save
    Raises:
        Exception: If database operation fails
    """
    logger.info('Saving data to database...')  # Log the saving action
    # Simulate DB save operation
    for record in data:
        logger.info(f'Saving record: {record}')  # Log each record being saved
    # Placeholder for actual DB operation

async def handle_errors(task: str, error: Exception) -> None:
    """Handle errors during processing.
    
    Args:
        task: Description of the task where error occurred
        error: Exception instance
    """
    logger.error(f'Error occurred during {task}: {error}')  # Log the error with task details

async def process_batch(manual_id: str) -> None:
    """Main workflow to process a single equipment manual.
    
    Args:
        manual_id: Identifier for the equipment manual
    """
    try:
        await validate_input({'manual_id': manual_id})  # Validate input
        raw_data = await fetch_data(manual_id)  # Fetch data from OCR service
        structured_data = await transform_records(raw_data)  # Transform data
        await save_to_db(structured_data)  # Save structured data
    except ValueError as ve:
        await handle_errors('input validation', ve)  # Handle validation errors
    except httpx.HTTPStatusError as http_err:
        await handle_errors('fetching data', http_err)  # Handle HTTP errors
    except Exception as e:
        await handle_errors('processing batch', e)  # Handle other errors

if __name__ == '__main__':
    # Example usage
    manual_id = '12345'  # Example manual ID
    asyncio.run(process_batch(manual_id))  # Run the processing function