add a page to allow bulk updating of responses to a question

This is mostly to allow people to download a CSV of responses, edit that
and re-upload it to correct the responses. It will only update an
existing response if it has changed.

Author: struan

ceuk-marking/urls.py

@@ -328,6 +328,11 @@
         questions.SectionList.as_view(),
         name="question_sections",
     ),
+    path(
+        "questions/update/<section_name>/<question>/",
+        questions.QuestionBulkUpdateView.as_view(),
+        name="question_bulk_update",
+    ),
 ]
 
 urlpatterns = [

crowdsourcer/fixtures/responses.json

@@ -10,7 +10,7 @@
     "response_type": 1,
     "public_notes": "public notrs",
     "page_number": "0",
-    "evidence": "",
+    "evidence": null,
     "private_notes": "private notes",
     "revision_type": null,
     "revision_notes": null,

crowdsourcer/forms.py

@@ -584,3 +584,75 @@ def __init__(self, properties: {}, **kwargs):
 QuestionFormset = modelformset_factory(
     Question, fields=["question_type", "weighting"], extra=0, can_delete=False
 )
+
+
+class QuestionBulkUploadForm(Form):
+    question = CharField(widget=HiddenInput)
+    stage = ChoiceField(required=True, choices=[])
+    updated_responses = FileField()
+
+    def clean(self):
+        data = self.cleaned_data.get("updated_responses")
+
+        try:
+            df = pd.read_csv(
+                data,
+                usecols=[
+                    "authority",
+                    "answer",
+                    "score",
+                    "public_notes",
+                    "page_number",
+                    "evidence",
+                    "private_notes",
+                ],
+            )
+        except ValueError as v:
+            raise ValidationError(f"Problem processing csv file: {v}")
+
+        self.responses_df = df
+
+        try:
+            question = Question.objects.get(id=self.cleaned_data["question"])
+        except Question.DoesNotExist:
+            raise ValidationError(f"Bad question id: {self.cleaned_data['question']}")
+
+        is_multi = question.question_type == "multiple_choice"
+
+        file_errors = []
+        for _, row in self.responses_df.iterrows():
+            desc = row["answer"].strip()
+            try:
+                PublicAuthority.objects.get(
+                    name=row["authority"],
+                    marking_session=self.session,
+                )
+            except PublicAuthority.DoesNotExist:
+                file_errors.append(f"No such authority: {row['authority']}")
+                continue
+
+            if desc == "-":
+                continue
+
+            if not is_multi:
+                answers = [desc]
+            else:
+                answers = desc.split("|")
+
+            for answer in answers:
+                try:
+                    Option.objects.get(question=question, description=answer)
+                except Option.DoesNotExist:
+                    file_errors.append(
+                        f"No such answer for {row['authority']}: {answer}"
+                    )
+                    continue
+
+        if len(file_errors) > 0:
+            raise ValidationError(file_errors)
+
+    def __init__(self, question_id, stage_choices, session, **kwargs):
+        super().__init__(**kwargs)
+        self.session = session
+        self.initial["question"] = question_id
+        self.fields["stage"].choices = stage_choices

crowdsourcer/templates/crowdsourcer/questions/question_bulk_upload.html

@@ -0,0 +1,21 @@
+{% extends 'crowdsourcer/base.html' %}
+
+{% load crowdsourcer_tags django_bootstrap5 %}
+
+{% block content %}
+{% if show_login %}
+    <h1 class="mb-4">Sign in</h1>
+    <a href="{% url 'login' %}">Sign in</a>
+{% else %}
+<h1 class="mb-4">Update Responses for {{ section.title }} {{ question.number_and_part }}</h1>
+<h4 class="mb-4">{{ question.description }}</h4>
+
+
+    <form enctype="multipart/form-data" action="" method="post">
+        {% csrf_token %}
+        {% bootstrap_form form %}
+        <input type="submit" value="Update">
+    </form>
+
+{% endif %}
+{% endblock %}

crowdsourcer/tests/data/test_question_upload.csv

@@ -0,0 +1,4 @@
+authority,answer,score,public_notes,page_number,evidence,private_notes
+Aberdeen City Council,Yes,1,uploaded public notes,99,uploaded evidence,uploaded private notes
+Aberdeenshire Council,No,0,,,,
+Adur District Council,-,-,-,-,-,-

crowdsourcer/tests/data/test_question_upload_one_unchanged.csv

@@ -0,0 +1,4 @@
+authority,answer,score,public_notes,page_number,evidence,private_notes
+Aberdeen City Council,Yes,1,uploaded public notes,99,uploaded evidence,uploaded private notes
+Aberdeenshire Council,Yes,0,public notrs,0,,private notes
+Adur District Council,-,-,-,-,-,-

crowdsourcer/tests/test_question_views.py

@@ -0,0 +1,125 @@
+import pathlib
+
+from django.contrib.auth.models import Permission, User
+from django.test import TestCase
+from django.urls import reverse
+
+from crowdsourcer.models import Question, Response
+
+
+class BaseTestCase(TestCase):
+    fixtures = [
+        "authorities.json",
+        "basics.json",
+        "users.json",
+        "questions.json",
+        "options.json",
+        "assignments.json",
+        "responses.json",
+    ]
+
+    def setUp(self):
+        p = Permission.objects.get(codename="can_manage_users")
+        u = User.objects.get(username="volunteer_admin")
+        u.user_permissions.add(p)
+
+        self.client.force_login(u)
+        self.user = u
+
+
+class TestBulkUpload(BaseTestCase):
+    def test_one_update_one_new(self):
+        url = reverse("question_bulk_update", args=("Transport", "1"))
+        response = self.client.get(url)
+
+        q = Question.objects.get(
+            section__title="Transport",
+            section__marking_session__label="Default",
+            number=1,
+        )
+
+        all_r = Response.objects.filter(question=q, response_type__type="First Mark")
+        self.assertEqual(all_r.count(), 1)
+
+        r = Response.objects.get(question=q, authority__name="Aberdeenshire Council")
+
+        self.assertEqual(r.option.description, "Yes")
+        self.assertEqual(r.page_number, "0")
+
+        upload_file = (
+            pathlib.Path(__file__).parent.resolve()
+            / "data"
+            / "test_question_upload.csv"
+        )
+
+        with open(upload_file, "rb") as fp:
+            response = self.client.post(
+                url,
+                data={
+                    "question": 281,
+                    "updated_responses": fp,
+                    "stage": "First Mark",
+                },
+            )
+
+        self.assertRedirects(response, "/Default" + url)
+        self.assertEqual(all_r.count(), 2)
+
+        r = Response.objects.get(question=q, authority__name="Aberdeen City Council")
+
+        self.assertEqual(r.option.description, "Yes")
+        self.assertEqual(r.page_number, "99")
+
+        r = Response.objects.get(question=q, authority__name="Aberdeenshire Council")
+
+        self.assertEqual(r.option.description, "No")
+        self.assertEqual(r.page_number, None)
+
+    def test_one_new_one_unchanged(self):
+        url = reverse("question_bulk_update", args=("Transport", "1"))
+        response = self.client.get(url)
+
+        q = Question.objects.get(
+            section__title="Transport",
+            section__marking_session__label="Default",
+            number=1,
+        )
+
+        all_r = Response.objects.filter(question=q, response_type__type="First Mark")
+        self.assertEqual(all_r.count(), 1)
+
+        r = Response.objects.get(question=q, authority__name="Aberdeenshire Council")
+
+        last_update = r.last_update
+        self.assertEqual(r.option.description, "Yes")
+        self.assertEqual(r.page_number, "0")
+
+        upload_file = (
+            pathlib.Path(__file__).parent.resolve()
+            / "data"
+            / "test_question_upload_one_unchanged.csv"
+        )
+
+        with open(upload_file, "rb") as fp:
+            response = self.client.post(
+                url,
+                data={
+                    "question": 281,
+                    "updated_responses": fp,
+                    "stage": "First Mark",
+                },
+            )
+
+        self.assertRedirects(response, "/Default" + url)
+        self.assertEqual(all_r.count(), 2)
+
+        r = Response.objects.get(question=q, authority__name="Aberdeen City Council")
+
+        self.assertEqual(r.option.description, "Yes")
+        self.assertEqual(r.page_number, "99")
+
+        r = Response.objects.get(question=q, authority__name="Aberdeenshire Council")
+
+        self.assertEqual(r.option.description, "Yes")
+        self.assertEqual(r.page_number, "0")
+        self.assertEqual(last_update, r.last_update)