Files
Ross fa98541e31 .
2026-07-06 22:06:58 +01:00

1380 lines
45 KiB
Python
Executable File

import json
from django.shortcuts import render, get_object_or_404, redirect
from django import forms
# from django.contrib.auth.models import User
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth.models import User
from django.core.exceptions import PermissionDenied
from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic.detail import DetailView
from django.views.generic.edit import CreateView, UpdateView, DeleteView
from django.views.generic import ListView
from django.views.decorators.csrf import csrf_exempt
from django.urls import reverse_lazy, reverse
from django.http import Http404, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from django.utils.safestring import mark_safe
from django.db.models import Count
from .forms import (
ExamAuthorForm,
ExamGroupsForm,
ExamMarkerForm,
RapidForm,
MarkRapidQuestionForm,
ImageFormSet,
RapidQuestionAnswerForm,
RegionForm,
AbnormalityForm,
ExaminationForm,
AnswerFormSet,
AnswerUpdateFormSet,
ExamForm,
)
from .models import (
Rapid,
Abnormality,
RapidImage,
Region,
Examination,
Exam,
Answer,
UserAnswer,
)
from generic.views import (
AuthorRequiredMixin,
ExamCloneMixin,
ExamCreateBase,
ExamDeleteBase,
ExamGroupsUpdateBase,
ExamUpdateBase,
ExamViews,
GenericViewBase,
UpdateQuestionMixin,
)
from reversion.views import RevisionMixin
from .tables import RapidTable, RapidUserAnswerTable
from .filters import RapidFilter, RapidUserAnswerFilter
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
from .decorators import user_is_author_or_rapid_checker
from collections import defaultdict
import json
import statistics
import plotly.express as px
from django.core.cache import cache
from helpers.images import image_as_base64
import logging
from django.contrib import messages
from copy import deepcopy
from django.forms.models import model_to_dict
from rapids.forms import RapidCreationDefaultForm
from rapids.models import RapidCreationDefault
from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
from dal import autocomplete
logger = logging.getLogger(__name__)
def normaliseScore(score):
if score == 49:
return 4.5
elif score in [50, 51]:
return 5
elif score in [52, 53]:
return 5.5
elif score in [54]:
return 6
elif score in [55, 56]:
return 6.5
elif score in [57, 58]:
return 7
elif score in [59]:
return 7.5
elif score in [60]:
return 8
return 4
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if self.request.user.is_superuser:
return obj
if self.request.user.groups.filter(name="rapid_checker").exists():
return obj
if self.request.user not in obj.author.all():
raise PermissionDenied() # or Http404
return obj
@login_required
def rapid_split(request, pk):
rapid = get_object_or_404(Rapid, pk=pk)
if (
not request.user.groups.filter(name="rapid_checker").exists()
and request.user not in rapid.author.all()
):
raise PermissionDenied()
images = rapid.images.all()
old_abnormality = rapid.abnormality.all()
old_region = rapid.region.all()
old_examination = rapid.examination.all()
# old_site = rapid.site.all()
old_author = rapid.author.all()
if not images:
raise Http404
for n in range(len(images) - 1):
rapid.pk = None
rapid.save()
rapid.abnormality.set(old_abnormality)
rapid.region.set(old_region)
rapid.examination.set(old_examination)
# rapid.site.set(old_site)
rapid.author.set(old_author)
images[n].rapid = rapid
images[n].save()
# images[-1].rapid
# if request.user not in rapid.author.all():
# raise PermissionDenied
# logging.debug(rapid.subspecialty.first().name.all())
return render(request, "rapids/question_detail.html", {"question": rapid})
class RapidCreationDefaultView(LoginRequiredMixin, UpdateView):
model = RapidCreationDefault
form_class = RapidCreationDefaultForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
# def dispatch(self, request, *args, **kwargs):
# """
# Overridden so we can make sure the `Rapid` instance exists
# before going any further.
# """
# self.pk = get_object_or_404(Rapid, pk=kwargs['pk'])
# return super().dispatch(request, *args, **kwargs)
def get_object(self, queryset=None):
obj, create = RapidCreationDefault.objects.get_or_create(
author=self.request.user
)
return obj
def form_valid(self, form):
model = form.save(commit=False)
model.author = self.request.user
model.save()
response = super().form_valid(form)
return response
# form.instance.author.add(self.request.user.id)
#@login_required
#def rapid_clone(request, pk):
# new_item = get_object_or_404(Rapid, pk=pk)
# new_item.pk = None # autogen a new pk (item_id)
# # new_item.name = "Copy of " + new_item.name #need to change uniques
#
# if (
# request.user not in new_item.get_author_objects()
# and not request.user.is_superuser
# ):
# raise PermissionDenied() # or Http404
#
# form = RapidForm(request.POST or None, instance=new_item)
#
# image_formset = ImageFormSet()
# answer_formset = AnswerFormSet()
#
# if form.is_valid():
# form.instance.author.add(request.user.id)
#
# # logger.debug(formset.is_valid())
# if image_formset.is_valid() and answer_formset.is_valid():
# response = super().form_valid(form)
# image_formset.instance = obj
# image_formset.save()
#
# answer_formset.instance = obj
# answer_formset.save()
# return response
# else:
# return super().form_invalid(form)
#
# context = {
# "form": form,
# "image_formset": image_formset,
# "answer_formset": answer_formset,
# # other context
# }
#
# return render(request, "rapids/rapid_form.html", context)
class RapidCreateBase(RevisionMixin, LoginRequiredMixin, CreateView):
model = Rapid
form_class = RapidForm
# Sending user object to the form, to verify which fields to display/remove (depending on group)
def get_form_kwargs(self):
kwargs = super(RapidCreateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(RapidCreateBase, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = ImageFormSet(
self.request.POST, self.request.FILES
)
context["image_formset"].full_clean()
context["answer_formset"] = AnswerFormSet(self.request.POST)
context["answer_formset"].full_clean()
else:
context["image_formset"] = ImageFormSet()
context["answer_formset"] = AnswerFormSet()
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
image_formset = context["image_formset"]
answer_formset = context["answer_formset"]
if image_formset.is_valid() and answer_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = self.object
image_formset.save()
answer_formset.instance = self.object
answer_formset.save()
# If the normal submit button is pressed we save as normal
if "submit" in self.request.POST:
return response
# else we redirect to the clone url
else:
return redirect("rapids:rapid_clone", pk=self.object.pk)
else:
return super().form_invalid(form)
# @login_required
class RapidCreate(RapidCreateBase):
# initial = {"laterality": Rapid.NONE}
def get_initial(self):
if "pk" in self.kwargs:
initial = super().get_initial()
exam = get_object_or_404(Exam, pk=self.kwargs["pk"])
initial["exams"] = [exam.id]
return initial
## There has to be a better way...
# try:
# s = (i.pk for i in self.request.user.rapid_default.site.all())
# self.initial.update({"site": s})
# except AttributeError:
# pass
# return self.initial
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
# self.object = form.save(commit=False)
# self.object.save()
# form.instance.author.add(self.request.user.id)
# return super().form_valid(form)
class RapidAnswerUpdate(RevisionMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = Rapid
form_class = RapidQuestionAnswerForm
template_name = "rapids/rapidquestionanswer_form.html"
def get_context_data(self, **kwargs):
context = super(RapidAnswerUpdate, self).get_context_data(**kwargs)
if self.request.POST:
context["answer_formset"] = AnswerUpdateFormSet(
self.request.POST, self.request.FILES, instance=self.object
)
context["answer_formset"].full_clean()
else:
context["answer_formset"] = AnswerUpdateFormSet(instance=self.object)
context["question"] = context["object"]
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
# form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
formset = context["answer_formset"]
# logger.debug(formset.is_valid())
if formset.is_valid():
response = super().form_valid(form)
formset.instance = self.object
formset.save()
return response
else:
return super().form_invalid(form)
class RapidUpdate(
UpdateQuestionMixin
):
model = Rapid
form_class = RapidForm
# fields = '__all__'
# #fields = [ 'condition' ]
# #initial = {'date_of_death': '05/01/2018'}
# exclude = [ 'created_date', 'published_date' ]
def get_context_data(self, **kwargs):
context = super(RapidUpdate, self).get_context_data(**kwargs)
if self.request.POST:
context["image_formset"] = ImageFormSet(
self.request.POST, self.request.FILES, instance=self.object
)
context["image_formset"].full_clean()
context["answer_formset"] = AnswerFormSet(
self.request.POST, instance=self.object
)
context["answer_formset"].full_clean()
else:
context["image_formset"] = ImageFormSet(instance=self.object)
context["answer_formset"] = AnswerFormSet(instance=self.object)
context["question"] = context["object"]
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
context = self.get_context_data(form=form)
image_formset = context["image_formset"]
answer_formset = context["answer_formset"]
# logger.debug(formset.is_valid())
if image_formset.is_valid() and answer_formset.is_valid():
response = super().form_valid(form)
image_formset.instance = self.object
image_formset.save()
answer_formset.instance = self.object
answer_formset.save()
# restore exam orders
# wanted_exams = form["exams"].data
# for exam in self.object.exams.all():
# if exam in exam_orders and self.object in exam_orders[exam]:
# print(exam_orders[exam])
# exam.exam_questions.set(exam_orders[exam])
# exam.save()
return response
else:
return super().form_invalid(form)
class RapidClone(RapidCreateBase):
"""Clones a existing rapid"""
def get_initial(self):
# print(self.request)
old_object = get_object_or_404(Rapid, pk=self.kwargs["pk"])
initial_data = model_to_dict(old_object, exclude=["id"])
return initial_data
@login_required
@user_is_author_or_rapid_checker
def rapid_scrap(request, pk):
try:
rapid = Rapid.objects.get(pk=pk)
except Rapid.DoesNotExist:
raise Http404("Rapid does not exist")
if (
not request.user.groups.filter(name="rapid_checker").exists()
and request.user not in rapid.author.all()
):
raise PermissionDenied()
rapid.scrapped = not rapid.scrapped
rapid.save()
return HttpResponseRedirect(reverse("rapids:question_detail", args=(pk,)))
# @login_required
# def edit_abnormality_popup(request):
# instance = get_object_or_404(Abnormality, pk = pk)
# form = AbnormalityForm(request.POST or None)
# if form.is_valid():
# instance = form.save()
# return HttpResponse('<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>' % (instance.pk, instance))
# return render(request, "rapids/create_simple.html", {'form': form})
@login_required
def create_abnormality(request):
form = AbnormalityForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_abnormality");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Abnormality"}
)
@login_required
def get_abnormality_id(request):
"""Return the primary key for an Abnormality looked up by name (admin popup helper).
Scope: authenticated users only.
Functionality: looks up an Abnormality by exact name and returns its ID as JSON;
used by inline admin form popups.
"""
if request.accepts("application/json"):
abnormality_name = request.GET.get("abnormality_name", "")
abnormality_id = Abnormality.objects.get(name=abnormality_name).id
data = {
"abnormality_id": abnormality_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
@login_required
def create_examination(request):
form = ExaminationForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_examination");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Examination"}
)
@login_required
def get_examination_id(request):
"""Return the primary key for an Examination looked up by name (admin popup helper).
Scope: authenticated users only.
Functionality: looks up an Examination by exact name and returns its ID as JSON;
used by inline admin form popups.
"""
if request.accepts("application/json"):
examination_name = request.GET.get("examination_name", "")
examination_id = Examination.objects.get(name=examination_name).id
data = {
"examination_id": examination_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
@login_required
def create_region(request):
form = RegionForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_region");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Region"}
)
@login_required
def get_region_id(request):
"""Return the primary key for a Region looked up by name (admin popup helper).
Scope: authenticated users only.
Functionality: looks up a Region by exact name and returns its ID as JSON;
used by inline admin form popups.
"""
if request.accepts("application/json"):
region_name = request.GET.get("region_name", "")
region_id = Region.objects.get(name=region_name).id
data = {
"region_id": region_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
class RapidView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = Rapid
table_class = RapidTable
template_name = "question_table_view.html"
filterset_class = RapidFilter
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["app_name"] = "rapids"
return context
@login_required
def mark_all(request, exam_pk, sk):
return mark(request, exam_pk, sk, unmarked_exam_answers_only=False)
@login_required
def mark_review(request, exam_pk, sk):
return mark(request, exam_pk, sk, unmarked_exam_answers_only=False, review=True)
# @user_passes_test(user_is_admin, login_url="/accounts/login")
@login_required
def mark(request, exam_pk, sk, unmarked_exam_answers_only=True, review=False):
exam = get_object_or_404(Exam, pk=exam_pk)
if not GenericExamViews.check_user_marker_access(request.user, exam_pk):
raise PermissionDenied
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
mark_url = "rapids:mark"
if review:
mark_url = "rapids:mark_review"
elif not unmarked_exam_answers_only:
mark_url = "rapids:mark_all"
questions = exam.get_questions()
n = sk
question_details = {
"total": len(questions),
"current": n + 1,
}
try:
question = questions[sk]
except IndexError:
raise Http404("Exam question does not exist")
if request.method == "POST":
form = MarkRapidQuestionForm(request.POST)
# *******************************
# TODO: convert to JSON request
# *******************************
if form.is_valid():
# If skip button is pressed skip the question without marking
if "skip" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n + 1)
cd = form.cleaned_data
marked_answers = json.loads(cd.get("marked_answers"))
# This should probably use json (or something else...)
# ajax.....
to_run = (
("correct", Answer.MarkOptions.CORRECT),
("half-correct", Answer.MarkOptions.HALF_MARK),
("incorrect", Answer.MarkOptions.INCORRECT),
)
for status_string, status in to_run:
for ans in marked_answers[status_string]:
ans = ans.strip()
if ans == "":
continue
a = Answer.objects.filter(
answer__iexact=ans, question_id=question.pk
).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = ans
a.status = status
a.save()
if "next" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n + 1)
elif "previous" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n - 1)
# Reset user answers (relies on the functional javascript)
unmarked_user_answers = set()
else:
form = MarkRapidQuestionForm()
# answers_dict = {}
# for ans in question.answers.all():
# answers_dict[ans.answer_compare] = ans
correct_answers = []
half_mark_answers = []
incorrect_answers = []
review_user_answers = []
unmarked_answers_bool = False
unmarked_user_answers = []
# this could be improved
if question.normal:
incorrect_answers = question.get_user_answers(exam_pk=exam_pk)
if "" in incorrect_answers:
incorrect_answers.remove("")
if "normal" in incorrect_answers:
incorrect_answers.remove("normal")
else:
if review:
unmarked_user_answers = question.get_user_answers(
exam_pk=exam_pk, include_normal=False
)
elif unmarked_exam_answers_only:
unmarked_user_answers = question.get_unmarked_user_answers(exam_pk=exam_pk)
else:
unmarked_user_answers = question.get_unmarked_user_answers()
if not unmarked_exam_answers_only:
correct_answers = question.answers.filter(status=Answer.MarkOptions.CORRECT)
half_mark_answers = question.answers.filter(
status=Answer.MarkOptions.HALF_MARK
)
incorrect_answers = question.answers.filter(
status=Answer.MarkOptions.INCORRECT
)
if review:
for ans in unmarked_user_answers:
# duplicated from user answer model....
marked_ans = question.answers.filter(answer_compare__iexact=ans).first()
mark = "unmarked"
if marked_ans is not None:
if marked_ans.status == Answer.MarkOptions.CORRECT:
mark = "correct"
elif marked_ans.status == Answer.MarkOptions.HALF_MARK:
mark = "half-correct"
elif marked_ans.status == Answer.MarkOptions.INCORRECT:
mark = "incorrect"
if mark == "unmarked":
unmarked_answers_bool = True
review_user_answers.append((ans, mark))
return render(
request,
"rapids/mark.html",
{
"exam": exam,
"form": form,
"review": review,
"question": question,
"question_number": n,
"question_details": question_details,
"unmarked_answers_bool": unmarked_answers_bool,
"user_answers": unmarked_user_answers,
"review_user_answers": review_user_answers,
"correct_answers": correct_answers,
"half_mark_answers": half_mark_answers,
"incorrect_answers": incorrect_answers,
"unmarked_exam_answers_only": unmarked_exam_answers_only,
},
)
class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView):
model = Rapid
success_url = reverse_lazy("rapids:question_view")
GenericExamViews = ExamViews(
Exam, Rapid, Answer, UserAnswer, "rapids", "rapid", normalise_score=normaliseScore
)
class ExamCreate(ExamCreateBase):
model = Exam
form_class = ExamForm
class ExamUpdate(ExamUpdateBase, AuthorOrCheckerRequiredMixin):
model = Exam
form_class = ExamForm
class ExamDelete(AuthorOrCheckerRequiredMixin, ExamDeleteBase):
model = Exam
class ExamAuthorUpdate(
RevisionMixin,
CheckCanEditMixin,
LoginRequiredMixin,
AuthorRequiredMixin,
UpdateView,
):
model = Exam
form_class = ExamAuthorForm
template_name = "author_form.html"
class ExamMarkersUpdate(
RevisionMixin, CheckCanEditMixin, LoginRequiredMixin, AuthorRequiredMixin, UpdateView
):
model = Exam
form_class = ExamMarkerForm
template_name = "markers_form.html"
class ExamGroupsUpdate(ExamGroupsUpdateBase):
model = Exam
form_class = ExamGroupsForm
class UserAnswerView(LoginRequiredMixin, DetailView):
model = UserAnswer
class UserAnswerTableView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = UserAnswer
table_class = RapidUserAnswerTable
template_name = "rapids/user_answer_question_view.html"
filterset_class = RapidUserAnswerFilter
class UserAnswerDelete(SuperuserRequiredMixin, DeleteView):
model = UserAnswer
template_name = "user_answer_delete.html"
success_url = reverse_lazy("rapids:user_answer_table_view")
GenericViews = GenericViewBase("rapids", Rapid, UserAnswer, Exam)
@user_is_author_or_rapid_checker
def question_anonymise_dicom(request, pk):
question = get_object_or_404(Rapid, pk=pk)
question.anonymise_images()
return redirect("rapids:question_detail", pk=pk)
@user_is_author_or_rapid_checker
def question_save_annotation(request, pk):
if request.method == "POST":
question = get_object_or_404(Rapid, pk=pk)
image_annotations = json.loads(request.POST.get("annotation"))
question_images = question.images.filter(feedback_image=False)
for n, image in enumerate(question_images):
annotation = image_annotations[n]
if annotation != "undefined":
image.image_annotations = annotation
else:
image.image_annotations = ""
image.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@user_is_author_or_rapid_checker
def question_migrate_to_shorts(request, pk):
"""Migrate a single Rapid question to a shorts.Question.
Accessible to question authors and members of the rapid_checker group (or superusers).
"""
rapid = get_object_or_404(Rapid, pk=pk)
from shorts.models import Question as ShortQuestion
try:
q = ShortQuestion.from_rapid(rapid)
except Exception as e:
messages.error(request, f"Migration failed: {e}")
return redirect("rapids:question_detail", pk=rapid.pk)
if q is None:
messages.info(request, "Question skipped (marked normal) — not migrated.")
return redirect("rapids:question_detail", pk=rapid.pk)
messages.success(request, "Migration complete — redirected to new Short question.")
return redirect(q.get_absolute_url())
@login_required
def exam_migrate_rapids_to_shorts(request, pk):
"""Migrate all rapids associated with an Exam to shorts.Question.
Accessible to exam authors and superusers.
"""
exam = get_object_or_404(Exam, pk=pk)
# check permission: exam author or superuser
if not (request.user.is_superuser or exam.author.filter(id=request.user.id).exists()):
raise PermissionDenied()
from shorts.models import Question as ShortQuestion
created = []
skipped = []
failed = []
for rapid in exam.exam_questions.all():
try:
q = ShortQuestion.from_rapid(rapid)
if q is None:
skipped.append(str(rapid.id))
else:
created.append(str(q.id))
except Exception as e:
failed.append(str(rapid.id))
continue
if created:
messages.success(request, f"Created shorts questions: {', '.join(created)}")
if skipped:
messages.info(request, f"Skipped normal rapids (not migrated): {', '.join(skipped)}")
if failed:
messages.error(request, f"Failed to migrate rapids: {', '.join(failed)}")
# Redirect back to the Rapids exam overview (shorts exam may not exist)
return redirect("rapids:exam_overview", pk=exam.pk)
class ExamClone(ExamCloneMixin, ExamCreate):
"""Clone exam view"""
@login_required
def get_question_by_hash(request):
if request.method == "POST":
hash = request.POST.get("hash")
try:
question = RapidImage.objects.get(image_md5_hash=hash)
data = {
"status": "success",
"id": question.pk,
"url": question.rapid.get_absolute_url(),
}
return JsonResponse(data, status=200)
except RapidImage.DoesNotExist:
data = {"status": "success", "id": False}
return JsonResponse(data, status=200)
# TODO: better permissions
@login_required
def question_json_unbased(request, pk):
"""
No (file based) caching is enabled for unbased quesitons
"""
question = get_object_or_404(Rapid, pk=pk)
question_json = question.get_question_json(based=False)
return JsonResponse(question_json)
@login_required
def question_json(request, pk):
question = get_object_or_404(Rapid, pk=pk)
question_json = question.get_question_json(based=True)
return JsonResponse(question_json)
@login_required
def question_viewer(request):
# question = get_object_or_404(Rapid, pk=pk)
ids_param = request.GET.get("ids")
ids = ids_param.split(",") if ids_param else []
return render(
request,
"rapids/question_viewer.html",
{
"ids": ids,
},
)
class AnswerAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# Don't forget to filter out results depending on the visitor !
if not self.request.user.is_authenticated:
return Answer.objects.none()
qs = Answer.objects.all()
if self.q:
qs = qs.filter(answer__istartswith=self.q)
return qs
def confirm_answer(request, answer_id: int):
if request.htmx:
answer = get_object_or_404(Answer, pk=answer_id)
# Check if the user has permission to confirm the answer
if not answer.can_edit(request.user):
return HttpResponse("Invalid permissions", status=403)
answer.clean()
answer.proposed = False
answer.save()
return HttpResponse("Answer accepted")
raise PermissionDenied()
def delete_answer(request, answer_id: int):
if request.htmx:
answer = get_object_or_404(Answer, pk=answer_id)
# Check if the user has permission to confirm the answer
if not answer.can_edit(request.user):
return HttpResponse("Invalid permissions", status=403)
answer.delete()
return HttpResponse("Answer deleted")
raise PermissionDenied()
def question_add_exam(request, question_id: int):
if request.htmx:
question = get_object_or_404(Rapid, pk=question_id)
if not question.can_edit(request.user):
return HttpResponse("Invalid permissions", status=403)
if request.POST:
exam_id = request.POST.get("exam_id", None)
if exam_id is None:
return HttpResponse("No exam id provided", status=400)
exam = get_object_or_404(Exam, pk=exam_id)
if request.POST.get("remove", False) == "true":
question.exams.remove(exam)
return HttpResponse("Question removed from exam.")
else:
question.exams.add(exam)
return HttpResponse("Question added to exam.")
else:
# Return a list of exams that we can add to the question
exams = Exam.objects.filter(author=request.user)
exams = exams.difference(question.exams.all())
if not exams:
return HttpResponse("No exams available to add")
html = "Exams to add to question: <br>"
for exam in exams:
html = html + (f"<span id='htmx-exam-list'><form><input name='exam_id' value='{exam.id}' type='hidden'><button hx-post=\"{request.path}\""
f">{exam.name}: {exam.id}</button></form>"
)
html = html + "<br>Exams to remove from question: <br>"
for exam in question.exams.all():
html = html + (f"<span id='htmx-exam-list'><form><input name='exam_id' value='{exam.id}' type='hidden'><input name='remove' value='true' type='hidden'><button hx-post=\"{request.path}\""
f"hx-confirm='Are you sure you want to remove this exam from the question?'"
f">{exam.name}: {exam.id}</button></form>"
)
html = html + "<button _='on click remove #htmx-exam-list'>Cancel</button>"
return HttpResponse(mark_safe(html))
raise PermissionDenied()
@login_required
def mark2(request, exam_pk, sk, unmarked_exam_answers_only=True, review=False):
"""Improved marking UI (mark2) for rapids."""
exam = get_object_or_404(Exam, pk=exam_pk)
if not GenericExamViews.check_user_marker_access(request.user, exam_pk):
raise PermissionDenied
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
mark_url = "rapids:mark2"
if review:
mark_url = "rapids:mark2"
questions = exam.get_questions()
n = sk
question_details = {
"total": len(questions),
"current": n + 1,
}
try:
question = questions[sk]
except IndexError:
raise Http404("Exam question does not exist")
if request.method == "POST":
form = MarkRapidQuestionForm(request.POST)
if form.is_valid():
if "skip" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n + 1)
cd = form.cleaned_data
marked_answers = json.loads(cd.get("marked_answers"))
to_run = (
("correct", Answer.MarkOptions.CORRECT),
("half-correct", Answer.MarkOptions.HALF_MARK),
("incorrect", Answer.MarkOptions.INCORRECT),
)
for status_string, status in to_run:
for ans in marked_answers[status_string]:
ans = ans.strip()
if ans == "":
continue
a = Answer.objects.filter(answer__iexact=ans, question_id=question.pk).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = ans
a.status = status
a.save()
if "next" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n + 1)
elif "previous" in request.POST:
return redirect(mark_url, exam_pk=exam_pk, sk=n - 1)
else:
form = MarkRapidQuestionForm()
correct_answers = []
half_mark_answers = []
incorrect_answers = []
review_user_answers = []
unmarked_answers_bool = False
unmarked_user_answers = []
if question.normal:
incorrect_answers = question.get_user_answers(exam_pk=exam_pk)
if "" in incorrect_answers:
incorrect_answers.remove("")
if "normal" in incorrect_answers:
incorrect_answers.remove("normal")
else:
if review:
unmarked_user_answers = question.get_user_answers(exam_pk=exam_pk, include_normal=False)
elif unmarked_exam_answers_only:
unmarked_user_answers = question.get_unmarked_user_answers(exam_pk=exam_pk)
else:
unmarked_user_answers = question.get_unmarked_user_answers()
if not unmarked_exam_answers_only:
correct_answers = question.answers.filter(status=Answer.MarkOptions.CORRECT)
half_mark_answers = question.answers.filter(status=Answer.MarkOptions.HALF_MARK)
incorrect_answers = question.answers.filter(status=Answer.MarkOptions.INCORRECT)
if review and not question.normal:
for ans in unmarked_user_answers:
marked_ans = question.answers.filter(answer_compare__iexact=ans).first()
mark = "unmarked"
if marked_ans is not None:
if marked_ans.status == Answer.MarkOptions.CORRECT:
mark = "correct"
elif marked_ans.status == Answer.MarkOptions.HALF_MARK:
mark = "half-correct"
elif marked_ans.status == Answer.MarkOptions.INCORRECT:
mark = "incorrect"
if mark == "unmarked":
unmarked_answers_bool = True
review_user_answers.append((ans, mark))
context = {
"exam": exam,
"form": form,
"review": review,
"question": question,
"question_number": n,
"question_details": question_details,
"unmarked_answers_bool": unmarked_answers_bool,
"user_answers": unmarked_user_answers,
"review_user_answers": review_user_answers,
"correct_answers": correct_answers,
"half_mark_answers": half_mark_answers,
"incorrect_answers": incorrect_answers,
"unmarked_exam_answers_only": unmarked_exam_answers_only,
}
return render(request, "rapids/mark2.html", context)
@login_required
def mark2_overview(request, pk):
"""Overview page for mark2 workflow: lists questions and unmarked counts."""
exam = get_object_or_404(Exam, pk=pk)
if not GenericExamViews.check_user_marker_access(request.user, pk):
raise PermissionDenied
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
questions = exam.get_questions()
question_unmarked_map = []
for q in questions:
count = int(q.get_unmarked_user_answer_count(exam_pk=exam.pk))
question_unmarked_map.append((q, count, count))
return render(request, "rapids/mark2_overview.html", {"exam": exam, "question_unmarked_map": question_unmarked_map})
@login_required
def mark2_toggle_answer(request):
"""Endpoint to toggle/set a single answer mark quickly."""
if request.method != "POST":
return JsonResponse({"status": "error", "message": "POST required"}, status=400)
question_pk = request.POST.get("question_pk") or request.POST.get("question")
answer_text = request.POST.get("answer")
newmark = request.POST.get("newmark")
if not question_pk or not answer_text or not newmark:
return JsonResponse({"status": "error", "message": "missing parameters"}, status=400)
try:
question = Rapid.objects.get(pk=question_pk)
except Rapid.DoesNotExist:
return JsonResponse({"status": "error", "message": "question not found"}, status=404)
exams = list(question.exams.all())
if exams:
exam_pk = exams[0].pk
if not GenericExamViews.check_user_marker_access(request.user, exam_pk):
return JsonResponse({"status": "error", "message": "forbidden"}, status=403)
mark_map = {
"correct": Answer.MarkOptions.CORRECT,
"half-correct": Answer.MarkOptions.HALF_MARK,
"incorrect": Answer.MarkOptions.INCORRECT,
}
exam_pk = request.POST.get('exam_pk')
sk = request.POST.get('sk')
if newmark == 'clear':
a = Answer.objects.filter(answer_compare__iexact=answer_text, question_id=question.pk).first()
if a is not None:
a.delete()
if request.htmx and exam_pk and sk is not None:
try:
return mark2_exam_marked(request, int(exam_pk), int(sk))
except Exception:
return JsonResponse({"status": "ok"})
return JsonResponse({"status": "ok"})
if newmark not in mark_map:
return JsonResponse({"status": "error", "message": "invalid mark"}, status=400)
status_val = mark_map[newmark]
a = Answer.objects.filter(answer_compare__iexact=answer_text, question_id=question.pk).first()
if a is None:
a = Answer()
a.question_id = question.pk
a.answer = answer_text
a.status = status_val
a.save()
if request.htmx and exam_pk and sk is not None:
try:
return mark2_exam_marked(request, int(exam_pk), int(sk))
except Exception:
return JsonResponse({"status": "ok"})
return JsonResponse({"status": "ok"})
@login_required
def mark2_exam_marked(request, exam_pk, sk):
"""Return a partial listing of answers submitted for this question within the given exam."""
exam = get_object_or_404(Exam, pk=exam_pk)
questions = exam.get_questions()
try:
question = questions[sk]
except Exception:
raise Http404("Question not found in exam")
ua_qs = question.cid_user_answers.filter(exam=exam)
order = request.GET.get('order', 'count')
grouped = ua_qs.values('answer_compare').annotate(count=Count('id'))
items = []
for row in grouped:
ans = row['answer_compare']
if ans == 'normal' or ans == '':
continue
cnt = row['count']
a = question.answers.filter(answer_compare__iexact=ans).first()
mark = None
if a is not None:
if a.status == Answer.MarkOptions.CORRECT:
mark = 'correct'
elif a.status == Answer.MarkOptions.HALF_MARK:
mark = 'half-correct'
elif a.status == Answer.MarkOptions.INCORRECT:
mark = 'incorrect'
items.append({'answer': ans, 'count': cnt, 'mark': mark})
if order == 'mark':
weight = {'correct': 3, 'half-correct': 2, 'incorrect': 1, None: 0}
items.sort(key=lambda x: (weight.get(x.get('mark')), x.get('count', 0)), reverse=True)
else:
items.sort(key=lambda x: x.get('count', 0), reverse=True)
context = {'exam': exam, 'question': question, 'items': items, 'sk': sk}
return render(request, 'rapids/partials/_mark2_exam_marked_fragment.html', context)
@login_required
def mark2_edit_answer(request):
"""HTMX-friendly endpoint to edit/create a stored Answer for a question."""
if request.method == 'GET' and request.htmx:
question_pk = request.GET.get('question_pk')
original = request.GET.get('original', '')
exam_pk = request.GET.get('exam_pk')
sk = request.GET.get('sk')
question = get_object_or_404(Rapid, pk=question_pk)
existing = question.answers.filter(answer_compare__iexact=original).first()
context = {
'question': question,
'original': original,
'existing': existing,
'exam_pk': exam_pk,
'sk': sk,
}
return render(request, 'rapids/partials/_mark2_edit_answer_fragment.html', context)
if request.method == 'POST' and request.htmx:
question_pk = request.POST.get('question_pk')
original = request.POST.get('original', '')
new_text = request.POST.get('new_answer', '').strip()
exam_pk = request.POST.get('exam_pk')
sk = request.POST.get('sk')
if not question_pk or not new_text:
return HttpResponse('Missing parameters', status=400)
question = get_object_or_404(Rapid, pk=question_pk)
a = question.answers.filter(answer_compare__iexact=original).first()
if a is None:
a = Answer()
a.question = question
a.answer = new_text
else:
a.answer = new_text
a.save()
if exam_pk and sk is not None:
try:
return mark2_exam_marked(request, int(exam_pk), int(sk))
except Exception:
return HttpResponse('OK')
return HttpResponse('OK')
raise PermissionDenied()
@login_required
def mark2_question_marked(request, pk):
"""Return a partial listing of all stored Answer choices for this question (grouped by status)."""
question = get_object_or_404(Rapid, pk=pk)
correct_answers = question.answers.filter(status=Answer.MarkOptions.CORRECT)
half_mark_answers = question.answers.filter(status=Answer.MarkOptions.HALF_MARK)
incorrect_answers = question.answers.filter(status=Answer.MarkOptions.INCORRECT)
context = {
'question': question,
'correct_answers': correct_answers,
'half_mark_answers': half_mark_answers,
'incorrect_answers': incorrect_answers,
}
return render(request, 'rapids/partials/_mark2_question_marked_fragment.html', context)