Files
penracourses/anatomy/views.py
T

1762 lines
64 KiB
Python

from django.forms.models import model_to_dict
from django.shortcuts import render, get_object_or_404, redirect
from django.views.decorators.csrf import csrf_exempt
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 FieldError, 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.db.models.functions import Lower
from django.db.models import Prefetch
from django.core.cache import cache
from django.urls import reverse_lazy, reverse
from django.http import Http404, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from dal import autocomplete
from django.conf import settings
from django.utils.html import format_html_join
from django.utils.safestring import mark_safe
from django.template.loader import render_to_string
from .forms import (
AnatomyAnswerForm,
AnatomyQuestionAnswerForm,
AnswerFormSet,
AnswerUpdateFormSet,
BodyPartForm,
ExamAuthorForm,
ExamGroupsForm,
ExamMarkerForm,
ExaminationForm,
MarkAnatomyQuestionForm,
AnatomyQuestionForm,
ExamForm,
SuggestIncorrectAnswerForm,
PrimaryAnswerForm,
)
from .models import (
AnatomyQuestion,
BodyPart,
UserAnswer,
# Examination,
Structure,
Exam,
Answer,
# HalfMarkAnswers,
# IncorrectAnswers,
)
from generic.widgets import ExamSearchWidget
class AddExamForm(forms.Form):
exam = forms.ModelChoiceField(queryset=Exam.objects.none(), required=True, widget=ExamSearchWidget(exam_model=Exam), label="Exam to add")
def __init__(self, *args, user=None, **kwargs):
super().__init__(*args, **kwargs)
if user is None:
qs = Exam.objects.none()
elif user.groups.filter(name="anatomy_checker").exists():
qs = Exam.objects.all()
else:
qs = Exam.objects.filter(author__id=user.id) | Exam.objects.filter(open_access=True)
self.fields["exam"].queryset = qs.distinct()
from generic.models import CidUser, Examination
from generic.views import AuthorRequiredMixin, ExamCloneMixin, ExamCreateBase, ExamDeleteBase, ExamGroupsUpdateBase, ExamUpdateBase, ExamViews, GenericViewBase, UpdateQuestionMixin
from reversion.views import RevisionMixin
from .decorators import user_is_author_or_anatomy_checker
from .tables import AnatomyQuestionTable, AnatomyUserAnswerTable
from .filters import AnatomyQuestionFilter, AnatomyUserAnswerFilter
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
from collections import defaultdict
import os
import json
import statistics
import plotly.express as px
from helpers.images import image_as_base64
from django.template.defaulttags import register
from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
from django.db.models import Count
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if self.request.user.groups.filter(name="anatomy_checker").exists():
return obj
if self.request.user not in obj.author.all():
raise PermissionDenied() # or Http404
return obj
@register.filter
def get_item(dictionary, key, default=None):
"""Safe template filter to fetch an item by key/index from various containers.
Supports dict-like objects (uses .get), lists/tuples (indexing by integer
keys), and other indexable objects. Returns ``default`` when the lookup
fails or the input is None.
"""
if dictionary is None:
return default
# If it's dict-like, prefer .get
try:
if hasattr(dictionary, "get") and callable(getattr(dictionary, "get")):
# Allow .get default behaviour
try:
return dictionary.get(key, default)
except Exception:
# Fall through to other strategies
pass
# If it's a list/tuple and key is integer-like, return by index
if isinstance(dictionary, (list, tuple)):
try:
idx = int(key)
except Exception:
return default
try:
return dictionary[idx]
except Exception:
return default
# Generic attempt: try __getitem__ with key (handles QueryDict, lists with int keys, etc.)
try:
return dictionary[key]
except Exception:
return default
except Exception:
return default
def user_is_admin(user):
if user:
if user.pk == 1:
return True
return False
def question_view(request):
questions = AnatomyQuestion.objects.all()
if not request.user.groups.filter(name="anatomy_checker").exists():
questions = questions.filter(open_access=False)
# raise PermissionDenied()
return render(request, "anatomy/question_view.html", {"questions": questions})
@login_required
def answer_question(request, pk):
question = get_object_or_404(AnatomyQuestion, pk=pk)
answer = question.user_answers.filter(
user=request.user
).first() # .filter(user=User)
if request.method == "POST":
if answer:
form = AnatomyAnswerForm(request.POST, instance=answer)
else:
form = AnatomyAnswerForm(request.POST)
if form.is_valid():
answer = form.save(commit=False)
answer.user = request.user
answer.question = question
# answer.published_date = timezone.now()
answer.save()
return redirect("question_detail", pk=pk)
else:
form = AnatomyAnswerForm(instance=answer)
return render(
request, "anatomy/answer_question.html", {"form": form, "question": question}
)
# @login_required
# def exam_take(request, pk, sk):
# """
# Allows taking of the exam on the django server (when logged in)
#
# No longer used (deprecated in favour of using RTS)
# """
# exam = get_object_or_404(Exam, pk=pk)
#
# questions = exam.exam_questions.all()
#
# try:
# question = questions[sk]
# except IndexError:
# raise Http404("Exam question does not exist")
#
# n = sk
#
# question_details = {
# "total": len(questions),
# "current": n + 1,
# }
#
# # Get data for flagged
# # answered_questions = UserAnswer.objects.filter(user=request.user).values_list("question__pk", flat=True)
# user_answer_data = UserAnswer.objects.filter(user=request.user).values_list(
# "question__pk", "answer", "flagged"
# )
# # flagged_questions = UserAnswer.objects.filter(user=request.user, flagged=True).values_list("question__pk", flat=True)
# # flagged_questions I= UserAnswer.filter(user=request.user, ).values_list("question__pk", flat=True)
#
# answered_questions = set()
# flagged_questions = set()
# for ans_pk, answer, flagged in user_answer_data:
# if len(answer.strip()) > 0:
# answered_questions.add(ans_pk)
# if flagged:
# flagged_questions.add(ans_pk)
#
# # u = question.user_answers
# answer = question.user_answers.filter(
# user=request.user
# ).first() # .filter(user=User)
# if request.method == "POST":
# if answer:
# form = AnatomyAnswerForm(request.POST, instance=answer)
# else:
# form = AnatomyAnswerForm(request.POST)
# if form.is_valid():
# answer = form.save(commit=False)
# answer.user = request.user
# answer.question = question
# # answer.published_date = timezone.now()
# answer.save()
# if "next" in request.POST:
# return redirect("anatomy:exam_take", pk=pk, sk=n + 1)
# elif "previous" in request.POST:
# return redirect("anatomy:exam_take", pk=pk, sk=n - 1)
# else:
# form = AnatomyAnswerForm(instance=answer)
# return render(
# request,
# "anatomy/exam.html",
# {
# "exam": exam,
# "form": form,
# "question": question,
# "question_details": question_details,
# "questions": questions,
# "flagged_questions": flagged_questions,
# "answered_questions": answered_questions,
# "answer": answer,
# },
# )
# def flag_question(request):
# pk = request.GET.get("pk", None)
# ans = UserAnswer.objects.filter(user=request.user, question__pk=pk).first()
# ans.flagged = not ans.flagged
# ans.save()
# data = {"flagged": ans.flagged}
# return JsonResponse(data)
@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 = "anatomy:mark"
if review:
mark_url = "anatomy:mark_review"
elif not unmarked_exam_answers_only:
mark_url = "anatomy:mark_all"
questions = exam.get_questions().prefetch_related("answers")
n = sk
question_details = {
"total": len(questions),
"current": n + 1,
}
try:
question: AnatomyQuestion = questions[sk]
except IndexError:
raise Http404("Exam question does not exist")
if request.method == "POST":
form = MarkAnatomyQuestionForm(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 = MarkAnatomyQuestionForm()
# 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)
correct_answers = []
half_mark_answers = []
incorrect_answers = []
review_user_answers = []
unmarked_answers_bool = False
unmarked_user_answers = []
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)
answer_suggest_incorrect: list = []
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))
else:
for ans in unmarked_user_answers:
for suggest_incorrect in question.answer_suggest_incorrect:
if suggest_incorrect in ans:
answer_suggest_incorrect.append(ans)
break
words_present: set = set()
words_present_in_correct: set = set()
for ans in correct_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_present_in_correct.update(answer_words)
for ans in half_mark_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_present_in_correct.update(answer_words)
for ans in incorrect_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_suggest_incorrect = words_present - words_present_in_correct - set(question.answer_suggest_incorrect)
return render(
request,
"anatomy/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,
"answer_suggest_incorrect": answer_suggest_incorrect,
"words_suggest_incorrect": words_suggest_incorrect,
},
)
@login_required
def exam_review_question_responses(request, pk: int, q_index: int):
"""Return the responses partial for a given anatomy exam question (HTMX-friendly)."""
exam = get_object_or_404(Exam, pk=pk)
questions = exam.get_questions()
try:
question = questions[q_index]
except Exception:
raise Http404("Question not found in exam")
try:
exam_user_answers_qs = question.cid_user_answers.filter(exam=exam).select_related("user")
except Exception:
exam_user_answers_qs = question.cid_user_answers.all()
# Allow an HTMX-driven reveal of respondent names. Default: anonymised.
reveal = request.GET.get("reveal", "0")
reveal_respondents = str(reveal).lower() in ("1", "true", "yes", "on")
# Build a small list that includes a sensible display name for each respondent.
# If the answer is from a regular user use their full name/username. If it's a CID
# candidate try to resolve the CidUser and show CID + name if available. If neither
# is present fall back to empty/anonymous.
# Optimize CID lookups: collect distinct CID values and fetch all CidUser rows
# in a single query to avoid an N+1 query pattern.
cids_qs = exam_user_answers_qs.values_list("cid", flat=True).distinct()
cids = [c for c in cids_qs if c is not None and str(c).strip() != ""]
cid_map = {}
if cids:
for cu in CidUser.objects.filter(cid__in=cids):
cid_map[cu.cid] = cu
exam_user_answers = []
for ua in exam_user_answers_qs:
# default display
display = None
is_cid = False
try:
if getattr(ua, "user", None):
user = ua.user
# prefer full name where available
full = getattr(user, "get_full_name", None)
if full:
name = user.get_full_name() or user.username
else:
name = getattr(user, "username", str(user))
display = name
elif getattr(ua, "cid", None) is not None:
cid_val = ua.cid
cu = cid_map.get(cid_val)
if cu is not None and cu.name:
display = f"CID{cid_val} ({cu.name})"
else:
display = f"CID{cid_val}"
is_cid = True
except Exception:
display = None
exam_user_answers.append({"ua": ua, "display_name_raw": display, "is_cid": is_cid})
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"app_name": "anatomy",
"exam_user_answers": exam_user_answers,
"reveal_respondents": reveal_respondents,
}
return render(request, "anatomy/partials/exam_review_question_responses_fragment.html", context)
@login_required
def exam_review_question_summary(request, pk: int, q_index: int):
"""Return aggregated response summary partial for anatomy question (HTMX-friendly)."""
exam = get_object_or_404(Exam, pk=pk)
questions = exam.get_questions()
try:
question = questions[q_index]
except Exception:
raise Http404("Question not found in exam")
ua_qs = question.cid_user_answers.filter(exam=exam)
total_responses = ua_qs.count()
# counts by score values ('2','1','0','')
score_counts = {}
for s, c in ua_qs.values_list("score").annotate(count=Count("id")) if False else []:
pass
# simpler: compute via values and normalise unmarked -> 'empty'
def _norm_score_key(v):
# Treat None or empty-string scores as the 'empty' bucket
if v is None:
return "empty"
if isinstance(v, str) and v.strip() == "":
return "empty"
return str(v)
score_counts = {_norm_score_key(row["score"]): row["count"] for row in ua_qs.values("score").annotate(count=Count("id"))}
# top submitted answers (by answer_compare) with dominant score per answer
# First, get counts grouped by answer_compare and score
grouped = ua_qs.values("answer_compare", "score").annotate(count=Count("id"))
# Aggregate totals and score breakdown per answer_compare
answer_map = {}
for row in grouped:
ans = row["answer_compare"]
sc = _norm_score_key(row["score"])
cnt = row["count"]
if ans not in answer_map:
answer_map[ans] = {"total": 0, "score_counts": {}}
answer_map[ans]["total"] += cnt
answer_map[ans]["score_counts"][sc] = answer_map[ans]["score_counts"].get(sc, 0) + cnt
# Build list sorted by total count desc, and choose a dominant score per answer
top_answers = []
for ans, meta in answer_map.items():
# pick dominant score by highest count; tie-break prefer higher numeric score
ans_score_counts = meta["score_counts"]
# sort by (count desc, score desc) where 'empty' maps to -1
def score_key(item):
s, c = item
# Map 'empty' (or any non-numeric key) to a low value so numeric scores win ties
if s == "empty":
s_val = -1
else:
try:
s_val = int(s)
except Exception:
s_val = -1
return (c, s_val)
dominant = max(ans_score_counts.items(), key=score_key)[0]
top_answers.append({"answer_compare": ans, "count": meta["total"], "dominant_score": dominant})
top_answers.sort(key=lambda x: x["count"], reverse=True)
top_answers = top_answers[:10]
correct_count = None
correct_pct = None
# compute correct based on matching marked answers if available
if total_responses:
# count scores equal to '2' as correct
correct_count = score_counts.get("2", 0)
correct_pct = round(100.0 * correct_count / total_responses, 1) if total_responses else None
# compute percentage distribution for score buckets for the summary bar chart
score_keys = ["2", "1", "0", "empty"]
score_pcts = {}
if total_responses:
for k in score_keys:
score_pcts[k] = round(100.0 * score_counts.get(k, 0) / total_responses, 1)
else:
for k in score_keys:
score_pcts[k] = 0.0
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"app_name": "anatomy",
"exam_response_total": total_responses,
"exam_response_score_counts": score_counts,
"exam_response_score_pcts": score_pcts,
"exam_response_correct_count": correct_count,
"exam_response_correct_pct": correct_pct,
"top_answers": top_answers,
}
return render(request, "anatomy/partials/exam_review_question_summary_fragment.html", context)
@login_required
def mark2(request, exam_pk, sk, unmarked_exam_answers_only=True, review=False):
"""Improved marking UI (mark2).
Same core behaviour as `mark` but renders a modern, HTMX-friendly template
and exposes a small toggle endpoint for quick per-answer marking.
"""
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 = "anatomy:mark2"
if review:
mark_url = "anatomy:mark2"
questions = exam.get_questions().prefetch_related("answers")
n = sk
question_details = {
"total": len(questions),
"current": n + 1,
}
try:
question: AnatomyQuestion = questions[sk]
except IndexError:
raise Http404("Exam question does not exist")
# For consistency with the existing mark view we accept a standard POST form
# but the preferred workflow is immediate HTMX/ajax toggles via mark2_toggle_answer.
if request.method == "POST":
form = MarkAnatomyQuestionForm(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 = MarkAnatomyQuestionForm()
# Reuse existing collections for display
correct_answers = []
half_mark_answers = []
incorrect_answers = []
review_user_answers = []
unmarked_answers_bool = False
unmarked_user_answers = []
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)
answer_suggest_incorrect: list = []
if review:
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))
else:
for ans in unmarked_user_answers:
for suggest_incorrect in question.answer_suggest_incorrect:
if suggest_incorrect in ans:
answer_suggest_incorrect.append(ans)
break
words_present: set = set()
words_present_in_correct: set = set()
for ans in correct_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_present_in_correct.update(answer_words)
for ans in half_mark_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_present_in_correct.update(answer_words)
for ans in incorrect_answers:
answer_words = set(ans.get_constituent_words())
words_present.update(answer_words)
words_suggest_incorrect = words_present - words_present_in_correct - set(question.answer_suggest_incorrect)
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,
"answer_suggest_incorrect": answer_suggest_incorrect,
"words_suggest_incorrect": words_suggest_incorrect,
}
return render(request, "anatomy/mark2.html", context)
@login_required
def mark2_overview(request, pk):
"""Overview page for mark2 workflow: lists questions and unmarked counts and links into mark2."""
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")
# Only allow authors or designated markers
if request.user not in exam.author.all():
if not GenericExamViews.check_user_marker_access(request.user, pk):
raise PermissionDenied
# For anatomy prefer prefetch similar to generic.mark_overview
questions = (
exam.exam_questions.select_related(
"modality",
"structure",
"body_part",
"region",
"examination",
"question_type",
)
.all()
.prefetch_related(
Prefetch(
"answers",
queryset=Answer.objects.filter(proposed=False, status=Answer.MarkOptions.CORRECT),
to_attr="prefetched_primary_answer",
),
)
.order_by("examquestiondetail__sort_order")
)
# Handle double marking which needs per-marker counts
try:
if exam.double_mark:
question_unmarked_map = []
for q in questions:
question_unmarked_map.append(
(
q,
int(q.get_unmarked_user_answer_count(exam_pk=exam.pk)),
int(q.get_unmarked_user_answer_count(exam_pk=exam.pk, marker=request.user)),
)
)
return render(request, "anatomy/mark2_overview.html", {"exam": exam, "question_unmarked_map": question_unmarked_map})
except AttributeError:
pass
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, "anatomy/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.
Expects POST with form fields: question_pk, answer (text), newmark ("correct"|"half-correct"|"incorrect").
Returns JSON status.
"""
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 = AnatomyQuestion.objects.get(pk=question_pk)
except AnatomyQuestion.DoesNotExist:
return JsonResponse({"status": "error", "message": "question not found"}, status=404)
# permission checks: ensure user can mark the exam(s) this question belongs to
# We can't easily infer exam_pk here, but ask GenericExamViews.check_user_marker_access
# using any exam that contains this question is adequate if present.
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)
# Map mark strings to model constants
mark_map = {
"correct": Answer.MarkOptions.CORRECT,
"half-correct": Answer.MarkOptions.HALF_MARK,
"incorrect": Answer.MarkOptions.INCORRECT,
}
# Support clearing a stored answer (delete) via 'clear'
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:
# permission: ensure user can edit this answer
if not a.can_edit(request.user):
return JsonResponse({"status": "error", "message": "forbidden"}, status=403)
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]
# Find existing Answer by answer_compare or create
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.
Shows distinct submitted answers with counts and any existing mark/status.
"""
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)
# Allow client to request ordering by 'count' (default) or by 'mark' (badge)
order = request.GET.get('order', 'count')
# Group distinct submitted answers and count occurrences
grouped = ua_qs.values('answer_compare').annotate(count=Count('id'))
items = []
for row in grouped:
ans = row['answer_compare']
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})
# Sort items according to requested ordering
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, 'anatomy/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.
GET: returns a small form fragment.
POST: updates/creates the Answer and, if exam_pk/sk provided, returns the
refreshed exam-marked fragment so the UI updates in-place.
"""
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(AnatomyQuestion, pk=question_pk)
# Try to find an existing Answer by compare
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, 'anatomy/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(AnatomyQuestion, 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 we have exam context, refresh the exam-marked fragment for that question
if exam_pk and sk is not None:
# delegate to mark2_exam_marked to re-render the card
try:
return mark2_exam_marked(request, int(exam_pk), int(sk))
except Exception:
return HttpResponse('OK')
return HttpResponse('OK')
# Fallback: forbidden for non-HTMX or other methods
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(AnatomyQuestion, 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, 'anatomy/partials/mark2_question_marked_fragment.html', context)
class AnatomyQuestionCreateBase(RevisionMixin, LoginRequiredMixin, CreateView):
model = AnatomyQuestion
form_class = AnatomyQuestionForm
def get_form_kwargs(self):
kwargs = super(AnatomyQuestionCreateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super(AnatomyQuestionCreateBase, self).get_context_data(**kwargs)
if "object" in context:
context["question"] = context["object"]
if self.request.POST:
context["answer_formset"] = AnswerFormSet(
self.request.POST, self.request.FILES
)
else:
context["answer_formset"] = AnswerFormSet()
return context
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
response = super().form_valid(form)
form.instance.author.add(self.request.user.id)
#context = self.get_context_data(form=form)
#formset = context["answer_formset"]
#if formset.is_valid():
# response = super().form_valid(form)
# formset.instance = self.object
# 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("anatomy:question_clone", pk=self.object.pk)
#else:
# return super().form_invalid(form)
# 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("anatomy:question_clone", pk=self.object.pk)
def get_success_url(self) -> str:
return reverse("anatomy:question_detail", kwargs={"pk":self.object.pk})
class AnatomyQuestionCreate(AnatomyQuestionCreateBase):
# initial = {'laterality': AnatomyQuestion.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.AnatomyQuestion_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 AnatomyQuestionAnswerUpdate(RevisionMixin, AuthorOrCheckerRequiredMixin, UpdateView):
model = AnatomyQuestion
form_class = AnatomyQuestionAnswerForm
template_name = "anatomy/anatomyquestionanswer_form.html"
def get_context_data(self, **kwargs):
context = super(AnatomyQuestionAnswerUpdate, 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 AnatomyQuestionUpdate(LoginRequiredMixin, AuthorOrCheckerRequiredMixin,
class AnatomyQuestionUpdate(UpdateQuestionMixin):
model = AnatomyQuestion
form_class = AnatomyQuestionForm
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"]
return super().form_valid(form)
class QuestionClone(AnatomyQuestionCreateBase):
"""Clones a existing question"""
def get_initial(self):
# print(self.request)
old_object = get_object_or_404(AnatomyQuestion, pk=self.kwargs["pk"])
if self.request.user not in old_object.get_author_objects() and not self.request.user.is_superuser:
raise PermissionDenied() # or Http404
initial_data = model_to_dict(old_object, exclude=["id"])
# We don't want to clone the exam details
# exams = old_object.exams.all().values_list("id", flat=True)
# initial_data["exams"] = list(exams)
# TODO: clone answers?
return initial_data
@login_required
def create_body_part(request):
form = BodyPartForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_body_part");</script>'
% (instance.pk, instance)
)
return render(
request, "anatomy/create_simple.html", {"form": form, "name": "BodyPart"}
)
@csrf_exempt
def get_body_part_id(request):
if request.accepts("application/json")():
body_part_name = request.GET["body_part_name"]
body_part_id = BodyPart.objects.get(name=body_part_name).id
data = {
"body_part_id": body_part_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, "anatomy/create_simple.html", {"form": form, "name": "Examination"}
)
@csrf_exempt
def get_examination_id(request):
if request.accepts("application/json")():
examination_name = request.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("/")
@csrf_exempt
def get_structure_id(request):
if request.accepts("application/json")():
structure_name = request.GET["structure_name"]
structure_id = Structure.objects.get(name=structure_name).id
data = {
"structure_id": structure_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
class AnatomyQuestionView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = AnatomyQuestion
table_class = AnatomyQuestionTable
template_name = "question_table_view.html"
filterset_class = AnatomyQuestionFilter
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["app_name"] = "anatomy"
return context
# def get_queryset(self):
# qs = super().get_queryset()
# qs = qs.select_related('modality', 'question_type')
# return qs
@user_is_author_or_anatomy_checker
def question_suggest_incorrect_answers(request, pk):
question = get_object_or_404(AnatomyQuestion, pk=pk)
# helper: build candidate suggestions (n-grams + prefixes)
def build_candidates(question_obj):
from django.db.models import Count
import re
def tokenize(s: str):
return re.findall(r"\w+", (s or "").lower())
candidates_qs = (
question_obj.cid_user_answers.values("answer_compare")
.annotate(count=Count("id"))
.order_by("-count")
)
existing = set([s for s in question_obj.answer_suggest_incorrect if s])
marked_texts = [m.lower() for m in question_obj.get_marked_answers()]
ngram_map = {}
ngram_weight = {}
prefix_map = {}
prefix_weight = {}
for row in candidates_qs:
ans = (row.get("answer_compare") or "").strip()
cnt = int(row.get("count") or 0)
if not ans:
continue
tokens = tokenize(ans)
max_n = min(4, len(tokens))
for n in range(1, max_n + 1):
for i in range(len(tokens) - n + 1):
ngram = " ".join(tokens[i : i + n])
ngram_map.setdefault(ngram, set()).add(ans)
ngram_weight[ngram] = ngram_weight.get(ngram, 0) + cnt
for tok in tokens:
tlen = len(tok)
for plen in range(3, min(6, tlen) + 1):
pref = tok[:plen]
prefix_map.setdefault(pref, set()).add(ans)
prefix_weight[pref] = prefix_weight.get(pref, 0) + cnt
raw_candidates = []
for ngram, ans_set in ngram_map.items():
if ngram in existing:
continue
excluded = False
for mt in marked_texts:
if ngram in mt:
excluded = True
break
if excluded:
continue
coverage = len(ans_set)
if coverage < 1:
continue
raw_candidates.append(
{
"text": ngram,
"coverage": coverage,
"count": ngram_weight.get(ngram, 0),
"words": len(ngram.split()),
"type": "ngram",
}
)
for pref, ans_set in prefix_map.items():
if pref in existing:
continue
excluded = False
for mt in marked_texts:
if pref in mt:
excluded = True
break
if excluded:
continue
coverage = len(ans_set)
if coverage < 1:
continue
raw_candidates.append(
{
"text": pref,
"coverage": coverage,
"count": prefix_weight.get(pref, 0),
"words": 1,
"type": "prefix",
}
)
raw_candidates.sort(key=lambda x: (x["words"], -x["coverage"], -x["count"]))
selected = []
selected_texts = []
for c in raw_candidates:
t = c["text"]
skip = False
for s in selected_texts:
if s in t:
skip = True
break
if not skip:
selected.append({"answer": t, "count": c["count"], "type": c.get("type", "ngram")})
selected_texts.append(t)
return selected
# Support HTMX partial rendering with suggestion candidates and inline add
if request.method == "POST":
# Support remove action from the edit fragment
if request.htmx and request.POST.get("remove"):
to_remove = request.POST.get("remove").strip()
if to_remove:
existing = [s for s in question.answer_suggest_incorrect if s]
if to_remove in existing:
existing = [s for s in existing if s != to_remove]
question.answer_suggest_incorrect = existing
question.save()
# Recompute candidates and render the edit fragment
form = SuggestIncorrectAnswerForm(instance=question)
candidates = build_candidates(question)
context = {"form": form, "question": question, "candidates": candidates}
return render(request, "anatomy/partials/suggest_incorrect_fragment.html", context)
# Support single-candidate add via HTMX (button posts 'suggestion').
# Edits should only be performed via the edit fragment; when a
# suggestion is posted we re-render the edit fragment so the
# user remains in edit mode and sees the updated candidate list.
if request.htmx and request.POST.get("suggestion"):
suggestion = request.POST.get("suggestion").strip()
if suggestion:
existing = [s for s in question.answer_suggest_incorrect if s]
if suggestion not in existing:
existing.append(suggestion)
question.answer_suggest_incorrect = existing
question.save()
# If this is a quick add (from the answers table) return the collapsed partial
if request.POST.get("quick"):
return render(request, "anatomy/partials/suggest_incorrect_collapsed.html", {"question": question})
# Recompute candidates and render the edit fragment
form = SuggestIncorrectAnswerForm(instance=question)
candidates = build_candidates(question)
context = {"form": form, "question": question, "candidates": candidates}
return render(request, "anatomy/partials/suggest_incorrect_fragment.html", context)
# Otherwise treat as the full form submission
form = SuggestIncorrectAnswerForm(request.POST, instance=question)
if form.is_valid():
obj = form.save()
obj.save()
# If HTMX requested, return the fragment or collapsed partial
if request.htmx:
# If the close button was used, return the collapsed small partial
if request.POST.get("close"):
return render(request, "anatomy/partials/suggest_incorrect_collapsed.html", {"question": question})
# Otherwise re-render the edit fragment so the user stays in edit mode
form = SuggestIncorrectAnswerForm(instance=question)
candidates = build_candidates(question)
context = {"form": form, "question": question, "candidates": candidates}
return render(request, "anatomy/partials/suggest_incorrect_fragment.html", context)
return render(request, "anatomy/question_detail.html#suggest-incorrect-form", {"question": question})
return HttpResponse("Invalid")
else:
form = SuggestIncorrectAnswerForm(instance=question)
candidates = build_candidates(question)
context = {"form": form, "question": question, "candidates": candidates}
if request.htmx:
return render(request, "anatomy/partials/suggest_incorrect_fragment.html", context)
return render(request, "anatomy/suggest_incorrect.html", context)
return HttpResponse("Error", status=400)
@user_is_author_or_anatomy_checker
def question_suggest_incorrect_collapsed(request, pk):
"""Return the small collapsed 'Edit suggestions' box as a partial."""
question = get_object_or_404(AnatomyQuestion, pk=pk)
context = {"question": question}
return render(request, "anatomy/partials/suggest_incorrect_collapsed.html", context)
@user_is_author_or_anatomy_checker
def question_save_annotation(request, pk):
if request.method == "POST":
question = get_object_or_404(AnatomyQuestion, pk=pk)
question.image_annotations = request.POST.get("annotation")
question.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@user_is_author_or_anatomy_checker
def question_clear_annotation(request, pk):
"""Clear the stored image annotations for a question."""
if request.method == "POST":
question = get_object_or_404(AnatomyQuestion, pk=pk)
question.image_annotations = ""
question.save(update_fields=["image_annotations"])
return JsonResponse({"status": "success"}, status=200)
return JsonResponse({"status": "error"}, status=400)
GenericExamViews = ExamViews(
Exam, AnatomyQuestion, Answer, UserAnswer, "anatomy", "anatomy"
)
class UserAnswerView(SuperuserRequiredMixin, DetailView):
model = UserAnswer
class UserAnswerTableView(SuperuserRequiredMixin, SingleTableMixin, FilterView):
model = UserAnswer
table_class = AnatomyUserAnswerTable
template_name = "anatomy/user_answer_question_view.html"
filterset_class = AnatomyUserAnswerFilter
class UserAnswerDelete(SuperuserRequiredMixin, DeleteView):
model = UserAnswer
template_name = "user_answer_delete.html"
success_url = reverse_lazy("anatomy:user_answer_table_view")
class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView):
model = AnatomyQuestion
success_url = reverse_lazy("anatomy:question_view")
class ExamCreate(ExamCreateBase):
model = Exam
form_class = ExamForm
class ExamUpdate(ExamUpdateBase, AuthorOrCheckerRequiredMixin):
model = Exam
form_class = ExamForm
#def get_context_data(self, **kwargs):
# context = super().get_context_data(**kwargs)
# context["can_edit"] = self.object.check_user_can_edit(self.request.user)
# return context
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
GenericViews = GenericViewBase("anatomy", AnatomyQuestion, UserAnswer, Exam)
class ExamClone(ExamCloneMixin, ExamCreate):
"""Clone exam view"""
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(AnatomyQuestion, pk=question_id)
if not question.can_edit(request.user):
return HttpResponse("Invalid permissions", status=403)
# POST: handle add/remove; if exam_id provided use existing behaviour
if request.method == "POST":
exam_id = request.POST.get("exam_id", None)
if exam_id:
exam = get_object_or_404(Exam, pk=exam_id)
if request.POST.get("remove", False) == "true":
question.exams.remove(exam)
else:
question.exams.add(exam)
# return out-of-band swap: update exam links and remove the editor
links_html = render_to_string("anatomy/partials/exam_links.html", {"question": question}, request=request)
oob_remove = '<div id="exam-list" hx-swap-oob="outerHTML"></div>'
return HttpResponse(mark_safe(links_html + oob_remove))
# else try to process from the search form or widget-provided values
# The ExamSearchWidget may submit a list; accept first value if present.
exam_vals = request.POST.getlist('exam') if hasattr(request.POST, 'getlist') else []
if not exam_vals:
# also accept single 'exam' param
single = request.POST.get('exam')
if single:
exam_vals = [single]
if exam_vals:
try:
exam = get_object_or_404(Exam, pk=int(exam_vals[0]))
except (ValueError, TypeError):
exam = None
if exam:
if request.POST.get("remove", False) == "true":
question.exams.remove(exam)
else:
question.exams.add(exam)
links_html = render_to_string("anatomy/partials/exam_links.html", {"question": question}, request=request)
oob_remove = '<div id="exam-list" hx-swap-oob="outerHTML"></div>'
return HttpResponse(mark_safe(links_html + oob_remove))
# else try to process via bound form (fallback)
form = AddExamForm(request.POST, user=request.user)
if form.is_valid():
exam = form.cleaned_data.get("exam")
if request.POST.get("remove", False) == "true":
question.exams.remove(exam)
else:
question.exams.add(exam)
links_html = render_to_string("anatomy/partials/exam_links.html", {"question": question}, request=request)
oob_remove = '<div id="exam-list" hx-swap-oob="outerHTML"></div>'
return HttpResponse(mark_safe(links_html + oob_remove))
else:
# render form with errors
return render(request, "anatomy/partials/exam_edit_fragment.html", {"form": form, "question": question})
# GET: render the fragment with the add form and current exams
form = AddExamForm(user=request.user)
return render(request, "anatomy/partials/exam_edit_fragment.html", {"form": form, "question": question})
raise PermissionDenied()
@login_required
def primary_answer_htmx(request, pk: int):
"""HTMX endpoint to render or update a question's primary answer.
GET: render the form for selecting/creating primary answer.
POST: create/select answer and set `question.primary_answer`, then
return the display fragment.
"""
if not request.htmx:
raise PermissionDenied()
question = get_object_or_404(AnatomyQuestion, pk=pk)
if not question.can_edit(request.user):
return HttpResponse("Invalid permissions", status=403)
if request.method == "POST":
form = PrimaryAnswerForm(request.POST, question=question)
if form.is_valid():
new_text = form.cleaned_data.get("new_answer", "").strip()
existing = form.cleaned_data.get("existing_answer", "")
if new_text:
# create a new Answer and mark it CORRECT
ans = Answer(question=question, answer=new_text, proposed=False, status=Answer.MarkOptions.CORRECT)
ans.save()
question.primary_answer = ans
question.save(update_fields=["primary_answer"])
elif existing:
try:
aid = int(existing)
ans = Answer.objects.get(pk=aid, question=question)
question.primary_answer = ans
question.save(update_fields=["primary_answer"])
except (ValueError, Answer.DoesNotExist):
return HttpResponse("Invalid answer choice", status=400)
# return display fragment
return render(request, "anatomy/partials/primary_answer_display.html", {"question": question, "can_edit": question.can_edit(request.user)})
else:
return HttpResponse("Invalid form", status=400)
# Handle cancel request to restore original state
if request.method == "GET" and request.GET.get("cancel"):
if question.primary_answer:
return render(request, "anatomy/partials/primary_answer_display.html", {"question": question, "can_edit": question.can_edit(request.user)})
else:
return render(request, "anatomy/partials/primary_answer_empty.html", {"question": question, "can_edit": question.can_edit(request.user)})
# GET - render form
form = PrimaryAnswerForm(question=question)
return render(request, "anatomy/partials/primary_answer_form.html", {"form": form, "question": question, "can_edit": question.can_edit(request.user)})