bulk reponse editing interface

Allows uploading a CSV with updated responses to a question for bulk
editing

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/forms.py

@@ -584,3 +584,76 @@ 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"
+        print(question.question_type, is_multi)
+
+        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/scoring.py

@@ -614,7 +614,7 @@ def get_response_data(
         for opt in response.multi_option.all():
             descs.append(opt.description)
             score += opt.score
-        answer = ",".join(descs)
+        answer = "|".join(descs)
     elif response.option is not None:
         score = response.option.score
         answer = response.option.description

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/views/questions.py

@@ -1,10 +1,22 @@
+import re
+
 from django.contrib.auth.mixins import UserPassesTestMixin
+from django.db import transaction
 from django.shortcuts import get_object_or_404
 from django.urls import reverse
 from django.views.generic import FormView, ListView
 
-from crowdsourcer.forms import OptionFormset, QuestionFormset
-from crowdsourcer.models import Option, Question, Section
+import pandas as pd
+
+from crowdsourcer.forms import OptionFormset, QuestionBulkUploadForm, QuestionFormset
+from crowdsourcer.models import (
+    Option,
+    PublicAuthority,
+    Question,
+    Response,
+    ResponseType,
+    Section,
+)
 
 
 class SectionList(ListView):
@@ -27,7 +39,7 @@ def get_success_url(self):
             "session_urls:edit_options",
             kwargs={
                 "marking_session": self.request.current_session.label,
-                "section_name": "Buildings & Heating",
+                "section_name": self.kwargs["section_name"],
             },
         )
 
@@ -97,3 +109,129 @@ def get_context_data(self, **kwargs):
     def form_valid(self, form):
         form.save()
         return super().form_valid(form)
+
+
+class QuestionBulkUpdateView(UserPassesTestMixin, FormView):
+    template_name = "crowdsourcer/questions/question_bulk_upload.html"
+    form_class = QuestionBulkUploadForm
+
+    def test_func(self):
+        return self.request.user.has_perm("crowdsourcer.can_manage_users")
+
+    def get_success_url(self):
+        return reverse(
+            "session_urls:question_bulk_update",
+            kwargs={
+                "marking_session": self.request.current_session.label,
+                "section_name": self.kwargs["section_name"],
+                "question": self.kwargs["question"],
+            },
+        )
+
+    def get_form(self):
+        self.section = get_object_or_404(
+            Section,
+            title=self.kwargs["section_name"],
+            marking_session=self.request.current_session,
+        )
+
+        q_parts = re.match(r"(\d+)([a-z]?)", self.kwargs["question"])
+        q_kwargs = {
+            "section": self.section,
+            "number": q_parts.groups()[0],
+        }
+
+        if len(q_parts.groups()) == 2 and q_parts.groups()[1] != "":
+            q_kwargs["number_part"] = q_parts.groups()[1]
+
+        self.question = get_object_or_404(Question, **q_kwargs)
+        return self.form_class(
+            self.question.pk,
+            [(rt.type, rt.type) for rt in ResponseType.objects.all()],
+            self.request.current_session,
+            **self.get_form_kwargs(),
+        )
+
+    def get_context_data(self, **kwargs):
+        context = super().get_context_data(**kwargs)
+
+        context["section"] = self.section
+        context["question"] = self.question
+
+        return context
+
+    def form_valid(self, form):
+        data = form.cleaned_data
+
+        question = get_object_or_404(Question, id=data["question"])
+        stage = get_object_or_404(ResponseType, type=data["stage"])
+        is_multi = question.question_type == "multiple_choice"
+
+        with transaction.atomic():
+            for index, row in form.responses_df.iterrows():
+                answer = row["answer"].strip()
+                if answer == "-":
+                    continue
+
+                if is_multi:
+                    answers = answer.split("|")
+
+                authority = PublicAuthority.objects.get(
+                    name=row["authority"],
+                    marking_session=self.request.current_session,
+                )
+                if not is_multi:
+                    option = Option.objects.get(question=question, description=answer)
+
+                try:
+                    response = Response.objects.get(
+                        question=question, response_type=stage, authority=authority
+                    )
+                    changed = False
+                    opts = {}
+                    for col in ["page_number", "evidence", "public_notes"]:
+                        if pd.isna(row[col]) is False:
+                            opts[col] = row[col]
+                            if row[col] != getattr(response, col):
+                                changed = True
+                    if not is_multi and response.option != option:
+                        changed = True
+                        opts["option"] = option
+
+                    if changed:
+                        response.user = self.request.user
+                        for k, v in opts.items():
+                            setattr(response, k, v)
+                        response.save()
+
+                        if is_multi:
+                            response.multi_option.clear()
+                            for a in answers:
+                                option = Option.objects.get(
+                                    question=question, description=a
+                                )
+                                response.multi_option.add(option.id)
+
+                except Response.DoesNotExist:
+                    opts = {
+                        "question": question,
+                        "response_type": stage,
+                        "authority": authority,
+                        "user": self.request.user,
+                    }
+                    for col in ["page_number", "evidence", "public_notes"]:
+                        if pd.isna(row[col]) is False:
+                            opts[col] = row[col]
+                    if not is_multi:
+                        opts["option"] = option
+
+                    response = Response.objects.create(**opts)
+
+                    if is_multi:
+                        for a in answers:
+                            option = Option.objects.get(
+                                question=question, description=a
+                            )
+                            response.multi_option.add(option.id)
+
+        return super().form_valid(form)