Files
penracourses/generic/views.py
T
Ross 79e2caa9df .
2022-05-31 23:46:33 +01:00

2109 lines
74 KiB
Python

from collections import defaultdict
import statistics
import threading
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.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, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from django.contrib.auth.mixins import LoginRequiredMixin, UserPassesTestMixin
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_filters.views import FilterView
from django_filters import FilterSet
from django_tables2.views import SingleTableMixin
from reversion.views import RevisionMixin
from atlas.models import CaseCollection
from generic.decorators import user_is_cid_user_manager
from generic.filters import CidUserFilter
from generic.tables import CidUserExamTable, CidUserTable
from .forms import (
CidUserForm,
ExaminationForm,
CidUserGroupForm,
UserUserGroupForm,
)
from .models import CidUser, CidUserGroup, Examination, QuestionNote, UserUserGroup
from rapids.models import Rapid as RapidQuestion
from rapids.models import Exam as RapidExam
from longs.models import Long as LongQuestion
from longs.models import Exam as LongExam
from anatomy.models import AnatomyQuestion as AnatomyQuestion
from anatomy.models import Exam as AnatomyExam
from sbas.models import Question as SbasQuestion
from sbas.models import Exam as SbasExam
from physics.models import Question as PhysicsQuestion
from physics.models import Exam as PhysicsExam
from django.db.models import Case, When
from django.conf import settings
from datetime import datetime
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 AuthorRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
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 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(
'<script>opener.closePopup(window, "%s", "%s", "#id_examination");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Examination"}
)
@csrf_exempt
def get_examination_id(request):
if request.is_ajax():
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.is_ajax() and request.method == "POST":
question_type = request.POST.get("type")
if question_type == "rapid":
Exam = RapidExam
Question = RapidQuestion
elif question_type == "long":
Exam = LongExam
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,
):
self.Exam = exam
self.Question = question
self.Answer = answer
self.CidUserAnswer = cid_user_answer
self.app_name = app
self.question_type = question_type
# 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 ExamFilter(FilterSet):
class Meta:
model = exam
fields = {
"name": ["contains"],
"exam_mode": ["exact"],
"active": ["exact"],
"archive": ["exact"],
"open_access": ["exact"],
#'release_date': ['isnull'],
}
# filter_overrides = {
# models.CharField: {
# 'filter_class': django_filters.CharFilter,
# 'extra': lambda f: {
# 'lookup_expr': 'icontains',
# },
# },
# models.BooleanField: {
# 'filter_class': django_filters.BooleanFilter,
# 'extra': lambda f: {
# 'widget': forms.CheckboxInput,
# },
# },
# }
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
super().__init__(data, *args, **kwargs)
self.ExamFilter = ExamFilter
def check_user_access(self, user: User, exam_id: int = 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
"""
print("****", user, exam_id)
if user.is_superuser:
return True
if exam_id is not None:
exam = 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 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__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_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)
if exam.open_access or 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 True
@method_decorator(login_required)
def exam_toggle_results_published(self, request, pk):
if request.is_ajax() and 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.publish_results = (
True if request.POST.get("publish_results") == "true" else False
)
exam.save()
data = {
"status": "success",
"publish_results": exam.publish_results,
"name": exam.name,
"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.is_ajax() and 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.active = True if request.POST.get("active") == "true" else False
exam.save()
data = {
"status": "success",
"active": exam.active,
"name": exam.name,
"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)
@method_decorator(login_required)
def exam_list(self, request, all=False):
if all:
exams = self.Exam.objects.all().order_by("name")
else:
exams = self.Exam.objects.filter(exam_mode=True, archive=False).order_by(
"name"
)
if not self.check_user_access(request.user):
raise PermissionDenied
marking = False
if self.app_name in ("rapids", "anatomy", "longs"):
marking = True
return render(
request,
"exam_list.html".format(self.app_name),
{
"exams": exams,
"app_name": self.app_name,
"view_all": all,
"marking": marking,
},
)
@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():
if not self.check_user_access(request.user, pk):
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(),
),
)
)
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",
)
)
# 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()
# .prefetch_related("images", "abnormality", "region", "examination")
)
else:
questions = exam.exam_questions.select_related().all()
question_number = len(questions)
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,
},
)
@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")
questions = exam.exam_questions.all()
cids = self.CidUserAnswer.objects.filter(question__in=questions, exam__id=pk)
# Force a score update
for c in cids:
c.get_answer_score(cached=False)
return render(
request,
f"{self.app_name}/base.html",
{
"simple_content": format_html(
"""Answer scores updated <br/>
<a href='{}'>Return</a>""",
reverse(f"{self.app_name}:exam_scores_cid", kwargs={"pk": exam.pk}),
)
},
)
def exam_cids(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():
if not request.user.is_superuser:
if not self.check_user_access(request.user, exam_id):
raise PermissionDenied
cid_users = exam.valid_cid_users.all()
cid_user_count = cid_users.count()
user_users = exam.valid_user_users.all()
user_user_count = user_users.count()
context = {
"exam": exam,
"cid_users": cid_users,
"cid_user_count": cid_user_count,
"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_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():
if not self.check_user_access(request.user, exam_id):
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_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():
if not self.check_user_access(request.user, exam_id):
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
@method_decorator(login_required)
def exam_json_edit(self, request, pk):
if request.is_ajax() and request.method == "POST":
if not self.check_user_edit_access(request.user, exam_id=pk):
data = {"status": "invalid permisions"}
return JsonResponse(data, status=400)
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.long_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["atlas"] = 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)
exam.exam_questions.set(objects)
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 = 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_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.CidUserAnswer.objects.filter(score=self.Answer.MarkOptions.UNMARKED, exam__id=pk),
# to_attr="prefetched_unmarked_cid_answers"
# #queryset=self.Answer.objects.filter(),
# ),
)
)
else:
questions = exam.exam_questions.all()
# 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()
else:
exams = self.Exam.objects.filter(author__id=request.user.id)
filter = self.ExamFilter(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_detail(self, request, pk, sk):
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
question = exam.exam_questions.all()[sk]
exam_length = len(exam.exam_questions.all())
pos = exam.get_question_index(question) + 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
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,
},
)
def active_exams(self, request, json=True, based=True, cid=None, passcode=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)
for exam in exams:
if exam.active or self.check_user_access(request.user, exam.pk):
print(exam.name, cid, passcode)
if exam.exam_mode and not exam.check_cid_user(
cid, passcode, user_id=request.user.pk
):
print(exam.name, "fail")
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())
# 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 == False:
return active_exams["exams"]
return JsonResponse(active_exams)
@method_decorator(csrf_exempt)
def postExamAnswers(self, request):
if request.is_ajax and request.method == "POST":
n = 0
for answer in json.loads(request.POST.get("answers")):
eid = int(answer["eid"].split("/")[1])
uid = False
try:
cid = int(answer["cid"])
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 = get_object_or_404(self.Exam, pk=eid)
if not exam.check_cid_user(cid, answer["passcode"], user_id=uid):
return JsonResponse(
{"success": False, "error": "invalid cid / passcode"}
)
if uid:
existing_answers = self.CidUserAnswer.objects.filter(
question__id=answer["qid"], exam__id=eid, user__id=uid
)
else:
existing_answers = self.CidUserAnswer.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.CidUserAnswer(
answer=posted_answer,
normal=normal,
user=User.objects.get(id=uid),
)
else:
ans = self.CidUserAnswer(
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
ans.answer = ""
elif answer["qidn"] == "2" and not ans.normal:
ans.answer = posted_answer
case "anatomy":
if not existing_answers:
if uid:
ans = self.CidUserAnswer(
answer=posted_answer, uid=User.objects.get(id=uid)
)
else:
ans = self.CidUserAnswer(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.CidUserAnswer(user=User.objects.get(id=uid))
else:
ans = self.CidUserAnswer(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")))
)
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
)
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_unbased(self, request, pk):
"""
No (file based) caching is enabled for unbased exams
"""
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")
time = datetime.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
return JsonResponse(exam_json)
def exam_json(self, request, pk):
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")
# exam_json_cache = cache.get("{}_exam_json_{}".format(self.app_name, pk))
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)
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 = datetime.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_question_json(self, request, pk, sk):
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")
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(self, request, pk, sk):
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")
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)
def exam_scores_cid(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.exam_mode:
raise Http404("Packet not in exam mode")
# 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()
if self.app_name in ("physics", "sbas"):
questions = exam.exam_questions.all()
else:
questions = exam.exam_questions.all().prefetch_related("answers")
# We could prefect the CidUserAnswers.answers here (if we didn't cache them)
cid_user_answers = self.CidUserAnswer.objects.select_related("question").filter(
question__in=questions, exam__id=pk
)
cids = set()
# Loop through all candidates
for cid_user_answer in cid_user_answers:
# Convoluted (probably...)
if cid_user_answer.user is None:
cid = cid_user_answer.cid
# cid_passcodes[cid] = cid_user_answer.passcode
cids.add(cid)
else:
cid = f"u/{cid_user_answer.user.username}"
cids.add(cid)
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)
# user_answers_and_marks[cid].append((ans, answer_score))
if self.app_name in ("rapids", "anatomy", "sbas"):
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 = {}
for user in user_answers_marks:
if self.app_name in ("rapids", "anatomy", "sbas"):
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]
# 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 = len(questions) * 2
elif self.app_name == "physics":
max_score = len(questions) * 5
else:
max_score = len(questions)
if len(user_scores_list) < 1:
mean = 0
median = 0
mode = 0
fig_html = ""
else:
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
exam.user_scores = user_scores
exam.save()
return render(
request,
f"{self.app_name}/exam_scores_new.html",
{
"cids": sorted(cids),
# "cid_passcodes": cid_passcodes,
"exam": exam,
"unmarked": unmarked,
"questions": questions,
"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,
"max_score": max_score,
"mean": mean,
"median": median,
"mode": mode,
"plot": fig_html,
},
)
class GenericViewBase:
def __init__(self, app_name, QuestionObject, CidUserAnswerObject, ExamObject):
self.question_object = QuestionObject
self.cid_user_answer_object = CidUserAnswerObject
self.exam_object = ExamObject
self.app_name = app_name
g = {
"rapids": "rapid_checker",
"anatomy": "anatomy_checker",
"longs": "longs_checker",
"sbas": "sbas_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
"""
if request.is_ajax():
exam = 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.exam_questions.all()[int(request.POST["question_number"])]
question_json = question.get_question_json(based=False, feedback=True)
return JsonResponse(question_json)
return JsonResponse({"status": "error"})
@method_decorator(user_passes_test(lambda u: u.is_superuser))
def user_answer_delete_multiple(self, request):
if request.is_ajax():
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"})
return JsonResponse({"status": "error"})
@method_decorator(login_required)
def question_detail(self, request, pk):
question = 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},
)
@method_decorator(login_required)
def question_user_answers(self, request, pk):
question = get_object_or_404(self.question_object, pk=pk)
answers = self.cid_user_answer_object.objects.filter(question__id=question.pk)
return render(
request,
f"{self.app_name}/question_user_answers.html",
{"question": question, "answers": answers},
)
@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_initial(self):
old_object = get_object_or_404(self.model, pk=self.kwargs["exam_id"])
initial_data = model_to_dict(old_object, exclude=["id"])
questions = old_object.exam_questions.all().values_list("id", flat=True)
self.exam_questions = list(questions)
return initial_data
def form_valid(self, form):
object = form.save()
object.exam_questions.set(self.exam_questions)
object.save()
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 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 RapidExam.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 LongExam.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 RapidExam.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 LongExam.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_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_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 manage_cid_users(request):
if request.is_ajax() and 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[0]
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 = CidUser.objects.all().order_by("-cid")[0].cid + 1
except IndexError:
new_cid = 10000
if emails:
email_list = [i.strip() for i in emails.split()]
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(RevisionMixin, CidManagerRequiredMixin, CreateView):
model = CidUser
form_class = CidUserForm
class CidUserUpdate(RevisionMixin, CidManagerRequiredMixin, UpdateView):
model = CidUser
form_class = CidUserForm
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
class UserUserGroupCreate(RevisionMixin, CidManagerRequiredMixin, CreateView):
model = UserUserGroup
form_class = UserUserGroupForm
class UserUserGroupUpdate(RevisionMixin, CidManagerRequiredMixin, UpdateView):
model = UserUserGroup
form_class = UserUserGroupForm
class UserUserGroupDelete(RevisionMixin, CidManagerRequiredMixin, DeleteView):
model = UserUserGroup
template_name = "confirm_delete.html"
success_url = reverse_lazy("generic:user_group_view")
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)
class ExamUpdateBase(RevisionMixin, 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)