File size: 2,933 Bytes
fc3fdb8 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 |
import json
import time
import logging
from typing import Dict, Any, Optional
from datetime import datetime
from core.redis_client import redis_client
logger = logging.getLogger(__name__)
class UserLogger:
"""Comprehensive user interaction logging system"""
def __init__(self):
self.redis_client = redis_client.get_client()
def log_user_action(self, user_id: str, action: str, details: Dict[str, Any] = None):
"""Log a user action"""
try:
event_data = {
"user_id": user_id,
"action": action,
"timestamp": datetime.now().isoformat(),
"details": details or {}
}
# Store in Redis with expiration (30 days)
key = f"analytics:user_events:{user_id}:{int(time.time())}"
self.redis_client.setex(
key,
2592000, # 30 days in seconds
json.dumps(event_data)
)
# Also store in global events
global_key = f"analytics:global_events:{int(time.time())}"
self.redis_client.setex(
global_key,
2592000, # 30 days in seconds
json.dumps(event_data)
)
logger.info(f"Logged user action: {user_id} - {action}")
except Exception as e:
logger.error(f"Failed to log user action: {e}")
def log_ui_state(self, user_id: str, state: str, details: Dict[str, Any] = None):
"""Log UI state changes"""
self.log_user_action(user_id, f"ui_state_{state}", details)
def log_performance_metric(self, user_id: str, metric_name: str, value: float, details: Dict[str, Any] = None):
"""Log performance metrics"""
metric_details = {
"value": value,
"metric": metric_name
}
if details:
metric_details.update(details)
self.log_user_action(user_id, f"performance_{metric_name}", metric_details)
def log_error(self, user_id: str, error_type: str, error_message: str, details: Dict[str, Any] = None):
"""Log errors"""
error_details = {
"error_type": error_type,
"error_message": error_message
}
if details:
error_details.update(details)
self.log_user_action(user_id, f"error_{error_type}", error_details)
def log_feedback(self, user_id: str, rating: int, comment: str = "", details: Dict[str, Any] = None):
"""Log user feedback"""
feedback_details = {
"rating": rating,
"comment": comment
}
if details:
feedback_details.update(details)
self.log_user_action(user_id, "user_feedback", feedback_details)
# Global instance
user_logger = UserLogger()
|