# -*- coding: utf-8 -*- """ Management command to generate activity reports. Usage: python manage.py generate_activity_report --period daily python manage.py generate_activity_report --period weekly --format csv python manage.py generate_activity_report --period monthly --output /path/to/report.json """ from django.core.management.base import BaseCommand, CommandError from django.utils import timezone from datetime import datetime, timedelta import json import csv import os import logging logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'Generate activity reports' def add_arguments(self, parser): parser.add_argument( '--period', choices=['daily', 'weekly', 'monthly', 'yearly', 'custom'], default='daily', help='Report period (default: daily)' ) parser.add_argument( '--start-date', type=str, help='Start date for custom period (YYYY-MM-DD)' ) parser.add_argument( '--end-date', type=str, help='End date for custom period (YYYY-MM-DD)' ) parser.add_argument( '--format', choices=['json', 'csv', 'txt'], default='json', help='Output format (default: json)' ) parser.add_argument( '--output', type=str, help='Output file path (default: stdout)' ) parser.add_argument( '--user-id', type=int, help='Generate report for specific user ID' ) parser.add_argument( '--action', type=str, help='Filter by specific action type' ) parser.add_argument( '--include-details', action='store_true', help='Include detailed activity information' ) parser.add_argument( '--top-users', type=int, default=10, help='Number of top active users to include (default: 10)' ) def handle(self, *args, **options): period = options['period'] start_date = options['start_date'] end_date = options['end_date'] output_format = options['format'] output_file = options['output'] user_id = options['user_id'] action_filter = options['action'] include_details = options['include_details'] top_users_count = options['top_users'] # Validate custom period if period == 'custom': if not start_date or not end_date: raise CommandError( 'Custom period requires both --start-date and --end-date' ) # Import here to avoid circular imports from apps.activities.services import ActivityAnalyticsService try: # Parse dates for custom period if period == 'custom': start_dt = datetime.strptime(start_date, '%Y-%m-%d') end_dt = datetime.strptime(end_date, '%Y-%m-%d') start_dt = timezone.make_aware(start_dt) end_dt = timezone.make_aware(end_dt.replace(hour=23, minute=59, second=59)) else: start_dt, end_dt = self._get_period_dates(period) self.stdout.write( self.style.SUCCESS( f'Generating {period} activity report from {start_dt.strftime("%Y-%m-%d")} ' f'to {end_dt.strftime("%Y-%m-%d")}' ) ) # Generate report data report_data = self._generate_report_data( start_dt, end_dt, user_id, action_filter, include_details, top_users_count ) # Output report if output_file: self._save_report_to_file(report_data, output_file, output_format) self.stdout.write( self.style.SUCCESS(f'Report saved to: {output_file}') ) else: self._output_report_to_stdout(report_data, output_format) logger.info( f'Activity report generated: {period} period, ' f'{report_data.get("total_activities", 0)} activities' ) except ValueError as e: raise CommandError(f'Invalid date format: {str(e)}') except Exception as e: logger.error(f'Error generating activity report: {str(e)}') raise CommandError(f'Error generating report: {str(e)}') def _get_period_dates(self, period): """Get start and end dates for the specified period.""" now = timezone.now() if period == 'daily': start = now.replace(hour=0, minute=0, second=0, microsecond=0) end = now.replace(hour=23, minute=59, second=59, microsecond=999999) elif period == 'weekly': days_since_monday = now.weekday() start = (now - timedelta(days=days_since_monday)).replace( hour=0, minute=0, second=0, microsecond=0 ) end = (start + timedelta(days=6)).replace( hour=23, minute=59, second=59, microsecond=999999 ) elif period == 'monthly': start = now.replace(day=1, hour=0, minute=0, second=0, microsecond=0) if now.month == 12: next_month = start.replace(year=now.year + 1, month=1) else: next_month = start.replace(month=now.month + 1) end = (next_month - timedelta(days=1)).replace( hour=23, minute=59, second=59, microsecond=999999 ) elif period == 'yearly': start = now.replace( month=1, day=1, hour=0, minute=0, second=0, microsecond=0 ) end = now.replace( month=12, day=31, hour=23, minute=59, second=59, microsecond=999999 ) return start, end def _generate_report_data(self, start_dt, end_dt, user_id, action_filter, include_details, top_users_count): """Generate comprehensive report data.""" from apps.activities.models import Activity from apps.activities.services import ActivityAnalyticsService from django.contrib.auth import get_user_model from django.db.models import Count, Q User = get_user_model() # Base queryset activities = Activity.objects.filter( timestamp__gte=start_dt, timestamp__lte=end_dt ) # Apply filters if user_id: activities = activities.filter(user_id=user_id) if action_filter: activities = activities.filter(action__icontains=action_filter) # Basic statistics total_activities = activities.count() unique_users = activities.values('user').distinct().count() # Activity breakdown by action action_stats = activities.values('action').annotate( count=Count('id') ).order_by('-count') # Top active users top_users = activities.values( 'user__id', 'user__email', 'user__first_name', 'user__last_name' ).annotate( activity_count=Count('id') ).order_by('-activity_count')[:top_users_count] # Daily activity distribution daily_stats = ActivityAnalyticsService.get_daily_activity_stats( start_date=start_dt.date(), end_date=end_dt.date() ) # Hourly distribution (for shorter periods) period_days = (end_dt - start_dt).days hourly_stats = None if period_days <= 7: # Only for week or less hourly_stats = activities.extra( select={'hour': 'EXTRACT(hour FROM timestamp)'} ).values('hour').annotate( count=Count('id') ).order_by('hour') # Build report data report_data = { 'report_info': { 'generated_at': timezone.now().isoformat(), 'period': { 'start': start_dt.isoformat(), 'end': end_dt.isoformat(), 'days': period_days }, 'filters': { 'user_id': user_id, 'action': action_filter } }, 'summary': { 'total_activities': total_activities, 'unique_users': unique_users, 'avg_activities_per_user': round( total_activities / unique_users if unique_users > 0 else 0, 2 ), 'avg_activities_per_day': round( total_activities / max(period_days, 1), 2 ) }, 'action_breakdown': list(action_stats), 'top_users': list(top_users), 'daily_stats': daily_stats } if hourly_stats: report_data['hourly_distribution'] = list(hourly_stats) # Include detailed activities if requested if include_details: detailed_activities = activities.select_related('user').values( 'id', 'user__email', 'action', 'timestamp', 'ip_address', 'user_agent', 'details' ).order_by('-timestamp')[:1000] # Limit to prevent huge outputs report_data['detailed_activities'] = list(detailed_activities) return report_data def _save_report_to_file(self, report_data, output_file, output_format): """Save report data to file.""" # Ensure directory exists os.makedirs(os.path.dirname(output_file), exist_ok=True) if output_format == 'json': with open(output_file, 'w', encoding='utf-8') as f: json.dump(report_data, f, indent=2, default=str) elif output_format == 'csv': with open(output_file, 'w', newline='', encoding='utf-8') as f: writer = csv.writer(f) # Write summary writer.writerow(['Summary']) for key, value in report_data['summary'].items(): writer.writerow([key.replace('_', ' ').title(), value]) writer.writerow([]) # Empty row # Write action breakdown writer.writerow(['Action Breakdown']) writer.writerow(['Action', 'Count']) for item in report_data['action_breakdown']: writer.writerow([item['action'], item['count']]) writer.writerow([]) # Empty row # Write top users writer.writerow(['Top Users']) writer.writerow(['Email', 'Name', 'Activity Count']) for user in report_data['top_users']: name = f"{user.get('user__first_name', '')} {user.get('user__last_name', '')}".strip() writer.writerow([ user.get('user__email', ''), name or 'N/A', user['activity_count'] ]) elif output_format == 'txt': with open(output_file, 'w', encoding='utf-8') as f: self._write_text_report(f, report_data) def _output_report_to_stdout(self, report_data, output_format): """Output report data to stdout.""" if output_format == 'json': self.stdout.write(json.dumps(report_data, indent=2, default=str)) elif output_format == 'txt': self._write_text_report(self.stdout, report_data) elif output_format == 'csv': # For CSV to stdout, just show a summary self.stdout.write('CSV format is better suited for file output.') self._write_text_report(self.stdout, report_data) def _write_text_report(self, output, report_data): """Write report in text format.""" output.write('\n' + '='*60) output.write('\nACTIVITY REPORT') output.write('\n' + '='*60) # Report info info = report_data['report_info'] output.write(f"\nGenerated: {info['generated_at']}") output.write(f"\nPeriod: {info['period']['start']} to {info['period']['end']}") output.write(f"\nDays: {info['period']['days']}") # Summary output.write('\n\nSUMMARY:') output.write('\n' + '-'*20) summary = report_data['summary'] output.write(f"\nTotal Activities: {summary['total_activities']:,}") output.write(f"\nUnique Users: {summary['unique_users']:,}") output.write(f"\nAvg Activities/User: {summary['avg_activities_per_user']}") output.write(f"\nAvg Activities/Day: {summary['avg_activities_per_day']}") # Action breakdown output.write('\n\nTOP ACTIONS:') output.write('\n' + '-'*20) for i, action in enumerate(report_data['action_breakdown'][:10], 1): output.write(f"\n{i:2d}. {action['action']}: {action['count']:,}") # Top users output.write('\n\nTOP USERS:') output.write('\n' + '-'*20) for i, user in enumerate(report_data['top_users'], 1): name = f"{user.get('user__first_name', '')} {user.get('user__last_name', '')}".strip() display_name = name or user.get('user__email', 'Unknown') output.write(f"\n{i:2d}. {display_name}: {user['activity_count']:,} activities") output.write('\n' + '='*60 + '\n')