chat_agent/utils/error_handler.py

"""

Centralized error handling utilities for the chat agent.



This module provides comprehensive error handling, logging, and fallback

response mechanisms for all chat agent components.

"""

import logging
import traceback
import functools
from typing import Dict, Any, Optional, Callable, Union
from datetime import datetime
from enum import Enum

from flask import jsonify, request
from flask_socketio import emit


class ErrorSeverity(Enum):
    """Error severity levels for categorization and handling."""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


class ErrorCategory(Enum):
    """Error categories for better classification and handling."""
    API_ERROR = "api_error"
    DATABASE_ERROR = "database_error"
    VALIDATION_ERROR = "validation_error"
    AUTHENTICATION_ERROR = "authentication_error"
    RATE_LIMIT_ERROR = "rate_limit_error"
    NETWORK_ERROR = "network_error"
    SYSTEM_ERROR = "system_error"
    USER_ERROR = "user_error"


class ChatAgentError(Exception):
    """Base exception class for all chat agent errors."""
    
    def __init__(self, message: str, category: ErrorCategory = ErrorCategory.SYSTEM_ERROR,

                 severity: ErrorSeverity = ErrorSeverity.MEDIUM, 

                 user_message: Optional[str] = None, 

                 error_code: Optional[str] = None,

                 context: Optional[Dict[str, Any]] = None):
        """

        Initialize chat agent error.

        

        Args:

            message: Technical error message for logging

            category: Error category for classification

            severity: Error severity level

            user_message: User-friendly error message

            error_code: Unique error code for tracking

            context: Additional context information

        """
        super().__init__(message)
        self.category = category
        self.severity = severity
        self.user_message = user_message or self._get_default_user_message()
        self.error_code = error_code or self._generate_error_code()
        self.context = context or {}
        self.timestamp = datetime.utcnow()
    
    def _get_default_user_message(self) -> str:
        """Get default user-friendly message based on category."""
        default_messages = {
            ErrorCategory.API_ERROR: "I'm having trouble connecting to my services. Please try again in a moment.",
            ErrorCategory.DATABASE_ERROR: "I'm experiencing some technical difficulties. Please try again later.",
            ErrorCategory.VALIDATION_ERROR: "There seems to be an issue with your request. Please check your input and try again.",
            ErrorCategory.AUTHENTICATION_ERROR: "Authentication failed. Please check your credentials.",
            ErrorCategory.RATE_LIMIT_ERROR: "I'm currently experiencing high demand. Please try again in a moment.",
            ErrorCategory.NETWORK_ERROR: "I'm having trouble connecting right now. Please check your connection and try again.",
            ErrorCategory.SYSTEM_ERROR: "I encountered an unexpected error. Please try again.",
            ErrorCategory.USER_ERROR: "There's an issue with your request. Please check your input and try again."
        }
        return default_messages.get(self.category, "An unexpected error occurred. Please try again.")
    
    def _generate_error_code(self) -> str:
        """Generate unique error code for tracking."""
        import uuid
        return f"{self.category.value.upper()}_{str(uuid.uuid4())[:8]}"
    
    def to_dict(self) -> Dict[str, Any]:
        """Convert error to dictionary for JSON serialization."""
        return {
            'error_code': self.error_code,
            'category': self.category.value,
            'severity': self.severity.value,
            'message': self.user_message,
            'timestamp': self.timestamp.isoformat(),
            'context': self.context
        }


