""" Enhanced Error Handling and Logging Utilities Provides comprehensive error handling, logging, and monitoring for the Vibeon Translator """ import logging import traceback import functools import time import asyncio from typing import Any, Callable, Dict, Optional, Union from datetime import datetime from enum import Enum import json import sys from config.settings import settings class ErrorSeverity(Enum): """Error severity levels""" LOW = "low" MEDIUM = "medium" HIGH = "high" CRITICAL = "critical" class ErrorCategory(Enum): """Error categories for better organization""" TRANSLATION = "translation" AUTHENTICATION = "authentication" DATABASE = "database" CACHE = "cache" NETWORK = "network" MEMORY = "memory" MODEL_LOADING = "model_loading" API = "api" VALIDATION = "validation" SYSTEM = "system" @functools.lru_cache(maxsize=1) def get_logger(name: str = "vibeon_translator") -> logging.Logger: """Get configured logger instance""" logger = logging.getLogger(name) if not logger.handlers: # Create formatter formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) # Console handler console_handler = logging.StreamHandler(sys.stdout) console_handler.setFormatter(formatter) console_handler.setLevel(logging.INFO) # File handler try: file_handler = logging.FileHandler('logs/vibeon_translator.log') file_handler.setFormatter(formatter) file_handler.setLevel(logging.DEBUG) logger.addHandler(file_handler) except Exception: pass # File logging is optional logger.addHandler(console_handler) logger.setLevel(logging.INFO) return logger class ErrorHandler: """Enhanced error handling and monitoring""" def __init__(self): self.logger = get_logger("error_handler") self.error_counts = {} self.error_history = [] self.max_history = 1000 def handle_error( self, error: Exception, context: Optional[Dict[str, Any]] = None, severity: ErrorSeverity = ErrorSeverity.MEDIUM, category: ErrorCategory = ErrorCategory.SYSTEM, user_id: Optional[str] = None, request_id: Optional[str] = None ) -> Dict[str, Any]: """ Handle and log errors with comprehensive context Args: error: The exception that occurred context: Additional context information severity: Error severity level category: Error category user_id: User ID if available request_id: Request ID if available Returns: Error information dictionary """ error_info = { "timestamp": datetime.utcnow().isoformat(), "error_type": type(error).__name__, "error_message": str(error), "severity": severity.value, "category": category.value, "user_id": user_id, "request_id": request_id, "context": context or {}, "traceback": traceback.format_exc(), "stack_trace": traceback.format_stack() } # Log based on severity log_level = self._get_log_level(severity) self.logger.log(log_level, f"[{category.value.upper()}] {error_info['error_message']}", extra={"error_info": error_info}) # Update error counts error_key = f"{category.value}:{type(error).__name__}" self.error_counts[error_key] = self.error_counts.get(error_key, 0) + 1 # Add to history self.error_history.append(error_info) if len(self.error_history) > self.max_history: self.error_history.pop(0) # Handle critical errors if severity == ErrorSeverity.CRITICAL: self._handle_critical_error(error_info) return error_info def _get_log_level(self, severity: ErrorSeverity) -> int: """Get log level based on severity""" mapping = { ErrorSeverity.LOW: logging.INFO, ErrorSeverity.MEDIUM: logging.WARNING, ErrorSeverity.HIGH: logging.ERROR, ErrorSeverity.CRITICAL: logging.CRITICAL } return mapping.get(severity, logging.ERROR) def _handle_critical_error(self, error_info: Dict[str, Any]): """Handle critical errors with additional actions""" self.logger.critical(f"CRITICAL ERROR: {error_info['error_message']}") # Could add additional actions here: # - Send alerts # - Restart services # - Notify administrators # - Save to external monitoring system def get_error_stats(self) -> Dict[str, Any]: """Get error statistics""" return { "total_errors": len(self.error_history), "error_counts": self.error_counts, "recent_errors": self.error_history[-10:] if self.error_history else [], "error_rate": self._calculate_error_rate() } def _calculate_error_rate(self) -> float: """Calculate error rate per minute""" if not self.error_history: return 0.0 now = datetime.utcnow() recent_errors = [ error for error in self.error_history if (now - datetime.fromisoformat(error["timestamp"])).total_seconds() < 60 ] return len(recent_errors) def clear_error_history(self): """Clear error history""" self.error_history.clear() self.error_counts.clear() def error_handler( severity: ErrorSeverity = ErrorSeverity.MEDIUM, category: ErrorCategory = ErrorCategory.SYSTEM, reraise: bool = False, return_default: Any = None ): """ Decorator for automatic error handling Args: severity: Error severity level category: Error category reraise: Whether to reraise the exception return_default: Default value to return on error """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): handler = ErrorHandler() try: return func(*args, **kwargs) except Exception as e: error_info = handler.handle_error( error=e, context={ "function": func.__name__, "args": str(args)[:200], # Limit size "kwargs": str(kwargs)[:200] }, severity=severity, category=category ) if reraise: raise return return_default @functools.wraps(func) async def async_wrapper(*args, **kwargs): handler = ErrorHandler() try: return await func(*args, **kwargs) except Exception as e: error_info = handler.handle_error( error=e, context={ "function": func.__name__, "args": str(args)[:200], "kwargs": str(kwargs)[:200] }, severity=severity, category=category ) if reraise: raise return return_default return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper return decorator def performance_monitor(func: Callable) -> Callable: """Decorator for performance monitoring""" @functools.wraps(func) def wrapper(*args, **kwargs): start_time = time.time() handler = ErrorHandler() try: result = func(*args, **kwargs) execution_time = time.time() - start_time handler.logger.info( f"Function {func.__name__} executed in {execution_time:.3f}s", extra={ "function": func.__name__, "execution_time": execution_time, "args_count": len(args), "kwargs_count": len(kwargs) } ) return result except Exception as e: execution_time = time.time() - start_time handler.handle_error( error=e, context={ "function": func.__name__, "execution_time": execution_time }, severity=ErrorSeverity.MEDIUM, category=ErrorCategory.SYSTEM ) raise @functools.wraps(func) async def async_wrapper(*args, **kwargs): start_time = time.time() handler = ErrorHandler() try: result = await func(*args, **kwargs) execution_time = time.time() - start_time handler.logger.info( f"Async function {func.__name__} executed in {execution_time:.3f}s", extra={ "function": func.__name__, "execution_time": execution_time, "args_count": len(args), "kwargs_count": len(kwargs) } ) return result except Exception as e: execution_time = time.time() - start_time handler.handle_error( error=e, context={ "function": func.__name__, "execution_time": execution_time }, severity=ErrorSeverity.MEDIUM, category=ErrorCategory.SYSTEM ) raise return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper def validate_input( required_fields: Optional[list] = None, field_types: Optional[Dict[str, type]] = None, custom_validators: Optional[Dict[str, Callable]] = None ): """ Decorator for input validation Args: required_fields: List of required field names field_types: Dictionary mapping field names to expected types custom_validators: Dictionary mapping field names to custom validation functions """ def decorator(func: Callable) -> Callable: @functools.wraps(func) def wrapper(*args, **kwargs): handler = ErrorHandler() # Get function signature import inspect sig = inspect.signature(func) bound_args = sig.bind(*args, **kwargs) bound_args.apply_defaults() # Validate required fields if required_fields: for field in required_fields: if field not in bound_args.arguments: error = ValueError(f"Required field '{field}' is missing") handler.handle_error( error=error, context={"missing_field": field, "function": func.__name__}, severity=ErrorSeverity.MEDIUM, category=ErrorCategory.VALIDATION ) raise error # Validate field types if field_types: for field, expected_type in field_types.items(): if field in bound_args.arguments: value = bound_args.arguments[field] if not isinstance(value, expected_type): error = TypeError( f"Field '{field}' must be of type {expected_type.__name__}, " f"got {type(value).__name__}" ) handler.handle_error( error=error, context={ "field": field, "expected_type": expected_type.__name__, "actual_type": type(value).__name__, "function": func.__name__ }, severity=ErrorSeverity.MEDIUM, category=ErrorCategory.VALIDATION ) raise error # Run custom validators if custom_validators: for field, validator in custom_validators.items(): if field in bound_args.arguments: try: validator(bound_args.arguments[field]) except Exception as e: handler.handle_error( error=e, context={ "field": field, "function": func.__name__, "validation_error": str(e) }, severity=ErrorSeverity.MEDIUM, category=ErrorCategory.VALIDATION ) raise return func(*args, **kwargs) return wrapper return decorator # Global error handler instance error_handler_instance = ErrorHandler() # Convenience functions def log_error(error: Exception, context: Optional[Dict[str, Any]] = None, severity: ErrorSeverity = ErrorSeverity.MEDIUM, category: ErrorCategory = ErrorCategory.SYSTEM): """Log an error with the global error handler""" return error_handler_instance.handle_error(error, context, severity, category) def get_error_stats() -> Dict[str, Any]: """Get error statistics from the global error handler""" return error_handler_instance.get_error_stats() def clear_errors(): """Clear error history from the global error handler""" error_handler_instance.clear_error_history()