Update utils/logging.py

utils/logging.py

@@ -1,163 +1,232 @@
-"""
-Logging module for the MONA application.
-
-This module provides a centralized logging system for the application,
-replacing print statements with proper logging for better debugging and monitoring.
-"""
-
 import logging
-import logging.config
+import os
+import sys
 from datetime import datetime
+from typing import Optional, Dict, Any
+import json
 
-from utils.config import LOGGING_CONFIG
-
-# Configure logging based on the configuration in config.py
-logging.config.dictConfig(LOGGING_CONFIG)
-
-# Create a logger for the application
-logger = logging.getLogger("mona")
-
-
-def get_logger(name=None):
+# Configure logging
+def setup_logger(name: str = "mona_app", level: str = "INFO") -> logging.Logger:
+    """
+    Set up a logger with both file and console handlers
     """
-    Get a logger instance with the specified name.
+    logger = logging.getLogger(name)
     
-    Args:
-        name (str, optional): The name of the logger. If None, returns the root logger.
-            For module-specific loggers, use __name__ as the name.
+    # Clear existing handlers to avoid duplicates
+    if logger.handlers:
+        logger.handlers.clear()
     
-    Returns:
-        logging.Logger: A logger instance.
-    """
-    if name:
-        return logging.getLogger(f"mona.{name}")
-    return logger
-
-
-def log_function_call(func):
-    """
-    Decorator to log function calls with parameters and return values.
+    # Set log level
+    log_level = getattr(logging, level.upper(), logging.INFO)
+    logger.setLevel(log_level)
     
-    Args:
-        func (callable): The function to decorate.
+    # Create formatters
+    detailed_formatter = logging.Formatter(
+        '%(asctime)s - %(name)s - %(levelname)s - %(funcName)s:%(lineno)d - %(message)s'
+    )
+    simple_formatter = logging.Formatter('%(levelname)s - %(message)s')
     
-    Returns:
-        callable: The decorated function.
-    """
-    def wrapper(*args, **kwargs):
-        func_logger = get_logger(func.__module__)
-        func_name = func.__qualname__
-        
-        # Log function call with arguments
-        args_repr = [repr(a) for a in args]
-        kwargs_repr = [f"{k}={v!r}" for k, v in kwargs.items()]
-        signature = ", ".join(args_repr + kwargs_repr)
-        func_logger.debug(f"Calling {func_name}({signature})")
+    # Console handler
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(simple_formatter)
+    
+    # File handler (if in a writable environment)
+    try:
+        log_dir = "/tmp/logs" if os.path.exists("/tmp") else "./logs"
+        os.makedirs(log_dir, exist_ok=True)
         
-        # Call the function
-        try:
-            result = func(*args, **kwargs)
-            func_logger.debug(f"{func_name} returned {result!r}")
-            return result
-        except Exception as e:
-            func_logger.exception(f"Exception in {func_name}: {str(e)}")
-            raise
+        log_file = os.path.join(log_dir, f"{name}.log")
+        file_handler = logging.FileHandler(log_file)
+        file_handler.setLevel(logging.DEBUG)
+        file_handler.setFormatter(detailed_formatter)
+        logger.addHandler(file_handler)
+    except Exception:
+        # If we can't create file handler, continue with just console
+        pass
     
-    return wrapper
-
-
-def log_error(error_msg, exc_info=None):
-    """
-    Log an error message.
+    logger.addHandler(console_handler)
     
-    Args:
-        error_msg (str): The error message to log.
-        exc_info (Exception, optional): The exception information to include in the log.
-    """
-    if exc_info:
-        logger.error(error_msg, exc_info=exc_info)
-    else:
-        logger.error(error_msg)
-
+    return logger
 
-def log_warning(warning_msg):
-    """
-    Log a warning message.
+# Global logger instance
+_logger = None
+
+def get_logger() -> logging.Logger:
+    """Get the global logger instance"""
+    global _logger
+    if _logger is None:
+        _logger = setup_logger()
+    return _logger
+
+def log_error(message: str, error: Exception = None) -> None:
+    """Log error messages with optional exception details"""
+    logger = get_logger()
+    if error:
+        logger.error(f"{message}: {str(error)}", exc_info=True)
+    else:
+        logger.error(message)
+
+def log_warning(message: str) -> None:
+    """Log warning messages"""
+    logger = get_logger()
+    logger.warning(message)
+
+def log_info(message: str) -> None:
+    """Log info messages"""
+    logger = get_logger()
+    logger.info(message)
+
+def log_debug(message: str) -> None:
+    """Log debug messages"""
+    logger = get_logger()
+    logger.debug(message)
+
+def log_performance(operation: str, duration: float, additional_info: Dict[str, Any] = None) -> None:
+    """Log performance metrics"""
+    logger = get_logger()
+    perf_data = {
+        "operation": operation,
+        "duration_seconds": round(duration, 4),
+        "timestamp": datetime.now().isoformat()
+    }
     
-    Args:
-        warning_msg (str): The warning message to log.
-    """
-    logger.warning(warning_msg)
-
-
-def log_info(info_msg):
-    """
-    Log an info message.
+    if additional_info:
+        perf_data.update(additional_info)
     
-    Args:
-        info_msg (str): The info message to log.
-    """
-    logger.info(info_msg)
-
+    logger.info(f"PERFORMANCE: {json.dumps(perf_data)}")
 
-def log_debug(debug_msg):
-    """
-    Log a debug message.
+def log_user_action(action: str, user_id: str = None, details: Dict[str, Any] = None) -> None:
+    """Log user actions for analytics"""
+    logger = get_logger()
+    action_data = {
+        "action": action,
+        "user_id": user_id or "anonymous",
+        "timestamp": datetime.now().isoformat()
+    }
     