class ErrorHandler:
    """Centralized error handler for the chat agent."""
    
    def __init__(self, logger: Optional[logging.Logger] = None):
        """

        Initialize error handler.

        

        Args:

            logger: Logger instance to use for error logging

        """
        self.logger = logger or logging.getLogger(__name__)
        self.fallback_responses = self._initialize_fallback_responses()
    
    def _initialize_fallback_responses(self) -> Dict[ErrorCategory, str]:
        """Initialize fallback responses for different error categories."""
        return {
            ErrorCategory.API_ERROR: "I'm having trouble with my AI services right now. Here are some general programming tips that might help:\n\n• Break down complex problems into smaller steps\n• Use descriptive variable names\n• Add comments to explain your logic\n• Test your code frequently\n\nPlease try your question again in a moment!",
            
            ErrorCategory.DATABASE_ERROR: "I'm experiencing some technical difficulties accessing my knowledge base. In the meantime, I recommend:\n\n• Checking official documentation for your programming language\n• Looking for similar examples online\n• Breaking your problem into smaller parts\n\nPlease try again shortly!",
            
            ErrorCategory.RATE_LIMIT_ERROR: "I'm currently helping many students and need a moment to catch up. While you wait:\n\n• Review your code for any obvious syntax errors\n• Try running your code to see what happens\n• Think about what you're trying to accomplish\n\nPlease try your question again in a few seconds!",
            
            ErrorCategory.NETWORK_ERROR: "I'm having connection issues right now. Here's what you can try:\n\n• Check your internet connection\n• Refresh the page and try again\n• Make sure your code follows proper syntax\n\nI'll be back online shortly!",
            
            ErrorCategory.SYSTEM_ERROR: "I encountered an unexpected issue. Don't worry - this happens sometimes! Try:\n\n• Rephrasing your question\n• Being more specific about what you need help with\n• Checking if your code has any obvious errors\n\nPlease try again!",
            
            ErrorCategory.USER_ERROR: "I need a bit more information to help you effectively. Please:\n\n• Be specific about what you're trying to do\n• Include any error messages you're seeing\n• Share the relevant code if possible\n\nThen I can give you much better assistance!"
        }
    
    def handle_error(self, error: Exception, context: Optional[Dict[str, Any]] = None) -> ChatAgentError:
        """

        Handle and classify an error, returning a standardized ChatAgentError.

        

        Args:

            error: The original exception

            context: Additional context information

            

        Returns:

            ChatAgentError: Standardized error object

        """
        # Convert to ChatAgentError if not already
        if isinstance(error, ChatAgentError):
            chat_error = error
        else:
            chat_error = self._classify_error(error, context)
        
        # Log the error
        self._log_error(chat_error, error)
        
        return chat_error
    
    def _classify_error(self, error: Exception, context: Optional[Dict[str, Any]] = None) -> ChatAgentError:
        """Classify an error and convert to ChatAgentError."""
        error_str = str(error).lower()
        error_type = type(error).__name__
        
        # API-related errors
        if any(keyword in error_str for keyword in ['groq', 'api', 'langchain']):
            if 'rate limit' in error_str or '429' in error_str:
                return ChatAgentError(
                    message=str(error),
                    category=ErrorCategory.RATE_LIMIT_ERROR,
                    severity=ErrorSeverity.MEDIUM,
                    context=context
                )
            elif 'authentication' in error_str or '401' in error_str:
                return ChatAgentError(
                    message=str(error),
                    category=ErrorCategory.AUTHENTICATION_ERROR,
                    severity=ErrorSeverity.HIGH,
                    context=context
                )
            else:
                return ChatAgentError(
                    message=str(error),
                    category=ErrorCategory.API_ERROR,
                    severity=ErrorSeverity.MEDIUM,
                    context=context
                )
        
        # Database-related errors
        elif any(keyword in error_str for keyword in ['database', 'sql', 'connection', 'postgresql', 'redis']):
            return ChatAgentError(
                message=str(error),
                category=ErrorCategory.DATABASE_ERROR,
                severity=ErrorSeverity.HIGH,
                context=context
            )
        
        # Network-related errors
        elif any(keyword in error_str for keyword in ['network', 'connection', 'timeout', 'unreachable']):
            return ChatAgentError(
                message=str(error),
                category=ErrorCategory.NETWORK_ERROR,
                severity=ErrorSeverity.MEDIUM,
                context=context
            )
        
        # Validation errors
        elif any(keyword in error_str for keyword in ['validation', 'invalid', 'malformed']):
            return ChatAgentError(
                message=str(error),
                category=ErrorCategory.VALIDATION_ERROR,
                severity=ErrorSeverity.LOW,
                context=context
            )
        
        # Default to system error
        else:
            return ChatAgentError(
                message=str(error),
                category=ErrorCategory.SYSTEM_ERROR,
                severity=ErrorSeverity.MEDIUM,
                context=context
            )
    
    def _log_error(self, chat_error: ChatAgentError, original_error: Exception):
        """Log error with appropriate level and context."""
        log_data = {
            'error_code': chat_error.error_code,
            'category': chat_error.category.value,
            'severity': chat_error.severity.value,
            'original_error': str(original_error),
            'error_type': type(original_error).__name__,
            'context': chat_error.context,
            'timestamp': chat_error.timestamp.isoformat()
        }
        
        # Add traceback for higher severity errors
        if chat_error.severity in [ErrorSeverity.HIGH, ErrorSeverity.CRITICAL]:
            log_data['traceback'] = traceback.format_exc()
        
        # Log with appropriate level
        if chat_error.severity == ErrorSeverity.CRITICAL:
            self.logger.critical(f"Critical error: {chat_error.message}", extra=log_data)
        elif chat_error.severity == ErrorSeverity.HIGH:
            self.logger.error(f"High severity error: {chat_error.message}", extra=log_data)
        elif chat_error.severity == ErrorSeverity.MEDIUM:
            self.logger.warning(f"Medium severity error: {chat_error.message}", extra=log_data)
        else:
            self.logger.info(f"Low severity error: {chat_error.message}", extra=log_data)
    
    def get_fallback_response(self, error: ChatAgentError) -> str:
        """Get fallback response for an error category."""
        return self.fallback_responses.get(error.category, self.fallback_responses[ErrorCategory.SYSTEM_ERROR])
    
    def handle_api_response_error(self, error: Exception, context: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Handle error for API responses."""
        chat_error = self.handle_error(error, context)
        
        return {
            'success': False,
            'error': chat_error.to_dict(),
            'fallback_response': self.get_fallback_response(chat_error)
        }
    
    def handle_websocket_error(self, error: Exception, context: Optional[Dict[str, Any]] = None):
        """Handle error for WebSocket responses."""
        chat_error = self.handle_error(error, context)
        
        emit('error', {
            'error': chat_error.to_dict(),
            'fallback_response': self.get_fallback_response(chat_error)
        })


def error_handler_decorator(error_handler: ErrorHandler, 

                          return_fallback: bool = False,

                          emit_websocket_error: bool = False):
    """

    Decorator for automatic error handling in functions.

    

    Args:

        error_handler: ErrorHandler instance to use

        return_fallback: Whether to return fallback response on error

        emit_websocket_error: Whether to emit WebSocket error on exception

    """
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            try:
                return func(*args, **kwargs)
            except Exception as e:
                context = {
                    'function': func.__name__,
                    'args': str(args)[:200],  # Limit context size
                    'kwargs': str(kwargs)[:200]
                }
                
                chat_error = error_handler.handle_error(e, context)
                
                if emit_websocket_error:
                    error_handler.handle_websocket_error(e, context)
                    return None
                elif return_fallback:
                    return error_handler.get_fallback_response(chat_error)
                else:
                    raise chat_error
        
        return wrapper
    return decorator


# Global error handler instance
_global_error_handler = None


def get_error_handler() -> ErrorHandler:
    """Get global error handler instance."""
    global _global_error_handler
    if _global_error_handler is None:
        _global_error_handler = ErrorHandler()
    return _global_error_handler


def set_error_handler(error_handler: ErrorHandler):
    """Set global error handler instance."""
    global _global_error_handler
    _global_error_handler = error_handler