from __future__ import annotations from datetime import date, timedelta from typing import Any from django.db.models import Avg, Count, Q from django.db.models.functions import TruncDate from django.http import Http404, HttpRequest, HttpResponse from django.shortcuts import get_object_or_404 from django.utils import timezone from django.views import View from inertia import render from labs.models import Exam, ExamAttempt, ExamVariant from .base import PermissionRequiredAdminMixin def parse_date_range(request: HttpRequest) -> tuple[date, date]: """Parse and validate date range from request parameters. Returns (date_from, date_to) tuple. Defaults to last 30 days. """ today = timezone.now().date() # Check for preset range (7, 30, 90 days) days_str = request.GET.get("days") if days_str and days_str.isdigit(): days = int(days_str) if days in (7, 30, 90): return today - timedelta(days=days - 1), today # Check for custom date range date_from_str = request.GET.get("date_from") date_to_str = request.GET.get("date_to") if date_from_str and date_to_str: try: date_from = date.fromisoformat(date_from_str) date_to = date.fromisoformat(date_to_str) # Limit to max 365 days to prevent heavy queries if (date_to - date_from).days > 365: date_from = date_to - timedelta(days=365) return date_from, date_to except ValueError: pass # Default: last 30 days return today - timedelta(days=29), today def fill_date_gaps(data: dict[date, Any], date_from: date, date_to: date) -> list[date]: """Generate a complete list of dates between date_from and date_to.""" dates = [] current = date_from while current <= date_to: dates.append(current) current += timedelta(days=1) return dates class ExamAttemptsView(PermissionRequiredAdminMixin, View): """View for exam attempts analytics and list.""" permission_required = ["labs.view_exam", "labs.view_examattempt"] component = "exams/attempts/Index" def get(self, request: HttpRequest, pk: int) -> HttpResponse: exam = get_object_or_404(Exam, pk=pk) date_from, date_to = parse_date_range(request) # Parse filters variant_id = request.GET.get("variant_id") status = request.GET.get("status", "all") result = request.GET.get("result", "all") search_query = request.GET.get("q", "").strip() page = int(request.GET.get("page", 1)) page_size = 20 # Base queryset for this exam base_qs = ExamAttempt.objects.filter(exam=exam) # Apply date filter based on started_at datetime_from = timezone.make_aware( timezone.datetime.combine(date_from, timezone.datetime.min.time()) ) datetime_to = timezone.make_aware( timezone.datetime.combine(date_to, timezone.datetime.max.time()) ) date_filtered_qs = base_qs.filter( started_at__gte=datetime_from, started_at__lte=datetime_to ) # Apply variant filter if variant_id and variant_id.isdigit(): date_filtered_qs = date_filtered_qs.filter(variant_id=int(variant_id)) # Apply status filter if status == "in_progress": date_filtered_qs = date_filtered_qs.filter( status=ExamAttempt.Status.IN_PROGRESS ) elif status == "completed": date_filtered_qs = date_filtered_qs.filter( status=ExamAttempt.Status.COMPLETED ) elif status == "expired": date_filtered_qs = date_filtered_qs.filter(status=ExamAttempt.Status.EXPIRED) # Apply result filter if result == "passed": date_filtered_qs = date_filtered_qs.filter( submitted_at__isnull=False, is_passed=True ) elif result == "not_passed": date_filtered_qs = date_filtered_qs.filter( submitted_at__isnull=False, is_passed=False ) # Summary stats (from filtered queryset) now = timezone.now() summary = self._compute_summary(date_filtered_qs, now) # Charts data (from filtered queryset, excluding search) charts = self._compute_charts(date_filtered_qs, date_from, date_to) # Variant breakdown (from filtered queryset, excluding variant filter and search) breakdown_qs = base_qs.filter( started_at__gte=datetime_from, started_at__lte=datetime_to ) if status == "in_progress": breakdown_qs = breakdown_qs.filter(status=ExamAttempt.Status.IN_PROGRESS) elif status == "completed": breakdown_qs = breakdown_qs.filter(status=ExamAttempt.Status.COMPLETED) elif status == "expired": breakdown_qs = breakdown_qs.filter(status=ExamAttempt.Status.EXPIRED) if result == "passed": breakdown_qs = breakdown_qs.filter(submitted_at__isnull=False, is_passed=True) elif result == "not_passed": breakdown_qs = breakdown_qs.filter( submitted_at__isnull=False, is_passed=False ) variant_breakdown = self._compute_variant_breakdown(breakdown_qs, exam) # Attempts list with search and pagination list_qs = date_filtered_qs.select_related("user", "variant").order_by( "-started_at" ) if search_query: # Search by email, user ID, or attempt ID search_conditions = Q(user__email__icontains=search_query) if search_query.isdigit(): search_conditions |= Q(id=int(search_query)) | Q( user_id=int(search_query) ) list_qs = list_qs.filter(search_conditions) total = list_qs.count() total_pages = (total + page_size - 1) // page_size page = max(1, min(page, total_pages or 1)) offset = (page - 1) * page_size attempts_data = list(list_qs[offset : offset + page_size]) # Serialize variants for filter dropdown variants = exam.variants.values("id", "title", "language_code").order_by( "language_code" ) return render( request, self.component, { "exam": {"id": exam.id, "code": exam.code}, "variants": list(variants), "filters": { "date_from": date_from.isoformat(), "date_to": date_to.isoformat(), "days": request.GET.get("days", "30"), "variant_id": variant_id, "status": status, "result": result, "q": search_query, }, "summary": summary, "charts": charts, "variant_breakdown": variant_breakdown, "attempts": { "data": [ self._serialize_attempt(a, now) for a in attempts_data ], "pagination": { "page": page, "page_size": page_size, "total": total, "total_pages": total_pages, }, }, }, ) def _compute_summary(self, qs, now) -> dict[str, Any]: """Compute summary statistics.""" # Basic counts attempts_started = qs.count() participants = qs.values("user_id").distinct().count() submitted_qs = qs.filter(submitted_at__isnull=False) submitted = submitted_qs.count() passed = submitted_qs.filter(is_passed=True).count() pass_rate_pct = round(passed / submitted * 100, 1) if submitted > 0 else None # Active now: IN_PROGRESS, not submitted, not expired active_now = qs.filter( status=ExamAttempt.Status.IN_PROGRESS, submitted_at__isnull=True, expires_at__gte=now, ).count() # Overdue pending: IN_PROGRESS, not submitted, expired overdue_pending = qs.filter( status=ExamAttempt.Status.IN_PROGRESS, submitted_at__isnull=True, expires_at__lt=now, ).count() return { "attempts_started": attempts_started, "participants": participants, "submitted": submitted, "passed": passed, "pass_rate_pct": pass_rate_pct, "active_now": active_now, "overdue_pending": overdue_pending, } def _compute_charts( self, qs, date_from: date, date_to: date ) -> dict[str, Any]: """Compute chart data.""" # Participation trend (by started_at) participation_data = ( qs.annotate(date=TruncDate("started_at")) .values("date") .annotate( attempts=Count("id"), participants=Count("user_id", distinct=True), ) .order_by("date") ) participation_by_date = { item["date"]: item for item in participation_data if item["date"] } # Pass rate trend (by submitted_at) pass_rate_data = ( qs.filter(submitted_at__isnull=False) .annotate(date=TruncDate("submitted_at")) .values("date") .annotate( submitted=Count("id"), passed=Count("id", filter=Q(is_passed=True)), ) .order_by("date") ) pass_rate_by_date = { item["date"]: item for item in pass_rate_data if item["date"] } # Fill in missing dates all_dates = fill_date_gaps({}, date_from, date_to) participation = { "dates": [d.isoformat() for d in all_dates], "attempts": [ participation_by_date.get(d, {}).get("attempts", 0) for d in all_dates ], "participants": [ participation_by_date.get(d, {}).get("participants", 0) for d in all_dates ], } pass_rate = { "dates": [d.isoformat() for d in all_dates], "submitted": [ pass_rate_by_date.get(d, {}).get("submitted", 0) for d in all_dates ], "passed": [ pass_rate_by_date.get(d, {}).get("passed", 0) for d in all_dates ], "pass_rate_pct": [], } for d in all_dates: item = pass_rate_by_date.get(d, {}) submitted = item.get("submitted", 0) passed = item.get("passed", 0) if submitted > 0: pass_rate["pass_rate_pct"].append(round(passed / submitted * 100, 1)) else: pass_rate["pass_rate_pct"].append(None) return {"participation": participation, "pass_rate": pass_rate} def _compute_variant_breakdown(self, qs, exam: Exam) -> list[dict[str, Any]]: """Compute per-variant breakdown statistics.""" breakdown_data = ( qs.values("variant_id") .annotate( started=Count("id"), submitted=Count("id", filter=Q(submitted_at__isnull=False)), passed=Count("id", filter=Q(is_passed=True)), avg_score=Avg("score", filter=Q(submitted_at__isnull=False)), ) .order_by("variant_id") ) # Get variant info variants = { v.id: v for v in exam.variants.only("id", "title", "language_code", "total_score") } result = [] for item in breakdown_data: variant = variants.get(item["variant_id"]) if not variant: continue submitted = item["submitted"] passed = item["passed"] pass_rate_pct = round(passed / submitted * 100, 1) if submitted > 0 else None avg_score = round(item["avg_score"], 1) if item["avg_score"] else None result.append( { "variant_id": variant.id, "variant_title": variant.title, "language_code": variant.language_code, "total_score": variant.total_score, "started": item["started"], "submitted": submitted, "passed": passed, "pass_rate_pct": pass_rate_pct, "avg_score": avg_score, } ) return result def _serialize_attempt(self, attempt: ExamAttempt, now) -> dict[str, Any]: """Serialize an attempt for the list view.""" # Determine if overdue (IN_PROGRESS, not submitted, expired) is_overdue = ( attempt.status == ExamAttempt.Status.IN_PROGRESS and attempt.submitted_at is None and attempt.expires_at and attempt.expires_at < now ) return { "id": attempt.id, "user": { "id": attempt.user_id, "email": attempt.user.email, }, "variant": { "id": attempt.variant_id, "title": attempt.variant.title, "language_code": attempt.variant.language_code, "total_score": attempt.variant.total_score, }, "status": { ExamAttempt.Status.IN_PROGRESS: "in_progress", ExamAttempt.Status.COMPLETED: "completed", ExamAttempt.Status.EXPIRED: "expired", }.get(attempt.status, "unknown"), "is_passed": attempt.is_passed if attempt.submitted_at else None, "score": attempt.score if attempt.submitted_at else None, "started_at": attempt.started_at.isoformat() if attempt.started_at else None, "submitted_at": ( attempt.submitted_at.isoformat() if attempt.submitted_at else None ), "is_overdue": is_overdue, } class ExamVariantAttemptsView(PermissionRequiredAdminMixin, View): """View for variant-specific attempts analytics and list.""" permission_required = ["labs.view_examvariant", "labs.view_examattempt"] component = "exams/variants/attempts/Index" def get( self, request: HttpRequest, exam_id: int, variant_id: int ) -> HttpResponse: variant = get_object_or_404(ExamVariant, pk=variant_id) if variant.exam_id != exam_id: raise Http404("Variant does not belong to this exam") date_from, date_to = parse_date_range(request) # Parse filters (no variant filter since we're already scoped to one) status = request.GET.get("status", "all") result = request.GET.get("result", "all") search_query = request.GET.get("q", "").strip() page = int(request.GET.get("page", 1)) page_size = 20 # Base queryset for this variant base_qs = ExamAttempt.objects.filter(variant=variant) # Apply date filter datetime_from = timezone.make_aware( timezone.datetime.combine(date_from, timezone.datetime.min.time()) ) datetime_to = timezone.make_aware( timezone.datetime.combine(date_to, timezone.datetime.max.time()) ) date_filtered_qs = base_qs.filter( started_at__gte=datetime_from, started_at__lte=datetime_to ) # Apply status filter if status == "in_progress": date_filtered_qs = date_filtered_qs.filter( status=ExamAttempt.Status.IN_PROGRESS ) elif status == "completed": date_filtered_qs = date_filtered_qs.filter( status=ExamAttempt.Status.COMPLETED ) elif status == "expired": date_filtered_qs = date_filtered_qs.filter(status=ExamAttempt.Status.EXPIRED) # Apply result filter if result == "passed": date_filtered_qs = date_filtered_qs.filter( submitted_at__isnull=False, is_passed=True ) elif result == "not_passed": date_filtered_qs = date_filtered_qs.filter( submitted_at__isnull=False, is_passed=False ) # Summary stats now = timezone.now() summary = self._compute_summary(date_filtered_qs, now) # Charts data charts = self._compute_charts(date_filtered_qs, date_from, date_to) # Attempts list with search and pagination list_qs = date_filtered_qs.select_related("user").order_by("-started_at") if search_query: search_conditions = Q(user__email__icontains=search_query) if search_query.isdigit(): search_conditions |= Q(id=int(search_query)) | Q( user_id=int(search_query) ) list_qs = list_qs.filter(search_conditions) total = list_qs.count() total_pages = (total + page_size - 1) // page_size page = max(1, min(page, total_pages or 1)) offset = (page - 1) * page_size attempts_data = list(list_qs[offset : offset + page_size]) return render( request, self.component, { "exam": {"id": variant.exam_id, "code": variant.exam.code}, "variant": { "id": variant.id, "title": variant.title, "language_code": variant.language_code, "duration_seconds": variant.duration_seconds, "pass_score": variant.pass_score, "total_score": variant.total_score, }, "filters": { "date_from": date_from.isoformat(), "date_to": date_to.isoformat(), "days": request.GET.get("days", "30"), "status": status, "result": result, "q": search_query, }, "summary": summary, "charts": charts, "attempts": { "data": [ self._serialize_attempt(a, variant, now) for a in attempts_data ], "pagination": { "page": page, "page_size": page_size, "total": total, "total_pages": total_pages, }, }, }, ) def _compute_summary(self, qs, now) -> dict[str, Any]: """Compute summary statistics.""" attempts_started = qs.count() participants = qs.values("user_id").distinct().count() submitted_qs = qs.filter(submitted_at__isnull=False) submitted = submitted_qs.count() passed = submitted_qs.filter(is_passed=True).count() pass_rate_pct = round(passed / submitted * 100, 1) if submitted > 0 else None active_now = qs.filter( status=ExamAttempt.Status.IN_PROGRESS, submitted_at__isnull=True, expires_at__gte=now, ).count() overdue_pending = qs.filter( status=ExamAttempt.Status.IN_PROGRESS, submitted_at__isnull=True, expires_at__lt=now, ).count() return { "attempts_started": attempts_started, "participants": participants, "submitted": submitted, "passed": passed, "pass_rate_pct": pass_rate_pct, "active_now": active_now, "overdue_pending": overdue_pending, } def _compute_charts( self, qs, date_from: date, date_to: date ) -> dict[str, Any]: """Compute chart data.""" participation_data = ( qs.annotate(date=TruncDate("started_at")) .values("date") .annotate( attempts=Count("id"), participants=Count("user_id", distinct=True), ) .order_by("date") ) participation_by_date = { item["date"]: item for item in participation_data if item["date"] } pass_rate_data = ( qs.filter(submitted_at__isnull=False) .annotate(date=TruncDate("submitted_at")) .values("date") .annotate( submitted=Count("id"), passed=Count("id", filter=Q(is_passed=True)), ) .order_by("date") ) pass_rate_by_date = { item["date"]: item for item in pass_rate_data if item["date"] } all_dates = fill_date_gaps({}, date_from, date_to) participation = { "dates": [d.isoformat() for d in all_dates], "attempts": [ participation_by_date.get(d, {}).get("attempts", 0) for d in all_dates ], "participants": [ participation_by_date.get(d, {}).get("participants", 0) for d in all_dates ], } pass_rate = { "dates": [d.isoformat() for d in all_dates], "submitted": [ pass_rate_by_date.get(d, {}).get("submitted", 0) for d in all_dates ], "passed": [ pass_rate_by_date.get(d, {}).get("passed", 0) for d in all_dates ], "pass_rate_pct": [], } for d in all_dates: item = pass_rate_by_date.get(d, {}) submitted = item.get("submitted", 0) passed = item.get("passed", 0) if submitted > 0: pass_rate["pass_rate_pct"].append(round(passed / submitted * 100, 1)) else: pass_rate["pass_rate_pct"].append(None) return {"participation": participation, "pass_rate": pass_rate} def _serialize_attempt( self, attempt: ExamAttempt, variant: ExamVariant, now ) -> dict[str, Any]: """Serialize an attempt for the list view.""" is_overdue = ( attempt.status == ExamAttempt.Status.IN_PROGRESS and attempt.submitted_at is None and attempt.expires_at and attempt.expires_at < now ) return { "id": attempt.id, "user": { "id": attempt.user_id, "email": attempt.user.email, }, "status": { ExamAttempt.Status.IN_PROGRESS: "in_progress", ExamAttempt.Status.COMPLETED: "completed", ExamAttempt.Status.EXPIRED: "expired", }.get(attempt.status, "unknown"), "is_passed": attempt.is_passed if attempt.submitted_at else None, "score": attempt.score if attempt.submitted_at else None, "total_score": variant.total_score, "started_at": attempt.started_at.isoformat() if attempt.started_at else None, "submitted_at": ( attempt.submitted_at.isoformat() if attempt.submitted_at else None ), "is_overdue": is_overdue, }