feat: #152 Add AMP handling for quizzes

Author: payamnj

django_email_learning/personalised/api/urls.py

@@ -3,13 +3,15 @@
     FileUploadView,
     AssignmentSubmissionView,
     QuizSubmissionView,
+    AmpQuizSubmissionView,
     SubmitCertificateFormView,
 )
 
 app_name = "django_email_learning"
 
 urlpatterns = [
     path("quiz/", QuizSubmissionView.as_view(), name="quiz_submission"),
+    path("quiz-amp/", AmpQuizSubmissionView.as_view(), name="quiz_amp_submission"),
     path(
         "assignment/",
         AssignmentSubmissionView.as_view(),

django_email_learning/personalised/api/views.py

@@ -1,11 +1,14 @@
 from django.http import JsonResponse
+from django.conf import settings
 from django.views import View
 from django.urls import reverse
 from django.utils import timezone
 from django_email_learning.personalised.api.serializers import (
     QuizSubmissionRequest,
     QuestionResponse,
 )
+from django.utils.decorators import method_decorator
+from django.views.decorators.csrf import csrf_exempt
 from django_email_learning.services.metrics_service import metric_service
 from django_email_learning.services import jwt_service
 from django.utils.translation import gettext as _
@@ -204,15 +207,28 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
         token = serializer.token
         answers = serializer.answers
 
-        submited_question_ids = {response.id for response in answers}
+        response_payload, error_response = self.process_quiz_submission(
+            token=token,
+            answers=answers,
+        )
+        if error_response:
+            return error_response
+
+        return JsonResponse(response_payload, status=200)
+
+    @classmethod
+    def process_quiz_submission(
+        cls, token: str, answers: list[QuestionResponse]
+    ) -> tuple[dict | None, JsonResponse | None]:
+        submitted_question_ids = {response.id for response in answers}
         response_map = {response.id: response.answers for response in answers}
 
         try:
             decoded = jwt_service.decode_jwt(token=token)
         except jwt_service.InvalidTokenException as jde:
-            return JsonResponse({"error": str(jde)}, status=400)
+            return None, JsonResponse({"error": str(jde)}, status=400)
         except jwt_service.ExpiredTokenException as ete:
-            return JsonResponse({"error": str(ete)}, status=410)
+            return None, JsonResponse({"error": str(ete)}, status=410)
 
         delivery_id = decoded["delivery_id"]
 
@@ -221,7 +237,7 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
                 id=delivery_id, hash_value=decoded["delivery_hash"]
             )
         except ContentDelivery.DoesNotExist:
-            return JsonResponse(
+            return None, JsonResponse(
                 {
                     "error": "The content delivery associated with this token does not exist."
                 },
@@ -230,23 +246,25 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
 
         enrollment = delivery.enrollment
         if enrollment.status != EnrollmentStatus.ACTIVE:
-            return JsonResponse({"error": "Quiz is not valid anymore"}, status=400)
+            return None, JsonResponse(
+                {"error": "Quiz is not valid anymore"}, status=400
+            )
 
         quiz = delivery.course_content.quiz
         if not quiz:
-            return JsonResponse(
+            return None, JsonResponse(
                 {"error": "No quiz associated with this link"}, status=500
             )
 
         try:
-            score, passed = self.calculate_score_and_passed(
+            score, passed = cls.calculate_score_and_passed(
                 quiz, answers, decoded.get("question_ids")
             )
             logger.info(
                 f"Learner ID {enrollment.learner.id} submitted quiz for Course {enrollment.course.title} with score {score}. Passed: {passed}"
             )
         except ValueError as ve:
-            return JsonResponse({"error": str(ve)}, status=500)
+            return None, JsonResponse({"error": str(ve)}, status=500)
 
         QuizSubmission.objects.create(
             delivery=delivery,
@@ -271,9 +289,9 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
             if (quiz.is_blocking and passed) or QuizSubmission.objects.filter(
                 delivery=delivery
             ).count() == 1:
-                delivery = delivery.schedule_next_delivery()
+                new_delivery = delivery.schedule_next_delivery()
 
-            if not delivery:
+            if not new_delivery:
                 enrollment.graduate()
         else:
             failed_submissions_count = QuizSubmission.objects.filter(
@@ -328,7 +346,7 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
             is_passed=passed,
             is_blocking=quiz.is_blocking,
         )
-        return JsonResponse(
+        return (
             {
                 "score": score,
                 "passed": passed,
@@ -353,14 +371,14 @@ def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
                             ],
                         }
                         for question in quiz.questions.filter(
-                            id__in=submited_question_ids
+                            id__in=submitted_question_ids
                         )
                     ],
                 }
                 if not quiz.is_blocking
                 else None,
             },
-            status=200,
+            None,
         )
 
     @staticmethod
@@ -420,6 +438,101 @@ def calculate_score_and_passed(
         return score, passed
 
 
+@method_decorator(csrf_exempt, name="dispatch")
+class AmpQuizSubmissionView(View):
+    @staticmethod
+    def _set_amp_headers(response: JsonResponse, source_origin: str) -> JsonResponse:
+        response["Access-Control-Allow-Origin"] = source_origin
+        response["AMP-Access-Control-Allow-Source-Origin"] = source_origin
+        response["Access-Control-Allow-Credentials"] = "true"
+        return response
+
+    def _amp_json_response(
+        self, payload: dict, status: int, source_origin: str
+    ) -> JsonResponse:
+        response = JsonResponse(payload, status=status)
+        return self._set_amp_headers(response=response, source_origin=source_origin)
+
+    def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
+        source_origin = request.GET.get("__amp_source_origin")
+        if not source_origin:
+            return JsonResponse(
+                {"error": _("Missing __amp_source_origin query parameter.")},
+                status=400,
+            )
+
+        trusted_origins = {
+            trusted_origin.rstrip("/")
+            for trusted_origin in settings.CSRF_TRUSTED_ORIGINS
+        }
+        normalized_source_origin = source_origin.rstrip("/")
+        if normalized_source_origin not in trusted_origins:
+            return JsonResponse(
+                {"error": _("Invalid __amp_source_origin. Origin is not trusted.")},
+                status=400,
+            )
+
+        token = request.POST.get("token")
+        if not token:
+            return self._amp_json_response(
+                {"error": _("Token is required.")},
+                status=400,
+                source_origin=normalized_source_origin,
+            )
+
+        answers_map: dict[int, set[int]] = {}
+
+        for key, values in request.POST.lists():
+            if key == "token":
+                continue
+
+            try:
+                question_id = int(key)
+            except (TypeError, ValueError):
+                return self._amp_json_response(
+                    {"error": _("Invalid question ID in form payload.")},
+                    status=400,
+                    source_origin=normalized_source_origin,
+                )
+
+            parsed_answers: set[int] = set()
+            for value in values:
+                if value in (None, ""):
+                    continue
+
+                try:
+                    parsed_answers.add(int(value))
+                except (TypeError, ValueError):
+                    return self._amp_json_response(
+                        {"error": _("Invalid answer ID in form payload.")},
+                        status=400,
+                        source_origin=normalized_source_origin,
+                    )
+
+            answers_map[question_id] = parsed_answers
+
+        answers = [
+            QuestionResponse(id=question_id, answers=answer_ids)
+            for question_id, answer_ids in answers_map.items()
+        ]
+
+        response_payload, error_response = QuizSubmissionView.process_quiz_submission(
+            token=token,
+            answers=answers,
+        )
+        if error_response:
+            return self._set_amp_headers(
+                response=error_response,
+                source_origin=normalized_source_origin,
+            )
+
+        return self._amp_json_response(
+            response_payload,
+            status=200,
+            source_origin=normalized_source_origin,
+        )
+
+
 class SubmitCertificateFormView(View):
     def post(self, request, *args, **kwargs):  # type: ignore[no-untyped-def]
         payload = json.loads(request.body)

django_email_learning/services/command_models/send_quiz_command.py

@@ -6,6 +6,8 @@
 from django_email_learning.services.metrics_service import metric_service
 from django.core.mail import EmailMultiAlternatives
 from django.template.loader import render_to_string
+from django.conf import settings
+from django.urls import reverse
 from typing import Literal
 
 from django_email_learning.services.utils import mask_email
@@ -15,6 +17,9 @@ class QuizNotFoundError(Exception):
     pass
 
 
+DJANGO_EMAIL_LEARNING_CONF = settings.DJANGO_EMAIL_LEARNING
+
+
 class SendQuizCommand(AbstractCommand):
     command_name: Literal["send_quiz"] = "send_quiz"
     link: str
@@ -31,11 +36,17 @@ def execute(self) -> None:
             f"Sending quiz with ID {content.quiz.id} to email {mask_email(self.email)}"
         )
 
+        token = self.link.split("token=")[-1] if "token=" in self.link else None
+        if token:
+            token = token.split("&")[0]  # In case there are other query parameters
+
         quiz = content.quiz
         subject = quiz.title
         context = {
             "quiz": quiz,
             "link": self.link,
+            "amp_action_url": f"{DJANGO_EMAIL_LEARNING_CONF['SITE_BASE_URL']}{reverse('django_email_learning:api_personalised:quiz_amp_submission')}",
+            "token": token,
             "unsubscribe_link": content.course.generate_unsubscribe_link(self.email),
         }
         payload = render_to_string("emails/quiz.txt", context)
@@ -49,6 +60,10 @@ def execute(self) -> None:
         email_message.attach_alternative(
             render_to_string("emails/quiz.html", context), "text/html"
         )
+        if DJANGO_EMAIL_LEARNING_CONF.get("AMP_ENABLED"):
+            email_message.attach_alternative(
+                render_to_string("emails/quiz_amp.html", context), "text/x-amp-html"
+            )
 
         email_sender_service.send(email_message)
         metric_service.quiz_sent(

django_email_learning/templates/emails/base_amp.html

@@ -0,0 +1,71 @@
+{% load i18n %}
+{% with brand_color="#636eec" %}
+<!doctype html>
+<html amp4email data-css-strict>
+    <head>
+        <meta charset="utf-8">
+        <script async src="https://cdn.ampproject.org/v0.js"></script>
+        {% block extra_head %}{% endblock %}
+        <style amp4email-boilerplate>body{visibility:hidden}</style>
+        <style amp-custom>
+            .background {
+                padding: 10px;
+                width: 100%;
+                box-sizing: border-box;
+                font-family: Helvetica, Arial, sans-serif;
+                line-height: 1.8;
+                font-size: 15px;
+            }
+            .paper {
+                margin: 10px;
+                padding: 8px;
+                border-radius: 8px;
+                border: solid 1px #eee;
+            }
+            .button {
+                padding: 14px 30px;
+                border-radius: 8px;
+                font-size: 16px;
+                font-weight: 600;
+                box-shadow: 0 4px 14px rgba(36, 48, 110, 0.16);
+                background-color: #636eec;
+                color: #fff;
+            }
+            .button:hover {
+                background-color: #6e00a5;
+                cursor: pointer;
+            }
+            .email-title {
+                margin: 10px;
+                text-align: center;
+                border-bottom: 2px solid #d8d9ee;
+                padding-bottom: 10px;
+                line-height: 1.35;
+                font-size: 19px;
+            }
+            .locale {
+                {% get_current_language_bidi as IS_RTL %}
+                {% if IS_RTL %}
+                    direction: rtl;
+                    text-align: right;
+                {% else %}
+                    direction: ltr;
+                    text-align: left;
+                {% endif %}
+            }
+            p {
+                padding: 0 15px;
+            }
+            {% block extra_styles %}{% endblock %}
+        </style>
+    </head>
+<body>
+    <div class="background">
+        <div class="locale">
+            {% block content %}{% endblock %}
+        </div>
+        {% block footer %}{% endblock %}
+    </div>
+</body>
+</html>
+{% endwith %}