Classify and Route Industrial Maintenance Tickets with Unstructured and LlamaIndex

"""
Production implementation for Classifying and Routing Industrial Maintenance Tickets.
This module provides secure, scalable operations for handling unstructured ticket data.
"""
from typing import Dict, Any, List, Optional
import os
import logging
import requests
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from time import sleep

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

# Configuration class for environment variables
class Config:
    DATABASE_URL: str = os.getenv('DATABASE_URL', 'sqlite:///./test.db')

# Database setup using SQLAlchemy
Base = declarative_base()

class Ticket(Base):
    __tablename__ = 'tickets'

    id = Column(Integer, primary_key=True, index=True)
    description = Column(String)
    category = Column(String)

# Initialize the database connection
engine = create_engine(Config.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

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

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

# Function to normalize ticket categories
async def normalize_data(data: Dict[str, Any]) -> Dict[str, Any]:
    """Normalize ticket data for consistency.
    
    Args:
        data: Raw ticket data
    Returns:
        Normalized ticket data
    """
    data['category'] = data['category'].lower()
    return data

# Function to transform ticket records
async def transform_records(data: Dict[str, Any]) -> Ticket:
    """Transform raw input into a Ticket object.
    
    Args:
        data: Raw input data
    Returns:
        Ticket object
    """
    return Ticket(description=data['description'], category=data['category'])

# Function to save ticket data to the database
async def save_to_db(ticket: Ticket, db: Session) -> Ticket:
    """Save the ticket to the database.
    
    Args:
        ticket: Ticket object to save
        db: Database session
    Returns:
        Saved ticket object
    """
    db.add(ticket)
    db.commit()
    db.refresh(ticket)
    return ticket

# Function to fetch ticket data from an external API
async def fetch_data(api_url: str) -> List[Dict[str, Any]]:
    """Fetch ticket data from an external API.
    
    Args:
        api_url: URL of the API to fetch data from
    Returns:
        List of ticket data
    Raises:
        HTTPException: If the API call fails
    """
    try:
        response = requests.get(api_url)
        response.raise_for_status()
        return response.json()
    except requests.RequestException as e:
        logger.error(f'Error fetching data from API: {e}')
        raise HTTPException(status_code=500, detail='Error fetching data')

# Main class for ticket processing
class TicketProcessor:
    def __init__(self, session: Session):
        self.session = session

    async def process_ticket(self, data: Dict[str, Any]) -> Ticket:
        """Main workflow for processing a ticket.
        
        Args:
            data: Raw ticket data
        Returns:
            Processed Ticket object
        """
        await validate_input(data)
        sanitized_data = await sanitize_fields(data)
        normalized_data = await normalize_data(sanitized_data)
        ticket = await transform_records(normalized_data)
        return await save_to_db(ticket, self.session)

# FastAPI application setup
app = FastAPI()

@app.post('/tickets/', response_model=Ticket)
async def create_ticket(data: Dict[str, Any]):
    """API endpoint to create a ticket.
    
    Args:
        data: Ticket data to create
    Returns:
        Created Ticket object
    Raises:
        HTTPException: For validation errors
    """
    db = SessionLocal()
    processor = TicketProcessor(db)
    try:
        ticket = await processor.process_ticket(data)
        return ticket
    except ValueError as ve:
        logger.warning(f'Validation error: {ve}')
        raise HTTPException(status_code=422, detail=str(ve))
    finally:
        db.close()  # Ensure the database session is closed

if __name__ == '__main__':
    # Example usage: You can run the FastAPI server and test the endpoint
    import uvicorn
    uvicorn.run(app, host='0.0.0.0', port=8000)