250 lines
9.0 KiB
Python
250 lines
9.0 KiB
Python
"""
|
|
Reporting views: completion, progress, and quiz-attempt reports with CSV export.
|
|
|
|
All views accept optional query parameters:
|
|
- course_id (UUID)
|
|
- org_id (UUID — acts as department filter)
|
|
- date_from (ISO-8601 date)
|
|
- date_to (ISO-8601 date)
|
|
"""
|
|
import csv
|
|
from datetime import datetime, timezone
|
|
|
|
from django.http import StreamingHttpResponse
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
|
|
from tracking.models import Enrollment, PageProgress
|
|
from quizzes.models import QuizAttempt
|
|
|
|
|
|
def _parse_date(value):
|
|
if not value:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(value).replace(tzinfo=timezone.utc)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _enrollment_queryset(params):
|
|
qs = Enrollment.objects.select_related("user", "course")
|
|
if course_id := params.get("course_id"):
|
|
qs = qs.filter(course_id=course_id)
|
|
if org_id := params.get("org_id"):
|
|
qs = qs.filter(org_id=org_id)
|
|
if date_from := _parse_date(params.get("date_from")):
|
|
qs = qs.filter(enrolled_at__gte=date_from)
|
|
if date_to := _parse_date(params.get("date_to")):
|
|
qs = qs.filter(enrolled_at__lte=date_to)
|
|
return qs
|
|
|
|
|
|
class CompletionReportView(APIView):
|
|
"""Enrollment-level completion status report."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request):
|
|
qs = _enrollment_queryset(request.query_params)
|
|
rows = [
|
|
{
|
|
"enrollment_id": str(e.id),
|
|
"user_email": e.user.email,
|
|
"user_display_name": e.user.display_name,
|
|
"course_title": e.course.title,
|
|
"org_id": str(e.org_id),
|
|
"enrolled_at": e.enrolled_at.isoformat(),
|
|
"completed_at": e.completed_at.isoformat() if e.completed_at else None,
|
|
"is_complete": e.completed_at is not None,
|
|
}
|
|
for e in qs
|
|
]
|
|
return Response(rows)
|
|
|
|
|
|
class CompletionReportCSVView(APIView):
|
|
"""CSV export of the completion report."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
FIELDS = [
|
|
"enrollment_id", "user_email", "user_display_name",
|
|
"course_title", "org_id", "enrolled_at", "completed_at", "is_complete",
|
|
]
|
|
|
|
def get(self, request):
|
|
qs = _enrollment_queryset(request.query_params)
|
|
|
|
def _rows():
|
|
yield self.FIELDS
|
|
for e in qs.iterator(chunk_size=500):
|
|
yield [
|
|
str(e.id),
|
|
e.user.email,
|
|
e.user.display_name,
|
|
e.course.title,
|
|
str(e.org_id),
|
|
e.enrolled_at.isoformat(),
|
|
e.completed_at.isoformat() if e.completed_at else "",
|
|
"true" if e.completed_at else "false",
|
|
]
|
|
|
|
response = StreamingHttpResponse(_csv_stream(_rows()), content_type="text/csv")
|
|
response["Content-Disposition"] = 'attachment; filename="completion_report.csv"'
|
|
return response
|
|
|
|
|
|
class ProgressReportView(APIView):
|
|
"""Page-level dwell-time and completion aggregated per enrollment."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request):
|
|
enrollment_ids = _enrollment_queryset(request.query_params).values_list("id", flat=True)
|
|
qs = (
|
|
PageProgress.objects
|
|
.filter(enrollment_id__in=enrollment_ids)
|
|
.select_related("enrollment__user", "enrollment__course", "page")
|
|
)
|
|
rows = [
|
|
{
|
|
"enrollment_id": str(pp.enrollment_id),
|
|
"user_email": pp.enrollment.user.email,
|
|
"course_title": pp.enrollment.course.title,
|
|
"page_title": pp.page.title,
|
|
"accumulated_seconds": pp.accumulated_seconds,
|
|
"is_complete": pp.is_complete,
|
|
"last_seen_at": pp.last_seen_at.isoformat() if pp.last_seen_at else None,
|
|
}
|
|
for pp in qs
|
|
]
|
|
return Response(rows)
|
|
|
|
|
|
class ProgressReportCSVView(APIView):
|
|
"""CSV export of the progress report."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
FIELDS = [
|
|
"enrollment_id", "user_email", "course_title",
|
|
"page_title", "accumulated_seconds", "is_complete", "last_seen_at",
|
|
]
|
|
|
|
def get(self, request):
|
|
enrollment_ids = _enrollment_queryset(request.query_params).values_list("id", flat=True)
|
|
qs = (
|
|
PageProgress.objects
|
|
.filter(enrollment_id__in=enrollment_ids)
|
|
.select_related("enrollment__user", "enrollment__course", "page")
|
|
)
|
|
|
|
def _rows():
|
|
yield self.FIELDS
|
|
for pp in qs.iterator(chunk_size=500):
|
|
yield [
|
|
str(pp.enrollment_id),
|
|
pp.enrollment.user.email,
|
|
pp.enrollment.course.title,
|
|
pp.page.title,
|
|
str(pp.accumulated_seconds),
|
|
"true" if pp.is_complete else "false",
|
|
pp.last_seen_at.isoformat() if pp.last_seen_at else "",
|
|
]
|
|
|
|
response = StreamingHttpResponse(_csv_stream(_rows()), content_type="text/csv")
|
|
response["Content-Disposition"] = 'attachment; filename="progress_report.csv"'
|
|
return response
|
|
|
|
|
|
class AttemptReportView(APIView):
|
|
"""Quiz attempt report: score, pass/fail, timestamps."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
def get(self, request):
|
|
qs = QuizAttempt.objects.select_related("quiz", "user")
|
|
if org_id := request.query_params.get("org_id"):
|
|
# Quiz attempts don't carry org_id directly; filter via enrollments
|
|
enrolled_user_ids = (
|
|
Enrollment.objects.filter(org_id=org_id).values_list("user_id", flat=True)
|
|
)
|
|
qs = qs.filter(user_id__in=enrolled_user_ids)
|
|
if course_id := request.query_params.get("course_id"):
|
|
# Quiz belongs to a Page which belongs to a Lesson → Module → Course
|
|
qs = qs.filter(quiz__page__lesson__module__course_id=course_id)
|
|
if date_from := _parse_date(request.query_params.get("date_from")):
|
|
qs = qs.filter(started_at__gte=date_from)
|
|
if date_to := _parse_date(request.query_params.get("date_to")):
|
|
qs = qs.filter(started_at__lte=date_to)
|
|
|
|
rows = [
|
|
{
|
|
"attempt_id": str(a.id),
|
|
"user_email": a.user.email,
|
|
"quiz_title": a.quiz.title,
|
|
"attempt_number": a.attempt_number,
|
|
"score_percent": float(a.score_percent) if a.score_percent is not None else None,
|
|
"is_passed": a.is_passed,
|
|
"status": a.status,
|
|
"started_at": a.started_at.isoformat(),
|
|
"submitted_at": a.submitted_at.isoformat() if a.submitted_at else None,
|
|
}
|
|
for a in qs
|
|
]
|
|
return Response(rows)
|
|
|
|
|
|
class AttemptReportCSVView(APIView):
|
|
"""CSV export of the attempt report."""
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
FIELDS = [
|
|
"attempt_id", "user_email", "quiz_title", "attempt_number",
|
|
"score_percent", "is_passed", "status", "started_at", "submitted_at",
|
|
]
|
|
|
|
def get(self, request):
|
|
# Reuse the JSON view logic to build the queryset
|
|
json_view = AttemptReportView()
|
|
json_view.request = request
|
|
qs = QuizAttempt.objects.select_related("quiz", "user")
|
|
if org_id := request.query_params.get("org_id"):
|
|
enrolled_user_ids = (
|
|
Enrollment.objects.filter(org_id=org_id).values_list("user_id", flat=True)
|
|
)
|
|
qs = qs.filter(user_id__in=enrolled_user_ids)
|
|
if course_id := request.query_params.get("course_id"):
|
|
qs = qs.filter(quiz__page__lesson__module__course_id=course_id)
|
|
if date_from := _parse_date(request.query_params.get("date_from")):
|
|
qs = qs.filter(started_at__gte=date_from)
|
|
if date_to := _parse_date(request.query_params.get("date_to")):
|
|
qs = qs.filter(started_at__lte=date_to)
|
|
|
|
def _rows():
|
|
yield self.FIELDS
|
|
for a in qs.iterator(chunk_size=500):
|
|
yield [
|
|
str(a.id),
|
|
a.user.email,
|
|
a.quiz.title,
|
|
str(a.attempt_number),
|
|
str(float(a.score_percent)) if a.score_percent is not None else "",
|
|
"true" if a.is_passed else ("false" if a.is_passed is False else ""),
|
|
a.status,
|
|
a.started_at.isoformat(),
|
|
a.submitted_at.isoformat() if a.submitted_at else "",
|
|
]
|
|
|
|
response = StreamingHttpResponse(_csv_stream(_rows()), content_type="text/csv")
|
|
response["Content-Disposition"] = 'attachment; filename="attempt_report.csv"'
|
|
return response
|
|
|
|
|
|
class _EchoWriter:
|
|
def write(self, value):
|
|
return value
|
|
|
|
|
|
def _csv_stream(rows):
|
|
writer = csv.writer(_EchoWriter())
|
|
for row in rows:
|
|
yield writer.writerow(row)
|