import json from django.shortcuts import render, get_object_or_404, redirect from django import forms # 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.core.exceptions import PermissionDenied from django.contrib.auth.mixins import LoginRequiredMixin from django.views.generic.detail import DetailView from django.views.generic.edit import CreateView, UpdateView, DeleteView from django.views.generic import ListView from django.views.decorators.csrf import csrf_exempt from django.urls import reverse_lazy, reverse from django.http import Http404, JsonResponse from django.http import HttpResponseRedirect, HttpResponse from django.utils.safestring import mark_safe from .forms import ( ExamAuthorForm, ExamGroupsForm, ExamMarkerForm, RapidForm, MarkRapidQuestionForm, ImageFormSet, RapidQuestionAnswerForm, RegionForm, AbnormalityForm, ExaminationForm, AnswerFormSet, AnswerUpdateFormSet, ExamForm, ) from .models import ( Rapid, Abnormality, RapidImage, Region, Examination, Exam, Answer, UserAnswer, ) from generic.views import ( AuthorRequiredMixin, ExamCloneMixin, ExamCreateBase, ExamDeleteBase, ExamGroupsUpdateBase, ExamUpdateBase, ExamViews, GenericViewBase, UpdateQuestionMixin, ) from reversion.views import RevisionMixin from .tables import RapidTable, RapidUserAnswerTable from .filters import RapidFilter, RapidUserAnswerFilter from django_tables2 import SingleTableView, SingleTableMixin from django_filters.views import FilterView from .decorators import user_is_author_or_rapid_checker from collections import defaultdict import json import statistics import plotly.express as px from django.core.cache import cache from helpers.images import image_as_base64 import logging from django.contrib import messages from copy import deepcopy from django.forms.models import model_to_dict from rapids.forms import RapidCreationDefaultForm from rapids.models import RapidCreationDefault from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin from dal import autocomplete logger = logging.getLogger(__name__) def normaliseScore(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 class AuthorOrCheckerRequiredMixin(object): def get_object(self, *args, **kwargs): obj = super().get_object(*args, **kwargs) if self.request.user.is_superuser: return obj if self.request.user.groups.filter(name="rapid_checker").exists(): return obj if self.request.user not in obj.author.all(): raise PermissionDenied() # or Http404 return obj @login_required def rapid_split(request, pk): rapid = get_object_or_404(Rapid, pk=pk) if ( not request.user.groups.filter(name="rapid_checker").exists() and request.user not in rapid.author.all() ): raise PermissionDenied() images = rapid.images.all() old_abnormality = rapid.abnormality.all() old_region = rapid.region.all() old_examination = rapid.examination.all() # old_site = rapid.site.all() old_author = rapid.author.all() if not images: raise Http404 for n in range(len(images) - 1): rapid.pk = None rapid.save() rapid.abnormality.set(old_abnormality) rapid.region.set(old_region) rapid.examination.set(old_examination) # rapid.site.set(old_site) rapid.author.set(old_author) images[n].rapid = rapid images[n].save() # images[-1].rapid # if request.user not in rapid.author.all(): # raise PermissionDenied # logging.debug(rapid.subspecialty.first().name.all()) return render(request, "rapids/question_detail.html", {"question": rapid}) class RapidCreationDefaultView(LoginRequiredMixin, UpdateView): model = RapidCreationDefault form_class = RapidCreationDefaultForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] # def dispatch(self, request, *args, **kwargs): # """ # Overridden so we can make sure the `Rapid` instance exists # before going any further. # """ # self.pk = get_object_or_404(Rapid, pk=kwargs['pk']) # return super().dispatch(request, *args, **kwargs) def get_object(self, queryset=None): obj, create = RapidCreationDefault.objects.get_or_create( author=self.request.user ) return obj def form_valid(self, form): model = form.save(commit=False) model.author = self.request.user model.save() response = super().form_valid(form) return response # form.instance.author.add(self.request.user.id) #@login_required #def rapid_clone(request, pk): # new_item = get_object_or_404(Rapid, pk=pk) # new_item.pk = None # autogen a new pk (item_id) # # new_item.name = "Copy of " + new_item.name #need to change uniques # # if ( # request.user not in new_item.get_author_objects() # and not request.user.is_superuser # ): # raise PermissionDenied() # or Http404 # # form = RapidForm(request.POST or None, instance=new_item) # # image_formset = ImageFormSet() # answer_formset = AnswerFormSet() # # if form.is_valid(): # form.instance.author.add(request.user.id) # # # logger.debug(formset.is_valid()) # if image_formset.is_valid() and answer_formset.is_valid(): # response = super().form_valid(form) # image_formset.instance = obj # image_formset.save() # # answer_formset.instance = obj # answer_formset.save() # return response # else: # return super().form_invalid(form) # # context = { # "form": form, # "image_formset": image_formset, # "answer_formset": answer_formset, # # other context # } # # return render(request, "rapids/rapid_form.html", context) class RapidCreateBase(RevisionMixin, LoginRequiredMixin, CreateView): model = Rapid form_class = RapidForm # Sending user object to the form, to verify which fields to display/remove (depending on group) def get_form_kwargs(self): kwargs = super(RapidCreateBase, self).get_form_kwargs() kwargs.update({"user": self.request.user}) return kwargs def get_context_data(self, **kwargs): context = super(RapidCreateBase, self).get_context_data(**kwargs) if self.request.POST: context["image_formset"] = ImageFormSet( self.request.POST, self.request.FILES ) context["image_formset"].full_clean() context["answer_formset"] = AnswerFormSet(self.request.POST) context["answer_formset"].full_clean() else: context["image_formset"] = ImageFormSet() context["answer_formset"] = AnswerFormSet() return context def form_valid(self, form): self.object = form.save(commit=False) self.object.save() form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) image_formset = context["image_formset"] answer_formset = context["answer_formset"] if image_formset.is_valid() and answer_formset.is_valid(): response = super().form_valid(form) image_formset.instance = self.object image_formset.save() answer_formset.instance = self.object answer_formset.save() # If the normal submit button is pressed we save as normal if "submit" in self.request.POST: return response # else we redirect to the clone url else: return redirect("rapids:rapid_clone", pk=self.object.pk) else: return super().form_invalid(form) # @login_required class RapidCreate(RapidCreateBase): # initial = {"laterality": Rapid.NONE} def get_initial(self): if "pk" in self.kwargs: initial = super().get_initial() exam = get_object_or_404(Exam, pk=self.kwargs["pk"]) initial["exams"] = [exam.id] return initial ## There has to be a better way... # try: # s = (i.pk for i in self.request.user.rapid_default.site.all()) # self.initial.update({"site": s}) # except AttributeError: # pass # return self.initial # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] # self.object = form.save(commit=False) # self.object.save() # form.instance.author.add(self.request.user.id) # return super().form_valid(form) class RapidAnswerUpdate(RevisionMixin, AuthorOrCheckerRequiredMixin, UpdateView): model = Rapid form_class = RapidQuestionAnswerForm template_name = "rapids/rapidquestionanswer_form.html" def get_context_data(self, **kwargs): context = super(RapidAnswerUpdate, self).get_context_data(**kwargs) if self.request.POST: context["answer_formset"] = AnswerUpdateFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["answer_formset"].full_clean() else: context["answer_formset"] = AnswerUpdateFormSet(instance=self.object) context["question"] = context["object"] return context def form_valid(self, form): self.object = form.save(commit=False) self.object.save() # form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) formset = context["answer_formset"] # logger.debug(formset.is_valid()) if formset.is_valid(): response = super().form_valid(form) formset.instance = self.object formset.save() return response else: return super().form_invalid(form) class RapidUpdate( UpdateQuestionMixin ): model = Rapid form_class = RapidForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_context_data(self, **kwargs): context = super(RapidUpdate, self).get_context_data(**kwargs) if self.request.POST: context["image_formset"] = ImageFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["image_formset"].full_clean() context["answer_formset"] = AnswerFormSet( self.request.POST, instance=self.object ) context["answer_formset"].full_clean() else: context["image_formset"] = ImageFormSet(instance=self.object) context["answer_formset"] = AnswerFormSet(instance=self.object) context["question"] = context["object"] return context def form_valid(self, form): self.object = form.save(commit=False) self.object.save() form.instance.author.add(self.request.user.id) context = self.get_context_data(form=form) image_formset = context["image_formset"] answer_formset = context["answer_formset"] # logger.debug(formset.is_valid()) if image_formset.is_valid() and answer_formset.is_valid(): response = super().form_valid(form) image_formset.instance = self.object image_formset.save() answer_formset.instance = self.object answer_formset.save() # restore exam orders # wanted_exams = form["exams"].data # for exam in self.object.exams.all(): # if exam in exam_orders and self.object in exam_orders[exam]: # print(exam_orders[exam]) # exam.exam_questions.set(exam_orders[exam]) # exam.save() return response else: return super().form_invalid(form) class RapidClone(RapidCreateBase): """Clones a existing rapid""" def get_initial(self): # print(self.request) old_object = get_object_or_404(Rapid, pk=self.kwargs["pk"]) initial_data = model_to_dict(old_object, exclude=["id"]) return initial_data @login_required @user_is_author_or_rapid_checker def rapid_scrap(request, pk): try: rapid = Rapid.objects.get(pk=pk) except Rapid.DoesNotExist: raise Http404("Rapid does not exist") if ( not request.user.groups.filter(name="rapid_checker").exists() and request.user not in rapid.author.all() ): raise PermissionDenied() rapid.scrapped = not rapid.scrapped rapid.save() return HttpResponseRedirect(reverse("rapids:question_detail", args=(pk,))) # @login_required # def edit_abnormality_popup(request): # instance = get_object_or_404(Abnormality, pk = pk) # form = AbnormalityForm(request.POST or None) # if form.is_valid(): # instance = form.save() # return HttpResponse('' % (instance.pk, instance)) # return render(request, "rapids/create_simple.html", {'form': form}) @login_required def create_abnormality(request): form = AbnormalityForm(request.POST or None) if form.is_valid(): instance = form.save() return HttpResponse( '' % (instance.pk, instance) ) return render( request, "rapids/create_simple.html", {"form": form, "name": "Abnormality"} ) @login_required def get_abnormality_id(request): """Return the primary key for an Abnormality looked up by name (admin popup helper). Scope: authenticated users only. Functionality: looks up an Abnormality by exact name and returns its ID as JSON; used by inline admin form popups. """ if request.accepts("application/json"): abnormality_name = request.GET.get("abnormality_name", "") abnormality_id = Abnormality.objects.get(name=abnormality_name).id data = { "abnormality_id": abnormality_id, } return HttpResponse(json.dumps(data), content_type="application/json") return HttpResponse("/") @login_required def create_examination(request): form = ExaminationForm(request.POST or None) if form.is_valid(): instance = form.save() return HttpResponse( '' % (instance.pk, instance) ) return render( request, "rapids/create_simple.html", {"form": form, "name": "Examination"} ) @login_required def get_examination_id(request): """Return the primary key for an Examination looked up by name (admin popup helper). Scope: authenticated users only. Functionality: looks up an Examination by exact name and returns its ID as JSON; used by inline admin form popups. """ if request.accepts("application/json"): examination_name = request.GET.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("/") @login_required def create_region(request): form = RegionForm(request.POST or None) if form.is_valid(): instance = form.save() return HttpResponse( '' % (instance.pk, instance) ) return render( request, "rapids/create_simple.html", {"form": form, "name": "Region"} ) @login_required def get_region_id(request): """Return the primary key for a Region looked up by name (admin popup helper). Scope: authenticated users only. Functionality: looks up a Region by exact name and returns its ID as JSON; used by inline admin form popups. """ if request.accepts("application/json"): region_name = request.GET.get("region_name", "") region_id = Region.objects.get(name=region_name).id data = { "region_id": region_id, } return HttpResponse(json.dumps(data), content_type="application/json") return HttpResponse("/") class RapidView(LoginRequiredMixin, SingleTableMixin, FilterView): model = Rapid table_class = RapidTable template_name = "question_table_view.html" filterset_class = RapidFilter def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context["app_name"] = "rapids" return context @login_required def mark_all(request, exam_pk, sk): return mark(request, exam_pk, sk, unmarked_exam_answers_only=False) @login_required def mark_review(request, exam_pk, sk): return mark(request, exam_pk, sk, unmarked_exam_answers_only=False, review=True) # @user_passes_test(user_is_admin, login_url="/accounts/login") @login_required def mark(request, exam_pk, sk, unmarked_exam_answers_only=True, review=False): exam = get_object_or_404(Exam, pk=exam_pk) if not GenericExamViews.check_user_marker_access(request.user, exam_pk): raise PermissionDenied if not exam.exam_mode: raise Http404("Packet not in exam mode") mark_url = "rapids:mark" if review: mark_url = "rapids:mark_review" elif not unmarked_exam_answers_only: mark_url = "rapids:mark_all" questions = exam.get_questions() n = sk question_details = { "total": len(questions), "current": n + 1, } try: question = questions[sk] except IndexError: raise Http404("Exam question does not exist") if request.method == "POST": form = MarkRapidQuestionForm(request.POST) # ******************************* # TODO: convert to JSON request # ******************************* if form.is_valid(): # If skip button is pressed skip the question without marking if "skip" in request.POST: return redirect(mark_url, exam_pk=exam_pk, sk=n + 1) cd = form.cleaned_data marked_answers = json.loads(cd.get("marked_answers")) # This should probably use json (or something else...) # ajax..... to_run = ( ("correct", Answer.MarkOptions.CORRECT), ("half-correct", Answer.MarkOptions.HALF_MARK), ("incorrect", Answer.MarkOptions.INCORRECT), ) for status_string, status in to_run: for ans in marked_answers[status_string]: ans = ans.strip() if ans == "": continue a = Answer.objects.filter( answer__iexact=ans, question_id=question.pk ).first() if a is None: a = Answer() a.question_id = question.pk a.answer = ans a.status = status a.save() if "next" in request.POST: return redirect(mark_url, exam_pk=exam_pk, sk=n + 1) elif "previous" in request.POST: return redirect(mark_url, exam_pk=exam_pk, sk=n - 1) # Reset user answers (relies on the functional javascript) unmarked_user_answers = set() else: form = MarkRapidQuestionForm() # answers_dict = {} # for ans in question.answers.all(): # answers_dict[ans.answer_compare] = ans correct_answers = [] half_mark_answers = [] incorrect_answers = [] review_user_answers = [] unmarked_answers_bool = False unmarked_user_answers = [] # this could be improved if question.normal: incorrect_answers = question.get_user_answers(exam_pk=exam_pk) if "" in incorrect_answers: incorrect_answers.remove("") if "normal" in incorrect_answers: incorrect_answers.remove("normal") else: if review: unmarked_user_answers = question.get_user_answers( exam_pk=exam_pk, include_normal=False ) elif unmarked_exam_answers_only: unmarked_user_answers = question.get_unmarked_user_answers(exam_pk=exam_pk) else: unmarked_user_answers = question.get_unmarked_user_answers() if not unmarked_exam_answers_only: correct_answers = question.answers.filter(status=Answer.MarkOptions.CORRECT) half_mark_answers = question.answers.filter( status=Answer.MarkOptions.HALF_MARK ) incorrect_answers = question.answers.filter( status=Answer.MarkOptions.INCORRECT ) if review: for ans in unmarked_user_answers: # duplicated from user answer model.... marked_ans = question.answers.filter(answer_compare__iexact=ans).first() mark = "unmarked" if marked_ans is not None: if marked_ans.status == Answer.MarkOptions.CORRECT: mark = "correct" elif marked_ans.status == Answer.MarkOptions.HALF_MARK: mark = "half-correct" elif marked_ans.status == Answer.MarkOptions.INCORRECT: mark = "incorrect" if mark == "unmarked": unmarked_answers_bool = True review_user_answers.append((ans, mark)) return render( request, "rapids/mark.html", { "exam": exam, "form": form, "review": review, "question": question, "question_number": n, "question_details": question_details, "unmarked_answers_bool": unmarked_answers_bool, "user_answers": unmarked_user_answers, "review_user_answers": review_user_answers, "correct_answers": correct_answers, "half_mark_answers": half_mark_answers, "incorrect_answers": incorrect_answers, "unmarked_exam_answers_only": unmarked_exam_answers_only, }, ) class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView): model = Rapid success_url = reverse_lazy("rapids:question_view") GenericExamViews = ExamViews( Exam, Rapid, Answer, UserAnswer, "rapids", "rapid", normalise_score=normaliseScore ) class ExamCreate(ExamCreateBase): model = Exam form_class = ExamForm class ExamUpdate(ExamUpdateBase, AuthorOrCheckerRequiredMixin): model = Exam form_class = ExamForm class ExamDelete(AuthorOrCheckerRequiredMixin, ExamDeleteBase): model = Exam class ExamAuthorUpdate( RevisionMixin, CheckCanEditMixin, LoginRequiredMixin, AuthorRequiredMixin, UpdateView, ): model = Exam form_class = ExamAuthorForm template_name = "author_form.html" class ExamMarkersUpdate( RevisionMixin, CheckCanEditMixin, LoginRequiredMixin, AuthorRequiredMixin, UpdateView ): model = Exam form_class = ExamMarkerForm template_name = "markers_form.html" class ExamGroupsUpdate(ExamGroupsUpdateBase): model = Exam form_class = ExamGroupsForm class UserAnswerView(LoginRequiredMixin, DetailView): model = UserAnswer class UserAnswerTableView(LoginRequiredMixin, SingleTableMixin, FilterView): model = UserAnswer table_class = RapidUserAnswerTable template_name = "rapids/user_answer_question_view.html" filterset_class = RapidUserAnswerFilter class UserAnswerDelete(SuperuserRequiredMixin, DeleteView): model = UserAnswer template_name = "user_answer_delete.html" success_url = reverse_lazy("rapids:user_answer_table_view") GenericViews = GenericViewBase("rapids", Rapid, UserAnswer, Exam) @user_is_author_or_rapid_checker def question_anonymise_dicom(request, pk): question = get_object_or_404(Rapid, pk=pk) question.anonymise_images() return redirect("rapids:question_detail", pk=pk) @user_is_author_or_rapid_checker def question_save_annotation(request, pk): if request.method == "POST": question = get_object_or_404(Rapid, pk=pk) image_annotations = json.loads(request.POST.get("annotation")) question_images = question.images.filter(feedback_image=False) for n, image in enumerate(question_images): annotation = image_annotations[n] if annotation != "undefined": image.image_annotations = annotation else: image.image_annotations = "" image.save() data = {"status": "success"} return JsonResponse(data, status=200) else: data = {"status": "error"} return JsonResponse(data, status=400) @user_is_author_or_rapid_checker def question_migrate_to_shorts(request, pk): """Migrate a single Rapid question to a shorts.Question. Accessible to question authors and members of the rapid_checker group (or superusers). """ rapid = get_object_or_404(Rapid, pk=pk) from shorts.models import Question as ShortQuestion try: q = ShortQuestion.from_rapid(rapid) except Exception as e: messages.error(request, f"Migration failed: {e}") return redirect("rapids:question_detail", pk=rapid.pk) if q is None: messages.info(request, "Question skipped (marked normal) — not migrated.") return redirect("rapids:question_detail", pk=rapid.pk) messages.success(request, "Migration complete — redirected to new Short question.") return redirect(q.get_absolute_url()) @login_required def exam_migrate_rapids_to_shorts(request, pk): """Migrate all rapids associated with an Exam to shorts.Question. Accessible to exam authors and superusers. """ exam = get_object_or_404(Exam, pk=pk) # check permission: exam author or superuser if not (request.user.is_superuser or exam.author.filter(id=request.user.id).exists()): raise PermissionDenied() from shorts.models import Question as ShortQuestion created = [] skipped = [] failed = [] for rapid in exam.exam_questions.all(): try: q = ShortQuestion.from_rapid(rapid) if q is None: skipped.append(str(rapid.id)) else: created.append(str(q.id)) except Exception as e: failed.append(str(rapid.id)) continue if created: messages.success(request, f"Created shorts questions: {', '.join(created)}") if skipped: messages.info(request, f"Skipped normal rapids (not migrated): {', '.join(skipped)}") if failed: messages.error(request, f"Failed to migrate rapids: {', '.join(failed)}") # Redirect back to the Rapids exam overview (shorts exam may not exist) return redirect("rapids:exam_overview", pk=exam.pk) class ExamClone(ExamCloneMixin, ExamCreate): """Clone exam view""" @login_required def get_question_by_hash(request): if request.method == "POST": hash = request.POST.get("hash") try: question = RapidImage.objects.get(image_md5_hash=hash) data = { "status": "success", "id": question.pk, "url": question.rapid.get_absolute_url(), } return JsonResponse(data, status=200) except RapidImage.DoesNotExist: data = {"status": "success", "id": False} return JsonResponse(data, status=200) # TODO: better permissions @login_required def question_json_unbased(request, pk): """ No (file based) caching is enabled for unbased quesitons """ question = get_object_or_404(Rapid, pk=pk) question_json = question.get_question_json(based=False) return JsonResponse(question_json) @login_required def question_json(request, pk): question = get_object_or_404(Rapid, pk=pk) question_json = question.get_question_json(based=True) return JsonResponse(question_json) @login_required def question_viewer(request): # question = get_object_or_404(Rapid, pk=pk) ids_param = request.GET.get("ids") ids = ids_param.split(",") if ids_param else [] return render( request, "rapids/question_viewer.html", { "ids": ids, }, ) class AnswerAutocomplete(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 Answer.objects.none() qs = Answer.objects.all() if self.q: qs = qs.filter(answer__istartswith=self.q) return qs def confirm_answer(request, answer_id: int): if request.htmx: answer = get_object_or_404(Answer, pk=answer_id) # Check if the user has permission to confirm the answer if not answer.can_edit(request.user): return HttpResponse("Invalid permissions", status=403) answer.clean() answer.proposed = False answer.save() return HttpResponse("Answer accepted") raise PermissionDenied() def delete_answer(request, answer_id: int): if request.htmx: answer = get_object_or_404(Answer, pk=answer_id) # Check if the user has permission to confirm the answer if not answer.can_edit(request.user): return HttpResponse("Invalid permissions", status=403) answer.delete() return HttpResponse("Answer deleted") raise PermissionDenied() def question_add_exam(request, question_id: int): if request.htmx: question = get_object_or_404(Rapid, pk=question_id) if not question.can_edit(request.user): return HttpResponse("Invalid permissions", status=403) if request.POST: exam_id = request.POST.get("exam_id", None) if exam_id is None: return HttpResponse("No exam id provided", status=400) exam = get_object_or_404(Exam, pk=exam_id) if request.POST.get("remove", False) == "true": question.exams.remove(exam) return HttpResponse("Question removed from exam.") else: question.exams.add(exam) return HttpResponse("Question added to exam.") else: # Return a list of exams that we can add to the question exams = Exam.objects.filter(author=request.user) exams = exams.difference(question.exams.all()) if not exams: return HttpResponse("No exams available to add") html = "Exams to add to question:
" for exam in exams: html = html + (f"
" ) html = html + "
Exams to remove from question:
" for exam in question.exams.all(): html = html + (f"
" ) html = html + "" return HttpResponse(mark_safe(html)) raise PermissionDenied()