-    Args:
-        debug_msg (str): The debug message to log.
-    """
-    logger.debug(debug_msg)
-
+    if details:
+        action_data.update(details)
+    
+    logger.info(f"USER_ACTION: {json.dumps(action_data)}")
 
-def log_user_activity(user_id, activity_type, details=None):
-    """
-    Log user activity for analytics purposes.
+def log_data_operation(operation: str, table_name: str = None, record_count: int = None, success: bool = True) -> None:
+    """Log database/data operations"""
+    logger = get_logger()
+    data_op = {
+        "operation": operation,
+        "table": table_name,
+        "record_count": record_count,
+        "success": success,
+        "timestamp": datetime.now().isoformat()
+    }
     
-    Args:
-        user_id (str): The user identifier.
-        activity_type (str): The type of activity (e.g., 'page_view', 'task_create').
-        details (dict, optional): Additional details about the activity.
-    """
-    activity_log = {
-        "timestamp": datetime.now().isoformat(),
-        "user_id": user_id,
-        "activity_type": activity_type,
+    level = logging.INFO if success else logging.ERROR
+    logger.log(level, f"DATA_OP: {json.dumps(data_op)}")
+
+def log_api_call(endpoint: str, method: str = "GET", status_code: int = None, duration: float = None) -> None:
+    """Log API calls and responses"""
+    logger = get_logger()
+    api_data = {
+        "endpoint": endpoint,
+        "method": method,
+        "status_code": status_code,
+        "duration_seconds": round(duration, 4) if duration else None,
+        "timestamp": datetime.now().isoformat()
+    }
+    
+    logger.info(f"API_CALL: {json.dumps(api_data)}")
+
+def log_security_event(event_type: str, details: Dict[str, Any] = None, severity: str = "INFO") -> None:
+    """Log security-related events"""
+    logger = get_logger()
+    security_data = {
+        "event_type": event_type,
+        "severity": severity,
+        "timestamp": datetime.now().isoformat()
     }
     
     if details:
-        activity_log["details"] = details
+        security_data.update(details)
+    
+    level = getattr(logging, severity.upper(), logging.INFO)
+    logger.log(level, f"SECURITY: {json.dumps(security_data)}")
+
+def log_system_health(component: str, status: str, metrics: Dict[str, Any] = None) -> None:
+    """Log system health checks"""
+    logger = get_logger()
+    health_data = {
+        "component": component,
+        "status": status,
+        "timestamp": datetime.now().isoformat()
+    }
     
-    logger.info(f"User activity: {activity_log}")
-
-
-def log_performance(operation, duration_ms):
-    """
-    Log performance metrics for monitoring.
+    if metrics:
+        health_data.update(metrics)
     
-    Args:
-        operation (str): The operation being measured.
-        duration_ms (float): The duration of the operation in milliseconds.
-    """
-    logger.info(f"Performance: {operation} took {duration_ms:.2f}ms")
-
+    level = logging.INFO if status == "healthy" else logging.WARNING
+    logger.log(level, f"HEALTH: {json.dumps(health_data)}")
 
-def log_ai_model_usage(model_name: str, operation: str, tokens_used: Optional[int] = None):
-    """
-    Log AI model usage for monitoring and quota management.
-    
-    Args:
-        model_name (str): The name of the AI model used.
-        operation (str): The operation performed with the model.
-        tokens_used (int, optional): The number of tokens used in the operation.
-    """
-    usage_log = {
-        "timestamp": datetime.now().isoformat(),
+def log_ai_model_usage(model_name: str, operation: str, tokens_used: Optional[int] = None) -> None:
+    """Log AI model usage for monitoring and billing"""
+    logger = get_logger()
+    ai_data = {
         "model": model_name,
         "operation": operation,
+        "tokens_used": tokens_used,
+        "timestamp": datetime.now().isoformat()
     }
     
-    if tokens_used is not None:
-        usage_log["tokens_used"] = tokens_used
+    logger.info(f"AI_USAGE: {json.dumps(ai_data)}")
+
+def log_business_metric(metric_name: str, value: float, unit: str = None, tags: Dict[str, str] = None) -> None:
+    """Log business metrics for analytics"""
+    logger = get_logger()
+    metric_data = {
+        "metric": metric_name,
+        "value": value,
+        "unit": unit,
+        "timestamp": datetime.now().isoformat()
+    }
     
-    logger.info(f"AI model usage: {usage_log}")
+    if tags:
+        metric_data["tags"] = tags
+    
+    logger.info(f"METRIC: {json.dumps(metric_data)}")
+
+# Context manager for performance logging
+class PerformanceLogger:
+    def __init__(self, operation: str, additional_info: Dict[str, Any] = None):
+        self.operation = operation
+        self.additional_info = additional_info or {}
+        self.start_time = None
+    
+    def __enter__(self):
+        self.start_time = datetime.now()
+        return self
+    
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if self.start_time:
+            duration = (datetime.now() - self.start_time).total_seconds()
+            success = exc_type is None
+            
+            perf_info = self.additional_info.copy()
+            perf_info["success"] = success
+            
+            if not success and exc_val:
+                perf_info["error"] = str(exc_val)
+            
+            log_performance(self.operation, duration, perf_info)
+
+# Decorator for automatic performance logging
+def log_performance_decorator(operation_name: str = None):
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            op_name = operation_name or f"{func.__module__}.{func.__name__}"
+            with PerformanceLogger(op_name):
+                return func(*args, **kwargs)
+        return wrapper
+    return decorator
+
+# Initialize logger on module import
+get_logger().info("Logging system initialized")