Ross
1234 lines
40 KiB
Python
1234 lines
40 KiB
Python
import secrets
|
|
from typing import Any, Optional
|
|
from django.conf import settings
|
|
|
|
from django_tables2 import SingleTableMixin
|
|
from atlas.models import CaseCollection, CidReportAnswer, Case as AtlasCase
|
|
from generic.decorators import user_is_cid_user_manager
|
|
from generic.filters import UserUserFilter
|
|
from generic.tables import UserUserTable
|
|
from generic.views import (
|
|
CidManagerRequiredMixin,
|
|
RedirectMixin,
|
|
get_question_and_content_type,
|
|
get_user_exams,
|
|
get_exam_model_from_app_name,
|
|
)
|
|
from django.core.exceptions import PermissionDenied, ObjectDoesNotExist
|
|
from django.shortcuts import render, get_object_or_404, redirect
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django import forms
|
|
|
|
from django.core.exceptions import ValidationError
|
|
from django.core.validators import validate_email
|
|
|
|
from django.utils.html import format_html
|
|
|
|
# from django.contrib.auth.models import User
|
|
from django.contrib.auth.decorators import login_required, user_passes_test
|
|
from django.contrib.auth.models import User
|
|
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from django.contrib.auth.mixins import LoginRequiredMixin
|
|
|
|
from django.views.generic.edit import CreateView, UpdateView, DeleteView
|
|
from django.views.generic import ListView, TemplateView
|
|
|
|
from django.db.models.functions import Lower
|
|
|
|
from django.core.cache import cache
|
|
|
|
from django.urls import reverse_lazy, reverse
|
|
|
|
from django.http import Http404, JsonResponse
|
|
|
|
from django.http import HttpResponseRedirect, HttpResponse
|
|
|
|
from physics.models import UserAnswer as PhysicsUserAnswer
|
|
from physics.models import Exam as PhysicsExam
|
|
|
|
from anatomy.models import UserAnswer as AnatomyUserAnswer
|
|
from anatomy.models import Exam as AnatomyExam
|
|
from anatomy.models import AnatomyQuestion
|
|
from anatomy.models import Answer as AnatomyAnswer
|
|
from rad.filters import UserListFilter
|
|
|
|
from helpers.cimar import CimarAPI, InvalidCredentials, NoSession
|
|
from rapids.models import UserAnswer as RapidsUserAnswer
|
|
from rapids.models import Exam as RapidsExam
|
|
from rapids.models import Rapid
|
|
from rapids.models import Answer as RapidAnswer
|
|
|
|
from longs.models import UserAnswer as LongsUserAnswer
|
|
from longs.models import Exam as LongsExam
|
|
from longs.models import Long
|
|
|
|
from sbas.models import UserAnswer as SbasUserAnswer
|
|
from sbas.models import Exam as SbasExam
|
|
|
|
from anatomy.views import GenericExamViews as AnatomyExamViews
|
|
from rapids.views import GenericExamViews as RapidsExamsViews
|
|
from shorts.views import GenericExamViews as ShortsExamsViews
|
|
from longs.views import GenericExamViews as LongsExamsViews
|
|
|
|
from generic.forms import QuestionNoteForm
|
|
from generic.models import (
|
|
CidUser,
|
|
CimarCase,
|
|
ExamCollection,
|
|
QuestionNote,
|
|
Supervisor,
|
|
UserGrades,
|
|
UserProfile,
|
|
USER_EXAM_TYPES,
|
|
)
|
|
from django.contrib.auth.models import Group
|
|
from generic.models import UserUserGroup
|
|
|
|
from django_filters.views import FilterView
|
|
|
|
import json
|
|
|
|
from django.template import RequestContext
|
|
|
|
from django.db.models import Q
|
|
|
|
from dal import autocomplete
|
|
|
|
from loguru import logger
|
|
|
|
import psutil
|
|
|
|
|
|
|
|
|
|
def feedback_checker(user):
|
|
return user.groups.filter(name="feedback_checker").exists()
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def server(request):
|
|
return render(request, "server.html")
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def people(request):
|
|
return render(request, "people.html", {})
|
|
|
|
|
|
@login_required
|
|
def profile(request):
|
|
user = request.user
|
|
available_groups = UserUserGroup.objects.filter(archive=False).exclude(
|
|
pk__in=user.user_groups.values_list("pk", flat=True)
|
|
)
|
|
available_auth_groups = Group.objects.exclude(
|
|
pk__in=user.groups.values_list("pk", flat=True)
|
|
)
|
|
is_cid_user_manager = request.user.groups.filter(name="cid_user_manager").exists()
|
|
return render(
|
|
request,
|
|
"profile.html",
|
|
{
|
|
"user": user,
|
|
"available_groups": available_groups,
|
|
"available_auth_groups": available_auth_groups,
|
|
"is_cid_user_manager": is_cid_user_manager,
|
|
},
|
|
)
|
|
|
|
|
|
def index(request):
|
|
collections = []
|
|
|
|
if request.user.is_authenticated:
|
|
collections = ExamCollection.objects.filter(archive=False, author=request.user)
|
|
|
|
rcr_assessor = request.user.groups.filter(
|
|
Q(name="rcr_radiology_assessor")
|
|
| Q(name="rcr_oncology_assessor")
|
|
| Q(name="rcr_assessor")
|
|
).exists()
|
|
return render(
|
|
request,
|
|
"index.html",
|
|
{"rcr_assessor": rcr_assessor, "collections": collections},
|
|
)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def account_profile(request, slug):
|
|
user = get_object_or_404(User, username=slug)
|
|
available_groups = UserUserGroup.objects.filter(archive=False).exclude(
|
|
pk__in=user.user_groups.values_list("pk", flat=True)
|
|
)
|
|
available_auth_groups = Group.objects.exclude(
|
|
pk__in=user.groups.values_list("pk", flat=True)
|
|
)
|
|
is_cid_user_manager = request.user.groups.filter(name="cid_user_manager").exists()
|
|
return render(
|
|
request,
|
|
"profile.html",
|
|
{
|
|
"user": user,
|
|
"available_groups": available_groups,
|
|
"available_auth_groups": available_auth_groups,
|
|
"is_cid_user_manager": is_cid_user_manager,
|
|
},
|
|
)
|
|
|
|
|
|
def cid_selector(request):
|
|
return render(
|
|
request,
|
|
"cid_selector.html",
|
|
)
|
|
|
|
|
|
@login_required
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def cid_scores_admin(request, cid):
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
|
|
if not cid_user:
|
|
raise Http404("CID not found")
|
|
|
|
return cid_user_overview(request, cid_user.cid, cid_user.passcode)
|
|
|
|
|
|
@login_required
|
|
# @user_passes_test(lambda u: u.is_superuser)
|
|
@user_is_cid_user_manager
|
|
def cid_results(request, cid):
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
|
|
if not cid_user:
|
|
raise Http404("CID not found")
|
|
|
|
report = cid_user.generate_exam_report()
|
|
|
|
return render(
|
|
request,
|
|
"cid_results.html",
|
|
{
|
|
"report": report,
|
|
},
|
|
)
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def user_scores_admin(request, user_id):
|
|
try:
|
|
user = User.objects.get(id=user_id)
|
|
except User.DoesNotExist:
|
|
raise Http404("Cid not found")
|
|
|
|
return user_scores(request, user=user)
|
|
|
|
|
|
@login_required
|
|
def user_scores(request, user=None):
|
|
admin = True
|
|
if user is None:
|
|
user = request.user
|
|
admin = False
|
|
|
|
exams = get_user_exams(user)
|
|
|
|
# case_collections = request.user.casecollection_exams.all()
|
|
|
|
available_exams = []
|
|
|
|
for n, t in USER_EXAM_TYPES:
|
|
exam_rel = getattr(user, t)
|
|
if exam_rel.exists():
|
|
temp_exams = exam_rel.filter(exam_mode=True, archive=False).order_by("name")
|
|
available_exams.append((n, temp_exams))
|
|
|
|
# available_exams = request.user.get_cid_exams()
|
|
|
|
return render(
|
|
request,
|
|
"user_scores.html",
|
|
{
|
|
# "physics_exams": physics_exams,
|
|
# "anatomy_exams": anatomy_exams,
|
|
# "rapid_exams": rapid_exams,
|
|
# "longs_exams": longs_exams,
|
|
# "sba_exams": sba_exams,
|
|
"all_exams": exams,
|
|
# "passcode": passcode,
|
|
"cid_user": user,
|
|
"available_exams": available_exams,
|
|
"case_collections": [], # case_collections,
|
|
"admin": admin,
|
|
},
|
|
)
|
|
|
|
|
|
@login_required
|
|
def user_marking(request):
|
|
"""Main page for user marking overview. Uses HTMX to fetch per-app partials."""
|
|
apps = ["physics", "anatomy", "rapids", "shorts", "longs", "sbas"]
|
|
return render(request, "rad/user_marking.html", {"apps": apps})
|
|
|
|
|
|
@login_required
|
|
def user_marking_partial(request, exam_type: str):
|
|
"""Return an HTMX partial listing exams the user can mark for a given exam type.
|
|
|
|
Each exam is checked for whether it has any unmarked answers (using
|
|
question.get_unmarked_user_answer_count when available).
|
|
"""
|
|
user = request.user
|
|
|
|
try:
|
|
ExamModel = get_exam_model_from_app_name(exam_type)
|
|
except Exception:
|
|
return HttpResponse("Unknown exam type", status=400)
|
|
|
|
# Exams where user is an author or an authorised marker
|
|
exams_qs = (
|
|
ExamModel.objects.filter(archive=False, exam_mode=True)
|
|
.filter(Q(author=user) | Q(markers=user))
|
|
.distinct()
|
|
.order_by("name")
|
|
.prefetch_related("author", "markers")
|
|
)
|
|
|
|
exams_data = []
|
|
for exam in exams_qs:
|
|
has_unmarked = False
|
|
unmarked_count = 0
|
|
# iterate questions and use their helper to count unmarked answers when available
|
|
for q in exam.get_questions():
|
|
count = 0
|
|
# Prefer question-provided helper if present
|
|
helper = getattr(q, "get_unmarked_user_answer_count", None)
|
|
if callable(helper):
|
|
try:
|
|
count = helper(exam_pk=exam.pk, marker=user)
|
|
except TypeError:
|
|
count = helper(exam_pk=exam.pk)
|
|
except Exception:
|
|
count = 0
|
|
else:
|
|
# Fallback: inspect related answer manager if present
|
|
rel_manager = None
|
|
for rel_name in ("cid_user_answers", "user_answers", "answers", "useranswer_set"):
|
|
if hasattr(q, rel_name):
|
|
rel_manager = getattr(q, rel_name)
|
|
break
|
|
|
|
if rel_manager is not None:
|
|
try:
|
|
ans_model = rel_manager.model
|
|
# If the answer model has a `score` field, count answers that are unmarked
|
|
field_names = [f.name for f in ans_model._meta.get_fields()]
|
|
if "score" in field_names:
|
|
# Try to detect an UNMARKED constant on the model
|
|
unmarked_vals = [None, ""]
|
|
model_score_const = getattr(ans_model, "ScoreOptions", None) or getattr(ans_model, "MarkOptions", None)
|
|
if model_score_const is not None:
|
|
# attempt to use any UNMARKED attribute
|
|
unmarked_val = getattr(model_score_const, "UNMARKED", None)
|
|
if unmarked_val is not None:
|
|
unmarked_vals.append(unmarked_val)
|
|
|
|
count = rel_manager.filter(exam__id=exam.pk).filter(score__in=unmarked_vals).count()
|
|
else:
|
|
# No score field — likely auto-marked; treat as 0
|
|
count = 0
|
|
except Exception:
|
|
count = 0
|
|
|
|
try:
|
|
c = int(count)
|
|
except Exception:
|
|
c = 0
|
|
if c:
|
|
has_unmarked = True
|
|
unmarked_count += c
|
|
|
|
exams_data.append({"exam": exam, "has_unmarked": has_unmarked, "unmarked_count": unmarked_count})
|
|
|
|
return render(request, "generic/partials/user_marking_exam_list.html", {"exams_data": exams_data, "exam_type": exam_type})
|
|
|
|
|
|
def cid_user_overview(request, cid, passcode):
|
|
# exam = get_object_or_404(Exam, pk=pk)
|
|
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
if not cid_user or cid_user.passcode != passcode:
|
|
raise Http404("CID / Passcode combination not found")
|
|
|
|
EXAM_ANSWER_MAP = {
|
|
"physics": (PhysicsUserAnswer, PhysicsExam),
|
|
"anatomy": (AnatomyUserAnswer, AnatomyExam),
|
|
"rapids": (RapidsUserAnswer, RapidsExam),
|
|
"longs": (LongsUserAnswer, LongsExam),
|
|
"sbas": (SbasUserAnswer, SbasExam),
|
|
}
|
|
kwargs = {"exam_mode": True, "archive": False}
|
|
|
|
exams = []
|
|
for exam_type in EXAM_ANSWER_MAP:
|
|
UserAnswer, Exam = EXAM_ANSWER_MAP[exam_type]
|
|
exam_answers = UserAnswer.objects.filter(cid=cid)
|
|
exam_ids = exam_answers.values_list("exam").distinct()
|
|
exams_to_add = Exam.objects.filter(id__in=exam_ids, **kwargs)
|
|
if exams_to_add:
|
|
exams.append((exam_type, exams_to_add))
|
|
|
|
case_collections = cid_user.casecollection_exams.all()
|
|
|
|
available_exams = cid_user.get_cid_exams(include_case_collections=False)
|
|
|
|
return render(
|
|
request,
|
|
"cid_user_overview.html",
|
|
{
|
|
# "physics_exams": physics_exams,
|
|
# "anatomy_exams": anatomy_exams,
|
|
# "rapid_exams": rapid_exams,
|
|
# "longs_exams": longs_exams,
|
|
# "sba_exams": sba_exams,
|
|
"all_exams": exams,
|
|
"cid": cid,
|
|
"passcode": passcode,
|
|
"cid_user": cid_user,
|
|
"available_exams": available_exams,
|
|
"case_collections": case_collections,
|
|
},
|
|
)
|
|
|
|
|
|
def active_exams(request, cid=None, passcode=None, based=True):
|
|
active_exams = {}
|
|
|
|
if cid is not None:
|
|
active_exams["cid"] = cid
|
|
cid_user = CidUser.objects.filter(cid=cid).first()
|
|
if not cid_user or cid_user.passcode != passcode:
|
|
return JsonResponse({"status": "invalid"}, status=401)
|
|
|
|
anatomy_exams = AnatomyExamViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
rapid_exams = RapidsExamsViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
short_exams = ShortsExamsViews.active_exams(
|
|
request, False, cid=cid, passcode=passcode
|
|
)
|
|
long_exams = LongsExamsViews.active_exams(request, False, cid=cid, passcode=passcode)
|
|
|
|
exams = []
|
|
exams.extend(anatomy_exams)
|
|
exams.extend(rapid_exams)
|
|
exams.extend(short_exams)
|
|
exams.extend(long_exams)
|
|
|
|
active_exams["exams"] = exams
|
|
|
|
if request.user:
|
|
active_exams["user"] = request.user.username
|
|
active_exams["user_id"] = request.user.pk
|
|
else:
|
|
active_exams["user"] = False
|
|
|
|
return JsonResponse(active_exams)
|
|
|
|
|
|
def active_exams_unbased(request, cid=None, passcode=None):
|
|
return active_exams(request, cid, passcode, based=False)
|
|
|
|
|
|
@csrf_exempt
|
|
def exam_submit(request):
|
|
if request.method == "POST":
|
|
exam_type, exam_id = request.POST.get("eid").split("/")
|
|
|
|
try:
|
|
if exam_type == "anatomy":
|
|
return AnatomyExamViews.post_exam_answers(request)
|
|
elif exam_type == "rapid":
|
|
return RapidsExamsViews.post_exam_answers(request)
|
|
elif exam_type == "long":
|
|
return LongsExamsViews.post_exam_answers(request)
|
|
elif exam_type == "short":
|
|
return ShortsExamsViews.post_exam_answers(request)
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid exam type"})
|
|
except ObjectDoesNotExist:
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
# post_exam_answers
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
|
|
def feedback_create(request, question_type, pk):
|
|
if question_type == "anatomy":
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
question = Rapid
|
|
elif question_type == "long":
|
|
question = Long
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid exam type"})
|
|
|
|
|
|
class AddQuestionNote(CreateView):
|
|
model = QuestionNote
|
|
form_class = QuestionNoteForm
|
|
|
|
# fields = ("author", "note_type", "note")
|
|
|
|
def get_form_kwargs(self):
|
|
kwargs = super(AddQuestionNote, self).get_form_kwargs()
|
|
# update the kwargs for the form init method with yours
|
|
kwargs.update(self.kwargs) # self.kwargs contains all url conf params
|
|
kwargs.update({"user": self.request.user})
|
|
return kwargs
|
|
|
|
def get_initial(self):
|
|
pk = self.kwargs["pk"]
|
|
self.question_type = self.kwargs["question_type"]
|
|
|
|
question, content_type = get_question_and_content_type(self.question_type)
|
|
|
|
self.question = get_object_or_404(question, pk=pk)
|
|
return {"content_type": content_type, "object_id": pk}
|
|
|
|
def get_context_data(self, **kwargs):
|
|
# This can also be accessed via view.**** in the template
|
|
context = super(AddQuestionNote, self).get_context_data(**kwargs)
|
|
context["question"] = self.question
|
|
context["question_type"] = self.question_type
|
|
return context
|
|
|
|
def form_valid(self, form):
|
|
model = form.save(commit=False)
|
|
|
|
if self.request.user.is_authenticated:
|
|
model.user_author = self.request.user
|
|
|
|
model.save()
|
|
|
|
response = super().form_valid(form)
|
|
|
|
return render(self.request, "feedback_submitted.html")
|
|
return response
|
|
|
|
def get_success_url(self) -> str:
|
|
return super().get_success_url()
|
|
|
|
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
def feedback_note_delete(request, pk):
|
|
q = get_object_or_404(QuestionNote, pk=pk)
|
|
q.delete()
|
|
|
|
return redirect(q.get_object_url())
|
|
|
|
|
|
@login_required
|
|
def feedback_mark_complete(request, pk):
|
|
q = get_object_or_404(QuestionNote, pk=pk)
|
|
|
|
if (
|
|
request.user not in q.question.author.all()
|
|
and not request.user.groups.filter(name="feedback_checker").exists()
|
|
):
|
|
raise PermissionDenied()
|
|
|
|
q.complete = True
|
|
q.save()
|
|
return redirect(q.get_object_url())
|
|
|
|
|
|
@csrf_exempt
|
|
def answer_suggestion_submit(request):
|
|
if request.method == "POST":
|
|
post_data = json.loads(request.body)
|
|
question_type, qid = post_data["qid"].split("/")
|
|
answer_string = post_data["answer"]
|
|
status = post_data["status"]
|
|
|
|
if str(status) not in " 012":
|
|
return JsonResponse({"success": False, "error": "Invalid status"})
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
q = get_object_or_404(question, pk=qid)
|
|
if AnswerModel.objects.filter(answer=answer_string, question=q):
|
|
return JsonResponse({"success": False, "error": "Answer already exists"})
|
|
|
|
a = AnswerModel(question=q, answer=answer_string, status=status, proposed=True)
|
|
a.save()
|
|
return JsonResponse({"success": True, "error": "Answer submited"})
|
|
|
|
# post_exam_answers
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
|
|
@login_required
|
|
def answer_submit(request):
|
|
if request.method == "POST":
|
|
qid = request.POST.get("qid")
|
|
question_type = request.POST.get("question_type")
|
|
answer_string = request.POST.get("answer")
|
|
status = request.POST.get("status")
|
|
|
|
if str(status) not in "012":
|
|
return JsonResponse({"success": False, "error": "Invalid status"})
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
q = get_object_or_404(question, pk=qid)
|
|
if AnswerModel.objects.filter(answer=answer_string, question=q):
|
|
return JsonResponse({"success": False, "error": "Answer already exists"})
|
|
|
|
a = AnswerModel(question=q, answer=answer_string, status=status)
|
|
a.save()
|
|
return JsonResponse({"success": True, "error": "Answer added"})
|
|
|
|
return JsonResponse({"success": False, "error": "Invalid data"})
|
|
|
|
|
|
@login_required
|
|
def answer_suggestion_confirm(request):
|
|
if request.method == "POST":
|
|
question_type = request.POST.get("question_type")
|
|
aid = request.POST.get("aid")
|
|
status = request.POST.get("status")
|
|
|
|
if question_type == "anatomy":
|
|
AnswerModel = AnatomyAnswer
|
|
question = AnatomyQuestion
|
|
elif question_type == "rapid":
|
|
AnswerModel = RapidAnswer
|
|
question = Rapid
|
|
else:
|
|
return JsonResponse({"success": False, "error": "Invalid question type"})
|
|
|
|
answer = get_object_or_404(AnswerModel, pk=aid)
|
|
answer.proposed = False
|
|
answer.status = status
|
|
answer.save()
|
|
return JsonResponse({"success": True, "error": "Answer changed"})
|
|
|
|
|
|
@login_required
|
|
@user_passes_test(feedback_checker)
|
|
def view_feedback(request):
|
|
# exam = get_object_or_404(Exam, pk=pk)
|
|
|
|
feedback_notes = QuestionNote.objects.filter(complete=False)
|
|
|
|
return render(
|
|
request,
|
|
"view_feedback.html",
|
|
{
|
|
"notes": feedback_notes,
|
|
},
|
|
)
|
|
|
|
|
|
def privacy_view(request):
|
|
return render(request, "privacy.html")
|
|
|
|
|
|
def cimar(request):
|
|
if "link_id" in request.GET:
|
|
link_id = request.GET["link_id"]
|
|
return render(request, "cimar.html", {"link_id": link_id})
|
|
return render(request, "cimar.html")
|
|
|
|
|
|
def about_view(request):
|
|
return render(request, "about.html")
|
|
|
|
|
|
def stats(request):
|
|
collection_count = CaseCollection.objects.filter(archive=False).count()
|
|
case_count = AtlasCase.objects.filter().count()
|
|
|
|
anatomy_exam_count = AnatomyExam.objects.filter(archive=False).count()
|
|
anatomy_question_count = AnatomyQuestion.objects.filter().count()
|
|
return render(
|
|
request,
|
|
"stats.html",
|
|
{
|
|
"collection_count": collection_count,
|
|
"case_count": case_count,
|
|
"anatomy_exam_count": anatomy_exam_count,
|
|
"anatomy_question_count": anatomy_question_count,
|
|
},
|
|
)
|
|
|
|
|
|
# HTTP Error 400
|
|
def page_not_found(request, exception):
|
|
response = render(request, "404.html")
|
|
|
|
response.status_code = 404
|
|
|
|
return response
|
|
|
|
|
|
def page_forbidden(request, exception):
|
|
response = render(request, "403.html", context={"exception": exception})
|
|
|
|
response.status_code = 403
|
|
|
|
return response
|
|
|
|
|
|
def server_error(request):
|
|
response = render(request, "500.html")
|
|
|
|
response.status_code = 500
|
|
|
|
return response
|
|
|
|
|
|
class UserListView(CidManagerRequiredMixin, FilterView):
|
|
model = User
|
|
template_name = "user_list_view.html"
|
|
|
|
filterset_class = UserListFilter
|
|
|
|
|
|
class UserListTableView(CidManagerRequiredMixin, SingleTableMixin, FilterView):
|
|
model = User
|
|
table_class = UserUserTable
|
|
template_name = "generic/user_view.html"
|
|
|
|
filterset_class = UserUserFilter
|
|
|
|
# table_pagination = {"per_page": 5}
|
|
|
|
def get_paginate_by(self, table_data) -> int | None:
|
|
return 100
|
|
# return super().get_paginate_by(table_data)
|
|
|
|
def get_context_data(self, **kwargs):
|
|
context = super().get_context_data(**kwargs)
|
|
# user = self.request.user
|
|
|
|
# filters = {"archive": False, "exam_mode": True}
|
|
|
|
# physics_exams = [(i.name, i.pk) for i in PhysicsExam.objects.filter(**filters)]
|
|
# rapid_exams = [(i.name, i.pk) for i in RapidsExam.objects.filter(**filters)]
|
|
# sba_exams = [(i.name, i.pk) for i in SbasExam.objects.filter(**filters)]
|
|
# longs_exams = [(i.name, i.pk) for i in LongsExam.objects.filter(**filters)]
|
|
# anatomy_exams = [(i.name, i.pk) for i in AnatomyExam.objects.filter(**filters)]
|
|
# casecollection_exams = [
|
|
# (i.name, i.pk)
|
|
# for i in CaseCollection.objects.filter(archive=False, collection_type__gt=0)
|
|
# ]
|
|
|
|
# context["physics_exams"] = physics_exams
|
|
# context["rapid_exams"] = rapid_exams
|
|
# context["sba_exams"] = sba_exams
|
|
# context["longs_exams"] = longs_exams
|
|
# context["anatomy_exams"] = anatomy_exams
|
|
# context["casecollection_exams"] = casecollection_exams
|
|
|
|
# cid_user_groups = [
|
|
# (i.name, i.pk) for i in CidUserGroup.objects.filter(archive=False)
|
|
# ]
|
|
|
|
# context["cid_user_groups"] = cid_user_groups
|
|
return context
|
|
|
|
|
|
class DeleteUserView(CidManagerRequiredMixin, DeleteView):
|
|
model = User
|
|
template_name: str = "confirm_delete.html"
|
|
success_url = reverse_lazy("accounts_list")
|
|
|
|
|
|
class UpdateUserView(CidManagerRequiredMixin, UpdateView):
|
|
model = User
|
|
fields = [
|
|
"first_name",
|
|
"last_name",
|
|
"email",
|
|
"is_active",
|
|
] # Keep listing whatever fields
|
|
# the combined UserProfile and User exposes.
|
|
template_name = "user_update.html"
|
|
slug_field = "username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def get_success_url(self):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
|
|
|
|
class UpdateUserProfileAdminView(CidManagerRequiredMixin, UpdateView):
|
|
model = UserProfile
|
|
fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
"peninsula_trainee",
|
|
] # Keep listing whatever fields
|
|
template_name = "user_update_profile.html"
|
|
slug_field = "user__username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def get_success_url(self):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
|
|
|
|
class UpdateUserProfileView(UpdateView):
|
|
model = UserProfile
|
|
fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
# "peninsula_trainee",
|
|
] # Keep listing whatever fields
|
|
template_name = "user_update_profile.html"
|
|
slug_field = "user__username"
|
|
slug_url_kwarg = "slug"
|
|
|
|
def dispatch(self, request, *args, **kwargs):
|
|
if (
|
|
request.user.is_superuser
|
|
or request.user.groups.filter(name="cid_user_manager").exists()
|
|
):
|
|
self.fields = [
|
|
"supervisor",
|
|
"grade",
|
|
"registration_number",
|
|
"site",
|
|
"peninsula_trainee",
|
|
]
|
|
# Check permissions for the request.user here
|
|
elif request.user != self.get_object().user:
|
|
raise Http404
|
|
elif request.user.userprofile.peninsula_trainee:
|
|
self.fields = ["supervisor", "grade", "registration_number", "site"]
|
|
|
|
else:
|
|
self.fields = ["grade", "registration_number", "site"]
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
def get_context_data(self, **kwargs: reverse_lazy) -> dict[str, Any]:
|
|
context = super().get_context_data(**kwargs)
|
|
|
|
form = context["form"]
|
|
|
|
if "supervisor" in form.fields:
|
|
# form.fields["supervisor"].queryset = Supervisor.objects.all()
|
|
form.fields["supervisor"] = forms.ModelChoiceField(
|
|
Supervisor.objects.all().order_by("name"),
|
|
required=False,
|
|
widget=autocomplete.ModelSelect2(url="generic:supervisor-autocomplete"),
|
|
)
|
|
return context
|
|
|
|
def get_success_url(self):
|
|
if "redirect" in self.request.GET:
|
|
return self.request.GET["redirect"]
|
|
|
|
if (
|
|
self.request.user.is_superuser
|
|
or self.request.user.groups.filter(name="cid_user_manager").exists()
|
|
):
|
|
view_name = "account_profile"
|
|
# No need for reverse_lazy here, because it's called inside the method
|
|
return reverse(view_name, kwargs={"slug": self.object.username})
|
|
else:
|
|
return reverse("profile")
|
|
|
|
|
|
# class UpdateUser(TemplateView):
|
|
#
|
|
# user_form_class = UserForm
|
|
# comment_form_class = UserProfileForm
|
|
# template_name = 'user_update.html'
|
|
# slug_field = 'username'
|
|
# slug_url_kwarg = 'slug'
|
|
#
|
|
# def post(self, request):
|
|
# post_data = request.POST or None
|
|
# user_form = self.user_form_class(post_data, prefix='user')
|
|
# profile_form = self.profile_form_class(post_data, prefix='userprofile')
|
|
#
|
|
# context = self.get_context_data(post_form=user_form,
|
|
# profile_form=profile_form)
|
|
#
|
|
# if user_form.is_valid():
|
|
# self.form_save(user_form)
|
|
# if profile_form.is_valid():
|
|
# self.form_save(profile_form)
|
|
#
|
|
# return self.render_to_response(context)
|
|
#
|
|
# def form_save(self, form):
|
|
# obj = form.save()
|
|
# #messages.success(self.request, "{} saved successfully".format(obj))
|
|
# return obj
|
|
#
|
|
# def get(self, request, *args, **kwargs):
|
|
# return self.post(request, *args, **kwargs)
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_check_users(request):
|
|
if request.method == "POST":
|
|
users = json.loads(request.POST.get("user_list"))
|
|
|
|
existing_users = []
|
|
|
|
for email in users:
|
|
if User.objects.filter(username=email).exists():
|
|
existing_users.append(email)
|
|
|
|
return JsonResponse({"users": existing_users})
|
|
|
|
return
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_bulk_create_check(request):
|
|
if not request.method == "POST":
|
|
return HttpResponse("Invalid request")
|
|
|
|
user = {}
|
|
|
|
user["email"] = request.POST.get("email")
|
|
user["first_name"] = request.POST.get("first_name")
|
|
user["last_name"] = request.POST.get("last_name")
|
|
user["grade"] = request.POST.get("grade")
|
|
user["supervisor_email"] = request.POST.get("supervisor_email")
|
|
user["supervisor_name"] = request.POST.get("supervisor_name")
|
|
|
|
errors = []
|
|
info = []
|
|
|
|
if user["email"] == "undefined":
|
|
errors.append("No email provided")
|
|
|
|
if User.objects.filter(username=user["email"]).exists():
|
|
errors.append("User already exists")
|
|
|
|
if not UserGrades.objects.filter(name=user["grade"]):
|
|
errors.append("Invalid grade")
|
|
|
|
# Treat empty, missing or the string 'undefined' as no supervisor email provided
|
|
if not user.get("supervisor_email") or user["supervisor_email"] == "undefined":
|
|
# Supervisor email not provided; try to match by name
|
|
if not user.get("supervisor_name"):
|
|
errors.append("No supervisor provided")
|
|
else:
|
|
supervisor = Supervisor.objects.filter(name__iexact=user["supervisor_name"])
|
|
if not supervisor.exists():
|
|
info.append("Supervisor does not exist (will be created)")
|
|
else:
|
|
info.append("Supervisor exists (matched by name)")
|
|
if supervisor.count() > 1:
|
|
errors.append("More than one supervisor with that name")
|
|
# no further name check needed since matched by name
|
|
else:
|
|
supervisor = Supervisor.objects.filter(email=user["supervisor_email"])
|
|
|
|
if not supervisor:
|
|
info.append("Supervisor does not exist (will be created)")
|
|
else:
|
|
info.append("Supervisor exists")
|
|
|
|
if supervisor.count() > 1:
|
|
errors.append("More than one supervisor with that email")
|
|
|
|
# Compare names case-insensitively and trimmed to avoid false mismatches
|
|
try:
|
|
sup_name = (supervisor.first().name or '').strip().lower()
|
|
posted_name = (user.get("supervisor_name") or '').strip().lower()
|
|
if posted_name and sup_name != posted_name:
|
|
errors.append("Supervisor name does not match")
|
|
except Exception:
|
|
pass
|
|
|
|
if errors:
|
|
return HttpResponse(
|
|
format_html("<span class='error'>{}</span>", "<br/>".join(errors))
|
|
)
|
|
|
|
return HttpResponse(format_html("<span class='info'>{}</span>", "<br/>".join(info)))
|
|
|
|
|
|
@user_is_cid_user_manager
|
|
def accounts_bulk_create(request):
|
|
if request.method == "POST":
|
|
context = {}
|
|
|
|
users = json.loads(request.POST.get("usersjson"))
|
|
|
|
created_users = []
|
|
existing_users = []
|
|
error_text = ""
|
|
|
|
try:
|
|
users = json.loads(request.POST.get("usersjson"))
|
|
|
|
for user in users:
|
|
try:
|
|
existing_user = User.objects.get(username=user["email"])
|
|
existing_users.append(user["email"])
|
|
|
|
user_profile = UserProfile.objects.get(user=existing_user)
|
|
|
|
# Update grade and supervisor details
|
|
# This code is dup[licated]
|
|
if "grade" in user:
|
|
if not user["grade"].startswith("ST"):
|
|
user["grade"] = f"ST{user['grade']}"
|
|
|
|
grade = UserGrades.objects.get(name=user["grade"])
|
|
user_profile.grade = grade
|
|
|
|
if "supervisor_name" in user and user.get("supervisor_name"):
|
|
sup_email = user.get("supervisor_email")
|
|
# If supervisor email provided, prefer that. Otherwise try to match by name or create placeholder
|
|
if sup_email and sup_email != 'undefined':
|
|
s, created = Supervisor.objects.get_or_create(email=sup_email, defaults={"name": user.get("supervisor_name")})
|
|
else:
|
|
# try match by name
|
|
s_qs = Supervisor.objects.filter(name__iexact=user.get("supervisor_name"))
|
|
if s_qs.exists():
|
|
s = s_qs.first()
|
|
created = False
|
|
else:
|
|
import uuid
|
|
placeholder = f"no-email-{uuid.uuid4().hex}@local.invalid"
|
|
s = Supervisor.objects.create(email=placeholder, name=user.get("supervisor_name"))
|
|
created = True
|
|
|
|
if created:
|
|
s.save()
|
|
|
|
user_profile.supervisor = s
|
|
user_profile.save()
|
|
except User.DoesNotExist:
|
|
# Start by checking the email
|
|
try:
|
|
validate_email(user["email"])
|
|
except ValidationError:
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>{user['email']} does not appear to be a valid email (account for {user['first_name']} {user['last_name']} not created).</p>"
|
|
)
|
|
continue
|
|
|
|
# Try to create the user
|
|
try:
|
|
user_dict = {
|
|
"username": user["email"],
|
|
"first_name": user["first_name"],
|
|
"last_name": user["last_name"],
|
|
"email": user["email"],
|
|
"password": secrets.token_hex(nbytes=16),
|
|
}
|
|
new_user = User.objects.create_user(**user_dict)
|
|
except Exception as error:
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>Error creating user: {user['email']}<br/>{error}</p>"
|
|
)
|
|
# No need to try and create a profile if we can't create a user
|
|
continue
|
|
|
|
try:
|
|
user_profile = UserProfile.objects.get(user=new_user)
|
|
user_profile.peninsula_trainee = True
|
|
# TODO: check supervisor details are correct
|
|
if "supervisor_name" in user and user.get("supervisor_name"):
|
|
sup_email = user.get("supervisor_email")
|
|
if sup_email and sup_email != 'undefined':
|
|
s, created = Supervisor.objects.get_or_create(email=sup_email, defaults={"name": user.get("supervisor_name")})
|
|
else:
|
|
s_qs = Supervisor.objects.filter(name__iexact=user.get("supervisor_name"))
|
|
if s_qs.exists():
|
|
s = s_qs.first()
|
|
created = False
|
|
else:
|
|
import uuid
|
|
placeholder = f"no-email-{uuid.uuid4().hex}@local.invalid"
|
|
s = Supervisor.objects.create(email=placeholder, name=user.get("supervisor_name"))
|
|
created = True
|
|
|
|
if created:
|
|
s.save()
|
|
|
|
user_profile.supervisor = s
|
|
|
|
if "grade" in user:
|
|
if not user["grade"].startswith("ST"):
|
|
user["grade"] = f"ST{user['grade']}"
|
|
|
|
grade = UserGrades.objects.get(name=user["grade"])
|
|
user_profile.grade = grade
|
|
|
|
user_profile.save()
|
|
|
|
created_users.append(user["email"])
|
|
except Exception as error:
|
|
# If we fail to create the profile delete the user
|
|
new_user.delete()
|
|
error_text = (
|
|
error_text
|
|
+ f"<p>Error creating user profile: {user['email']}<br/>{error} (user not created)</p>"
|
|
)
|
|
|
|
except Exception as error:
|
|
error_text = error_text + f"<p>Error creating users.<br/>{error}</p>"
|
|
|
|
context["error"] = format_html(error_text)
|
|
context["created_users"] = created_users
|
|
|
|
if existing_users:
|
|
context["message"] = "The following users already exist: " + ", ".join(
|
|
existing_users
|
|
)
|
|
|
|
return render(request, "accounts_bulk_create.html", context)
|
|
else:
|
|
return render(request, "accounts_bulk_create.html", {})
|
|
|
|
|
|
def request_cid_details(request):
|
|
email = request.POST.get("email")
|
|
try:
|
|
validate_email(email)
|
|
except ValidationError:
|
|
return HttpResponse("Invalid email.")
|
|
|
|
cid_users = CidUser.objects.filter(email=email)
|
|
|
|
for cid_user in cid_users:
|
|
cid_user.email_details(resend=True)
|
|
cid_user.email_details()
|
|
|
|
return HttpResponse(
|
|
f"Candidate details will be sent to the requested email (if it is found in the system). If you have not received an email after a few minutes you may need to contact <a href='mailto:{settings.CONTACT_EMAIL}'>{settings.CONTACT_EMAIL}</a>"
|
|
)
|
|
|
|
|
|
def cimar_study_dicom_json(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
with CimarAPI(request.user.userprofile.cimar_sid) as api:
|
|
# studies = api.list_studies()
|
|
# test_study = studies["studies"][0]
|
|
|
|
# schema = api.get_study_schema(test_study)
|
|
|
|
schema = api.get_study_schema_by_uuid("c9417b53-87aa-4c40-aca0-3fa34c225536")
|
|
|
|
return JsonResponse(api.get_study_dicom_json(schema))
|
|
|
|
|
|
def cimar_study_viewer(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
viewer_link = case.viewer_link + f"?sid={request.user.userprofile.cimar_sid}"
|
|
return render(request, "cimar_embed.html", {"viewer_link": viewer_link})
|
|
|
|
|
|
def cimar_study_viewer_embed(request, uuid="c9417b53-87aa-4c40-aca0-3fa34c225536"):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
viewer_link = case.viewer_link + f"?sid={request.user.userprofile.cimar_sid}"
|
|
return render(request, "cimar_embed.html#embed", {"viewer_link": viewer_link})
|
|
|
|
|
|
def cimar_study_series_block(request, uuid):
|
|
case = get_object_or_404(CimarCase, uuid=uuid)
|
|
|
|
return render(
|
|
request,
|
|
"cimar_series.html",
|
|
{
|
|
"series_data": case.series_data["series"],
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|
|
|
|
|
|
def cimar_status(request):
|
|
login_status = CimarAPI(request.user.userprofile.cimar_sid).check_login_status()
|
|
|
|
if login_status:
|
|
return HttpResponse("")
|
|
else:
|
|
return render(
|
|
request,
|
|
"cimar_login.html#login-block",
|
|
{
|
|
"login_status": login_status,
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|
|
|
|
|
|
def cimar_logout(request):
|
|
CimarAPI(request.user.userprofile.cimar_sid).logout()
|
|
return HttpResponse("Logged out")
|
|
|
|
|
|
def cimar_login(request):
|
|
if request.htmx:
|
|
username = request.POST.get("username")
|
|
password = request.POST.get("password")
|
|
|
|
if not username and not password:
|
|
return HttpResponse("Please fill in login details")
|
|
|
|
try:
|
|
sid = CimarAPI().login(username, password)
|
|
|
|
request.user.userprofile.cimar_sid = sid
|
|
request.user.userprofile.save()
|
|
except InvalidCredentials:
|
|
return HttpResponse("Invalid login details")
|
|
|
|
return HttpResponse("Logged in")
|
|
|
|
else:
|
|
try:
|
|
login_status = CimarAPI(
|
|
request.user.userprofile.cimar_sid
|
|
).check_login_status()
|
|
except NoSession:
|
|
login_status = False
|
|
return render(
|
|
request,
|
|
"cimar_login.html",
|
|
{
|
|
"login_status": login_status,
|
|
"cimar_sid": request.user.userprofile.cimar_sid,
|
|
},
|
|
)
|