""" User Analytics Management Tracks user registrations, analytics events, and predictions for admin dashboard """ import os import json from datetime import datetime from typing import Dict, List, Optional from collections import defaultdict class UserAnalytics: """Manage user analytics data storage and retrieval""" def __init__(self, data_dir: str = '../data'): self.data_dir = os.path.abspath(data_dir) os.makedirs(self.data_dir, exist_ok=True) self.users_file = os.path.join(self.data_dir, 'users.json') self.analytics_file = os.path.join(self.data_dir, 'analytics.json') self.predictions_file = os.path.join(self.data_dir, 'predictions.json') # Initialize files if they don't exist self._init_files() def _init_files(self): """Initialize JSON files with empty structures""" if not os.path.exists(self.users_file): self._save_file(self.users_file, { 'users': [], 'last_updated': None }) if not os.path.exists(self.analytics_file): self._save_file(self.analytics_file, { 'events': [], 'last_updated': None }) if not os.path.exists(self.predictions_file): self._save_file(self.predictions_file, { 'predictions': [], 'last_updated': None }) def _load_file(self, filepath: str) -> Dict: """Load JSON file""" try: if os.path.exists(filepath): with open(filepath, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"Error loading {filepath}: {e}") return {} def _save_file(self, filepath: str, data: Dict): """Save JSON file""" try: data['last_updated'] = datetime.now().isoformat() with open(filepath, 'w', encoding='utf-8') as f: json.dump(data, f, indent=2, ensure_ascii=False) except Exception as e: print(f"Error saving {filepath}: {e}") def register_user(self, user_data: Dict) -> str: """Register a new user""" data = self._load_file(self.users_file) users = data.get('users', []) # Add registration timestamp if not present if 'registration_date' not in user_data: user_data['registration_date'] = datetime.now().isoformat() # Add to users list users.append(user_data) data['users'] = users self._save_file(self.users_file, data) return user_data.get('user_id', 'unknown') def track_event(self, event_name: str, event_data: Dict): """Track an analytics event""" data = self._load_file(self.analytics_file) events = data.get('events', []) event = { 'name': event_name, 'timestamp': datetime.now().isoformat(), 'data': event_data } events.append(event) # Keep only last 1000 events to prevent file from growing too large if len(events) > 1000: events = events[-1000:] data['events'] = events self._save_file(self.analytics_file, data) def track_prediction(self, prediction_data: Dict): """Track a prediction""" data = self._load_file(self.predictions_file) predictions = data.get('predictions', []) # Add timestamp if not present if 'timestamp' not in prediction_data: prediction_data['timestamp'] = datetime.now().isoformat() predictions.append(prediction_data) # Keep only last 1000 predictions if len(predictions) > 1000: predictions = predictions[-1000:] data['predictions'] = predictions self._save_file(self.predictions_file, data) def get_user_stats(self) -> Dict: """Get user statistics""" data = self._load_file(self.users_file) users = data.get('users', []) if not users: return { 'total_users': 0, 'active_users_7d': 0, 'active_users_30d': 0, 'new_users_7d': 0, 'new_users_30d': 0 } now = datetime.now() active_7d = [] active_30d = [] new_7d = [] new_30d = [] for user in users: reg_date_str = user.get('registration_date', '') if reg_date_str: try: reg_date = datetime.fromisoformat(reg_date_str.replace('Z', '+00:00')) days_ago = (now - reg_date.replace(tzinfo=None)).days if days_ago <= 30: active_30d.append(user.get('user_id')) if days_ago <= 7: active_7d.append(user.get('user_id')) new_7d.append(user.get('user_id')) elif days_ago <= 30: new_30d.append(user.get('user_id')) except: pass return { 'total_users': len(users), 'active_users_7d': len(set(active_7d)), 'active_users_30d': len(set(active_30d)), 'new_users_7d': len(new_7d), 'new_users_30d': len(new_30d) + len(new_7d) } def get_platform_distribution(self) -> Dict: """Get platform distribution""" data = self._load_file(self.users_file) users = data.get('users', []) platforms = defaultdict(int) for user in users: platform = user.get('platform', 'unknown') platforms[platform] += 1 return dict(platforms) def get_language_distribution(self) -> Dict: """Get language distribution""" data = self._load_file(self.users_file) users = data.get('users', []) languages = defaultdict(int) for user in users: language = user.get('language', 'unknown') languages[language] += 1 return dict(languages) def get_user_timeline(self, days: int = 30) -> List[Dict]: """Get user registration timeline""" data = self._load_file(self.users_file) users = data.get('users', []) now = datetime.now() timeline = defaultdict(int) for user in users: reg_date_str = user.get('registration_date', '') if reg_date_str: try: reg_date = datetime.fromisoformat(reg_date_str.replace('Z', '+00:00')) days_ago = (now - reg_date.replace(tzinfo=None)).days if days_ago <= days: date_str = reg_date.strftime('%Y-%m-%d') timeline[date_str] += 1 except: pass # Convert to list of dicts sorted by date result = [{'date': date, 'count': count} for date, count in sorted(timeline.items())] return result def get_recent_users(self, limit: int = 10) -> List[Dict]: """Get recent user registrations""" data = self._load_file(self.users_file) users = data.get('users', []) # Sort by registration date (most recent first) sorted_users = sorted(users, key=lambda x: x.get('registration_date', ''), reverse=True) return sorted_users[:limit]