utils/analytics_logger.py

from typing import Dict, Any, Optional
import json
import time
from datetime import datetime
from utils.log_manager import LogManager

class AnalyticsLogger:
    """Handles logging of analytics events and metrics"""
    
    def __init__(self):
        self.log_manager = LogManager()
        self.logger = self.log_manager.get_analytics_logger("events")
        self.metrics_logger = self.log_manager.get_analytics_logger("metrics")
        
    def log_user_interaction(self, 
                           user_id: str,
                           interaction_type: str,
                           agent_type: str,
                           duration: float,
                           success: bool,
                           details: Optional[Dict] = None):
        """Log user interaction events"""
        event = {
            "event_type": "user_interaction",
            "user_id": user_id,
            "interaction_type": interaction_type,
            "agent_type": agent_type,
            "duration": duration,
            "success": success,
            "timestamp": datetime.now().isoformat(),
            "details": details or {}
        }
        
        self.logger.info(f"User Interaction: {json.dumps(event, indent=2)}")
        
    def log_agent_performance(self,
                            agent_type: str,
                            operation: str,
                            response_time: float,
                            success: bool,
                            error: Optional[str] = None):
        """Log agent performance metrics"""
        metric = {
            "metric_type": "agent_performance",
            "agent_type": agent_type,
            "operation": operation,
            "response_time": response_time,
            "success": success,
            "error": error,
            "timestamp": datetime.now().isoformat()
        }
        
        self.metrics_logger.info(f"Agent Performance: {json.dumps(metric, indent=2)}")
        
    def log_system_health(self,
                         cpu_usage: float,
                         memory_usage: float,
                         active_users: int,
                         active_sessions: int):
        """Log system health metrics"""
        metric = {
            "metric_type": "system_health",
            "cpu_usage": cpu_usage,
            "memory_usage": memory_usage,
            "active_users": active_users,
            "active_sessions": active_sessions,
            "timestamp": datetime.now().isoformat()
        }
        
        self.metrics_logger.info(f"System Health: {json.dumps(metric, indent=2)}")
        
    def log_error(self,
                  error_type: str,
                  error_message: str,
                  severity: str,
                  context: Optional[Dict] = None):
        """Log error events"""
        event = {
            "event_type": "error",
            "error_type": error_type,
            "error_message": error_message,
            "severity": severity,
            "context": context or {},
            "timestamp": datetime.now().isoformat()
        }
        
        self.logger.error(f"Error Event: {json.dumps(event, indent=2)}")
        
    def log_security_event(self,
                          event_type: str,
                          user_id: str,
                          success: bool,
                          details: Optional[Dict] = None):
        """Log security-related events"""
        event = {
            "event_type": "security",
            "security_event_type": event_type,
            "user_id": user_id,
            "success": success,
            "details": details or {},
            "timestamp": datetime.now().isoformat()
        }
        
        self.logger.info(f"Security Event: {json.dumps(event, indent=2)}")
        
    def log_model_performance(self,
                            model_name: str,
                            operation: str,
                            input_tokens: int,
                            output_tokens: int,
                            response_time: float,
                            success: bool):
        """Log AI model performance metrics"""
        metric = {
            "metric_type": "model_performance",
            "model_name": model_name,
            "operation": operation,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "response_time": response_time,
            "success": success,
            "timestamp": datetime.now().isoformat()
        }
        
        self.metrics_logger.info(f"Model Performance: {json.dumps(metric, indent=2)}")
        
    def log_user_feedback(self,
                         user_id: str,
                         interaction_id: str,
                         rating: int,
                         feedback_text: Optional[str] = None):
        """Log user feedback"""
        event = {
            "event_type": "user_feedback",
            "user_id": user_id,
            "interaction_id": interaction_id,
            "rating": rating,
            "feedback_text": feedback_text,
            "timestamp": datetime.now().isoformat()
        }
        
        self.logger.info(f"User Feedback: {json.dumps(event, indent=2)}")
        
    def log_session_metrics(self,
                          session_id: str,
                          user_id: str,
                          session_type: str,
                          start_time: str,
                          end_time: str,
                          metrics: Dict[str, Any]):
        """Log session-specific metrics"""
        session_data = {
            "metric_type": "session_metrics",
            "session_id": session_id,
            "user_id": user_id,
            "session_type": session_type,
            "start_time": start_time,
            "end_time": end_time,
            "duration": self._calculate_duration(start_time, end_time),
            "metrics": metrics,
            "timestamp": datetime.now().isoformat()
        }
        
        self.metrics_logger.info(f"Session Metrics: {json.dumps(session_data, indent=2)}")
        
    def _calculate_duration(self, start_time: str, end_time: str) -> float:
        """Calculate duration between two ISO format timestamps"""
        start = datetime.fromisoformat(start_time)
        end = datetime.fromisoformat(end_time)
        return (end - start).total_seconds()