Build Intelligent Equipment Log Search Pipelines with DeepSeek-OCR-2 and LlamaIndex

"""
Production implementation for building intelligent equipment log search pipelines using DeepSeek-OCR-2 and LlamaIndex.
Provides secure, scalable operations for extracting and processing equipment logs.
"""
from typing import Dict, Any, List, Tuple
import os
import logging
import requests
from sqlalchemy import create_engine, Column, Integer, String, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.exc import SQLAlchemyError

# Logger setup to capture various levels of logs
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# SQLAlchemy setup for database interactions
Base = declarative_base()

class Config:
    database_url: str = os.getenv('DATABASE_URL', 'sqlite:///logs.db')
    retry_attempts: int = int(os.getenv('RETRY_ATTEMPTS', 5))
    retry_delay: int = int(os.getenv('RETRY_DELAY', 2))  # seconds

# Database model for logs
class EquipmentLog(Base):
    __tablename__ = 'equipment_logs'
    id = Column(Integer, primary_key=True)
    log_text = Column(Text)
    processed = Column(Integer, default=0)

# SQLAlchemy engine and session setup
engine = create_engine(Config.database_url)
Base.metadata.create_all(engine)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=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 'log_text' not in data:
        raise ValueError('Missing log_text field')
    return True

def sanitize_fields(data: Dict[str, Any]) -> Dict[str, Any]:
    """Sanitize log text to prevent injection.
    
    Args:
        data: Input data
    Returns:
        Sanitized data
    """
    data['log_text'] = data['log_text'].replace('<', '&lt;').replace('>', '&gt;')
    return data

def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize data for processing.
    
    Args:
        data: Input data
    Returns:
        Normalized data
    """
    data['log_text'] = data['log_text'].strip()
    return data

def fetch_data(session: Session) -> List[EquipmentLog]:
    """Fetch unprocessed logs from the database.
    
    Args:
        session: Database session
    Returns:
        List of EquipmentLog objects
    """
    return session.query(EquipmentLog).filter_by(processed=0).all()

def save_to_db(session: Session, log: EquipmentLog) -> None:
    """Save processed log to the database.
    
    Args:
        session: Database session
        log: EquipmentLog object to save
    """
    session.add(log)
    session.commit()

def call_api(log_text: str) -> Dict[str, Any]:
    """Call external API for OCR processing.
    
    Args:
        log_text: The text to process
    Returns:
        API response
    Raises:
        Exception: If API call fails
    """
    url = 'http://ocr-api.example.com/process'
    response = requests.post(url, json={'text': log_text})
    if response.status_code != 200:
        logger.error('API call failed with status %s', response.status_code)
        raise Exception('API call failed')
    return response.json()

def process_batch(session: Session) -> None:
    """Process a batch of logs.
    
    Args:
        session: Database session
    """
    logs = fetch_data(session)
    for log in logs:
        try:
            # Fetch log text and process it
            logger.info('Processing log ID: %d', log.id)
            result = call_api(log.log_text)
            log.processed = 1  # Mark as processed
            save_to_db(session, log)
            logger.info('Successfully processed log ID: %d', log.id)
        except Exception as e:
            logger.error('Error processing log ID %d: %s', log.id, str(e))

def aggregate_metrics(session: Session) -> Dict[str, int]:
    """Aggregate metrics from the logs.
    
    Args:
        session: Database session
    Returns:
        Dictionary with metrics
    """
    total_logs = session.query(EquipmentLog).count()
    processed_logs = session.query(EquipmentLog).filter_by(processed=1).count()
    return {'total': total_logs, 'processed': processed_logs}

class LogPipeline:
    """Main orchestrator class for log processing workflow."""
    def __init__(self):
        self.db_session = SessionLocal()

    def run(self) -> None:
        """Execute the log processing pipeline."""
        try:
            logger.info('Starting log processing pipeline')
            process_batch(self.db_session)
            metrics = aggregate_metrics(self.db_session)
            logger.info('Metrics: %s', metrics)
        except Exception as e:
            logger.error('Pipeline execution failed: %s', str(e))
        finally:
            self.db_session.close()  # Ensure session is closed

if __name__ == '__main__':
    # Example usage
    pipeline = LogPipeline()  # Create an instance of the pipeline
    pipeline.run()  # Run the log processing pipeline