"""Management command to display user statistics and reports. This command provides various statistics about users in the system, including registration trends, activity metrics, and user demographics. """ import json from django.db.models import Q from datetime import timedelta from django.utils import timezone from django.contrib.auth import get_user_model from django.core.management.base import BaseCommand from apps.accounts.models import UserSession, UserProfile User = get_user_model() class Command(BaseCommand): """Management command to display user statistics.""" help = 'Display user statistics and reports' def add_arguments(self, parser): """Add command arguments.""" parser.add_argument( '--format', choices=['table', 'json', 'csv'], default='table', help='Output format (default: table)' ) parser.add_argument( '--days', type=int, default=30, help='Number of days to include in recent activity stats (default: 30)' ) parser.add_argument( '--detailed', action='store_true', help='Show detailed statistics' ) parser.add_argument( '--export', type=str, help='Export statistics to file' ) def handle(self, *args, **options): """Handle the command execution.""" format_type = options['format'] days = options['days'] detailed = options['detailed'] export_file = options['export'] # Collect statistics stats = self._collect_statistics(days, detailed) # Format and display output if format_type == 'json': output = self._format_json(stats) elif format_type == 'csv': output = self._format_csv(stats) else: output = self._format_table(stats) # Export to file if requested if export_file: with open(export_file, 'w') as f: f.write(output) self.stdout.write( self.style.SUCCESS(f'Statistics exported to {export_file}') ) else: self.stdout.write(output) def _collect_statistics(self, days, detailed): """Collect user statistics. Args: days: Number of days for recent activity detailed: Whether to include detailed stats Returns: Dictionary containing statistics """ now = timezone.now() cutoff_date = now - timedelta(days=days) # Basic user counts total_users = User.objects.count() active_users = User.objects.filter(is_active=True).count() staff_users = User.objects.filter(is_staff=True).count() superusers = User.objects.filter(is_superuser=True).count() verified_users = User.objects.filter(is_verified=True).count() # Recent activity recent_registrations = User.objects.filter( date_joined__gte=cutoff_date ).count() recent_logins = User.objects.filter( last_login__gte=cutoff_date ).count() # Session statistics total_sessions = UserSession.objects.count() active_sessions = UserSession.objects.filter(is_active=True).count() recent_sessions = UserSession.objects.filter( created_at__gte=cutoff_date ).count() stats = { 'basic': { 'total_users': total_users, 'active_users': active_users, 'inactive_users': total_users - active_users, 'staff_users': staff_users, 'superusers': superusers, 'verified_users': verified_users, 'unverified_users': total_users - verified_users, }, 'recent_activity': { 'days': days, 'recent_registrations': recent_registrations, 'recent_logins': recent_logins, 'recent_sessions': recent_sessions, }, 'sessions': { 'total_sessions': total_sessions, 'active_sessions': active_sessions, 'inactive_sessions': total_sessions - active_sessions, }, 'generated_at': now.isoformat(), } if detailed: stats.update(self._collect_detailed_statistics(days)) return stats def _collect_detailed_statistics(self, days): """Collect detailed statistics. Args: days: Number of days for recent activity Returns: Dictionary containing detailed statistics """ now = timezone.now() cutoff_date = now - timedelta(days=days) # Registration trends (last 7 days) registration_trend = [] for i in range(7): date = now - timedelta(days=i) start_date = date.replace(hour=0, minute=0, second=0, microsecond=0) end_date = start_date + timedelta(days=1) count = User.objects.filter( date_joined__gte=start_date, date_joined__lt=end_date ).count() registration_trend.append({ 'date': start_date.strftime('%Y-%m-%d'), 'registrations': count }) # User profiles with complete information complete_profiles = UserProfile.objects.exclude( Q(job_title='') & Q(company='') & Q(bio='') ).count() # Most common domains email_domains = User.objects.values_list('email', flat=True) domain_counts = {} for email in email_domains: if '@' in email: domain = email.split('@')[1].lower() domain_counts[domain] = domain_counts.get(domain, 0) + 1 top_domains = sorted( domain_counts.items(), key=lambda x: x[1], reverse=True )[:10] # User activity by hour (last 24 hours) activity_by_hour = [] for hour in range(24): start_time = now.replace( hour=hour, minute=0, second=0, microsecond=0 ) - timedelta(days=1) end_time = start_time + timedelta(hours=1) login_count = User.objects.filter( last_login__gte=start_time, last_login__lt=end_time ).count() activity_by_hour.append({ 'hour': hour, 'logins': login_count }) return { 'detailed': { 'registration_trend': registration_trend, 'complete_profiles': complete_profiles, 'incomplete_profiles': UserProfile.objects.count() - complete_profiles, 'top_email_domains': top_domains, 'activity_by_hour': activity_by_hour, } } def _format_table(self, stats): """Format statistics as a table. Args: stats: Statistics dictionary Returns: Formatted table string """ output = [] output.append('=' * 60) output.append('USER STATISTICS REPORT') output.append('=' * 60) output.append(f"Generated at: {stats['generated_at']}") output.append('') # Basic statistics output.append('BASIC STATISTICS') output.append('-' * 20) basic = stats['basic'] output.append(f"Total Users: {basic['total_users']:,}") output.append(f"Active Users: {basic['active_users']:,}") output.append(f"Inactive Users: {basic['inactive_users']:,}") output.append(f"Staff Users: {basic['staff_users']:,}") output.append(f"Superusers: {basic['superusers']:,}") output.append(f"Verified Users: {basic['verified_users']:,}") output.append(f"Unverified Users: {basic['unverified_users']:,}") output.append('') # Recent activity output.append(f"RECENT ACTIVITY (Last {stats['recent_activity']['days']} days)") output.append('-' * 30) recent = stats['recent_activity'] output.append(f"New Registrations: {recent['recent_registrations']:,}") output.append(f"Recent Logins: {recent['recent_logins']:,}") output.append(f"Recent Sessions: {recent['recent_sessions']:,}") output.append('') # Session statistics output.append('SESSION STATISTICS') output.append('-' * 20) sessions = stats['sessions'] output.append(f"Total Sessions: {sessions['total_sessions']:,}") output.append(f"Active Sessions: {sessions['active_sessions']:,}") output.append(f"Inactive Sessions: {sessions['inactive_sessions']:,}") output.append('') # Detailed statistics if 'detailed' in stats: detailed = stats['detailed'] output.append('DETAILED STATISTICS') output.append('-' * 20) output.append(f"Complete Profiles: {detailed['complete_profiles']:,}") output.append(f"Incomplete Profiles: {detailed['incomplete_profiles']:,}") output.append('') # Top email domains output.append('TOP EMAIL DOMAINS') output.append('-' * 18) for domain, count in detailed['top_email_domains']: output.append(f"{domain:<30} {count:>5}") output.append('') output.append('=' * 60) return '\n'.join(output) def _format_json(self, stats): """Format statistics as JSON. Args: stats: Statistics dictionary Returns: JSON formatted string """ return json.dumps(stats, indent=2, default=str) def _format_csv(self, stats): """Format statistics as CSV. Args: stats: Statistics dictionary Returns: CSV formatted string """ output = [] output.append('Metric,Value') # Basic statistics basic = stats['basic'] for key, value in basic.items(): output.append(f'{key},{value}') # Recent activity recent = stats['recent_activity'] for key, value in recent.items(): if key != 'days': output.append(f'{key},{value}') # Session statistics sessions = stats['sessions'] for key, value in sessions.items(): output.append(f'{key},{value}') return '\n'.join(output)