ross/penracourses
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
321c7ef8951ffdf64790f7b2c34ec4ea91629d0c
penracourses/generic/views.py
T

5450 lines
196 KiB
Python
Raw Blame History

from collections import defaultdict, Counter
from pathlib import Path
import secrets
import statistics
import threading
from typing import Any
from dal import autocomplete
from django.contrib.auth.models import User, Group
from django.contrib.contenttypes.models import ContentType
from django.forms.models import model_to_dict
from django.db.models.functions import Coalesce, Lower
from django.shortcuts import render, get_object_or_404, redirect
from django.contrib.auth.decorators import login_required, user_passes_test
from django.urls import reverse_lazy
from django.urls.base import reverse
from django.utils import timezone
from django.utils.html import format_html
from django.views.decorators.csrf import csrf_exempt
from django.core.exceptions import PermissionDenied, FieldError
from django.http import Http404, HttpRequest, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
from django.core.exceptions import ValidationError
from django.core.validators import validate_email
from django.utils.decorators import method_decorator
from django.core.cache import cache
import json
import urllib
import urllib.parse
from django.views import View
from django.views.generic.edit import CreateView, UpdateView, DeleteView
from django.views.generic.list import ListView
from django.views.generic.detail import DetailView
from django_filters.views import FilterView
from django_filters import (
FilterSet,
OrderingFilter,
ModelMultipleChoiceFilter,
DateFromToRangeFilter,
)
from django_filters.widgets import DateRangeWidget
from django_tables2.views import SingleTableMixin
from reversion.views import RevisionMixin
from atlas.models import CaseCollection, CaseDetail
from generic.decorators import user_is_cid_user_manager
from generic.filters import CidUserFilter, ExaminationFilter, SupervisorFilter
from generic.models import UserUserGroup
from generic.tables import (
CidUserExamTable,
CidUserTable,
ExaminationTable,
SupervisorTable,
)
from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
import zipfile
from django.core.files.base import ContentFile
from django.db.models import Q, Count, F
from .forms import (
CidGroupExamForm,
CidUserForm,
CidUserExamForm,
ExamCollectionCloneForm,
ExamCollectionForm,
QuestionReviewForm,
ExaminationForm,
CidUserGroupForm,
ExaminationMergeForm,
SupervisorForm,
TraineeForm,
UserGroupExamForm,
CidGroupExamForm,
UserUserForm,
UserUserGroupForm,
UsersAutocompleteForm, # Added import
)
from .models import (
CidUser,
CidUserExam,
CidUserGroup,
CimarCase,
ExamBase,
ExamCollection,
ExamUserStatus,
Examination,
QuestionBase,
QuestionNote,
Supervisor,
UserGrades,
UserProfile,
UserUserGroup,
get_next_cid,
QuestionReview,
)
from rapids.models import (
Rapid as RapidQuestion,
ExamQuestionDetail as RapidsExamQuestionDetail,
)
from rapids.models import Exam as RapidsExam
from rapids.models import UserAnswer as RapidsUserAnswer
from longs.models import (
Long as LongQuestion,
LongSeries,
ExamQuestionDetail as LongsExamQuestionDetail,
)
from longs.models import Exam as LongsExam
from longs.models import UserAnswer as LongsUserAnswer
from anatomy.models import (
AnatomyQuestion as AnatomyQuestion,
ExamQuestionDetail as AnatomyExamQuestionDetail,
)
from anatomy.models import Exam as AnatomyExam
from sbas.models import (
Question as SbasQuestion,
ExamQuestionDetail as SbasExamQuestionDetail,
)
from anatomy.models import UserAnswer as AnatomyUserAnswer
from sbas.models import Exam as SbasExam
from sbas.models import UserAnswer as SbasUserAnswer
from physics.models import (
Question as PhysicsQuestion,
ExamQuestionDetail as PhysicsExamQuestionDetail,
)
from physics.models import Exam as PhysicsExam
from physics.models import UserAnswer as PhysicsUserAnswer
from shorts.models import Question as ShortsQuestion, ExamQuestionDetail as ShortsExamQuestionDetail, UserAnswer as ShortsUserAnswer, Exam as ShortsExam
from django.db.models import Case, When
from django.conf import settings
from datetime import datetime
from django.utils import timezone
import os
import string
import random
import plotly.express as px
# from rad.views import get_question_and_content_type
from django.db.models import Prefetch
from loguru import logger
@login_required
@user_passes_test(lambda u: u.is_superuser)
def bulk_delete_questions(request):
"""Generic bulk delete for question-like models.
Expects POST with 'app' (app_name) and 'selected' (list or comma-separated ids).
Returns a small HTML fragment summarising deletions (suitable for HTMX).
"""
if request.method != "POST":
return HttpResponse(status=405)
logger.debug(f"bulk_delete_questions called with POST: {request.POST}")
app = request.POST.get("app")
# Accept different submission patterns: checkboxes may be sent as 'selection' or 'selected'
selected_list = []
# common variants
for key in ("selected", "selection", "selected[]", "selection[]"):
vals = request.POST.getlist(key)
if vals:
# extend list; we'll parse ints below
selected_list.extend(vals)
# Also accept a raw CSV string under 'selected'
selected_raw = request.POST.get("selected")
ids = set()
if selected_list:
for s in selected_list:
# some checkbox libraries may give empty strings or 'on' for checked boxes
try:
ids.add(int(s))
except Exception:
# ignore non-int vals
continue
elif selected_raw:
for part in selected_raw.split(','):
try:
part = part.strip()
if not part:
continue
ids.add(int(part))
except Exception:
continue
# Map app name to model class
APP_MODEL_MAP = {
"sbas": SbasQuestion,
"physics": PhysicsQuestion,
"anatomy": AnatomyQuestion,
"rapids": RapidQuestion,
"rapid": RapidQuestion,
"shorts": ShortsQuestion,
"longs": LongQuestion,
"long": LongQuestion,
}
Model = APP_MODEL_MAP.get(app)
if Model is None:
logger.error(f"bulk_delete_questions called with unknown app: {app}")
return HttpResponse(f"Unknown app: {app}")
deleted = 0
errors = []
try:
qs = Model.objects.filter(pk__in=list(ids))
deleted = qs.count()
qs.delete()
except Exception as e:
logger.exception("bulk_delete failed for app=%s ids=%s", app, ids)
return HttpResponse("Unable to deleted items")
# Render a small fragment summarising deletions
context = {"deleted": deleted, "errors": errors}
return render(request, "generic/partials/bulk_delete_result.html", context)
class RedirectMixin:
def get_success_url(self) -> str:
if "redirect" in self.request.GET:
return self.request.GET["redirect"]
return super().get_success_url()
class AuthorRequiredMixin(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 not in obj.author.all():
raise PermissionDenied() # or Http404
return obj
class CidManagerRequiredMixin(UserPassesTestMixin):
def test_func(self):
return self.request.user.groups.filter(name="cid_user_manager").exists()
# def get_object(self, *args, **kwargs):
# obj = super().get_object(*args, **kwargs)
# if self.request.user.groups.filter(name="cid_user_manager").exists():
# return obj
# raise PermissionDenied() # or Http404
def get_user_exams(user, supervisor_view=False):
EXAM_ANSWER_MAP = {
"physics": (PhysicsUserAnswer, PhysicsExam, "physics_exams"),
"anatomy": (AnatomyUserAnswer, AnatomyExam, "anatomy_exams"),
"rapids": (RapidsUserAnswer, RapidsExam, "rapids_exams"),
"shorts": (ShortsUserAnswer, ShortsExam, "shorts_exams"),
"longs": (LongsUserAnswer, LongsExam, "longs_exams"),
"sbas": (SbasUserAnswer, SbasExam, "sbas_exams"),
}
kwargs = {"exam_mode": True, "archive": False}
exams = []
if supervisor_view:
kwargs["results_supervisor_visible"] = True
del kwargs["archive"]
for exam_type in EXAM_ANSWER_MAP:
UserAnswer, Exam, cid_rel = EXAM_ANSWER_MAP[exam_type]
exam_answers = UserAnswer.objects.filter(user=user)
exam_ids = exam_answers.values_list("exam").distinct()
exams_to_add = Exam.objects.filter(id__in=exam_ids, **kwargs)
if supervisor_view:
extra_exam_ids = CidUserExam.objects.filter(
user_user=user,
share_with_supervisor=True,
content_type=ContentType.objects.get_for_model(Exam),
).values_list(cid_rel)
exams_to_add = exams_to_add | Exam.objects.filter(id__in=extra_exam_ids)
if exams_to_add:
exams.append((exam_type, exams_to_add))
return exams
def get_exam_model_from_app_name(app_name: str) -> ExamBase:
EXAM_MAP = {
"physics": PhysicsExam,
"anatomy": AnatomyExam,
"rapids": RapidsExam,
"shorts": ShortsExam,
"longs": LongsExam,
"sbas": SbasExam,
}
return EXAM_MAP[app_name]
def normaliseRapidsScore(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
def get_question_and_content_type(question_type):
if question_type == "rapid":
question = RapidQuestion
content_type = ContentType.objects.get(model="rapid")
elif question_type == "shorts":
question = ShortsQuestion
content_type = ContentType.objects.get(model="rapid")
elif question_type == "anatomy":
question = AnatomyQuestion
content_type = ContentType.objects.get(model="anatomyquestion")
elif question_type == "long":
question = LongQuestion
content_type = ContentType.objects.get(model="long")
elif question_type == "sbas":
question = SbasQuestion
content_type = ContentType.objects.get(app_label="sbas", model="question")
elif question_type == "physics":
question = PhysicsQuestion
content_type = ContentType.objects.get(app_label="physics", model="question")
else:
raise PermissionError()
return question, content_type
# Create your views here.
@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
@user_passes_test(lambda u: u.is_superuser)
def examination_detail(request, pk):
examination = get_object_or_404(Examination, pk=pk)
merge_form = ExaminationMergeForm()
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"generic/examination_detail.html",
{"examination": examination, "merge_form": merge_form},
)
@login_required
@user_passes_test(lambda u: u.is_superuser)
def examination_merge(request, pk):
examination = get_object_or_404(Examination, pk=pk)
form = ExaminationMergeForm(request.POST)
if form.is_valid():
examination_to_merge_into = form.cleaned_data["examination_to_merge_into"]
success = False
if examination.merge_into(examination_to_merge_into):
success = True
else:
examination_to_merge_into = False
return render(
request,
"generic/examination_merge.html",
{
"form": form,
"examination": examination,
"examination_to_merge_into": examination_to_merge_into,
"success": success,
},
)
# logging.debug(atlas.subspecialty.first().name.all())
return render(
request,
"generic/examination_merge.html",
{
"examination": 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("/")
# Generic view to dispatch to indivuals app views
@login_required
def generic_exam_json_edit(request):
# GET: return a small fragment with an exam selector for the given type
if request.method == "GET":
question_type = request.GET.get("type")
if not question_type:
return JsonResponse({"status": "error", "message": "missing type"}, status=400)
try:
Exam = get_exam_model_from_app_name(question_type)
except Exception:
return JsonResponse({"status": "error", "message": "unknown type"}, status=400)
# Limit to exams that are not archived and are exam_mode
exams = Exam.objects.filter(archive=False, exam_mode=True).order_by("name")[:200]
return render(request, "generic/partials/exam_select_fragment.html", {"exams": exams, "type": question_type})
# Support POST additions via either an explicit JSON 'add_exam_questions' or
# via a list of checkboxes named 'selection' sent by HTMX from the question list.
if request.method == "POST":
question_type = request.POST.get("type")
if question_type == "rapids":
Exam = RapidsExam
Question = RapidQuestion
elif question_type == "shorts":
Exam = ShortsExam
Question = ShortsQuestion
elif question_type == "longs":
Exam = LongsExam
Question = LongQuestion
elif question_type == "anatomy":
Exam = AnatomyExam
Question = AnatomyQuestion
elif question_type == "physics":
Exam = PhysicsExam
Question = PhysicsQuestion
elif question_type == "sbas":
Exam = SbasExam
Question = SbasQuestion
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
exam = get_object_or_404(Exam, pk=request.POST.get("exam_id"))
# if "exam_toggle_active" in request.POST:
# active = json.loads(request.POST.get("active"))
# data = {"status": "success"}
# return JsonResponse(data, status=200)
if "add_exam_questions" in request.POST:
question_ids = json.loads(request.POST.get("add_exam_questions"))
else:
# Support HTMX flow where checkboxes named 'selection' are posted
# (multiple values). Convert to list of ints.
sel = request.POST.getlist("selection") or request.POST.getlist("selection[]")
question_ids = []
for s in sel:
try:
question_ids.append(int(s))
except Exception:
continue
# question_objects = Question.objects.filter(pk__in=question_ids)
add_count = 0
for i in question_ids:
# We get an integrity error if we try to add an object that already exists
if not exam.exam_questions.filter(pk=i).exists():
add_count += 1
exam.exam_questions.add(i)
if add_count > 0:
exam.save()
data = {"status": "success", "questions added": add_count}
return JsonResponse(data, status=200)
if "set_exam_order" in request.POST:
new_order = json.loads(request.POST.get("set_exam_order"))
preserved = Case(
*[When(pk=pk, then=pos) for pos, pk in enumerate(new_order)]
)
objects = Question.objects.filter(pk__in=new_order).order_by(preserved)
if len(objects) != exam.exam_questions.count():
data = {"status": "error, count does not match"}
return JsonResponse(data, status=400)
exam.exam_questions.set(objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
# Only checkers can do the following
if (
question_type == "rapid"
and not request.user.groups.filter(name="rapid_checker").exists()
):
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if (
question_type == "anatomy"
and not request.user.groups.filter(name="anatomy_checker").exists()
):
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if (
question_type == "long"
and not request.user.groups.filter(name="long_checker").exists()
):
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if "set_open_access" in request.POST:
if request.POST.get("set_open_access") == "true":
state = True
else:
state = False
questions_changed = []
for q in exam.exam_questions.all():
q.open_access = state
q.save()
questions_changed.append(q.pk)
data = {"status": "success", "questions": questions_changed, "state": state}
return JsonResponse(data, status=200)
data = {"status": "error, unknown request"}
return JsonResponse(data, status=400)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@login_required
def generic_exam_htmx(request):
"""HTMX-specific endpoint for selecting an exam and adding selected questions.
GET: returns the exam selection fragment (same as earlier partial)
POST: expects 'type' and checkbox list 'selection' (or selection[]) and 'exam_id'
adds selected question ids to the chosen exam and returns a small HTML fragment
or JSON suitable for HTMX swapping.
"""
if request.method == "GET":
question_type = request.GET.get("type")
if not question_type:
return HttpResponse("Missing type", status=400)
try:
Exam = get_exam_model_from_app_name(question_type)
except Exception:
return HttpResponse("Unknown type", status=400)
# For non-superusers, only show exams they can edit or that they authored
if request.user.is_superuser:
exams = Exam.objects.filter(archive=False, exam_mode=True).order_by("name")
else:
# authors or markers or assigned exams
exams = (
Exam.objects.filter(author__id=request.user.id)
| Exam.objects.filter(markers__id=request.user.id)
)
exams = exams.filter(archive=False).order_by("name")
return render(request, "generic/partials/exam_select_fragment.html", {"exams": exams, "type": question_type})
if request.method == "POST":
logger.debug(f"generic_exam_htmx POST: {request.POST}")
question_type = request.POST.get("type")
if not question_type:
return JsonResponse({"status": "error", "message": "missing type"}, status=400)
try:
Exam = get_exam_model_from_app_name(question_type)
except Exception:
return JsonResponse({"status": "error", "message": "unknown type"}, status=400)
exam = get_object_or_404(Exam, pk=request.POST.get("exam_id"))
# read checkbox selections
sel = request.POST.getlist("selection") or request.POST.getlist("selection[]")
question_ids = []
for s in sel:
try:
question_ids.append(int(s))
except Exception:
continue
add_count = 0
for i in question_ids:
if not exam.exam_questions.filter(pk=i).exists():
add_count += 1
exam.exam_questions.add(i)
if add_count > 0:
exam.save()
# return a small HTMX-friendly fragment summarising the result
return render(request, "generic/partials/exam_select_result.html", {"added": add_count, "exam": exam})
return HttpResponse(status=405)
@login_required
def generic_exam_set_open_access(request):
"""HTMX endpoint to set open_access on questions in an exam.
POST params:
- type: app name (sbas, rapids, anatomy, physics, shorts, longs)
- exam_id: id of the exam
- set_open_access: 'true' or 'false'
- selection / selection[]: optional list of question ids to apply to (if omitted apply to all)
Only questions that request.user can edit (question.can_edit(user)) or superusers will be modified.
Returns a small HTML fragment summarising the count of changed questions.
"""
if request.method != "POST":
return HttpResponse(status=405)
question_type = request.POST.get("type")
if not question_type:
return JsonResponse({"status": "error", "message": "missing type"}, status=400)
try:
Exam = get_exam_model_from_app_name(question_type)
except Exception:
return JsonResponse({"status": "error", "message": "unknown type"}, status=400)
exam = get_object_or_404(Exam, pk=request.POST.get("exam_id"))
# permission to edit exam required
# Many exam edits are limited to exam authors / checkers; reuse existing check where possible
# For now require the user to have edit access (ExamViews.check_user_edit_access equivalent)
# We'll do a lightweight check: if user is superuser OR user in exam.get_author_objects()
if not (request.user.is_superuser or request.user in exam.get_author_objects()):
return HttpResponse(format_html('<div class="alert alert-danger">Permission denied</div>'))
set_val = True if request.POST.get("set_open_access") == "true" else False
sel = request.POST.getlist("selection") or request.POST.getlist("selection[]")
if sel:
try:
ids = [int(s) for s in sel]
except Exception:
ids = []
qs = exam.exam_questions.filter(pk__in=ids)
else:
qs = exam.exam_questions.all()
changed = 0
for q in qs:
try:
# prefer per-question permission check if available
can_edit = request.user.is_superuser or getattr(q, 'can_edit', lambda u: False)(request.user)
except Exception:
can_edit = request.user.is_superuser
if not can_edit:
continue
if q.open_access != set_val:
q.open_access = set_val
q.save()
changed += 1
return render(request, "generic/partials/set_open_access_result.html", {"changed": changed, "exam": exam, "state": set_val})
class ExamViews(View, LoginRequiredMixin):
def __init__(
self,
exam,
question,
answer,
cid_user_answer,
app,
question_type,
normalise_score=None,
):
self.Exam: ExamBase = exam
self.Question = question
self.Answer = answer
self.UserAnswer = cid_user_answer
self.app_name = app
self.question_type = question_type
self.normalise_score = normalise_score
# THis may be better than check_user_access below
# group_map = {"rapids" : "rapid_checker", "anatomy" : "anatomy_checker", "longs":"long_checker"}
# exam_group = group_map[self.app_name]
class BasicExamFilter(FilterSet):
sort_order = OrderingFilter(
fields=(("name", "name"), ("exam_mode", "exam_mode"))
)
# Allow filtering by start/end date ranges (render as paired date inputs)
start_date = DateFromToRangeFilter(
field_name="start_date",
label="Start date",
widget=DateRangeWidget(attrs={"class": "datepicker", "autocomplete": "off", "placeholder": "dd/mm/yyyy"}),
)
end_date = DateFromToRangeFilter(
field_name="end_date",
label="End date",
widget=DateRangeWidget(attrs={"class": "datepicker", "autocomplete": "off", "placeholder": "dd/mm/yyyy"}),
)
class Meta:
model = exam
fields = {
"name": ["contains"],
"exam_mode": ["exact"],
"active": ["exact"],
"archive": ["exact"],
"open_access": ["exact"],
}
def __init__(self, data=None, *args, **kwargs):
# if filterset is bound, use initial values as defaults
if data is not None:
# get a mutable copy of the QueryDict
data = data.copy()
# for name, f in self.base_filters.items():
# filter param is either missing or empty, use initial as default
if not data.get("archive"):
data["archive"] = False
if not data.get("sort_order"):
data["sort_order"] = "name"
super().__init__(data, *args, **kwargs)
self.BasicExamFilter = BasicExamFilter
# We give some users the ability to filter by authors
class ExtraExamFilter(BasicExamFilter):
author = ModelMultipleChoiceFilter(
queryset=User.objects.all(), null_label="No author"
)
self.ExtraExamFilter = ExtraExamFilter
# TODO: these may be better implemented as decorators
def check_user_access(self, user: User, exam_id: int = None, marker=False):
"""Check if a user should be able to access a view
Args:
user ([user]): user object
exam_id ([int], optional): exam id. Defaults to None.
marker ([boolean], optional): Allow access if the user is a marker. Defaults to False. Requires that exam_id is set.
Returns:
[boolean]: True if the user has access
"""
print("****", user, exam_id)
if user.is_superuser:
return True
if marker and exam_id is None:
# raise error if exam_id is not provided
raise ValueError("exam_id must be provided if marker checking for a marker")
if exam_id is not None:
exam: ExamBase = get_object_or_404(self.Exam, pk=exam_id)
if (exam.open_access and exam.active) or user in exam.get_author_objects():
return True
if (
user.is_authenticated
and exam.active
and exam.cid_user_exam.filter(user_user=user)
):
return True
if marker and user in exam.markers.all():
return True
if exam.authors_only:
return False
if exam.open_access:
return True
if user.groups.filter(name="cid_user_manager").exists():
return True
if (
self.app_name == "rapids"
and user.groups.filter(name="rapid_checker").exists()
):
return True
if (
self.app_name == "shorts"
and user.groups.filter(name="shorts").exists()
):
return True
if (
self.app_name == "anatomy"
and user.groups.filter(
name__in=("anatomy_checker", "anatomy_marker")
).exists()
):
return True
if (
self.app_name == "longs"
and user.groups.filter(
name__in=("long_checker", "long_marker")
).exists()
):
return True
if (
self.app_name == "physics"
and user.groups.filter(name="physics_checker").exists()
):
return True
if (
self.app_name == "sbas"
and user.groups.filter(name="sba_checker").exists()
):
return True
if (
self.app_name == "atlas"
and user.groups.filter(name="casecollection_checker").exists()
):
return True
return False
def check_user_marker_access(self, user: User, exam_id: int = None):
return self.check_user_access(user, exam_id, marker=True)
def check_user_edit_access(self, user, exam_id=None):
"""Check if a user should be able to access a view
Args:
user ([user]): user object
Returns:
[boolean]: True if the user has access
"""
if user.is_superuser:
return True
# If a user is an exam author they should have acccess
if exam_id is not None:
exam = get_object_or_404(self.Exam, pk=exam_id)
# Remove open_access check for the momement
# if exam.open_access or user in exam.get_author_objects():
# return True
if user in exam.get_author_objects():
return True
if exam.authors_only:
return False
if (
self.app_name == "rapids"
and not user.groups.filter(name="rapid_checker").exists()
):
return False
if (
self.app_name == "anatomy"
and not user.groups.filter(name="anatomy_checker").exists()
):
return False
if (
self.app_name == "longs"
and not user.groups.filter(name__in=("long_checker",)).exists()
):
return False
if (
self.app_name == "physics"
and not user.groups.filter(name="physics_checker").exists()
):
return False
if (
self.app_name == "sbas"
and not user.groups.filter(name="sba_checker").exists()
):
return False
if (
self.app_name == "casecollection"
and not user.groups.filter(name="casecollection_checker").exists()
):
return False
return False
@method_decorator(login_required)
def exam_toggle_archived(self, request, pk):
if request.method == "POST":
if not self.check_user_edit_access(request.user):
data = {"status": "error, insufficient permission"}
return JsonResponse(data, status=400)
exam = get_object_or_404(self.Exam, pk=pk)
exam.archive = True if request.POST.get("archive") == "true" else False
print(f"{exam.archive=}")
exam.save()
data = {
"status": "success",
"archive": exam.archive,
"name": str(exam),
"id": exam.id,
}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def exam_toggle_results_published_htmx(self, request, pk):
if request.method == "POST":
if not self.check_user_edit_access(request.user, exam_id=pk):
return HttpResponse(
format_html(
'<div class="alert alert-danger" role="alert">Error toggling results published.</div>'
)
)
exam = get_object_or_404(self.Exam, pk=pk)
exam.publish_results = not exam.publish_results
exam.save()
return render(
request, "generic/partials/exams/exam_status.html#publish-results", context={
"exam": exam,
"collection": exam
})
else:
HttpResponse(
format_html(
'<div class="alert alert-danger" role="alert">Error toggling results published.</div>'
)
)
@method_decorator(login_required)
def exam_toggle_active_htmx(self, request, pk):
if request.method == "POST":
if not self.check_user_edit_access(request.user, exam_id=pk):
return HttpResponse(
format_html(
'<div class="alert alert-danger" role="alert">Error toggling exam active.</div>'
)
)
exam = get_object_or_404(self.Exam, pk=pk)
exam.active = not exam.active
exam.save()
return render(
request, "generic/partials/exams/exam_status.html#exam-active", context={
"exam": exam,
"collection": exam
})
else:
HttpResponse(
format_html(
'<div class="alert alert-danger" role="alert">Error toggling exam active.</div>'
)
)
@method_decorator(login_required)
def exam_bulk_update(self, request):
"""HTMX endpoint to bulk-update exams (currently supports updating the date).
Expects POST with 'exam_ids' (multiple) and optional 'date' (YYYY-MM-DD).
Returns the updated exam list partial.
"""
if request.method != "POST":
return HttpResponse(status=405)
exam_ids = request.POST.getlist("exam_ids")
date_str = request.POST.get("date", "")
# Parse date if provided
new_date = None
if date_str:
try:
from datetime import date as _date
new_date = _date.fromisoformat(date_str)
except Exception:
new_date = None
# Find the queryset for the current user as index() does
if (
request.user.is_superuser
or (
self.app_name == "rapids"
and request.user.groups.filter(name="rapid_checker").exists()
)
or (
self.app_name == "anatomy"
and request.user.groups.filter(name="anatomy_checker").exists()
)
or (
self.app_name == "longs"
and request.user.groups.filter(name="long_checker").exists()
)
):
exams_qs = self.Exam.objects.all()
filter = self.ExtraExamFilter(request.GET, queryset=exams_qs)
else:
exams_qs = self.Exam.objects.filter(author__id=request.user.id) | self.Exam.objects.filter(open_access=True)
filter = self.BasicExamFilter(request.GET, queryset=exams_qs)
# Apply changes
for eid in exam_ids:
try:
exam = self.Exam.objects.get(pk=eid)
except self.Exam.DoesNotExist:
continue
if not self.check_user_edit_access(request.user, exam_id=exam.pk):
# Skip exams the user cannot edit
continue
if new_date is not None:
exam.start_date = new_date
exam.save()
return render(request, "generic/partials/_exam_list.html", {"filter": filter, "app_name": self.app_name})
@method_decorator(login_required)
def exam_bulk_delete(self, request):
"""HTMX endpoint to bulk-delete selected exams. Expects POST 'exam_ids'.
Returns the updated exam list partial.
"""
if request.method != "POST":
return HttpResponse(status=405)
exam_ids = request.POST.getlist("exam_ids")
if (
request.user.is_superuser
or (
self.app_name == "rapids"
and request.user.groups.filter(name="rapid_checker").exists()
)
or (
self.app_name == "anatomy"
and request.user.groups.filter(name="anatomy_checker").exists()
)
or (
self.app_name == "longs"
and request.user.groups.filter(name="long_checker").exists()
)
):
exams_qs = self.Exam.objects.all()
filter = self.ExtraExamFilter(request.GET, queryset=exams_qs)
else:
exams_qs = self.Exam.objects.filter(author__id=request.user.id) | self.Exam.objects.filter(open_access=True)
filter = self.BasicExamFilter(request.GET, queryset=exams_qs)
for eid in exam_ids:
try:
exam = self.Exam.objects.get(pk=eid)
except self.Exam.DoesNotExist:
continue
if not self.check_user_edit_access(request.user, exam_id=exam.pk):
continue
try:
exam.delete()
except Exception:
# ignore individual delete failures
continue
return render(request, "generic/partials/_exam_list.html", {"filter": filter, "app_name": self.app_name})
@method_decorator(login_required)
def exam_toggle_results_published(self, request, pk):
if request.method == "POST":
if not self.check_user_edit_access(request.user, exam_id=pk):
data = {"status": "error, insufficient permission"}
return JsonResponse(data, status=400)
exam = get_object_or_404(self.Exam, pk=pk)
exam.publish_results = (
True if request.POST.get("publish_results") == "true" else False
)
exam.save()
data = {
"status": "success",
"publish_results": exam.publish_results,
"name": str(exam),
"id": exam.id,
}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def exam_toggle_active(self, request, pk):
if request.method == "POST":
if not self.check_user_edit_access(request.user, exam_id=pk):
data = {"status": "error, insufficient permission"}
return JsonResponse(data, status=400)
exam = get_object_or_404(self.Exam, pk=pk)
exam.active = True if request.POST.get("active") == "true" else False
exam.save()
data = {
"status": "success",
"active": exam.active,
"name": str(exam),
"id": exam.id,
}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def exam_review_start(self, request, pk):
"""Start a per-question review for an exam. Redirects to the first question."""
# Ensure the user can edit the exam (authors only)
if not self.check_user_edit_access(request.user, exam_id=pk):
raise PermissionDenied
exam = get_object_or_404(self.Exam, pk=pk)
# Materialise ordered questions
questions = list(exam.get_questions())
if not questions:
return render(request, "generic/exam_review_complete.html", {"exam": exam, "app_name": self.app_name})
# Redirect (render) the first question
return self.exam_review_question(request, pk, q_index=0)
@method_decorator(login_required)
def exam_review_question(self, request, pk, q_index=0):
"""Render a single question from an exam for review with navigation."""
if not self.check_user_edit_access(request.user, exam_id=pk):
raise PermissionDenied
exam = get_object_or_404(self.Exam, pk=pk)
questions = list(exam.get_questions())
total = len(questions)
try:
q_index = int(q_index)
except Exception:
q_index = 0
if q_index < 0 or q_index >= total:
# Out of range — render complete
return render(request, "generic/exam_review_complete.html", {"exam": exam, "app_name": self.app_name})
question = questions[q_index]
prev_index = q_index - 1 if q_index > 0 else None
next_index = q_index + 1 if q_index < total - 1 else None
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"total": total,
"prev_index": prev_index,
"next_index": next_index,
"app_name": self.app_name,
"can_edit": self.check_user_edit_access(request.user, exam_id=pk),
}
return render(request, "generic/exam_review_question.html", context)
@method_decorator(login_required)
def exam_json_recreate(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
exam.recreate_json = True
exam.save()
# exam.get_exam_json()
# TODO: check if mememory leak?
t = threading.Thread(target=self.exam_json, args=(request, pk))
t.setDaemon(True)
t.start()
return redirect("{}:exam_overview".format(self.app_name), pk=pk)
@method_decorator(login_required)
def question_json_recreate(self, request, pk):
question = get_object_or_404(self.Question, pk=pk)
question.recreate_json = True
question.save()
return redirect("{}:question_detail".format(self.app_name), pk=pk)
@method_decorator(login_required)
def exam_list_all(self, request):
return self.exam_list(request, all=True)
def order_questions(self, request, exam_id):
if not self.check_user_edit_access(request.user, exam_id=exam_id):
return HttpResponse("Permission denied")
exam = get_object_or_404(self.Exam, pk=exam_id)
exam.order_questions()
return HttpResponse("Done")
@method_decorator(login_required)
def exam_list_collection(self, request, collection_id):
collection = get_object_or_404(ExamCollection, pk=collection_id)
if not request.user in collection.author.all():
raise PermissionDenied
return self.exam_list(request, all=True, collection=collection)
@method_decorator(login_required)
def exam_list(self, request, all=False, collection=None):
if collection is None:
if not self.check_user_access(request.user):
# raise PermissionDenied
exam_list = self.Exam.objects.filter(
author__id=request.user.id, exam_mode=True
).order_by("name")
exam_list = exam_list | self.Exam.objects.filter(
markers__id=request.user.id, exam_mode=True
).order_by("name")
else:
exam_list = (
self.Exam.objects.prefetch_related(
"valid_user_users", "valid_cid_users"
)
.filter(exam_mode=True)
.order_by("name")
)
else:
exam_list = self.Exam.objects.filter(exam_mode=True, examcollection__in=[collection]).order_by("name")
if not all:
exam_list = exam_list.filter(archive=False).order_by("name")
marking = False
if self.app_name in ("rapids", "anatomy", "longs"):
marking = True
return render(
request,
"exam_list.html".format(self.app_name),
{
"exams": exam_list,
"app_name": self.app_name,
"view_all": all,
"marking": marking,
"collection": collection,
},
)
@method_decorator(login_required)
def exam_user(self, request, exam_id, user_id):
exam = get_object_or_404(
self.Exam.objects.prefetch_related("author"), pk=exam_id
)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, exam_id):
raise PermissionDenied
user = User.objects.get(pk=user_id)
user_exam = exam.get_or_create_cid_user_exam(user_user=user)
return JsonResponse(model_to_dict(user_exam))
@method_decorator(login_required)
def exam_report_email_status(self, request, exam_id):
exam = get_object_or_404(
self.Exam.objects.prefetch_related("author"), pk=exam_id
)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, exam_id):
raise PermissionDenied
users = exam.get_user_users_with_scores()
status = {}
for user in users:
# user = User.objects.get(pk=u)
user_exam = exam.get_or_create_cid_user_exam(user_user=user)
if user_exam.results_emailed_status:
status[user] = json.loads(user_exam.results_emailed_status)
return render(
request,
"generic/exam_report_email_status.html",
{
"exam": exam,
"status": status,
"app_name": self.app_name,
"users": users,
},
)
return JsonResponse(status)
@method_decorator(login_required)
def exam_report_email_unsent(self, request, exam_id):
return self.exam_report_email(request, exam_id, unsent_only=True)
@method_decorator(login_required)
def exam_report_email(self, request, exam_id, unsent_only=False, users=None):
exam = get_object_or_404(
self.Exam.objects.prefetch_related("author"), pk=exam_id
)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, exam_id):
raise PermissionDenied
additional_email = False
if request.htmx.prompt is not None:
additional_email = request.htmx.prompt
print(f"{additional_email=}")
# check addition email is valid
if additional_email:
try:
validate_email(additional_email)
except ValidationError:
return JsonResponse("Invalid additional email")
# We only need to send emails to those who have scores
# (who should be in exam.user_scores)
if users is None:
users = exam.get_user_users_with_scores()
time = timezone.now()
email_results = {}
for user in users:
# user = User.objects.get(pk=u)
user_exam = exam.get_or_create_cid_user_exam(user_user=user)
if unsent_only and user_exam.results_emailed_status:
if json.loads(user_exam.results_emailed_status)[0] == True:
continue
if additional_email:
email_results[user] = exam.email_user_results(
user, additional_emails=[additional_email]
)
else:
email_results[user] = exam.email_user_results(user)
email_results[user].append(f"{time}")
user_exam.results_emailed_status = json.dumps(email_results[user])
user_exam.save()
exam.exam_results_emailed = time
exam.save()
return render(
request,
"generic/exam_report_email_status.html",
{
"exam": exam,
"status": email_results,
"app_name": self.app_name,
},
)
return JsonResponse(email_results)
@method_decorator(login_required)
def exam_user_report_email(self, request, exam_id, user_id):
return self.exam_report_email(request, exam_id, users=[user_id])
# print(Exam.objects.all())
# exams = Exam.objects.all()
# print("test", Exam.objects.all().get(id=pk))
# exam = get_object_or_404(self.Exam.objects.prefetch_related("author"), pk=exam_id)
# u = get_object_or_404(User, pk=user_id)
# if request.user not in exam.author.all():
# if not self.check_user_access(request.user, exam_id):
# raise PermissionDenied
# res = exam.email_user_results(u)
# return HttpResponse(res)
@method_decorator(login_required)
def exam_user_report(self, request, exam_id, user_id):
# print(Exam.objects.all())
# exams = Exam.objects.all()
# print("test", Exam.objects.all().get(id=pk))
exam = get_object_or_404(
self.Exam.objects.prefetch_related("author"), pk=exam_id
)
u = get_object_or_404(User, pk=user_id)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, exam_id):
raise PermissionDenied
res = exam.generate_user_report(u)
print(res)
return HttpResponse(res)
@method_decorator(login_required)
def exam_overview(self, request, pk):
# print(Exam.objects.all())
# exams = Exam.objects.all()
# print("test", Exam.objects.all().get(id=pk))
exam = get_object_or_404(self.Exam.objects.prefetch_related("author"), pk=pk)
if request.user not in exam.author.all():
# We also let markers view the exam
if not self.check_user_access(request.user, pk, marker=True):
raise PermissionDenied
can_edit = (
self.check_user_edit_access(request.user, pk) | request.user.is_superuser
)
notes = []
if can_edit:
notes = self.exam_notes(request, pk)
if self.app_name == "anatomy":
questions = (
exam.exam_questions.select_related(
"modality",
"structure",
"body_part",
"region",
"examination",
"question_type",
)
.all()
.prefetch_related(
Prefetch(
"answers",
queryset=self.Answer.objects.filter(
proposed=False, status=self.Answer.MarkOptions.CORRECT
),
to_attr="prefetched_primary_answer",
# queryset=self.Answer.objects.filter(),
),
)
.order_by("examquestiondetail__sort_order")
)
elif self.app_name == "longs":
questions = (
exam.exam_questions.prefetch_related(
# Prefetch(
# "series",
# queryset=LongSeries.objects.select_related(
# "modality", "examination", "plane", "contrast"
# ).prefetch_related(Prefetch("images")),
# ),
"author",
# "examquestiondetail",
# "longsseries",
"series",
# "seriesdetail",
"series__images",
"series__plane",
"series__examination",
# to_attr="prefetched_primary_answer"
# queryset=self.Answer.objects.filter(),
# "series",
# "abnormality",
# "region",
# "examination",
)
.all()
.order_by("examquestiondetail__sort_order")
)
elif self.app_name == "rapids":
questions = (
exam.exam_questions.select_related()
.all()
.prefetch_related(
Prefetch(
"answers",
queryset=self.Answer.objects.filter(
proposed=False, status=self.Answer.MarkOptions.CORRECT
),
to_attr="prefetched_primary_answer",
# queryset=self.Answer.objects.filter(),
),
"images",
"abnormality",
"region",
"examination",
"author",
)
.order_by("examquestiondetail__sort_order")
)
# questions = (
# exam.exam_questions.select_related()
# .all()
# .prefetch_related("images", "abnormality", "region", "examination")
# )
elif self.app_name == "physics":
questions = (
exam.exam_questions.select_related("category")
.all()
.order_by("examquestiondetail__sort_order")
# .prefetch_related("images", "abnormality", "region", "examination")
)
else:
questions = (
exam.exam_questions.select_related()
.all()
.order_by("examquestiondetail__sort_order")
)
question_number = len(questions)
cid_candidates = exam.valid_cid_users.count()
users_candidates = exam.valid_user_users.count()
# question_orders = [q.examquestiondetail.sort_order for q in questions]
# Exam order can be missing when, this can be checked with
# exam.exam_questions.select_related().all().order_by("examquestiondetail__sort_order").values_list("examquestiondetail__sort_order")
# TODO decide if / how this should be flagged to a user (updating teh exam order will fix)
return render(
request,
"{}/exam_overview.html".format(self.app_name),
{
"exam": exam,
"questions": questions,
"question_number": question_number,
"can_edit": can_edit,
"notes": notes,
"app_name": self.app_name,
"candidate_count": (cid_candidates, users_candidates),
},
)
@method_decorator(login_required)
def exam_scores_refresh(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
# We don't worry about order here
questions = exam.exam_questions.all()
cids = self.UserAnswer.objects.filter(question__in=questions, exam__id=pk)
# Force a score update
for c in cids:
c.get_answer_score(cached=False)
return render(
request,
f"{self.app_name}/base.html",
{
"simple_content": format_html(
"""Answer scores updated <br/>
<a href='{}'>Return</a>""",
reverse(f"{self.app_name}:exam_scores_all", kwargs={"pk": exam.pk}),
)
},
)
def exam_cids(self, request, exam_id):
"""View that displays an overview of users that have been added to the exam.
Args:
request (_type_): _description_
exam_id (_type_): _description_
Raises:
PermissionDenied: _description_
Returns:
_type_: _description_
"""
exam = get_object_or_404(self.Exam, pk=exam_id)
# if not request.user.groups.filter(name="cid_user_manager").exists():
# raise PermissionDenied
if request.user not in exam.author.all():
if not request.user.is_superuser:
raise PermissionDenied
cid_users = (
exam.valid_cid_users.all()
) # .order_by("cid")#.prefetch_related("*")
cid_user_count = len(cid_users)
user_users = exam.valid_user_users.all() # .order_by("username")
user_user_count = len(user_users)
user_exam_data = exam.cid_users.all().prefetch_related("cid_user", "user_user")
context = {
"exam": exam,
"cid_users": cid_users,
"cid_user_count": cid_user_count,
"user_exam_data": user_exam_data,
"user_users": user_users,
"user_user_count": user_user_count,
"app_name": self.app_name,
}
if self.app_name == "atlas":
context["collection"] = exam
return render(request, "exam_cids.html", context)
# def exam_groups_edit(self, request, exam_id):
# exam = get_object_or_404(self.Exam, pk=exam_id)
# if not request.user.groups.filter(name="cid_user_manager").exists():
# # raise PermissionDenied
# if request.user not in exam.author.all():
# raise PermissionDenied
#
# context = {
# "exam": exam,
# #"groups": exam_groups,
# }
# return render(request, "exam_groups_edit.html", context)
def exam_users_edit(self, request, exam_id):
exam = get_object_or_404(self.Exam, pk=exam_id)
if not request.user.groups.filter(name="cid_user_manager").exists():
# raise PermissionDenied
if request.user not in exam.author.all():
raise PermissionDenied
current_user_users = exam.valid_user_users.all()
exam_groups = exam.user_user_groups.all()
available_user_users = list(
User.objects.filter(user_groups__in=exam_groups).difference(
current_user_users
)
)
print(available_user_users)
# available_user_users = User.objects.all()
context = {
"exam": exam,
"groups": exam_groups,
"current_user_users": current_user_users,
"available_user_users": available_user_users,
"app_name": self.app_name,
}
if self.app_name == "atlas":
context["collection"] = exam
return render(request, "exam_users_edit.html", context)
def exam_reset_answers(self, request, exam_id):
if request.htmx:
exam = get_object_or_404(self.Exam, pk=exam_id)
if not self.check_user_edit_access(request.user, exam_id):
raise PermissionDenied
## Delete all answers
exam.cid_user_answers.all().delete()
# User statuses
exam.exam_user_status.all().delete()
# CidUserExams
exam.cid_users.all().delete()
return HttpResponse("<b>Answers reset</b>")
else:
raise Http404("Invalid request")
def exam_cids_edit(self, request, exam_id):
exam = get_object_or_404(self.Exam, pk=exam_id)
if not request.user.groups.filter(name="cid_user_manager").exists():
# raise PermissionDenied
if request.user not in exam.author.all():
raise PermissionDenied
current_cid_users = exam.valid_cid_users.all()
exam_groups = exam.cid_user_groups.all()
available_cid_users = CidUser.objects.filter(group__in=exam_groups).difference(
current_cid_users
)
context = {
"exam": exam,
"groups": exam_groups,
"current_cid_users": current_cid_users,
"available_cid_users": available_cid_users,
"app_name": self.app_name,
}
if self.app_name == "atlas":
context["collection"] = exam
return render(request, "exam_cids_edit.html", context)
# @method_decorator(login_required)
def exam_notes(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, pk):
raise PermissionDenied
q, content_type = get_question_and_content_type(self.question_type)
question_pks = exam.exam_questions.values_list("pk", flat=True)
# Get active notes for type
notes = QuestionNote.objects.filter(
content_type=content_type, object_id__in=question_pks, complete=False
)
return notes
def exam_user_status_cid(self, request, pk, cid):
return self.exam_user_status(request, pk, cid=cid)
def exam_user_status_user(self, request, pk, user_id):
return self.exam_user_status(request, pk, user_id=user_id)
def exam_user_status(self, request, pk, cid=None, user_id=None):
# TODO: add filtering by user / cid
exam = get_object_or_404(self.Exam, pk=pk)
# Restrict just to exam authors
if request.user not in exam.author.all():
raise PermissionDenied
if cid is not None:
statuses = exam.exam_user_status.filter(cid_user_exam__cid_user__cid=cid)
elif user_id is not None:
statuses = exam.exam_user_status.filter(
cid_user_exam__user_user__id=user_id
)
else:
statuses = exam.exam_user_status.all()
return render(
request, "exam_user_status.html", {"exam": exam, "statuses": statuses}
)
# if not self.check_user_access(request.user, pk):
# raise PermissionDenied
# q, content_type = get_question_and_content_type(self.question_type)
## Get active notes for type
# statuses = ExamUserStatus.objects.filter(
# content_type=content_type, object_id__in=pk
# )
# return statuses
@method_decorator(login_required)
def exam_json_edit(self, request, pk):
if request.method == "POST":
if request.user.groups.filter(name="cid_user_manager").exists():
# This may give more permissions than we actually want
pass
elif not self.check_user_edit_access(request.user, exam_id=pk):
data = {"status": "invalid permisions"}
return JsonResponse(data, status=403)
exam = get_object_or_404(self.Exam, pk=pk)
if "edit_cid_user" in request.POST:
user_id = request.POST.get("edit_cid_user")
add = request.POST.get("add") == "true"
cid_user = CidUser.objects.get(pk=user_id)
app_exam_map = {}
app_exam_map["rapids"] = cid_user.rapid_exams
app_exam_map["shorts"] = cid_user.shorts_exams
app_exam_map["anatomy"] = cid_user.anatomy_exams
app_exam_map["longs"] = cid_user.longs_exams
app_exam_map["physics"] = cid_user.physics_exams
app_exam_map["sbas"] = cid_user.sba_exams
app_exam_map["atlas"] = cid_user.casecollection_exams
if add:
app_exam_map[self.app_name].add(pk)
else:
app_exam_map[self.app_name].remove(pk)
data = {"status": "success", "added": add}
return JsonResponse(data, status=200)
if "edit_user_user" in request.POST:
user_id = request.POST.get("edit_user_user")
add = request.POST.get("add") == "true"
user_user = User.objects.get(pk=user_id)
app_exam_map = {}
app_exam_map["rapids"] = user_user.user_rapid_exams
app_exam_map["shorts"] = user_user.user_shorts_exams
app_exam_map["anatomy"] = user_user.user_anatomy_exams
app_exam_map["longs"] = user_user.user_longs_exams
app_exam_map["physics"] = user_user.user_physics_exams
app_exam_map["sbas"] = user_user.user_sba_exams
app_exam_map["atlas"] = user_user.user_casecollection_exams
if add:
app_exam_map[self.app_name].add(pk)
else:
app_exam_map[self.app_name].remove(pk)
data = {"status": "success", "added": add}
return JsonResponse(data, status=200)
if "add_exam_questions" in request.POST:
question_ids = json.loads(request.POST.get("add_exam_questions"))
question_objects = self.Question.objects.filter(pk__in=question_ids)
exam.exam_questions.add(question_objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
if "set_exam_order" in request.POST:
new_order = json.loads(request.POST.get("set_exam_order"))
preserved = Case(
*[When(pk=pk, then=pos) for pos, pk in enumerate(new_order)]
)
objects = self.Question.objects.filter(pk__in=new_order).order_by(
preserved
)
# We now allow deleting exam questions
# if len(objects) != exam.exam_questions.count():
# data = {"status": "error, count does not match"}
# return JsonResponse(data, status=400)
if self.app_name == "atlas":
exam.cases.set(objects)
for n, case in enumerate(objects):
case_detail = CaseDetail.objects.get(case=case, collection=exam)
case_detail.sort_order = n
case_detail.save()
else:
exam.exam_questions.set(objects)
for n, question in enumerate(objects):
match self.app_name:
case "anatomy":
question_detail = AnatomyExamQuestionDetail.objects.get(
anatomyquestion=question, exam=exam
)
case "rapids":
question_detail = RapidsExamQuestionDetail.objects.get(
rapid=question, exam=exam
)
case "shorts":
question_detail = ShortsExamQuestionDetail.objects.get(
question=question, exam=exam
)
case "longs":
question_detail = LongsExamQuestionDetail.objects.get(
question=question, exam=exam
)
case "physics":
question_detail = PhysicsExamQuestionDetail.objects.get(
question=question, exam=exam
)
case "sbas":
question_detail = SbasExamQuestionDetail.objects.get(
question=question, exam=exam
)
case _:
data = {"status": "error"}
return JsonResponse(data, status=200)
question_detail.sort_order = n
question_detail.save()
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
if not self.check_user_access(request.user, pk):
raise PermissionDenied
if "set_open_access" in request.POST:
if request.POST.get("set_open_access") == "true":
state = True
else:
state = False
questions_changed = []
for q in exam.exam_questions.all():
q.open_access = state
q.save()
questions_changed.append(q.pk)
data = {
"status": "success",
"questions": questions_changed,
"state": state,
}
return JsonResponse(data, status=200)
data = {"status": "error, unknown request"}
return JsonResponse(data, status=400)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def mark_overview(self, request, pk):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
if request.user not in exam.author.all():
if not self.check_user_marker_access(request.user, pk):
raise PermissionDenied
if self.app_name == "anatomy":
questions = (
exam.exam_questions.select_related(
"modality",
"structure",
"body_part",
"region",
"examination",
"question_type",
)
.all()
.prefetch_related(
# "cid_user_answers",
Prefetch(
"answers",
queryset=self.Answer.objects.filter(
proposed=False, status=self.Answer.MarkOptions.CORRECT
),
to_attr="prefetched_primary_answer",
# queryset=self.Answer.objects.filter(),
),
# Prefetch(
# "cid_user_answers",
# queryset=self.UserAnswer.objects.filter(score=self.Answer.MarkOptions.UNMARKED, exam__id=pk),
# to_attr="prefetched_unmarked_cid_answers"
# #queryset=self.Answer.objects.filter(),
# ),
)
.order_by("examquestiondetail__sort_order")
)
else:
questions = exam.get_questions()
# Handle exams that require double marking
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,
"{}/mark_overview.html".format(self.app_name),
{"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,
"{}/mark_overview.html".format(self.app_name),
{"exam": exam, "question_unmarked_map": question_unmarked_map},
)
@method_decorator(login_required)
def index(self, request):
# self.check_user_access(request.user, exam.pk)
if (
request.user.is_superuser
or (
self.app_name == "rapids"
and request.user.groups.filter(name="rapid_checker").exists()
)
or (
self.app_name == "anatomy"
and request.user.groups.filter(name="anatomy_checker").exists()
)
or (
self.app_name == "longs"
and request.user.groups.filter(name="long_checker").exists()
)
):
exams = self.Exam.objects.all()
filter = self.ExtraExamFilter(request.GET, queryset=exams)
print("1")
else:
print("2")
exams = self.Exam.objects.filter(
author__id=request.user.id
) | self.Exam.objects.filter(open_access=True)
filter = self.BasicExamFilter(request.GET, queryset=exams)
return render(
request,
"generic/exam_index.html",
{"filter": filter, "app_name": self.app_name},
)
@method_decorator(login_required)
def exam_question_user_answer(
self, request, pk: int, sk: int, c_or_u: str, user_or_cid: str
):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if c_or_u == "u":
user = get_object_or_404(User, pk=user_or_cid)
answer = exam.get_question_user_user_answer(sk, user)
elif c_or_u == "c":
# cid_user = CidUser.objects.filter(cid=user_or_cid)
answer = exam.get_question_cid_user_answer(sk, user_or_cid)
else:
raise Http404
print(answer)
return HttpResponse(answer)
@method_decorator(login_required)
def exam_question_cid_user_answer(self, request, pk: int, sk: int, cid: str):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
cid_user = CidUser.objects.filter(cid=cid)
answer = exam.get_question_cid_user_answer(sk, cid_user)
print(answer)
@method_decorator(login_required)
def exam_question_detail(self, request, pk, sk):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user, pk, marker=True):
raise PermissionDenied
# question = exam.get_questions()[sk]
question = exam.get_question_by_index(sk)
exam_length = len(exam.exam_questions.all())
# pos = exam.get_question_index(question) + 1
pos = sk + 1
previous = -1
if sk > 0:
previous = sk - 1
next = sk + 1
if sk == exam_length - 1:
next = False
view_feedback = False
if (
# request.user.groups.filter(name=self.checker_group).exists() or
request.user.groups.filter(name="feedback_checker").exists()
or request.user in question.author.all()
):
view_feedback = True
can_edit = question.can_edit(request.user)
return render(
request,
"{}/question_detail.html".format(self.app_name),
{
"question": question,
"exam": exam,
"next": next,
"previous": previous,
"exam_length": exam_length,
"pos": pos,
"view_feedback": view_feedback,
"can_edit": can_edit,
"remote_url": settings.REMOTE_URL,
"app_name": self.app_name,
},
)
def active_exams(
self,
request: HttpRequest,
json: bool = True,
based: bool = True,
cid: int = None,
passcode: str = None,
):
exams = self.Exam.objects.filter(archive=False)
active_exams = {"exams": []}
if cid is not None:
cid_user = CidUser.objects.filter(cid=cid).first()
if not cid_user or cid_user.passcode != passcode:
if json == False:
return active_exams["exams"]
return JsonResponse({"status": "invalid"}, status=401)
exam: ExamBase
for exam in exams:
if exam.active or self.check_user_access(request.user, exam.pk):
if exam.exam_mode and not exam.check_cid_user(
cid, passcode, user=request.user, user_id=request.user.pk
):
continue
if exam.json_creation_time:
creation_time = exam.json_creation_time.isoformat()
else:
creation_time = "None"
url = request.build_absolute_uri(
exam.get_json_url(cid=cid, passcode=passcode)
)
# hacky
if not based:
url = url + "/unbased"
obj = {
"name": exam.get_exam_name(),
"url": url,
"type": self.question_type,
"eid": "{}/{}".format(self.question_type, exam.pk),
"json_creation_time": creation_time,
"exam_json_id": exam.exam_json_id,
"exam_active": exam.active,
"exam_mode": exam.exam_mode,
}
if self.question_type == "long":
h = {}
for question in exam.exam_questions.all():
# Generate json if needed
if question.json_creation_time is None:
question.get_question_json()
h[question.pk] = question.question_json_id
obj["multi_question_json"] = h
active_exams["exams"].append(obj)
if json is False:
return active_exams["exams"]
return JsonResponse(active_exams)
@method_decorator(csrf_exempt)
def post_exam_answers(self, request):
if request.method == "POST":
n = 0
for answer in json.loads(request.POST.get("answers")):
print("ANSWER", answer)
eid = int(answer["eid"].split("/")[1])
uid = False
passcode = None
try:
cid = int(answer["cid"])
passcode = answer["passcode"]
except ValueError:
if answer["cid"].startswith("u-"):
cid = False
uid = int(answer["cid"][2:])
if not uid == request.user.id:
return JsonResponse(
{"success": False, "error": "user / uid mismatch"}
)
else:
return JsonResponse(
{
"success": False,
"error": "cid or eid or answers not defined",
}
)
if not uid:
# As access is not restricted make sure the data appears valid
if (not isinstance(cid, int)) or (
not isinstance(eid, int) or (not isinstance(answer["ans"], str))
):
return JsonResponse(
{
"success": False,
"error": "cid or eid or answers not defined",
}
)
exam: ExamBase = get_object_or_404(self.Exam, pk=eid)
if not exam.check_cid_user(cid, passcode, user_id=uid):
return JsonResponse(
{"success": False, "error": "invalid cid / passcode"}
)
if uid:
existing_answers = self.UserAnswer.objects.filter(
question__id=answer["qid"], exam__id=eid, user__id=uid
)
else:
existing_answers = self.UserAnswer.objects.filter(
question__id=answer["qid"], exam__id=eid, cid=cid
)
posted_answer = answer["ans"]
match self.app_name:
case "rapids":
# Normal answers are just posted with the answer "Normal"
normal = False
if posted_answer == "Normal":
normal = True
if not existing_answers:
if uid:
ans = self.UserAnswer(
answer=posted_answer,
normal=normal,
user=User.objects.get(id=uid),
)
else:
ans = self.UserAnswer(
answer=posted_answer, normal=normal, cid=cid
)
ans.question_id = answer["qid"]
ans.exam_id = eid
else:
# Update an existing answer
# should never be more than one (famous last words)
ans = existing_answers[0]
if answer["qidn"] == "1":
ans.normal = normal
# Only wipe the answer text if the new answer is "Normal"
if normal:
ans.answer = ""
elif answer["qidn"] == "2" and not ans.normal:
ans.answer = posted_answer
case "shorts":
if not existing_answers:
if uid:
ans = self.UserAnswer(
answer=posted_answer, user=User.objects.get(id=uid)
)
else:
ans = self.UserAnswer(answer=posted_answer, cid=cid)
ans.question_id = answer["qid"]
ans.exam_id = eid
ans.score = None
else:
# Update an existing answer
# should never be more than one (famous last words)
ans = existing_answers[0]
ans.answer = posted_answer
ans.score = None
case "anatomy":
if not existing_answers:
if uid:
ans = self.UserAnswer(
answer=posted_answer, user=User.objects.get(id=uid)
)
else:
ans = self.UserAnswer(answer=posted_answer, cid=cid)
ans.question_id = answer["qid"]
ans.exam_id = eid
ans.score = self.Answer.MarkOptions.UNMARKED
else:
# Update an existing answer
# should never be more than one (famous last words)
ans = existing_answers[0]
ans.answer = posted_answer
ans.score = self.Answer.MarkOptions.UNMARKED
case "longs":
# Long cases seperate sections by qidn
qidn = answer["qidn"]
# If the user answer does not exist
if not existing_answers:
if uid:
ans = self.UserAnswer(user=User.objects.get(id=uid))
else:
ans = self.UserAnswer(cid=cid)
ans.question_id = answer["qid"]
ans.exam_id = eid
# If the answer already exists or we have started populating it
else:
# Update an existing answer
# should never be more than one (famous last words)
ans = existing_answers[0]
if qidn == "1":
ans.answer_observations = posted_answer
elif qidn == "2":
ans.answer_interpretation = posted_answer
elif qidn == "3":
ans.answer_principle_diagnosis = posted_answer
elif qidn == "4":
ans.answer_differential_diagnosis = posted_answer
elif qidn == "5":
ans.answer_management = posted_answer
# Mark as unmarked
ans.score = ans.ScoreOptions.UNMARKED
ans.full_clean()
ans.save()
# if ret is not True:
# return ret
n = n + 1
# print(UserAnswer.objects.filter(exam__id=q["eid"]))
# print(request.urlencode())
exam_type, exam_id = request.POST.get("eid").split("/")
exam = self.Exam.objects.get(pk=exam_id)
t = timezone.make_aware(
datetime.fromtimestamp(float(request.POST.get("start_time")))
)
# We currently store the submission time in two different places
if request.POST.get("cid").startswith("u-"):
cid_user_exam = exam.get_or_create_cid_user_exam(
user_user=request.user, start_time=t
)
else:
c = CidUser.objects.filter(cid=request.POST.get("cid")).first()
cid_user_exam = exam.get_or_create_cid_user_exam(
cid_user=c, start_time=t
)
exam.exam_user_status.create(
cid_user_exam=cid_user_exam,
status="submitted",
extra="manual submission",
)
cid_user_exam.end_time = timezone.now()
cid_user_exam.save()
return JsonResponse({"success": True, "question_count": n})
return JsonResponse({"success": False, "error": "Invalid data"})
def exam_json_cid(self, request, pk, cid, passcode):
return self.exam_json(request, pk, cid, passcode)
def exam_json_cid_unbased(self, request, pk, cid, passcode):
return self.exam_json_unbased(request, pk, cid, passcode)
def exam_json(self, request, pk, cid=None, passcode=None):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
# Check access (for logged in users when inactive)
if not exam.active and not self.check_user_access(request.user, pk):
raise Http404("No available exam")
# TODO: check access for logged in users
print("TEST", cid)
# Check access for users
if request.user.is_anonymous:
user_id = None
else:
user_id = request.user.pk
print(0)
if exam.exam_mode and not exam.check_cid_user(
cid, passcode, request.user, user_id
):
raise Http404("No available exam")
print(1)
# exam_json_cache = cache.get("{}_exam_json_{}".format(self.app_name, pk))
# Log exam access
if exam.exam_mode:
cid_user_exam = exam.get_or_create_cid_user_exam(
cid=cid, user_user=request.user
)
exam.exam_user_status.create(
cid_user_exam=cid_user_exam, status="downloaded"
)
path = "{0}{1}/exam/{2}.json".format(settings.MEDIA_ROOT, self.app_name, pk)
url = "{0}{1}/exam/{2}.json".format(settings.MEDIA_URL, self.app_name, pk)
Path(path).parent.mkdir(parents=True, exist_ok=True)
if os.path.isfile(path) and not exam.recreate_json:
# exam_json_cache["cached"] = True
# Make sure we pass on an argument if we are forcing a reload
if request.GET.get("_"):
query_string = urllib.parse.urlencode({"_": request.GET.get("_")})
url = "{}?{}".format(url, query_string)
return redirect(url)
return redirect(url)
return JsonResponse(exam_json_cache)
time = timezone.now()
exam.exam_json_id += 1
with open(path, "w+") as f:
exam_json = exam.get_exam_json()
exam_json["generated"] = time.isoformat()
exam_json["exam_json_id"] = exam.exam_json_id
f.write(json.dumps(exam_json))
exam.recreate_json = False
exam.json_creation_time = time
exam.save(recreate_json=False)
# We also try to clear the cache of any associated questions (longs)
# see exam_question_json
# n = exam.exam_questions.count()
# question_keys = ["{}_exam_json_{}_{}".format(self.app_name, pk, i) for i in range(1, n)]
# cache.delete_many(question_keys)
# cache.set("{}_exam_json_{}".format(self.app_name, pk), exam_json, 3600)
return JsonResponse(exam_json)
def exam_json_unbased(self, request, pk, cid=None, passcode=None):
"""
No (file based) caching is enabled for unbased exams
"""
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if (
exam.exam_mode
and not exam.active
and not self.check_user_access(request.user, pk)
):
raise Http404("No available exam")
# Check access for users
if request.user.is_anonymous:
user_id = None
else:
user_id = request.user.pk
if not exam.check_cid_user(cid, passcode, request.user, user_id):
raise Http404("No available exam")
time = timezone.now()
exam_json = exam.get_exam_json(based=False)
exam_json["generated"] = time.isoformat()
exam_json["exam_json_id"] = exam.exam_json_id
# exam_json["exam_active"] = False
# Log exam access
if exam.exam_mode:
cid_user_exam = exam.get_or_create_cid_user_exam(
cid=cid, user_user=request.user
)
exam.exam_user_status.create(
cid_user_exam=cid_user_exam, status="downloaded", extra="unbased"
)
return JsonResponse(exam_json)
def exam_question_json_cid(self, request, pk, sk, cid, passcode):
return self.exam_question_json(request, pk, sk, cid, passcode)
def exam_question_json(self, request, pk, sk, cid=None, passcode=None):
question = get_object_or_404(self.Question, pk=sk)
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.active and not self.check_user_access(request.user, pk):
raise Http404("No available exam")
# Check access for users
if request.user.is_anonymous:
user_id = None
else:
user_id = request.user.pk
if not exam.check_cid_user(cid, passcode, request.user, user_id):
raise Http404("No available exam")
if request.GET.get("_"):
base_url = redirect("{}:question_json".format(self.app_name), pk=sk)
query_string = urllib.parse.urlencode({"_": request.GET.get("_")})
url = "{}?{}".format(base_url, query_string)
return redirect(url)
return redirect("{}:question_json".format(self.app_name), pk=sk)
def exam_question_json_unbased_cid(self, request, pk, sk, cid, passcode):
return self.exam_question_json_unbased(request, pk, sk, cid, passcode)
def exam_question_json_unbased(self, request, pk, sk, cid=None, passcode=None):
question = get_object_or_404(self.Question, pk=sk)
exam = get_object_or_404(self.Exam, pk=pk)
print("CID", cid)
if not exam.active and not self.check_user_access(request.user, pk):
raise Http404("No available exam")
# Check access for users
if request.user.is_anonymous:
user_id = None
else:
user_id = request.user.pk
print("CID", cid)
if not exam.check_cid_user(cid, passcode, request.user, user_id):
raise Http404("No available exam")
return redirect("{}:question_json_unbased".format(self.app_name), pk=sk)
redirect()
question_json_cache = cache.get("{}_question_json_{}".format(self.app_name, sk))
if question_json_cache is not None and not question.recreate_json:
question_json_cache["cached"] = True
return JsonResponse(question_json_cache)
question_json = exam.get_exam_question_json(sk)
cache.set("{}_question_json_{}".format(self.app_name, sk), question_json, 3600)
return JsonResponse(question_json)
@method_decorator(login_required)
def exam_scores_user(self, request, pk):
return self.exam_scores_cid_user(request, pk)
@method_decorator(user_passes_test(lambda u: u.is_superuser))
def exam_scores_user_admin(self, request, pk, user_id):
user = User.objects.get(id=user_id)
return self.exam_scores_cid_user(request, pk, user=user)
def exam_scores_user_supervisor(self, request, pk, user_id):
user = User.objects.get(id=user_id)
if not request.user.is_superuser:
if not request.user.supervisor == user.userprofile.supervisor:
raise PermissionDenied
return self.exam_scores_cid_user(
request, pk, user=user, supervisor=user.userprofile.supervisor.user
)
@method_decorator(user_passes_test(lambda u: u.is_superuser))
def exam_scores_cid_user_admin(self, request, pk, cid):
user = CidUser.objects.get(cid=cid)
return self.exam_scores_cid_user(request, pk, cid, user.passcode)
def exam_scores_cid_user(
self, request, pk, cid=None, passcode="", user=None, supervisor=None
):
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
view_all_results = False
if request.user.groups.filter(name="view_all_results").exists():
view_all_results = True
if supervisor is None:
if not exam.check_cid_user(cid, passcode, request.user):
# TODO clear this up
# Should users be able to access results for exams they do not have access to (if they have results)?
if not self.check_user_access(request.user, pk):
# raise PermissionDenied
# if request.user.supervisor == user.userprofile.supervisor:
# pass
# check for supervisor access
raise Http404("Error accessing exam")
else:
if user is None:
raise Exception("user must be defined")
view_all_results = True
if user is None:
user = request.user
questions = (
exam.get_questions()
) # .prefetch_related("cid_user_answers", "answers")
# cid_user_answers = list(UserAnswer.objects.filter(cid=cid, exam__id=pk).prefetch_related("question"))
# cid_user_answers_q_map = {}
# for ans in cid_user_answers:
# cid_user_answers_q_map[ans.question] = ans
answers_and_marks = []
answers_marks = []
answers = []
for q in questions:
# Get user answer
if cid is not None:
user_answer = q.cid_user_answers.filter(cid=cid, exam__id=pk).first()
else:
user_answer = q.cid_user_answers.filter(
user=request.user, exam__id=pk
).first()
# user_answer = cid_user_answers_q_map[q]
feedback = None
if not user_answer or user_answer is None:
# skip if no answer
score, text = q.get_unanswered_mark_and_text()
# answers_marks.append(score)
# answers.append("")
answer_score = score
ans = text
else:
ans = user_answer.get_answer()
answer_score = user_answer.get_answer_score()
if self.app_name in ("longs", "shorts"):
feedback = user_answer.candidate_feedback
match self.app_name:
case "sbas":
correct_answer = q.get_correct_answer()
if not exam.publish_results and not view_all_results:
correct_answer = "*****"
answer_score = 0
answers.append(ans)
answers_marks.append(answer_score)
merged_ans = (ans, answer_score, correct_answer)
chosen_answer = q.get_answer_by_choice(ans)
answers_and_marks.append((q, *merged_ans, chosen_answer))
case "physics":
correct_answer = q.get_answers()
if not exam.publish_results and not view_all_results:
correct_answer = ("*****", "*****", "*****", "*****", "*****")
answer_score = (0, 0, 0, 0, 0)
answers.append(ans)
answers_marks.append(answer_score)
print(q.get_questions(), ans, answer_score, correct_answer)
merged_ans = zip(
q.get_questions(), ans, answer_score, correct_answer
)
answers_and_marks.append((q, merged_ans))
case _:
correct_answer = q.get_primary_answer()
if not exam.publish_results and not view_all_results:
correct_answer = "*****"
answer_score = 0
logger.debug(f"{ans=} {answer_score=} {correct_answer=}")
answers.append(ans)
answers_marks.append(answer_score)
if self.app_name in ("longs", "shorts"):
answers_and_marks.append(
(ans, answer_score, correct_answer, feedback)
)
else:
answers_and_marks.append((ans, answer_score, correct_answer))
print(answers_marks)
match self.app_name:
case "physics":
total_score = sum(sum(i) for i in answers_marks)
case _:
if "unmarked" in answers_marks or None in answers_marks:
answered = [i for i in answers_marks if type(i) == int]
total_score = sum(answered)
unmarked_number = len(answers_marks) - len(answered)
total_score = "{} ({} unmarked)".format(
total_score, unmarked_number
)
else:
total_score = sum(answers_marks)
max_score = self.get_max_score(questions)
#cid_user_exam = exam.cid_user_exam.filter(user_user=user).first()
#print(user)
#print(f"{cid_user_exam=}")
template_context = {
"exam": exam,
"cid": cid,
"passcode": passcode,
"questions": questions,
"answers": answers,
"answers_marks": answers_marks,
"total_score": total_score,
"max_score": max_score,
"answers_and_marks": answers_and_marks,
"view_all_results": view_all_results,
"supervisor": supervisor,
#"cid_user_exam": cid_user_exam,
}
if self.normalise_score is not None:
normalised_score = self.normalise_score(total_score)
template_context["normalised_score"] = normalised_score
return render(
request,
f"{self.app_name}/exam_scores_user.html",
template_context,
)
def get_max_score(self, questions):
match self.app_name:
case "rapids" | "anatomy":
max_score = len(questions) * 2
case "longs":
max_score = len(questions) * 8
case "physics" | "shorts":
max_score = len(questions) * 5
case _:
max_score = len(questions)
return max_score
def exam_scores_all(self, request, pk):
"""The exam scores pages. Displays all user scores in a tabular format.
Args:
request (_type_): _description_
pk (_type_): _description_
Raises:
Http404: _description_
PermissionDenied: _description_
Returns:
_type_: _description_
"""
exam: ExamBase = get_object_or_404(self.Exam, pk=pk)
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
# Restrict access to the scores page just to exam authors
if request.user not in exam.author.all():
raise PermissionDenied
# user_answers_and_marks = defaultdict(list)
user_answers_marks = defaultdict(list)
# user_answers = defaultdict(list)
# user_names = {}
# cid_passcodes = {}
by_question = defaultdict(dict)
unmarked = set()
cached_scores = True
valid_cid_users = set(exam.valid_cid_users.all().values_list("cid", flat=True))
valid_user_users = set(exam.valid_user_users.all().values_list("pk", flat=True))
if self.app_name in ("rapids"):
user_answers_callstates = defaultdict(list)
user_answers_callstates_counted = {}
if self.app_name in ("physics", "sbas", "longs", "shorts"):
cached_scores = False
questions_qs = exam.get_questions()
else:
questions_qs = exam.get_questions().prefetch_related("answers")
# Materialize questions into a list so we can index and build a question->index map
questions = list(questions_qs)
question_index_map = {q: i for i, q in enumerate(questions)}
# We could prefetch the UserAnswers.answers here (if we didn't cache them)
# Also select_related the user to avoid per-answer user lookups in the loop
cid_user_answers = (
self.UserAnswer.objects.select_related("question", "user")
.filter(question__in=questions, exam__id=pk)
)
question_number = len(questions)
cids = set()
plain_cids = set()
user_ids = set()
cids_user_id_map = {}
# Loop through all candidates
for cid_user_answer in cid_user_answers:
# Convoluted (probably...)
if cid_user_answer.user is None:
cid = f"c/{cid_user_answer.cid}"
cids_user_id_map[cid] = cid
# cid_passcodes[cid] = cid_user_answer.passcode
cids.add(cid)
plain_cids.add(cid_user_answer.cid)
else:
cid = f"u/{cid_user_answer.user.pk}"
# cids_user_id_map[cid] = cid_user_answer.user.username
name = f"{cid_user_answer.user.first_name} {cid_user_answer.user.last_name}"
if not name.strip():
name = cid_user_answer.user.username
cids_user_id_map[cid] = name
cids.add(cid)
user_ids.add(cid_user_answer.user.pk)
s = cid_user_answer
# user_names[cid] = cid
q = cid_user_answer.question
# if not s:
# # skip if no answer
# user_answers_marks[cid].append(0)
# user_answers[cid].append("")
# by_question[q].append(("", 0))
# continue
ans = s.get_answer_string()
answer_score = s.get_answer_score()
if answer_score == "unmarked":
# Use precomputed map to avoid repeated scanning/indexing of questions
index = question_index_map.get(q)
if index is not None:
unmarked.add(index)
# user_answers[cid].append(ans)
user_answers_marks[cid].append(answer_score)
if self.app_name == "rapids":
callstate = s.get_answer_callstate()
by_question[q][cid] = (ans, answer_score, callstate)
user_answers_callstates[cid].append(callstate)
elif self.app_name in ("anatomy", "sbas", "longs", "shorts"):
by_question[q][cid] = (ans, answer_score)
else:
zipped_ans_scores = zip(ans, answer_score)
by_question[q][cid] = zipped_ans_scores
user_scores = {}
user_scores_normalised = {}
user_answer_count = {}
for user in user_answers_marks:
# For longs we have a default score of 3 (not 0) if unanswered
# so we need to check for those
if self.app_name == "longs":
for question in questions:
# If either question or user do not exist we give a default
# not answered == score of 3
if user not in by_question[question]:
# print("NOT in")
by_question[question][user] = ("Not answered", 3.0)
user_answers_marks[user].append(3.0)
if self.app_name in ("physics",):
user_scores[user] = sum(
[sum(i) for i in user_answers_marks[user] if i != "unmarked"]
)
else:
user_scores[user] = sum(
[i for i in user_answers_marks[user] if i != "unmarked" and i != None]
)
if self.app_name == "rapids":
user_scores_normalised[user] = normaliseRapidsScore(
sum([i for i in user_answers_marks[user] if i != "unmarked"])
)
else:
user_scores_normalised[user] = user_scores[user]
user_answer_count[user] = len(user_answers_marks[user])
# ignore scores of 0 for stats
user_scores_list = [i for i in user_scores.values() if i > 0]
max_score = self.get_max_score(questions)
if len(user_scores_list) < 1:
mean = 0
median = 0
mode = 0
fig_html = ""
else:
# Exclude very low scores from statistics
lower_score_bound = max_score / 5
bounded_scores = [i for i in user_scores_list if i > lower_score_bound]
if bounded_scores:
user_scores_list = bounded_scores
mean = statistics.mean(user_scores_list)
median = statistics.median(user_scores_list)
try:
mode = statistics.mode(user_scores_list)
except statistics.StatisticsError:
mode = "No unique mode"
df = user_scores_list
fig = px.histogram(
df,
x=0,
title="{}: distribution of scores".format(exam),
labels={"0": "Score"},
height=400,
width=600,
)
fig_html = fig.to_html()
exam.stats_mean = mean
exam.stats_median = median
exam.stats_mode = mode
exam.stats_candidates = len(user_scores_list)
exam.stats_max_possible = max_score
exam.stats_min = min(user_scores_list)
exam.stats_max = max(user_scores_list)
exam.stats_graph = fig_html
user_data = {}
for u in cids:
# uid = cids_user_id_map[u]
user_data[u] = {
"score": user_scores[u],
"normalised_score": user_scores_normalised[u],
}
if self.app_name == "rapids":
counted_callstates = dict(Counter(user_answers_callstates[u]))
user_data[u]["callstates"] = str(counted_callstates)
user_answers_callstates_counted[u] = counted_callstates
exam.user_scores = user_data
exam.save()
missing_user_ids = valid_user_users - user_ids
missing_users = []
if missing_user_ids:
missing_users = User.objects.filter(pk__in=missing_user_ids)
template_variables = {
"cids": sorted(cids),
"missing_cids": valid_cid_users - plain_cids,
"missing_users": missing_users,
# "cid_passcodes": cid_passcodes,
"exam": exam,
"unmarked": unmarked,
"questions": questions,
"question_number": question_number,
"by_question": by_question,
# "user_answers": dict(user_answers),
# "user_answers_marks": dict(user_answers_marks),
"user_scores": user_scores,
"user_scores_normalised": user_scores_normalised,
# "user_scores_list": user_scores_list,
# "user_names": user_names,
# "user_answers_and_marks": user_answers_and_marks,
"user_answer_count": user_answer_count,
"cids_user_id_map": cids_user_id_map,
"max_score": max_score,
"mean": mean,
"median": median,
"mode": mode,
"plot": fig_html,
"cached_scores": cached_scores,
"can_edit": exam.check_user_can_edit(request.user),
}
if self.app_name == "rapids":
template_variables["user_answers_callstates"] = (
user_answers_callstates_counted
)
return render(
request,
f"{self.app_name}/exam_scores.html",
template_variables,
)
def exam_stats(self, request, exam_id):
exam: ExamBase = get_object_or_404(self.Exam, pk=exam_id)
if not self.check_user_edit_access(request.user, exam_id):
raise PermissionDenied()
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
template_variables = {"exam": exam}
return render(
request,
"exam_stats.html",
template_variables,
)
class GenericViewBase:
def __init__(self, app_name, QuestionObject, UserAnswerObject, ExamObject):
self.question_object = QuestionObject
self.cid_user_answer_object = UserAnswerObject
self.exam_object = ExamObject
self.app_name = app_name
g = {
"rapids": "rapid_checker",
"anatomy": "anatomy_checker",
"longs": "longs_checker",
"sbas": "sba_checker",
"physics": "physics_checker",
"shorts": "shorts_checker",
}
self.checker_group = g[app_name]
def check_user_edit_access(self, user, exam_id=None):
"""Check if a user should be able to access editing/reviewing an exam.
This mirrors the older permission helper used elsewhere in the file but
references `self.exam_object` and `self.app_name` (since GenericViewBase
is instantiated per app).
"""
if user.is_superuser:
return True
# If a user is an exam author they should have access
if exam_id is not None:
exam = get_object_or_404(self.exam_object, pk=exam_id)
if user in exam.get_author_objects():
return True
if getattr(exam, "authors_only", False):
return False
# App-specific group checks (require checker role for edit/review)
if (
self.app_name == "rapids"
and not user.groups.filter(name="rapid_checker").exists()
):
return False
if (
self.app_name == "anatomy"
and not user.groups.filter(name="anatomy_checker").exists()
):
return False
if (
self.app_name == "longs"
and not user.groups.filter(name__in=("long_checker",)).exists()
):
return False
if (
self.app_name == "physics"
and not user.groups.filter(name="physics_checker").exists()
):
return False
if (
self.app_name == "sbas"
and not user.groups.filter(name="sba_checker").exists()
):
return False
if (
self.app_name == "casecollection"
and not user.groups.filter(name="casecollection_checker").exists()
):
return False
return False
def help(self, request):
return render(request, f"{self.app_name}/help.html", {"app_name": self.app_name})
def question_review(self, request, pk):
"""
Return a json representation of the question when the exam is published
"""
exam: ExamBase = get_object_or_404(self.exam_object, pk=pk)
print(request.POST["question_number"])
print(type(request.POST["question_number"]))
if not exam.publish_results:
raise Http404
question = exam.get_questions()[int(request.POST["question_number"])]
question_json = question.get_question_json(based=False, feedback=True)
return JsonResponse(question_json)
def question_set_review(self, request, pk):
"""HTMX endpoint to get/set the latest QuestionReview for a question.
GET: renders the current review block or the form (if ?edit=1)
POST: creates a new QuestionReview and returns the rendered block
"""
logger.debug("question_set_review called")
question = get_object_or_404(self.question_object, pk=pk)
ct = ContentType.objects.get_for_model(question)
latest = (
QuestionReview.objects.filter(content_type=ct, object_id=question.pk)
.order_by("-created_on").first()
)
# If GET and edit requested, return the form. If `new` is set, render with no latest so the
# form is blank for creating a fresh review.
if request.method == "GET" and request.GET.get("edit"):
if request.GET.get("new"):
form = QuestionReviewForm()
return render(request, "generic/partials/question_review_form.html", {"question": question, "latest": None, "form": form, "app_name": self.app_name})
# Prefill the form with the latest review values (but we'll create a new review on POST)
if latest:
form = QuestionReviewForm(initial={"status": latest.status, "comment": latest.comment})
else:
form = QuestionReviewForm()
return render(request, "generic/partials/question_review_form.html", {"question": question, "latest": latest, "form": form, "app_name": self.app_name})
if request.method == "POST":
# Require login to post a review
if not request.user.is_authenticated:
return HttpResponse(status=403)
form = QuestionReviewForm(request.POST, user=request.user)
if not form.is_valid():
# Render the form back with errors
return render(request, "generic/partials/question_review_form.html", {"question": question, "latest": latest, "form": form, "app_name": self.app_name})
review = QuestionReview.objects.create(
content_type=ct,
object_id=question.pk,
author=request.user,
status=form.cleaned_data["status"],
comment=form.cleaned_data.get("comment", ""),
)
# Render the updated review block and include an HX-Trigger so the client
# can close the modal (HTMX will dispatch the event named below).
resp = render(request, "generic/partials/question_review_block.html", {"review": review, "question": question, "app_name": self.app_name})
# Trigger a client-side event; base template will listen for this and hide the modal
resp["HX-Trigger"] = "reviewSaved"
return resp
# Default: render the block (latest if exists, otherwise form)
if latest:
return render(request, "generic/partials/question_review_block.html", {"review": latest, "question": question, "app_name": self.app_name})
else:
# Ensure a form is available for the template
if latest:
form = QuestionReviewForm(initial={"status": latest.status, "comment": latest.comment})
else:
form = QuestionReviewForm()
return render(request, "generic/partials/question_review_form.html", {"question": question, "latest": latest, "form": form, "app_name": self.app_name})
def question_reviews_list(self, request, pk):
"""Return a partial listing all QuestionReview instances for a question."""
question = get_object_or_404(self.question_object, pk=pk)
ct = ContentType.objects.get_for_model(question)
reviews = (
QuestionReview.objects.filter(content_type=ct, object_id=question.pk)
.order_by("-created_on")
)
return render(request, "generic/partials/question_reviews_list.html", {"question": question, "reviews": reviews, "app_name": self.app_name})
@method_decorator(login_required)
def exam_review_start(self, request, pk):
"""Start a per-question review for an exam. Redirects to the first question."""
# Ensure the user can edit the exam (authors only)
if not self.check_user_edit_access(request.user, exam_id=pk):
raise PermissionDenied
exam = get_object_or_404(self.exam_object, pk=pk)
# Materialise ordered questions
questions = list(exam.get_questions())
if not questions:
return render(request, "generic/exam_review_complete.html", {"exam": exam, "app_name": self.app_name})
# Render the first question
return self.exam_review_question(request, pk, q_index=0)
@method_decorator(login_required)
def exam_review_question(self, request, pk, q_index=0):
"""Render a single question from an exam for review with navigation."""
if not self.check_user_edit_access(request.user, exam_id=pk):
raise PermissionDenied
exam = get_object_or_404(self.exam_object, pk=pk)
questions = list(exam.get_questions())
total = len(questions)
try:
q_index = int(q_index)
except Exception:
q_index = 0
if q_index < 0 or q_index >= total:
# Out of range — render complete
return render(request, "generic/exam_review_complete.html", {"exam": exam, "app_name": self.app_name})
question = questions[q_index]
prev_index = q_index - 1 if q_index > 0 else None
next_index = q_index + 1 if q_index < total - 1 else None
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"total": total,
"prev_index": prev_index,
"next_index": next_index,
"app_name": self.app_name,
"can_edit": self.check_user_edit_access(request.user, exam_id=pk),
}
return render(request, "generic/exam_review_question.html", context)
def question_review_start(self, request):
# Prepare category choices if the question model exposes a FK named 'category'
categories = None
try:
field = self.question_object._meta.get_field("category")
# Only handle FK-like fields
if hasattr(field, "related_model") and field.related_model is not None:
CategoryModel = field.related_model
# prefer 'category' or 'name' display field if present
order_by = "category" if hasattr(CategoryModel, "category") else "name" if hasattr(CategoryModel, "name") else "pk"
categories = CategoryModel.objects.all().order_by(order_by)
except Exception:
categories = None
# Status options map: label -> status code (or special 'UNREVIEWED')
status_options = [
("ANY", "Any"),
("UNREVIEWED", "Unreviewed"),
("AC", "Accepted"),
("OD", "Outdated"),
("ER", "Error"),
("RJ", "Rejected"),
("IP", "In Progress"),
]
return render(
request,
f"{self.app_name}/question_review_start.html",
{"app_name": self.app_name, "categories": categories, "status_options": status_options},
)
def question_review_next(self, request):
"""Find and redirect to the next question matching the supplied filters.
Accepts POST or GET parameters:
- category: integer primary key of category (optional)
- status: one of the status codes (AC, OD, ER, RJ, IP), or 'UNREVIEWED' or 'ANY'
"""
# Read filters
category = request.POST.get("category") or request.GET.get("category")
status = request.POST.get("status") or request.GET.get("status") or "ANY"
# How many matching questions to skip (useful for Skip button). Expected integer >= 0.
try:
skip = int(request.POST.get("skip") or request.GET.get("skip") or 0)
if skip < 0:
skip = 0
except Exception:
skip = 0
# Build base queryset of questions for this app
qs = self.question_object.objects.all().order_by("pk")
# Optionally restrict iteration to questions reviewed by the current user
reviewed_filter = (request.POST.get("reviewed") or request.GET.get("reviewed")) == "me"
if reviewed_filter:
try:
from django.db.models import Exists, OuterRef
# use the module-level ContentType (imported near file top)
ct = ContentType.objects.get_for_model(self.question_object)
reviewed_exists = QuestionReview.objects.filter(
content_type=ct, object_id=OuterRef("pk"), author=request.user
)
qs = qs.annotate(reviewed_by_me=Exists(reviewed_exists))
except Exception:
reviewed_filter = True
# Restrict by category if provided and the model has such a relation
if category:
try:
qs = qs.filter(category__id=int(category))
except Exception:
pass
# Apply permission filter similar to filters elsewhere: non-checkers see only open_access or their own
if not request.user.groups.filter(name=self.checker_group).exists():
try:
qs = qs.filter(open_access=True) | qs.filter(author__id=request.user.id)
except Exception:
# If model doesn't have those fields, ignore
pass
# Iterate questions and pick first matching status
for question in qs:
# Skip authors_only questions unless user is author or superuser
try:
if question.authors_only and not (
request.user.is_superuser or request.user in question.author.all()
):
continue
except Exception:
pass
ct = ContentType.objects.get_for_model(question)
latest = (
QuestionReview.objects.filter(content_type=ct, object_id=question.pk)
.order_by("-created_on").first()
)
match = False
if status == "ANY":
match = True
elif status == "UNREVIEWED":
match = latest is None
else:
# status is expected to be a QuestionReview.StatusChoices code
if latest is not None and latest.status == status:
match = True
if match:
# If caller asked us to skip N matches, decrement and continue until skip==0
if skip and skip > 0:
skip -= 1
continue
form = QuestionReviewForm()
return render(request, f"{self.app_name}/question_review_question.html", {"question": question, "form": form, "app_name": self.app_name, "filters": {"category": category, "status": status}})
# Nothing found - render complete page
return render(request, f"{self.app_name}/question_review_complete.html", {"app_name": self.app_name})
def question_review_list(self, request):
"""Return a page listing questions that match the supplied filters.
Accepts query parameters similar to question_review_next: category, status.
"""
# Read filters
category = request.GET.get("category") or request.POST.get("category")
status = request.GET.get("status") or request.POST.get("status") or "ANY"
# Build base queryset of questions for this app
qs = self.question_object.objects.all().order_by("pk")
# Optionally filter to questions reviewed by the current user.
# Use an Exists subquery to avoid fetching extra rows.
reviewed_filter = (request.GET.get("reviewed") or request.POST.get("reviewed")) == "me"
if reviewed_filter:
try:
from django.db.models import Exists, OuterRef
# use the module-level ContentType (imported near file top)
ct = ContentType.objects.get_for_model(self.question_object)
reviewed_exists = QuestionReview.objects.filter(
content_type=ct, object_id=OuterRef("pk"), author=request.user
)
qs = qs.annotate(reviewed_by_me=Exists(reviewed_exists))
except Exception:
# If annotation fails for any reason, fall back to no-op and we'll filter in Python loop
reviewed_filter = True
# Restrict by category if provided
if category:
try:
qs = qs.filter(category__id=int(category))
except Exception:
pass
# Apply permission filter similar to question_review_next
if not request.user.groups.filter(name=self.checker_group).exists():
try:
qs = qs.filter(open_access=True) | qs.filter(author__id=request.user.id)
except Exception:
pass
# Annotate with latest review status using a DB subquery for efficiency
try:
from django.db.models import OuterRef, Subquery
ct = ContentType.objects.get_for_model(self.question_object)
latest_status_subq = (
QuestionReview.objects.filter(content_type=ct, object_id=OuterRef("pk"))
.order_by("-created_on")
.values("status")[:1]
)
qs = qs.annotate(latest_review_status=Subquery(latest_status_subq))
except Exception:
# annotation failed — we'll fall back to per-object lookup below
pass
# Optionally filter by review status
results = []
for question in qs:
try:
if question.authors_only and not (
request.user.is_superuser or request.user in question.author.all()
):
continue
except Exception:
pass
# Determine latest status — prefer annotated value if present
latest_status = getattr(question, "latest_review_status", None)
match = False
if status == "ANY":
match = True
elif status == "UNREVIEWED":
match = latest_status is None
else:
if latest_status is not None and latest_status == status:
match = True
if match:
# If the caller requested only questions reviewed by the current user,
# skip any question that does not have a matching review.
if reviewed_filter:
# If annotation was applied this attribute will be present on the instance.
if hasattr(question, "reviewed_by_me"):
if not question.reviewed_by_me:
continue
else:
# Last-resort check via DB lookup
ct = ContentType.objects.get_for_model(question)
if not QuestionReview.objects.filter(content_type=ct, object_id=question.pk, author=request.user).exists():
continue
results.append(question)
# Provide human-readable labels for statuses to the template
status_choices = {code: label for code, label in QuestionReview.StatusChoices.choices}
return render(
request,
f"{self.app_name}/question_review_list.html",
{
"questions": results,
"app_name": self.app_name,
"filters": {"category": category, "status": status},
"status_choices": status_choices,
},
)
@method_decorator(user_passes_test(lambda u: u.is_superuser))
def user_answer_delete_multiple(self, request):
print(request.POST["answer_ids"])
answer_ids = json.loads(request.POST["answer_ids"])
# We could probably delete them all at once....
for id in answer_ids:
self.cid_user_answer_object.objects.get(pk=id).delete()
return JsonResponse({"status": "success"})
@method_decorator(login_required)
def question_thumbnail_fail(self, request, pk):
return self.question_thumbnail(request, pk=pk, fail_loudly=True)
@method_decorator(login_required)
def question_thumbnail(self, request, pk, fail_loudly=False):
question = get_object_or_404(self.question_object, pk=pk)
thumbnail, _ = question.get_thumbnail(recreate=True, fail_loudly=fail_loudly)
return HttpResponse(thumbnail)
@method_decorator(login_required)
def question_detail(self, request, pk):
question: QuestionBase = get_object_or_404(self.question_object, pk=pk)
if not question.open_access:
if (
not request.user.groups.filter(name=self.checker_group).exists()
and request.user not in question.author.all()
):
raise PermissionDenied()
# if request.user not in rapid.author.all():
# raise PermissionDenied
view_feedback = False
if (
request.user.groups.filter(name=self.checker_group).exists()
or request.user.groups.filter(name="feedback_checker").exists()
or request.user in question.author.all()
):
view_feedback = True
# logging.debug(rapid.subspecialty.first().name.all())
return render(
request,
f"{self.app_name}/question_detail.html",
{
"question": question,
"view_feedback": view_feedback,
"remote_url": settings.REMOTE_URL,
"can_edit": question.can_edit(request.user),
"app_name": self.app_name,
},
)
# TODO: improve permissions on these
@method_decorator(login_required)
def question_user_answers_by_compare(self, request, pk, answer_compare):
print(answer_compare)
answer_compare = urllib.parse.unquote(answer_compare)
return self.question_user_answers(request, pk, answer_compare)
@method_decorator(login_required)
def question_user_answers(self, request, pk, answer_compare=None):
print(locals())
question = get_object_or_404(self.question_object, pk=pk)
if answer_compare is None:
answers = self.cid_user_answer_object.objects.filter(
question__id=question.pk
).prefetch_related("question", "exam", "user")
else:
answers = self.cid_user_answer_object.objects.filter(
question__id=question.pk, answer_compare=answer_compare
).prefetch_related("question", "exam", "user")
return render(
request,
f"{self.app_name}/question_user_answers.html",
{
"question": question,
"answers": answers,
"answer_compare": answer_compare,
},
)
@method_decorator(login_required)
def author_detail(self, request, pk):
# logging.debug(Author.objects.all())
# author = get_object_or_404(Author, pk=pk)
author = User.objects.get(pk=pk)
questions = self.question_object.objects.filter(author=pk)
return render(
request,
f"{self.app_name}/category_detail.html",
{"category": author, "questions": questions},
)
@method_decorator(login_required)
def author_list(self, request):
authors = User.objects.all()
return render(
request, f"{self.app_name}/author_list.html", {"authors": authors}
)
class ExamCloneMixin:
def get_template_names(self) -> list[str]:
return "exam_clone_form.html"
# return super().get_template_names()
def get_initial(self):
old_object = get_object_or_404(self.model, pk=self.kwargs["exam_id"])
if not old_object.check_user_can_edit(self.request.user):
raise PermissionDenied() # or Http404
initial_data = model_to_dict(old_object, exclude=["id"])
# We manually transfer the forign keys / m2m relationships
questions = old_object.exam_questions.all().values_list("id", flat=True)
authors = old_object.author.all().values_list("id", flat=True)
# clear the associated groups
initial_data["valid_user_users"] = []
initial_data["valid_cid_users"] = []
initial_data["cid_user_groups"] = []
initial_data["user_user_groups"] = []
initial_data["active"] = False
initial_data["archive"] = False
initial_data["publish_results"] = False
self.exam_questions = list(questions)
self.author = list(authors)
self.old_object = old_object
return initial_data
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["can_edit"] = self.old_object.check_user_can_edit(self.request.user)
context["exam"] = self.old_object
return context
def form_valid(self, form):
object = form.save()
# Reapply these otherwise they get lost?
object.exam_questions.set(self.exam_questions)
object.author.set(self.author)
object.save()
object.order_questions()
return HttpResponseRedirect(object.get_absolute_url())
class ExaminationAutocomplete(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 Examination.objects.none()
qs = Examination.objects.all()
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(examination__icontains=self.q)
except FieldError:
return Examination.objects.none()
return qs
class SupervisorAutocomplete(autocomplete.Select2QuerySetView):
def get_queryset(self):
# TODO: we should probably filter this to only
# allow access by trainees / other suprevisors
if not self.request.user.is_authenticated:
return Supervisor.objects.none()
qs = Supervisor.objects.all().order_by("name")
if self.q:
# This raises a fielderror which breaks creating a new item if not caught
try:
qs = qs.filter(Q(email__icontains=self.q) | Q(name__icontains=self.q))
except FieldError:
return Supervisor.objects.none()
return qs
class CidUserExamView(CidManagerRequiredMixin, SingleTableMixin, FilterView):
model = CidUser
table_class = CidUserExamTable
template_name = "generic/cid_view.html"
filterset_class = CidUserFilter
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
# user = self.request.user
filters = {"archive": False, "exam_mode": True}
physics_exams = [(i.name, i.pk) for i in PhysicsExam.objects.filter(**filters)]
rapid_exams = [(i.name, i.pk) for i in RapidsExam.objects.filter(**filters)]
shorts_exams = [(i.name, i.pk) for i in ShortsExam.objects.filter(**filters)]
sba_exams = [(i.name, i.pk) for i in SbasExam.objects.filter(**filters)]
longs_exams = [(i.name, i.pk) for i in LongsExam.objects.filter(**filters)]
anatomy_exams = [(i.name, i.pk) for i in AnatomyExam.objects.filter(**filters)]
casecollection_exams = [
(i.name, i.pk)
for i in CaseCollection.objects.filter(archive=False, collection_type__gt=0)
]
cid_user_groups = [
(i.name, i.pk) for i in CidUserGroup.objects.filter(archive=False)
]
context["physics_exams"] = physics_exams
context["rapid_exams"] = rapid_exams
context["shorts_exams"] = shorts_exams
context["sba_exams"] = sba_exams
context["longs_exams"] = longs_exams
context["anatomy_exams"] = anatomy_exams
context["casecollection_exams"] = casecollection_exams
context["cid_user_groups"] = cid_user_groups
context["exams"] = True
return context
class CidUserView(CidManagerRequiredMixin, SingleTableMixin, FilterView):
model = CidUser
table_class = CidUserTable
template_name = "generic/cid_view.html"
filterset_class = CidUserFilter
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
# user = self.request.user
filters = {"archive": False, "exam_mode": True}
physics_exams = [(i.name, i.pk) for i in PhysicsExam.objects.filter(**filters)]
rapid_exams = [(i.name, i.pk) for i in RapidsExam.objects.filter(**filters)]
shorts_exams = [(i.name, i.pk) for i in ShortsExam.objects.filter(**filters)]
sba_exams = [(i.name, i.pk) for i in SbasExam.objects.filter(**filters)]
longs_exams = [(i.name, i.pk) for i in LongsExam.objects.filter(**filters)]
anatomy_exams = [(i.name, i.pk) for i in AnatomyExam.objects.filter(**filters)]
casecollection_exams = [
(i.name, i.pk)
for i in CaseCollection.objects.filter(archive=False, collection_type__gt=0)
]
context["physics_exams"] = physics_exams
context["rapid_exams"] = rapid_exams
context["shorts_exams"] = shorts_exams
context["sba_exams"] = sba_exams
context["longs_exams"] = longs_exams
context["anatomy_exams"] = anatomy_exams
context["casecollection_exams"] = casecollection_exams
cid_user_groups = [
(i.name, i.pk) for i in CidUserGroup.objects.filter(archive=False)
]
context["cid_user_groups"] = cid_user_groups
return context
@user_is_cid_user_manager
def cid_group_view(request):
groups = CidUserGroup.objects.filter(archive=False)
return render(
request,
"generic/cid_group_view.html",
{"groups": groups},
)
@user_is_cid_user_manager
def user_group_add_by_email_user(request, group_id):
if request.htmx:
group = get_object_or_404(UserUserGroup, pk=group_id)
emails = request.POST.get("emails")
if not emails:
return HttpResponse("No emails provided", content_type="text/plain")
added = []
invalid_emails = []
for email in emails.split():
try:
validate_email(email)
try:
user = User.objects.get(email=email)
group.users.add(user)
added.append(user.username)
except User.DoesNotExist:
invalid_emails.append(email)
except ValidationError:
# Ignore invalid emails
pass
res = f"<div class='alert alert-primary'>Users added to {group.name} [{','.join(added)} ({len(added)})]</div>"
if invalid_emails:
res += f"<br/><div class='alert alert-warning'>Invalid emails [{','.join(invalid_emails)}]</div>"
return HttpResponse(format_html(res), content_type="text/html")
raise Http404
@user_is_cid_user_manager
def cid_group_view_detail(request, group_id):
group = get_object_or_404(CidUserGroup, pk=group_id)
users = group.ciduser_set.filter(active=True)
return render(
request,
"generic/cid_group_view_detail.html",
{"group": group, "users": users, "group_type": "cid"},
)
@user_is_cid_user_manager
def cid_group_view_all(request):
groups = CidUserGroup.objects.filter()
return render(
request,
"generic/cid_group_view.html",
{"groups": groups, "view_all": True},
)
@user_is_cid_user_manager
def cid_group_add_candidates_to_exams(request, group_id):
if request.htmx:
group = get_object_or_404(CidUserGroup, pk=group_id)
exam_id = request.POST.get("exam_id")
exam_type = request.POST.get("exam_type")
ExamModel = get_exam_model_from_app_name(exam_type)
exam = get_object_or_404(ExamModel, pk=exam_id)
if request.POST.get("action") == "remove":
exam.valid_cid_users.remove(*group.ciduser_set.all())
return HttpResponse(f"Candidates removed")
else:
exam.valid_cid_users.add(*group.ciduser_set.all())
return HttpResponse(f"Candidates added")
# exam.save()
raise PermissionDenied() # or Http404
@user_is_cid_user_manager
def user_group_add_candidates_to_exams(request, group_id):
if request.htmx:
group = get_object_or_404(UserUserGroup, pk=group_id)
exam_id = request.POST.get("exam_id")
exam_type = request.POST.get("exam_type")
ExamModel = get_exam_model_from_app_name(exam_type)
exam = get_object_or_404(ExamModel, pk=exam_id)
if request.POST.get("action") == "remove":
exam.valid_user_users.remove(*group.users.all())
return HttpResponse(f"Candidates removed")
else:
exam.valid_user_users.add(*group.users.all())
return HttpResponse(f"Candidates added")
# exam.save()
raise PermissionDenied() # or Http404
pass
@user_is_cid_user_manager
def user_group_view_detail(request, group_id):
group = get_object_or_404(UserUserGroup, pk=group_id)
users = group.users.filter() # Filter inactive users?
return render(
request,
"generic/cid_group_view_detail.html",
{"group": group, "users": users, "group_type": "user"},
)
@user_is_cid_user_manager
def users_bulk_delete_supervisors(request):
if "selection" in request.POST:
selected_users = request.POST.getlist("selection")
user_models = User.objects.filter(id__in=selected_users)
for u in user_models:
u.userprofile.supervisor = None
u.save()
modified_users = ",".join([user.username for user in user_models])
return HttpResponse(
f"Supervisors removed from {modified_users}. Refresh page to see update",
content_type="text/plain",
)
else:
return HttpResponse("No users selected", content_type="text/plain")
class NoUsersSelected(Exception):
pass
def get_user_selection_from_request(request):
selected_users = request.POST.getlist("selection")
if not selected_users:
raise NoUsersSelected
user_models = User.objects.filter(id__in=selected_users)
return user_models
@user_is_cid_user_manager
def user_not_trainee(request, user_id):
user = get_object_or_404(User, pk=user_id)
user.userprofile.peninsula_trainee = False
user.userprofile.save()
return HttpResponse(
f"{user.username} is no longer a trainee", content_type="text/plain"
)
@user_is_cid_user_manager
def users_bulk_edit(request):
print(request.htmx.trigger)
print(request.htmx.trigger_name)
try:
match request.htmx.trigger:
case r if r is None:
html = "None"
case "bulk-deactivate-user-button":
user_models = get_user_selection_from_request(request)
# u: User
for u in user_models:
u.is_active = False
u.save()
html = "Users deactivated"
case "bulk-edit-grade-button":
user_grades = UserGrades.objects.all()
html = "".join(
[
f"""<button class='change-grade-button' id='add-grade--{grade.id}'
hx-post="{reverse('generic:users_bulk_edit')}"
hx-include="[name='selection']"
hx-confirm="This will modify grades of all selected users, are you sure you wish to continue?"
hx-target="#htmx-info"
_='on click put "" into #htmx-options'
>{grade.name}</button>"""
for grade in user_grades
]
)
# TODO: work out how to actually use hyperscript
html = (
html
+ f"""<button class='change-grade-button' id='add-grade--remove'
hx-post="{reverse('generic:users_bulk_edit')}"
hx-include="[name='selection']"
hx-confirm="This will remove grades of all selected users, are you sure you wish to continue?"
hx-target="#htmx-info"
_='on click put "" into #htmx-options'
>Remove</button>"""
)
html = (
html
+ "<button class='cancel-button' _='on click for el in .change-grade-button remove el end then remove me'>Cancel</button>"
)
case r if r.startswith("add-grade"):
user_models = get_user_selection_from_request(request)
grade = r.split("--")[-1]
if grade == "remove":
grade = None
# u: User
for u in user_models:
u.userprofile.grade_id = grade
u.save()
html = "Grades updated"
case "bulk-add-group-button":
user_groups = UserUserGroup.objects.all()
html = "".join(
[
f"""<button class='add-group-button' id='add-group--{group.id}'
hx-post="{reverse('generic:users_bulk_edit')}"
hx-include="[name='selection']"
hx-confirm="This will add the group '{group.name}' to all selected users, are you sure you wish to continue?"
hx-target="#htmx-info"
_='on click put "" into #htmx-options'
>{group.name}</button>"""
for group in user_groups
]
)
# TODO: work out how to actually use hyperscript
html = (
html
+ "<button class='cancel-button'_='on click for el in .add-group-button remove el end then remove me'>Cancel</button>"
)
case "bulk-remove-group-button":
user_groups = UserUserGroup.objects.all()
html = "".join(
[
f"""<button class='remove-group-button' id='remove-group--{group.id}'
hx-post="{reverse('generic:users_bulk_edit')}"
hx-include="[name='selection']"
hx-confirm="This will remove the group '{group.name}' from all selected users, are you sure you wish to continue?"
hx-target="#htmx-info"
_='on click put "" into #htmx-options'
>{group.name}</button>"""
for group in user_groups
]
)
# TODO: work out how to actually use hyperscript
html = (
html
+ "<button class='cancel-button'_='on click for el in .add-group-button remove el end then remove me'>Cancel</button>"
)
case r if r.startswith("add-group"):
user_models = get_user_selection_from_request(request)
# u: User
for u in user_models:
pass
u.user_groups.add(r.split("--")[-1])
u.save()
html = "Group added"
case r if r.startswith("remove-group"):
user_models = get_user_selection_from_request(request)
# u: User
for u in user_models:
group_id = r.split("--")[-1]
u.user_groups.remove(group_id)
html = "Group removed"
case _:
html = "None"
return HttpResponse(html, content_type="text/html")
except NoUsersSelected:
return HttpResponse("No users selected", content_type="text/html")
@user_is_cid_user_manager
def user_group_view(request):
groups = UserUserGroup.objects.filter(archive=False)
return render(
request,
"generic/user_group_view.html",
{"groups": groups},
)
@user_is_cid_user_manager
def user_group_view_all(request):
groups = UserUserGroup.objects.filter()
return render(
request,
"generic/user_group_view.html",
{"groups": groups, "view_all": True},
)
@login_required
def user_toggle_group(request, user_id, group_id):
"""Toggle membership of a user in either a UserUserGroup or an auth Group.
Visible to superusers and users in the `cid_user_manager` group. The
caller may pass a querystring `?type=auth` to toggle a Django auth Group; by
default it toggles the project's UserUserGroup.
Returns the rendered groups fragment for HTMX swaps.
"""
if request.method != "POST":
return HttpResponse("Method not allowed", status=405)
# Allow superusers or members of the cid_user_manager group
if not (
request.user.is_superuser
or request.user.groups.filter(name="cid_user_manager").exists()
):
return HttpResponse("Forbidden", status=403)
user = get_object_or_404(User, pk=user_id)
gtype = request.GET.get("type", "custom")
if gtype == "auth":
# Toggle Django auth Group membership
auth_group = get_object_or_404(Group, pk=group_id)
if auth_group.user_set.filter(pk=user.pk).exists():
auth_group.user_set.remove(user)
else:
auth_group.user_set.add(user)
else:
# Toggle project UserUserGroup membership
group = get_object_or_404(UserUserGroup, pk=group_id)
if group.users.filter(pk=user.pk).exists():
group.users.remove(user)
else:
group.users.add(user)
# Prepare available lists for the fragment
available_groups = UserUserGroup.objects.filter(archive=False).exclude(
pk__in=user.user_groups.values_list("pk", flat=True)
)
available_auth_groups = Group.objects.exclude(
pk__in=user.groups.values_list("pk", flat=True)
)
is_cid_user_manager = request.user.groups.filter(name="cid_user_manager").exists()
if gtype == "auth":
return render(
request,
"generic/partials/auth_groups_fragment.html",
{
"user": user,
"available_auth_groups": available_auth_groups,
"is_cid_user_manager": is_cid_user_manager,
},
)
return render(
request,
"generic/partials/user_groups_fragment.html",
{
"user": user,
"available_groups": available_groups,
"available_auth_groups": available_auth_groups,
"is_cid_user_manager": is_cid_user_manager,
},
)
@login_required
@user_is_cid_user_manager
def group_email_resend(request, pk):
return group_email(request, pk, resend=True)
@login_required
@user_is_cid_user_manager
def group_email_results_resend(request, pk):
return group_email_results(request, pk, resend=True)
@login_required
@user_is_cid_user_manager
def group_email(request, pk, resend=False):
group = get_object_or_404(CidUserGroup, pk=pk)
if resend:
users = group.ciduser_set.filter(active=True)
else:
users = group.ciduser_set.filter(active=True, login_email_sent=False)
return render(
request,
"generic/group_emailed.html",
# {"sent": sent, "not_sent": not_sent, "users_count": users.count()},
{"users": users, "users_count": users.count(), "results": False},
)
@user_is_cid_user_manager
def group_email_results(request, pk, resend=False):
group = get_object_or_404(CidUserGroup, pk=pk)
if resend:
users = group.ciduser_set.filter(active=True, internal_candidate=True)
else:
users = group.ciduser_set.filter(
active=True, internal_candidate=True, results_email_sent=False
)
# sent = []
# not_sent = []
# for u in users:
# success, msg = u.email_results(resend=resend)
# if success:
# sent.append(u)
# else:
# not_sent.append((u, msg))
return render(
request,
"generic/group_emailed.html",
# {"sent": sent, "not_sent": not_sent, "users_count": users.count()},
{"users": users, "users_count": users.count(), "results": True},
)
@user_is_cid_user_manager
def candidate_email_details(request, cid, resend=False):
user = CidUser.objects.get(cid=cid)
email_sent, comment = user.email_details(resend=False)
return JsonResponse({"sent": email_sent, "comment": comment})
@user_is_cid_user_manager
def candidate_email_results(request, cid, resend=False):
user = CidUser.objects.get(cid=cid)
email_sent, comment = user.email_results(
additional_emails=["priya.suresh@nhs.net"], resend=True
)
return JsonResponse({"sent": email_sent, "comment": comment})
@user_is_cid_user_manager
def candidate_email_results_resend(request, cid, resend=True):
return candidate_email_results(request, cid, resend=True)
@user_is_cid_user_manager
def create_cid_email(request):
pass
@user_is_cid_user_manager
def cid_details(request, cid: int):
print(cid)
if request.htmx:
cid_user = get_object_or_404(CidUser, cid=cid)
print(cid_user.name)
return HttpResponse(f"{cid_user.name} ({cid_user.email})")
raise PermissionDenied() # or Http404
@user_is_cid_user_manager
def manage_cid_users(request):
if request.method == "POST":
physics_exams = json.loads(request.POST.get("physics_exams"))
rapid_exams = json.loads(request.POST.get("rapid_exams"))
sba_exams = json.loads(request.POST.get("sba_exams"))
longs_exams = json.loads(request.POST.get("longs_exams"))
anatomy_exams = json.loads(request.POST.get("anatomy_exams"))
shorts_exams = json.loads(request.POST.get("shorts_exams"))
casecollection_exams = json.loads(request.POST.get("casecollection_exams"))
cid_groups = json.loads(request.POST.get("cid_groups"))
selected_cids = json.loads(request.POST.get("selected_cids"))
emails = json.loads(request.POST.get("emails"))
number_to_create = int(request.POST.get("number_to_create"))
add_group = request.POST.get("add_group")
delete = request.POST.get("delete")
activate = request.POST.get("activate")
deactivate = request.POST.get("deactivate")
add_exams = request.POST.get("add_exams")
change_exams = request.POST.get("change_exams")
clear_exams = request.POST.get("clear_exams")
toggle_internal = request.POST.get("toggle_internal")
if add_group == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.group_id = cid_groups
c.save()
return JsonResponse({"status": "success"})
if delete == "true":
if not request.user.is_superuser:
return JsonResponse({"status": "fail"})
cids = CidUser.objects.filter(pk__in=selected_cids).delete()
return JsonResponse({"status": "success"})
if activate == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.active = True
c.save()
return JsonResponse({"status": "success"})
if toggle_internal == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.internal_candidate = not c.internal_candidate
c.save()
return JsonResponse({"status": "success"})
if deactivate == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.active = False
c.save()
return JsonResponse({"status": "success"})
if clear_exams == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.physics_exams.clear()
c.rapid_exams.clear()
c.sba_exams.clear()
c.longs_exams.clear()
c.anatomy_exams.clear()
c.shorts_exams.clear()
c.casecollection_exams.clear()
return JsonResponse({"status": "success"})
if add_exams == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
for exam in physics_exams:
c.physics_exams.add(exam)
for exam in rapid_exams:
c.rapid_exams.add(exam)
for exam in sba_exams:
c.sba_exams.add(exam)
for exam in longs_exams:
c.longs_exams.add(exam)
for exam in anatomy_exams:
c.anatomy_exams.add(exam)
for exam in shorts_exams:
c.shorts_exams.add(exam)
for exam in casecollection_exams:
c.casecollection_exams.add(exam)
c.save()
return JsonResponse({"status": "success"})
if change_exams == "true":
cids = CidUser.objects.filter(pk__in=selected_cids)
for c in cids:
c.physics_exams.set(physics_exams)
c.rapid_exams.set(rapid_exams)
c.sba_exams.set(sba_exams)
c.longs_exams.set(longs_exams)
c.anatomy_exams.set(anatomy_exams)
c.shorts_exams.set(shorts_exams)
c.casecollection_exams.set(casecollection_exams)
c.save()
return JsonResponse({"status": "success"})
try:
new_cid = get_next_cid()
except IndexError:
new_cid = 10000
if "add_new_emails" in request.POST:
email_list = [i.strip() for i in emails.split()]
# Verify emails are all valid
invalid_emails = []
for email in email_list:
try:
validate_email(email)
except ValidationError as e:
invalid_emails.append(f"Invalid email: {email}")
print(f"Email list {email_list}")
if not email_list:
return JsonResponse({"status": "fail", "details": "No valid emails"})
if invalid_emails or not email_list:
e = "<br/>".join(invalid_emails)
return JsonResponse(
{"status": "fail", "details": f"Unable to create users<hr>{e}"}
)
for e in email_list:
passcode = "".join(random.choices(string.ascii_uppercase, k=4))
c = CidUser(cid=new_cid, passcode=passcode, email=e)
c.save()
if cid_groups:
c.group_id = cid_groups[0]
if physics_exams:
c.physics_exams.set(physics_exams)
if rapid_exams:
c.rapid_exams.set(rapid_exams)
if sba_exams:
c.sba_exams.set(sba_exams)
if longs_exams:
c.longs_exams.set(longs_exams)
if anatomy_exams:
c.anatomy_exams.set(anatomy_exams)
if shorts_exams:
c.shorts_exams.set(shorts_exams)
if casecollection_exams:
c.casecollection_exams.set(casecollection_exams)
c.save()
new_cid = new_cid + 1
for n in range(number_to_create):
passcode = "".join(random.choices(string.ascii_uppercase, k=4))
c = CidUser(cid=new_cid, passcode=passcode)
c.save()
if cid_groups:
c.group_id = cid_groups[0]
if physics_exams:
c.physics_exams.set(physics_exams)
if rapid_exams:
c.rapid_exams.set(rapid_exams)
if sba_exams:
c.sba_exams.set(sba_exams)
if longs_exams:
c.longs_exams.set(longs_exams)
if anatomy_exams:
c.anatomy_exams.set(anatomy_exams)
if shorts_exams:
c.shorts_exams.set(shorts_exams)
if casecollection_exams:
c.casecollection_exams.set(casecollection_exams)
c.save()
new_cid = new_cid + 1
return JsonResponse({"status": "success"})
return JsonResponse({"status": "fail"})
class CidUserCreate(CidManagerRequiredMixin, CreateView):
model = CidUser
form_class = CidUserForm
class CidUserUpdate(CidManagerRequiredMixin, UpdateView):
model = CidUser
form_class = CidUserForm
class CidUserExamUpdate(CidManagerRequiredMixin, UpdateView):
model = CidUser
form_class = CidUserExamForm
template_name = "generic/ciduserexam_form.html"
# def get_context_data(self, *args, **kwargs):
# context = super().get_context_data(**kwargs)
# context["test"] = "HELLO"
# return context
class CidUserGroupDelete(RevisionMixin, CidManagerRequiredMixin, DeleteView):
model = CidUserGroup
template_name = "confirm_delete.html"
success_url = reverse_lazy("generic:cid_group_view")
class CidUserGroupCreate(RevisionMixin, CidManagerRequiredMixin, CreateView):
model = CidUserGroup
form_class = CidUserGroupForm
class CidUserGroupUpdate(RevisionMixin, CidManagerRequiredMixin, UpdateView):
model = CidUserGroup
form_class = CidUserGroupForm
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
context = super().get_context_data(**kwargs)
context["group_type"] = "cid"
return context
class UserUserGroupCreate(RevisionMixin, CidManagerRequiredMixin, CreateView):
model = UserUserGroup
form_class = UserUserGroupForm
class UserUserGroupUpdate(RevisionMixin, CidManagerRequiredMixin, UpdateView):
model = UserUserGroup
form_class = UserUserGroupForm
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
context = super().get_context_data(**kwargs)
context["group_type"] = "user"
return context
class UserUserGroupDelete(RevisionMixin, CidManagerRequiredMixin, DeleteView):
model = UserUserGroup
template_name = "confirm_delete.html"
success_url = reverse_lazy("generic:user_group_view")
class UserGroupExamUpdate(CidManagerRequiredMixin, UpdateView):
model = UserUserGroup
form_class = UserGroupExamForm
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
context = super().get_context_data(**kwargs)
context["group_type"] = "user"
return context
class CidGroupExamUpdate(CidManagerRequiredMixin, UpdateView):
model = CidUserGroup
form_class = CidGroupExamForm
def get_context_data(self, **kwargs: Any) -> dict[str, Any]:
context = super().get_context_data(**kwargs)
context["group_type"] = "cid"
return context
class ExamCreateBase(RevisionMixin, LoginRequiredMixin, CreateView):
template_name = "exam_create_form.html"
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
return super().form_valid(form)
def get_context_data(self, **kwargs):
# Call the base implementation first to get a context
context = super().get_context_data(**kwargs)
print(dir(context["view"]))
print(context["view"].template_name_suffix)
print(context["view"].model.app_name)
return context
def get_form_kwargs(self):
kwargs = super(ExamCreateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
class ExamUpdateBase(RevisionMixin, CheckCanEditMixin, LoginRequiredMixin, UpdateView):
template_name = "exam_update_form.html"
def form_valid(self, form):
self.object = form.save(commit=False)
self.object.save()
form.instance.author.add(self.request.user.id)
return super().form_valid(form)
def get_form_kwargs(self):
kwargs = super(ExamUpdateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
class ExamDeleteBase(RevisionMixin, CheckCanEditMixin, DeleteView):
template_name = "exam_confirm_delete.html"
def get_success_url(self) -> str:
# return super().get_success_url()(self):
return reverse_lazy(f"{self.model.app_name}:index")
def exam_inactive(request, context):
return render(request, "exam_inactive.html", context)
# class UserUserCreate(CidManagerRequiredMixin, CreateView):
# model = User
# form_class = UserUserForm
# template_name: str = "generic/user_creation_form.html"
#
# success_url = reverse_lazy("accounts_list")
@user_is_cid_user_manager
def trainees(request, grade: None | str = None):
trainees = (
UserProfile.objects.filter(peninsula_trainee=True, user__is_active=True)
.order_by("grade__name", Lower("user__last_name"))
.prefetch_related("supervisor", "grade", "user")
)
if grade is not None:
trainees = trainees.filter(grade__name=grade)
context = {"trainees": trainees, "grade": grade}
return render(request, "generic/trainees.html", context)
@user_is_cid_user_manager
def trainees_bulk_update_from_spreadsheet(request):
"""Bulk update trainees based on pasted spreadsheet-like data.
The view works in two stages:
- Preview: parse the pasted text and show rows with any matched user / supervisor
- Apply: apply grade and supervisor updates for matched users
"""
import re
preview = []
report = None
if request.method == "POST":
action = request.POST.get("action", "preview")
data = request.POST.get("data", "")
lines = [l.strip() for l in data.splitlines()]
current_grade = None
for idx, line in enumerate(lines):
if not line:
continue
# detect grade headings like 'Year 1 - Group 1' -> ST1
m = re.search(r"Year\s*(\d+)", line, re.IGNORECASE)
if m:
num = m.group(1)
current_grade = f"ST{num}"
continue
# split columns by tab or two+ spaces
parts = re.split(r"\t| {2,}", line)
if len(parts) < 2:
# maybe single-space separated; try splitting on tab or single tab
parts = [p.strip() for p in line.split("\t") if p.strip()]
if not parts:
continue
name_part = parts[0].strip()
supervisor_part = parts[1].strip() if len(parts) > 1 else ""
# normalize name (remove parenthetical annotations)
norm_name = re.sub(r"\s*\([^)]*\)", "", name_part).strip()
# split into tokens; assume first token is first name, last token is last name
tokens = [t for t in re.split(r"\s+", norm_name) if t]
first = tokens[0] if tokens else ""
last = tokens[-1] if tokens else ""
matched_user = None
if first and last:
# Prefer an exact first+last match
qs = User.objects.filter(first_name__iexact=first, last_name__iexact=last)
if qs.exists():
matched_user = qs.first()
else:
# Fallback: only match by last name when the last-name query returns
# a single candidate AND the candidate's first name reasonably
# matches the provided first token. This avoids ambiguous matches
# (e.g. Maria vs Stephanie both 'Bailey'). We accept a match when
# the candidate first name startswith the first 3 chars of the
# provided first token, or contains it as a substring (case-ins).
candidates = User.objects.filter(last_name__iexact=last)
if candidates.count() == 1:
candidate = candidates.first()
cand_first = (candidate.first_name or "").strip()
if cand_first:
# take at most 3 chars for prefix comparison
prefix = first[:3].lower()
cand_first_low = cand_first.lower()
if cand_first_low.startswith(prefix) or first.lower() in cand_first_low:
matched_user = candidate
matched_supervisor = None
if supervisor_part:
sup_qs = Supervisor.objects.filter(name__icontains=supervisor_part)
if sup_qs.exists():
matched_supervisor = sup_qs.first()
preview.append(
{
"raw": line,
"row_index": idx,
"name": name_part,
"norm_name": norm_name,
"first": first,
"last": last,
"supervisor_text": supervisor_part,
"matched_user": matched_user,
"matched_supervisor": matched_supervisor,
"grade": current_grade,
}
)
if action == "apply":
updated = 0
skipped = 0
for row in preview:
# Check for a manual override from the submitted form
manual_key = f"manual_user_{row.get('row_index')}"
manual_val = request.POST.get(manual_key)
user = None
if manual_val:
try:
user = User.objects.get(pk=int(manual_val))
except Exception:
user = None
# Fall back to automatic match
if not user:
user = row.get("matched_user")
if not user:
skipped += 1
continue
profile = user.userprofile
# apply grade if present
grade_name = row.get("grade")
if grade_name:
grade_obj, _ = UserGrades.objects.get_or_create(name=grade_name)
profile.grade = grade_obj
# apply supervisor if matched
sup = row.get("matched_supervisor")
if sup:
profile.supervisor = sup
profile.save()
updated += 1
report = {"updated": updated, "skipped": skipped}
return render(
request,
"generic/trainees_bulk_update.html",
{"preview": preview, "report": report},
)
def create_trainee(request, context=None):
return create_user(request, context, trainee=True)
@user_is_cid_user_manager
def create_user(request, context=None, trainee: bool = False):
if trainee:
form_template = "generic/trainee_creation_form.html"
else:
form_template = "generic/user_creation_form.html"
# if this is a POST request we need to process the form data
if request.method == "POST":
# create a form instance and populate it with data from the request:
if trainee:
form = TraineeForm(request.POST)
else:
form = UserUserForm(request.POST)
# check whether it's valid:
if form.is_valid():
try:
user_dict = {
"username": request.POST["username"],
"first_name": request.POST["first_name"],
"last_name": request.POST["last_name"],
"email": request.POST["username"],
"password": secrets.token_hex(nbytes=16),
}
new_user = User.objects.create_user(**user_dict)
except Exception as error:
errors = f"Unable to create account: {error}"
return render(
request,
form_template,
{"form": form, "errors": errors},
)
try:
user_profile = UserProfile.objects.get(user=new_user)
user_profile.peninsula_trainee = trainee
if request.POST["grade"]:
grade = UserGrades.objects.get(pk=request.POST["grade"])
user_profile.grade = grade
if "supervisor" in request.POST and request.POST["supervisor"]:
supervisor = Supervisor.objects.get(pk=request.POST["supervisor"])
user_profile.supervisor = supervisor
user_profile.save()
except Exception as error:
errors = f"Unable to create account profile {error}"
return render(
request,
form_template,
{"form": form, "errors": errors},
)
return HttpResponseRedirect(reverse_lazy("accounts_list"))
return reverse_lazy("accounts_list")
# if a GET (or any other method) we'll create a blank form
else:
if trainee:
form = TraineeForm()
else:
form = UserUserForm()
return render(request, form_template, {"form": form})
class SupervisorDetail(CidManagerRequiredMixin, DetailView):
model = Supervisor
class SupervisorDelete(CidManagerRequiredMixin, DeleteView):
model = Supervisor
template_name: str = "confirm_delete.html"
success_url = reverse_lazy("generic:supervisor")
class SupervisorUpdate(RedirectMixin, CidManagerRequiredMixin, UpdateView):
model = Supervisor
form_class = SupervisorForm
# success_url = reverse_lazy("generic:supervisor_detail", kwargs={'pk': self.pk})
class SupervisorCreate(CidManagerRequiredMixin, CreateView):
model = Supervisor
form_class = SupervisorForm
# success_url = reverse_lazy("generic:supervisor_detail")
# class SupervisorList(CidManagerRequiredMixin, ListView):
# model = Supervisor
class SupervisorList(CidManagerRequiredMixin, SingleTableMixin, FilterView):
model = Supervisor
table_class = SupervisorTable
template_name = "generic/supervisor_list.html"
filterset_class = SupervisorFilter
class ExamCollectionList(ListView):
model = ExamCollection
def get_queryset(self):
"""Annotate exam counts per collection and prefetch authors to avoid N+1 queries in templates."""
qs = (
ExamCollection.objects.all()
.annotate(
anatomy_count=Count("anatomy_exams", filter=Q(anatomy_exams__archive=False)),
longs_count=Count("longs_exams", filter=Q(longs_exams__archive=False)),
physics_count=Count("physics_exams", filter=Q(physics_exams__archive=False)),
rapids_count=Count("rapids_exams", filter=Q(rapids_exams__archive=False)),
sbas_count=Count("sbas_exams", filter=Q(sbas_exams__archive=False)),
)
.annotate(
total_count=F("anatomy_count")
+F("longs_count")
+F("physics_count")
+F("rapids_count")
+F("sbas_count")
)
.prefetch_related("author")
.order_by("name")
)
return qs
class ExamCollectionDetail(DetailView, AuthorRequiredMixin):
model = ExamCollection
class ExamCollectionEdit(UpdateView, AuthorRequiredMixin):
model = ExamCollection
form_class = ExamCollectionForm
class ExamCollectionCreate(CreateView, AuthorRequiredMixin):
model = ExamCollection
form_class = ExamCollectionForm
class ExamCollectionClone(CreateView, AuthorRequiredMixin):
model = ExamCollection
template_name = "generic/examcollection_clone_form.html"
form_class = ExamCollectionCloneForm
def get_initial(self):
old_object = get_object_or_404(self.model, pk=self.kwargs["pk"])
self.initial_object = old_object
initial_data = model_to_dict(old_object, exclude=["id", "date"])
## We manually transfer the forign keys / m2m relationships
# questions = old_object.exam_questions.all().values_list("id", flat=True)
authors = old_object.author.all().values_list("id", flat=True)
## clear the associated groups
# initial_data["valid_user_users"] = []
# initial_data["valid_cid_users"] = []
# initial_data["cid_user_groups"] = []
# initial_data["user_user_groups"] = []
# initial_data["active"] = False
# initial_data["publish_results"] = False
# self.exam_questions = list(questions)
self.author = list(authors)
return initial_data
def form_valid(self, form):
object = form.save()
# Reapply these otherwise they get lost?
# object.exam_questions.set(self.exam_questions)
GROUP_TYPES = (
("Rapid Exams", "rapids_exams", RapidsExam),
("Short Exams", "shorts_exams", ShortsExam),
("Long Exams", "longs_exams", LongsExam),
("SBA Exams", "sbas_exams", SbasExam),
("Physic Exams", "physics_exams", PhysicsExam),
("Anatomy Exams", "anatomy_exams", AnatomyExam),
)
object.author.set(self.author)
object.save()
# Clone exams linked to the original collection into the new one.
# For each supported exam app, find exams related to the old collection
# and create shallow copies (questions/authors re-applied, but
# active/archive/publish flags cleared).
if hasattr(self, "initial_object") and self.initial_object:
for _label, rel_name, ExamModel in GROUP_TYPES:
try:
old_exams = list(getattr(self.initial_object, rel_name).all())
except Exception:
old_exams = []
for old_exam in old_exams:
# Build a dict of field values excluding the PK
data = model_to_dict(old_exam, exclude=["id"])
# Clear group membership and flags for cloned exams
data["valid_user_users"] = [] if "valid_user_users" in data else []
data["valid_cid_users"] = [] if "valid_cid_users" in data else []
data["cid_user_groups"] = [] if "cid_user_groups" in data else []
data["user_user_groups"] = [] if "user_user_groups" in data else []
data["active"] = False
# some models use `archive` or `archived` - try both safely
if "archive" in data:
data["archive"] = False
if "archived" in data:
data["archived"] = False
data["publish_results"] = False if "publish_results" in data else data.get("publish_results", False)
# Keep a copy of M2M fields to reapply after create
m2m_backup = {}
for m2m_field in ("exam_questions", "author", "markers", "valid_user_users", "valid_cid_users", "cid_user_groups", "user_user_groups"):
if m2m_field in data:
m2m_backup[m2m_field] = data.pop(m2m_field)
# If the user supplied a date on the new collection, apply it to
# cloned exams so the cloned exams have the requested date.
if getattr(object, "date", None):
data["date"] = object.date
# Set the new examcollection foreign key to the newly created collection
# model_to_dict returns the FK id; replace with instance for create()
data["examcollection"] = object
try:
new_exam = ExamModel.objects.create(**data)
except Exception:
# If creation failed for any model-specific field, skip cloning this exam
continue
# Reapply M2M relationships where possible
for field_name, ids in m2m_backup.items():
try:
getattr(new_exam, field_name).set(ids)
except Exception:
# If the field doesn't exist on this model, ignore
pass
try:
# Some exam models provide an ordering helper
if hasattr(new_exam, "order_questions"):
new_exam.order_questions()
except Exception:
pass
return HttpResponseRedirect(object.get_absolute_url())
class ExamCollectionDelete(DeleteView, AuthorRequiredMixin):
model = ExamCollection
template_name = "confirm_delete.html"
success_url = reverse_lazy("generic:examcollection_list")
class ExaminationView(SuperuserRequiredMixin, SingleTableMixin, FilterView):
model = Examination
table_class = ExaminationTable
template_name = "atlas/view.html"
filterset_class = ExaminationFilter
class ExaminationDelete(RevisionMixin, SuperuserRequiredMixin, DeleteView):
model = Examination
template_name = "confirm_delete.html"
success_url = reverse_lazy("generic:examination_view")
class ExaminationUpdate(UpdateView, SuperuserRequiredMixin):
model = Examination
form_class = ExaminationForm
class SeriesImagesZipViewBase(SuperuserRequiredMixin, View):
"""Download all images from an image series"""
http_method_names = ["get"]
series_object = None
def get_files(self):
series = self.series_object.objects.get(pk=self.kwargs["pk"])
return [i.image.file for i in series.get_images()]
def get_archive_name(self, request):
# series = Series.objects.get(pk=self.kwargs["pk"])
return f"Series {self.kwargs['pk']}.zip"
def get(self, request, *args, **kwargs):
temp_file = ContentFile(b"", name=self.get_archive_name(request))
with zipfile.ZipFile(
temp_file, mode="w", compression=zipfile.ZIP_DEFLATED
) as zip_file:
files = self.get_files()
for file_ in files:
path = os.path.split(file_.name)[-1]
zip_file.writestr(path, file_.read())
file_size = temp_file.tell()
temp_file.seek(0)
response = HttpResponse(temp_file, content_type="application/zip")
response["Content-Disposition"] = (
"attachment; filename=%s" % self.get_archive_name(request)
)
response["Content-Length"] = file_size
return response
class ExamGroupsUpdateBase(
RevisionMixin,
CheckCanEditMixin,
LoginRequiredMixin,
AuthorRequiredMixin,
UpdateView,
):
template_name = "generic/exam_groups_edit.html"
def get_success_url(self) -> str:
return reverse_lazy(
f"{self.object.get_app_name()}:exam_cids",
kwargs={"exam_id": self.object.pk},
)
def get_form_kwargs(self):
kwargs = super(ExamGroupsUpdateBase, self).get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
class UpdateQuestionMixin(RedirectMixin, RevisionMixin, UpdateView):
def get_form_kwargs(self):
kwargs = super().get_form_kwargs()
kwargs.update({"user": self.request.user})
return kwargs
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
context["question"] = context["object"]
# Do permission checks here
obj = context["object"]
if self.request.user in obj.get_author_objects():
return context
elif (
self.request.user.groups.filter(name="long_checker").exists()
or self.request.user.is_superuser
):
return context
review_group = f"{obj.get_app_name()}_reviewer"
logger.debug(f"Checking for group membership: {review_group}")
if self.request.user.groups.filter(name=review_group).exists():
return context
raise PermissionDenied() # or Http404
def supervisor_overview(request, pk):
supervisor = get_object_or_404(Supervisor, pk=pk)
if not request.user.is_superuser:
try:
if not request.user.supervisor == supervisor:
raise PermissionDenied()
except Supervisor.RelatedObjectDoesNotExist:
raise PermissionDenied()
trainees = User.objects.filter(userprofile__supervisor=supervisor)
return render(
request,
"generic/supervisor_overview.html",
{"supervisor": supervisor, "trainees": trainees},
)
def supervisor_trainee(request, pk, trainee_id):
supervisor = get_object_or_404(Supervisor, pk=pk)
if not request.user.is_superuser:
try:
if not request.user.supervisor == supervisor:
raise PermissionDenied()
except Supervisor.RelatedObjectDoesNotExist:
raise PermissionDenied()
# We do this to check that the supervisor can (should) see the trainee results
trainee = User.objects.get(userprofile__supervisor=supervisor, pk=trainee_id)
exams = get_user_exams(trainee, supervisor_view=True)
print(exams)
print(trainee)
return render(
request,
"generic/supervisor_trainee.html",
{"supervisor": supervisor, "trainee": trainee, "all_exams": exams},
)
def supervisor_request_account(request):
if request.method == "POST":
email = request.POST.get("email")
try:
validate_email(email)
except ValidationError:
return HttpResponse(f"Invalid email address ({email})")
try:
supervisor = Supervisor.objects.get(email=email)
except Supervisor.DoesNotExist:
return HttpResponse(f"Invalid email address ({email})")
if supervisor.user:
return HttpResponse(
format_html(
"User already exists. Reset your password <a href='{}'>here</a>",
reverse("password_reset"),
)
)
try:
user_account = User.objects.get(email=email)
except User.DoesNotExist:
try:
first, last = supervisor.name.split(" ", 1)
except ValueError:
first, last = supervisor.name, ""
user_account = User.objects.create_user(
email=email,
username=email,
password=secrets.token_hex(nbytes=16),
first_name=first,
last_name=last,
)
supervisor.user = user_account
supervisor.save()
return HttpResponse(
format_html(
"Account created, you now need to reset your password <a href='{}'>here</a>",
reverse("password_reset"),
)
)
else:
return render(request, "generic/supervisor_request_account.html", {})
def toggle_share_with_supervisor(request, pk):
if request.htmx:
cid_user_exam = get_object_or_404(CidUserExam, pk=pk)
print(request.user)
print(cid_user_exam)
print(cid_user_exam.user_user)
if request.user != cid_user_exam.user_user:
return HttpResponse("Incorrect permissions")
cid_user_exam.share_with_supervisor = not cid_user_exam.share_with_supervisor
cid_user_exam.save()
return HttpResponse(
f"Shared with supervisor: {cid_user_exam.share_with_supervisor}"
)
else:
raise PermissionDenied()
def exam_collection_add_author(request, collection_id):
if request.method != "POST":
form = UsersAutocompleteForm()
# Render the fixed form which properly includes form data in HTMX POSTs
return render(
request,
"generic/user_form_fixed.html",
{"form": form, "collection_id": collection_id},
)
return
collection = get_object_or_404(ExamCollection, pk=collection_id)
if not request.user in collection.author.all():
raise PermissionDenied
#user = get_object_or_404(User, pk=request.POST.get("user"))
users = User.objects.filter(pk__in=request.POST.getlist("users"))
collection.add_author_to_exams(users)
return HttpResponse("Author added to all exams")
def exam_collection_add_marker(request, collection_id, exam_type):
if request.method != "POST":
form = UsersAutocompleteForm()
return render(
request,
"generic/marker_form.html",
{"form": form, "collection_id": collection_id, "exam_type": exam_type},
)
return
collection = get_object_or_404(ExamCollection, pk=collection_id)
if not request.user in collection.author.all():
raise PermissionDenied
#user = get_object_or_404(User, pk=request.POST.get("user"))
users = User.objects.filter(pk__in=request.POST.getlist("users"))
collection.add_marker_to_exams(users, exam_types=(exam_type,))
return HttpResponse("Marker added to exams")
def cimar_case_refresh(request, uuid):
try:
CimarCase.objects.get(uuid=uuid).refresh_study()
except CimarCase.DoesNotExist:
CimarCase(uuid=uuid).save()
return HttpResponse("Case refreshed")
@login_required
def user_search(request):
"""HTMX endpoint to search users for the trainees bulk-update UI.
Expects GET params:
- q: query string
- row: optional row index (passed back to JS)
Returns a small HTML fragment listing up to 10 matching users. Each
result calls `setManualUser(row, id, text)` when clicked (see template
JS in `trainees_bulk_update.html`).
"""
q = (request.GET.get("q") or "").strip()
row = request.GET.get("row")
if not q:
return HttpResponse("")
users_qs = User.objects.filter(
Q(first_name__icontains=q)
| Q(last_name__icontains=q)
| Q(email__icontains=q)
| Q(username__icontains=q)
).order_by("last_name", "first_name")[:10]
results = [{"id": u.pk, "text": f"{u.get_full_name() or u.username} - {u.email or ''}"} for u in users_qs]
return render(request, "generic/partials/user_search_results.html", {"results": results, "row": row, "q": q})
@login_required
def supervisor_search(request):
"""HTMX endpoint to search supervisors for the trainees bulk-update UI.
Expects GET params:
- q: query string
- row: optional row index (passed back to JS)
Returns a small HTML fragment listing up to 10 matching supervisors. Each
result calls `setManualSupervisor(row, id, text)` when clicked (see
template JS in `trainees_bulk_update.html`).
"""
q = (request.GET.get("q") or "").strip()
row = request.GET.get("row")
if not q:
return HttpResponse("")
supervisors_qs = Supervisor.objects.filter(
Q(name__icontains=q) | Q(email__icontains=q)
).order_by("name")[:10]
results = [{"id": s.pk, "text": f"{s.name} - {s.email or ''}"} for s in supervisors_qs]
return render(request, "generic/partials/supervisor_search_results.html", {"results": results, "row": row, "q": q})
Reference in New Issue View Git Blame Copy Permalink