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 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 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 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.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 from .forms import ( CidGroupExamForm, CidUserForm, CidUserExamForm, ExamCollectionCloneForm, ExamCollectionForm, ExaminationForm, CidUserGroupForm, ExaminationMergeForm, SupervisorForm, TraineeForm, UserGroupExamForm, CidGroupExamForm, UserUserForm, UserUserGroupForm, ) from .models import ( CidUser, CidUserGroup, ExamBase, ExamCollection, ExamUserStatus, Examination, QuestionBase, QuestionNote, Supervisor, UserGrades, UserProfile, UserUserGroup, get_next_cid, ) from rapids.models import ( Rapid as RapidQuestion, ExamQuestionDetail as RapidsExamQuestionDetail, ) from rapids.models import Exam as RapidsExam from longs.models import ( Long as LongQuestion, LongSeries, ExamQuestionDetail as LongsExamQuestionDetail, ) from longs.models import Exam as LongsExam 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 sbas.models import Exam as SbasExam from physics.models import ( Question as PhysicsQuestion, ExamQuestionDetail as PhysicsExamQuestionDetail, ) from physics.models import Exam as PhysicsExam 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 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_exam_model_from_app_name(app_name: str) -> ExamBase: EXAM_MAP = { "physics": PhysicsExam, "anatomy": AnatomyExam, "rapids": RapidsExam, "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 == "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( '' % (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): if request.method == "POST": question_type = request.POST.get("type") if question_type == "rapids": Exam = RapidsExam Question = RapidQuestion 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")) # 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) 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")) ) 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 marker and user in exam.markers.all(): return True if exam.authors_only: return False if user.groups.filter(name="cid_user_manager").exists(): return True 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__in=("anatomy_checker", "anatomy_marker") ).exists() ): return False if ( self.app_name == "longs" and not user.groups.filter( name__in=("long_checker", "long_marker") ).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 == "atlas" and not user.groups.filter(name="casecollection_checker").exists() ): return False return True 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(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_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): raise PermissionDenied exam = get_object_or_404(self.Exam, pk=exam_id) exam.order_questions() return HttpResponse("Done") @method_decorator(login_required) def exam_list(self, request, all=False): 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") ) 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, }, ) @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
Return""", 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("Answers reset") 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["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["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 "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(), # ), ) ) 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) else: exams = self.Exam.objects.filter(author__id=request.user.id) 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, }, ) 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 "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 if exam.exam_mode and not exam.check_cid_user( cid, passcode, request.user, user_id ): raise Http404("No available exam") # 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) @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): exam: ExamBase = get_object_or_404(self.Exam, pk=pk) if not exam.exam_mode: raise Http404("Packet not in exam mode") if not exam.check_cid_user(cid, passcode, request.user): if not self.check_user_access(request.user, pk): # raise PermissionDenied raise Http404("Error accessing exam") if user is not None: request.user = 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 = [] view_all_results = False if request.user.groups.filter(name="view_all_results").exists(): view_all_results = True 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 == "longs": 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 answers.append(ans) answers_marks.append(answer_score) if self.app_name == "longs": 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: 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) match self.app_name: case "rapids" | "anatomy": max_score = len(questions) * 2 case "longs": max_score = len(questions) * 8 case "physics": max_score = len(questions) * 5 case _: max_score = len(questions) 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, } 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 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"): cached_scores = False questions = exam.get_questions() else: questions = exam.get_questions().prefetch_related("answers") # We could prefect the UserAnswers.answers here (if we didn't cache them) cid_user_answers = self.UserAnswer.objects.select_related("question").filter( question__in=questions, exam__id=pk ) question_number = questions.count() cids = set() plain_cids = set() user_ids = set() cids_user_id_map = {} cids_name_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": index = exam.get_question_index(q) 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"): 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 ("rapids", "anatomy", "sbas", "longs"): user_scores[user] = sum( [i for i in user_answers_marks[user] if i != "unmarked"] ) else: user_scores[user] = sum( [sum(i) for i in user_answers_marks[user] if i != "unmarked"] ) 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] if self.app_name in ("rapids", "anatomy"): max_score = question_number * 2 elif self.app_name == "physics": max_score = question_number * 5 else: max_score = question_number 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", } self.checker_group = g[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) @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_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), }, ) # TODO: improve permissions on these @method_decorator(login_required) def question_user_answers_by_compare(self, request, pk, answer_compare): print(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)] 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["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)] 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["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 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 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"""""" for grade in user_grades ] ) # TODO: work out how to actually use hyperscript html = ( html + f"""""" ) html = ( html + "" ) 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"""""" for group in user_groups ] ) # TODO: work out how to actually use hyperscript html = ( html + "" ) case "bulk-remove-group-button": user_groups = UserUserGroup.objects.all() html = "".join( [ f"""""" for group in user_groups ] ) # TODO: work out how to actually use hyperscript html = ( html + "" ) 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 @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")) 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.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 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.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 = "
".join(invalid_emails) return JsonResponse( {"status": "fail", "details": f"Unable to create users
{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 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 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) 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 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), ("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() 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.groups.filter(name="long_checker").exists() or self.request.user.is_superuser ): return context if self.request.user in obj.get_author_objects(): return context raise PermissionDenied() # or Http404