Ross
1791 lines
62 KiB
Python
1791 lines
62 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.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
|
|
from django_filters.views import FilterView
|
|
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,
|
|
)
|
|
|
|
from .models import CidUser, CidUserGroup, Examination, QuestionNote
|
|
|
|
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 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,
|
|
loadJsonAnswer,
|
|
):
|
|
self.Exam = exam
|
|
self.Question = question
|
|
self.Answer = answer
|
|
self.CidUserAnswer = cid_user_answer
|
|
self.app_name = app
|
|
self.question_type = question_type
|
|
self.loadJsonAnswer = loadJsonAnswer
|
|
|
|
# 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]
|
|
|
|
def check_user_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
|
|
"""
|
|
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_users.all()
|
|
|
|
cid_user_count = cid_users.count()
|
|
|
|
context = {
|
|
"exam": exam,
|
|
"cid_users": cid_users,
|
|
"cid_user_count": cid_user_count,
|
|
"app_name": self.app_name,
|
|
}
|
|
|
|
if self.app_name == "atlas":
|
|
context["collection"] = exam
|
|
|
|
return render(
|
|
request,
|
|
"exam_cids.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_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["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 "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):
|
|
exams = self.Exam.objects.filter(author__id=request.user.id)
|
|
|
|
if (
|
|
self.app_name == "rapids"
|
|
and request.user.groups.filter(name="rapid_checker").exists()
|
|
):
|
|
exams = self.Exam.objects.all()
|
|
if (
|
|
self.app_name == "anatomy"
|
|
and request.user.groups.filter(name="anatomy_checker").exists()
|
|
):
|
|
exams = self.Exam.objects.all()
|
|
if (
|
|
self.app_name == "longs"
|
|
and request.user.groups.filter(name="long_checker").exists()
|
|
):
|
|
exams = self.Exam.objects.all()
|
|
|
|
return render(request, "{}/index.html".format(self.app_name), {"exams": exams})
|
|
|
|
@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
|
|
|
|
return render(
|
|
request,
|
|
"{}/question_detail.html".format(self.app_name),
|
|
{
|
|
"question": question,
|
|
"exam": exam,
|
|
"next": next,
|
|
"previous": previous,
|
|
"exam_length": exam_length,
|
|
"pos": pos,
|
|
},
|
|
)
|
|
|
|
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):
|
|
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")):
|
|
ret_status, ret = self.loadJsonAnswer(answer)
|
|
# try:
|
|
# ret_status, ret = self.loadJsonAnswer(answer)
|
|
# except:
|
|
# # We catch the error here so users get an error (rather than 500)
|
|
# return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
# if ret is not True:
|
|
# return ret
|
|
|
|
if ret_status:
|
|
n = n + 1
|
|
else:
|
|
return ret
|
|
|
|
# print(UserAnswer.objects.filter(exam__id=q["eid"]))
|
|
# print(request.urlencode())
|
|
|
|
exam_type, exam_id = request.POST.get("eid").split("/")
|
|
|
|
c = CidUser.objects.filter(cid=request.POST.get("cid")).first()
|
|
exam = self.Exam.objects.get(pk=exam_id)
|
|
t = timezone.make_aware(
|
|
datetime.fromtimestamp(float(request.POST.get("start_time")))
|
|
)
|
|
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")
|
|
|
|
exam_json = exam.get_exam_json(based=False)
|
|
# 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()
|
|
|
|
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...)
|
|
cid = cid_user_answer.cid
|
|
# cid_passcodes[cid] = cid_user_answer.passcode
|
|
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
|
|
|
|
|
|
@login_required
|
|
@user_is_cid_user_manager
|
|
def group_view(request):
|
|
groups = CidUserGroup.objects.filter(archive=False)
|
|
|
|
return render(
|
|
request,
|
|
"generic/group_view.html",
|
|
{"groups": groups},
|
|
)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def group_view(request):
|
|
groups = CidUserGroup.objects.filter(archive=False)
|
|
|
|
return render(
|
|
request,
|
|
"generic/group_view.html",
|
|
{"groups": groups, "view_all": False},
|
|
)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def group_view_all(request):
|
|
groups = CidUserGroup.objects.filter()
|
|
|
|
return render(
|
|
request,
|
|
"generic/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},
|
|
)
|
|
|
|
# sent = []
|
|
# not_sent = []
|
|
# for u in users:
|
|
# success, msg = u.email_details(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()},
|
|
# )
|
|
|
|
|
|
@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 CidUserGroupCreate(RevisionMixin, CidManagerRequiredMixin, CreateView):
|
|
model = CidUserGroup
|
|
form_class = CidUserGroupForm
|
|
|
|
|
|
class CidUserGroupUpdate(RevisionMixin, CidManagerRequiredMixin, UpdateView):
|
|
model = CidUserGroup
|
|
form_class = CidUserGroupForm
|