Add ExamUserStatus logging for exam completion and start events

physics/views.py

@@ -1,6 +1,6 @@
 from django.contrib.contenttypes.models import ContentType
 from django.utils import timezone
-from generic.models import CidUser
+from generic.models import CidUser, ExamUserStatus, CidUserExam
 from physics.decorators import user_is_author_or_physics_checker
 from physics.filters import QuestionFilter, UserAnswerFilter
 from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
@@ -148,6 +148,18 @@ def exam_complete(request, pk, cid=None, passcode=None):
     cid_user_exam = exam.get_or_create_cid_user_exam(cid=cid, user_user=request.user)
 
     cid_user_exam.complete_exam()
+    try:
+        ct = ContentType.objects.get_for_model(exam)
+        ExamUserStatus.objects.create(
+            content_type=ct,
+            object_id=exam.pk,
+            cid_user_exam=cid_user_exam,
+            status="completed",
+            extra="physics",
+        )
+    except Exception:
+        # avoid crashing the user flow if logging fails
+        pass
 
     return HttpResponse("<div role='alert' class='alert alert-info'>Exam completed</div>")
 
@@ -178,8 +190,29 @@ def exam_take_overview(request, pk, cid=None, passcode=None):
         else:
             question_answer_tuples.append((q, None))
 
+    # detect whether a CidUserExam already exists so we only log the initial start once
+    content_type = ContentType.objects.get_for_model(exam)
+    existing = False
+    if cid is not None:
+        cid_user_obj = CidUser.objects.filter(cid=cid).first()
+        existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, cid_user=cid_user_obj).exists()
+    else:
+        existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, user_user=request.user).exists()
+
     cid_user_exam = exam.get_or_create_cid_user_exam(cid=cid, user_user=request.user)
 
+    if not existing:
+        try:
+            ExamUserStatus.objects.create(
+                content_type=content_type,
+                object_id=exam.pk,
+                cid_user_exam=cid_user_exam,
+                status="started",
+                extra="physics",
+            )
+        except Exception:
+            pass
+
     return render(
         request,
         "physics/exam_take_overview.html",

requirements.txt

@@ -59,3 +59,4 @@ pygments
 celery[redis]
 redis
 pywatchman
+numpy==2.4.0

sbas/views.py

@@ -1,7 +1,7 @@
 from reversion.views import RevisionMixin
 from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
 from django.views.generic.detail import DetailView
-from generic.models import CidUser
+from generic.models import CidUser, ExamUserStatus, CidUserExam
 from sbas.forms import UserAnswerForm, ExamAuthorForm, ExamForm
 from django.shortcuts import render, get_object_or_404, redirect
 from django.views.decorators.csrf import csrf_exempt
@@ -306,6 +306,17 @@ def exam_complete(request, pk, cid=None, passcode=None):
     cid_user_exam = exam.get_or_create_cid_user_exam(cid=cid, user_user=request.user)
 
     cid_user_exam.complete_exam()
+    try:
+        ct = ContentType.objects.get_for_model(exam)
+        ExamUserStatus.objects.create(
+            content_type=ct,
+            object_id=exam.pk,
+            cid_user_exam=cid_user_exam,
+            status="completed",
+            extra="sbas",
+        )
+    except Exception:
+        pass
 
     return HttpResponse("<div role='alert' class='alert alert-info'>Exam completed</div>")
 
@@ -366,8 +377,29 @@ def exam_take(request, pk: int, sk: int, cid: int = None, passcode: str = None):
 
     exam.check_user_can_take(cid, passcode, request.user)
 
+    # log start only when a new CidUserExam is created
+    content_type = ContentType.objects.get_for_model(exam)
+    existing = False
+    if cid is not None:
+        cid_user_obj = CidUser.objects.filter(cid=cid).first()
+        existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, cid_user=cid_user_obj).exists()
+    else:
+        existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, user_user=request.user).exists()
+
     cid_user_exam = exam.get_or_create_cid_user_exam(cid = cid, user_user=request.user)
 
+    if not existing:
+        try:
+            ExamUserStatus.objects.create(
+                content_type=content_type,
+                object_id=exam.pk,
+                cid_user_exam=cid_user_exam,
+                status="started",
+                extra="sbas",
+            )
+        except Exception:
+            pass
+
     question = exam.get_questions()[sk]
 
     exam_length = len(exam.exam_questions.all())