from django.utils import timezone from django.db.models import Q, Sum, Count, Avg, F from datetime import datetime, timedelta from decimal import Decimal import json class DashboardAnalytics: """Advanced analytics for dashboard""" @staticmethod def get_customer_satisfaction_score(): """Calculate customer satisfaction based on ticket resolution""" try: from tickets.models import Ticket from django.db.models import Case, When, IntegerField # Calculate satisfaction based on resolution time and ratings tickets = Ticket.objects.filter( status='closed', completed_at__isnull=False, created_at__gte=timezone.now() - timedelta(days=30) ).annotate( resolution_hours=F('completed_at') - F('created_at'), satisfaction_score=Case( When(resolution_hours__lt=timedelta(hours=24), then=5), When(resolution_hours__lt=timedelta(hours=48), then=4), When(resolution_hours__lt=timedelta(hours=72), then=3), When(resolution_hours__lt=timedelta(hours=168), then=2), default=1, output_field=IntegerField() ) ) avg_score = tickets.aggregate( avg_satisfaction=Avg('satisfaction_score') )['avg_satisfaction'] return round(avg_score or 0, 1) except ImportError: return 0 @staticmethod def get_revenue_forecast(): """Predict next month revenue based on trends""" from billing.models import Invoice from customers.models import Customer # Get last 3 months revenue revenues = [] for i in range(3): month_start = (timezone.now() - timedelta(days=30*i)).replace(day=1) month_end = (month_start + timedelta(days=32)).replace(day=1) - timedelta(days=1) revenue = Invoice.objects.filter( status='paid', updated_at__range=[month_start, month_end] ).aggregate(total=Sum('total_amount'))['total'] or 0 revenues.append(float(revenue)) # Simple linear forecast if len(revenues) >= 2: trend = (revenues[0] - revenues[1]) / 1 # Monthly change forecast = revenues[0] + trend return max(0, forecast) # Don't predict negative revenue return revenues[0] if revenues else 0 @staticmethod def get_service_performance(): """Analyze service package performance""" from customers.models import Service, Customer from billing.models import Invoice services = Service.objects.annotate( customer_count=Count('customer'), monthly_revenue=Sum( 'customer__invoices__total_amount', filter=Q( customer__invoices__status='paid', customer__invoices__issue_date__month=timezone.now().month, customer__invoices__issue_date__year=timezone.now().year ) ), churn_count=Count( 'customer', filter=Q(customer__is_active=False) ) ).filter(customer_count__gt=0) performance_data = [] for service in services: churn_rate = (service.churn_count / service.customer_count * 100) if service.customer_count > 0 else 0 performance_data.append({ 'name': service.name, 'customers': service.customer_count, 'revenue': float(service.monthly_revenue or 0), 'churn_rate': round(churn_rate, 2), 'avg_revenue_per_customer': float((service.monthly_revenue or 0) / service.customer_count) if service.customer_count > 0 else 0 }) return sorted(performance_data, key=lambda x: x['revenue'], reverse=True) @staticmethod def get_geographic_distribution(): """Get customer distribution by location""" from customers.models import Customer # This is a simplified version - you might want to integrate with actual geo-location customers_by_area = Customer.objects.filter(is_active=True).values( 'physical_address' ).annotate( customer_count=Count('id') ).order_by('-customer_count')[:10] return list(customers_by_area) @staticmethod def get_payment_health(): """Analyze payment patterns and health""" from billing.models import Invoice from customers.models import Customer # Payment timing analysis current_month = timezone.now().replace(day=1) # Get invoice statistics for current month all_invoices = Invoice.objects.filter(issue_date__gte=current_month) invoice_stats = all_invoices.aggregate( total_invoices=Count('id'), paid_invoices=Count('id', filter=Q(status='paid')), overdue_invoices=Count('id', filter=Q( status='unpaid', due_date__lt=timezone.now().date() )), total_amount=Sum('total_amount'), paid_amount=Sum('total_amount', filter=Q(status='paid')) ) # Calculate payment health metrics total = invoice_stats['total_invoices'] or 1 paid_rate = (invoice_stats['paid_invoices'] or 0) / total * 100 overdue_rate = (invoice_stats['overdue_invoices'] or 0) / total * 100 total_amount = invoice_stats['total_amount'] or Decimal('0') collection_rate = float((invoice_stats['paid_amount'] or Decimal('0')) / total_amount * 100) if total_amount > 0 else 0 return { 'payment_rate': round(paid_rate, 1), 'overdue_rate': round(overdue_rate, 1), 'collection_rate': round(collection_rate, 1), 'total_invoices': invoice_stats['total_invoices'], 'health_score': round((paid_rate + (100 - overdue_rate)) / 2, 1) } class DashboardSecurity: """Security features for dashboard""" @staticmethod def check_session_security(request): """Check session security and return warnings""" warnings = [] # Check session age if hasattr(request.user, 'last_login') and request.user.last_login: session_age = timezone.now() - request.user.last_login if session_age > timedelta(hours=8): warnings.append({ 'type': 'session_age', 'message': 'Long session detected. Consider re-authenticating.', 'severity': 'warning' }) # Check for suspicious activity user_agent = request.META.get('HTTP_USER_AGENT', '') if any(bot in user_agent.lower() for bot in ['bot', 'crawler', 'spider']): warnings.append({ 'type': 'suspicious_agent', 'message': 'Unusual user agent detected.', 'severity': 'high' }) return warnings @staticmethod def get_recent_login_attempts(): """Get recent login attempts for security monitoring""" from accounts.models import UserActivityLog recent_attempts = UserActivityLog.objects.filter( action__in=['login', 'failed_login'], timestamp__gte=timezone.now() - timedelta(hours=24) ).values( 'action', 'ip_address', 'timestamp', 'user__username' ).order_by('-timestamp')[:20] return list(recent_attempts) class DashboardCache: """Caching utilities for dashboard performance""" @staticmethod def get_cache_key(user, metric_type): """Generate cache key for user-specific metrics""" return f"dashboard_{user.id}_{metric_type}_{timezone.now().strftime('%Y%m%d_%H')}" @staticmethod def should_refresh_cache(cache_key, max_age_minutes=30): """Check if cache should be refreshed""" from django.core.cache import cache last_update = cache.get(f"{cache_key}_timestamp") if not last_update: return True age = timezone.now() - last_update return age > timedelta(minutes=max_age_minutes)