import json from django.shortcuts import render, get_object_or_404, redirect from django import forms from django.views.decorators.http import require_POST # 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 ( ObjectDoesNotExist, PermissionDenied, ViewDoesNotExist, ) from django.forms.models import model_to_dict 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 # Create your views here. from reversion.views import RevisionMixin from generic.views import ( AuthorRequiredMixin, ExamCloneMixin, ExamCreateBase, ExamDeleteBase, ExamGroupsUpdateBase, ExamUpdateBase, ExamViews, GenericViewBase, UpdateQuestionMixin, ) from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin from atlas.models import Finding, Structure, Condition from rad import settings from shorts.decorators import user_is_author_or_shorts_checker from .models import ( AnswerMarks, Exam, Question, QuestionFinding, QuestionImage, UserAnswer, SampleAnswer, ) from .forms import ( #AnswerFormSet, #AnswerUpdateFormSet, ExamAuthorForm, ExamGroupsForm, ExamMarkerForm, MarkQuestionDoubleForm, MarkQuestionSingleForm, QuestionFindingForm, QuestionForm, QuestionSampleAnswerForm, MarkQuestionForm, ExamForm, ImageFormSet, SampleAnswerUpdateFormSet, ) from .tables import QuestionTable, QuestionUserAnswerTable from .filters import QuestionFilter, QuestionUserAnswerFilter from django_tables2 import SingleTableView, SingleTableMixin from django_filters.views import FilterView from dal import autocomplete def normaliseScore(score): return 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="shorts_checker").exists(): return obj if self.request.user not in obj.author.all(): raise PermissionDenied() # or Http404 return obj @login_required def question_split(request, pk): question = get_object_or_404(Question, pk=pk) if ( not request.user.groups.filter(name="shorts_checker").exists() and request.user not in question.author.all() ): raise PermissionDenied() images = question.images.all() old_examination = question.examination.all() # old_site = rapid.site.all() old_author = question.author.all() if not images: raise Http404 for n in range(len(images) - 1): question.pk = None question.save() question.examination.set(old_examination) # question.site.set(old_site) question.author.set(old_author) images[n].question = question 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, "shorts/question_detail.html", {"question": question}) class QuestionCreateBase(RevisionMixin, LoginRequiredMixin, CreateView): model = Question form_class = QuestionForm # Sending user object to the form, to verify which fields to display/remove (depending on group) def get_form_kwargs(self): kwargs = super(QuestionCreateBase, self).get_form_kwargs() kwargs.update({"user": self.request.user}) return kwargs def get_context_data(self, **kwargs): context = super(QuestionCreateBase, self).get_context_data(**kwargs) if self.request.POST: context["image_formset"] = ImageFormSet( self.request.POST, self.request.FILES ) context["image_formset"].full_clean() else: context["image_formset"] = ImageFormSet() 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"] if image_formset.is_valid(): response = super().form_valid(form) image_formset.instance = self.object image_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("shorts:rapid_clone", pk=self.object.pk) else: return super().form_invalid(form) # @login_required class QuestionCreate(QuestionCreateBase): 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 QuestionSampleAnswerUpdate(AuthorOrCheckerRequiredMixin, UpdateView): model = Question form_class = QuestionSampleAnswerForm template_name = "shorts/sample_answer_form.html" def get_context_data(self, **kwargs): context = super(QuestionSampleAnswerUpdate, self).get_context_data(**kwargs) if self.request.POST: context["answer_formset"] = SampleAnswerUpdateFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["answer_formset"].full_clean() else: context["answer_formset"] = SampleAnswerUpdateFormSet(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 QuestionUpdate( UpdateQuestionMixin ): model = Question form_class = QuestionForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_context_data(self, **kwargs): context = super(QuestionUpdate, 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() else: context["image_formset"] = ImageFormSet(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"] # logger.debug(formset.is_valid()) if image_formset.is_valid(): response = super().form_valid(form) image_formset.instance = self.object image_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 QuestionClone(QuestionCreateBase): """Clones a existing shorts question""" def get_initial(self): # print(self.request) old_object = get_object_or_404(Question, pk=self.kwargs["pk"]) initial_data = model_to_dict(old_object, exclude=["id"]) return initial_data class QuestionView(LoginRequiredMixin, SingleTableMixin, FilterView): model = Question table_class = QuestionTable template_name = "question_table_view.html" filterset_class = QuestionFilter def get_context_data(self, **kwargs): context = super().get_context_data(**kwargs) context["app_name"] = "shorts" 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) def mark_answer_override(request, exam_id, question_number, answer_id): return mark_answer(request, exam_id, question_number, answer_id, override=True) def mark_answer(request, exam_id, question_number, answer_id, override=False): exam = get_object_or_404(Exam, pk=exam_id) if not GenericExamViews.check_user_marker_access(request.user, exam_id): raise PermissionDenied questions = exam.get_questions() n = question_number question_details = { "total": len(questions), "current": n + 1, } try: question = questions[question_number] except IndexError: raise Http404("Exam question does not exist") try: answer = question.cid_user_answers.get(pk=answer_id) except ObjectDoesNotExist: raise Http404("User answer does not exist") answer_list = list( question.cid_user_answers.filter(exam__id=exam_id).values_list("id", flat=True) ) answer_list.sort() previous_answer_id = False next_answer_id = False if len(answer_list) > 1: if answer_list[0] == answer_id: next_answer_id = answer_list[1] elif answer_list[-1] == answer_id: previous_answer_id = answer_list[-2] else: next_answer_id = answer_list[answer_list.index(answer_id) + 1] previous_answer_id = answer_list[answer_list.index(answer_id) - 1] try: if exam.double_mark: unmarked = question.get_unmarked_user_answers( exam_pk=exam.id, marker=request.user ) else: unmarked = question.get_unmarked_user_answers(exam_pk=exam.id) next_unmarked_id = unmarked[0].id if next_unmarked_id == answer_id: next_unmarked_id = unmarked[1].id except IndexError: next_unmarked_id = False # We use different forms if the exam should be single or double marked if not exam.double_mark or ( request.method == "POST" and request.POST["form_id"] == "discrepancy_form" ): if request.method == "POST": form = MarkQuestionSingleForm(request.POST) if form.is_valid(): # If skip button is pressed skip the question without marking # Skip is problematic if we skip to the next unmarked answer # if "skip" in request.POST: # return redirect("longs:mark_answer", pk=exam_id, sk=question_number, cid=next_unmarked_id) # Extract score from form and save it to the object answer.score = form.cleaned_data["score"] answer.candidate_feedback = form.cleaned_data["candidate_feedback"] answer.save() if "next" in request.POST: return redirect( "shorts:mark_answer", exam_id=exam_id, question_number=question_number, answer_id=next_unmarked_id, ) if "save" in request.POST: return redirect( "shorts:mark_answer", exam_id=exam_id, question_number=question_number, answer_id=answer_id, ) # elif "previous" in request.POST: # return redirect("longs:mark", pk=exam_id, sk=n - 1) else: form = MarkQuestionSingleForm( initial={ "score": answer.score, "candidate_feedback": answer.candidate_feedback, } ) else: if request.method == "POST": form = MarkQuestionDoubleForm(request.POST) if form.is_valid(): # If skip button is pressed skip the question without marking # Skip is problematic if we skip to the next unmarked answer # if "skip" in request.POST: # return redirect("longs:mark_answer", pk=exam_id, sk=question_number, cid=next_unmarked_id) mark_object = AnswerMarks.objects.get_or_create( user_answer=answer, marker=request.user )[0] # Extract score from form and save it to the object mark_object.score = form.cleaned_data["score"] mark_object.mark_reason = form.cleaned_data["mark_reason"] mark_object.candidate_feedback = form.cleaned_data["candidate_feedback"] mark_object.marker = request.user mark_object.save() # Form is saved, check if we have two marks (and if they match) if answer.mark.count() == 2: # if they do automatically update teh scoer marks = set(answer.mark.values_list("score", flat=True)) if len(marks) == 1: answer.score = marks.pop() else: answer.score = "" answer.save() if "next" in request.POST: return redirect( "shorts:mark_answer", exam_id=exam_id, question_number=question_number, answer_id=next_unmarked_id, ) if "save" in request.POST: return redirect( "shorts:mark_answer", exam_id=exam_id, question_number=question_number, answer_id=answer_id, ) # elif "previous" in request.POST: # return redirect("longs:mark", pk=exam_id, sk=n - 1) else: try: mark_object = AnswerMarks.objects.get( user_answer=answer, marker=request.user ) form = MarkQuestionDoubleForm( initial={ "score": mark_object.score, "mark_reason": mark_object.mark_reason, "candidate_feedback": mark_object.candidate_feedback, } ) except AnswerMarks.DoesNotExist: form = MarkQuestionDoubleForm() discrepancy_form = False if answer.mark.count() == 2 or override: # if they do automatically update teh scoer marks = set(answer.mark.values_list("score", flat=True)) if len(marks) > 1 or override: discrepancy_form = MarkQuestionSingleForm( initial={ "score": answer.score, "candidate_feedback": answer.candidate_feedback, } ) return render( request, "shorts/mark_answer.html", { "exam": exam, "form": form, "discrepancy_form": discrepancy_form, "answer": answer, "question": question, "question_details": question_details, "next_unmarked_id": next_unmarked_id, "unmarked": unmarked, "previous_answer_id": previous_answer_id, "next_answer_id": next_answer_id, "answer_id": answer_id, "remote_url": settings.REMOTE_URL, }, ) @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") 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") user_answers = question.cid_user_answers.filter(exam__id=exam_pk) unmarked_count = user_answers.filter(score__isnull=True).count() if exam.double_mark: marker_unmarked_count = question.get_unmarked_user_answer_count( exam.pk, marker=request.user ) return render( request, "shorts/mark_question_double_overview.html", { "exam": exam, "question": question, "question_details": question_details, "user_answers": user_answers, "unmarked_count": unmarked_count, "marker_unmarked_count": marker_unmarked_count, "review": review, }, ) else: return render( request, "shorts/mark_question_single_overview.html", { "exam": exam, "question": question, "question_details": question_details, "user_answers": user_answers, "unmarked_count": unmarked_count, "review": review, }, ) class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView): model = Question success_url = reverse_lazy("shorts:question_view") GenericExamViews = ExamViews( Exam, Question, None, UserAnswer, "shorts", "shorts", 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 = QuestionUserAnswerTable template_name = "shorts/user_answer_question_view.html" filterset_class = QuestionUserAnswerFilter class UserAnswerDelete(SuperuserRequiredMixin, DeleteView): model = UserAnswer template_name = "user_answer_delete.html" success_url = reverse_lazy("shorts:user_answer_table_view") GenericViews = GenericViewBase("shorts", Question, UserAnswer, Exam) @user_is_author_or_shorts_checker def question_anonymise_dicom(request, pk): question = get_object_or_404(Question, pk=pk) question.anonymise_images() return redirect("shorts:question_detail", pk=pk) @user_is_author_or_shorts_checker def question_save_annotation(request, pk): if request.method == "POST": question = get_object_or_404(Question, 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) 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 = QuestionImage.objects.get(image_md5_hash=hash) data = { "status": "success", "id": question.pk, "url": question.question.get_absolute_url(), } return JsonResponse(data, status=200) except QuestionImage.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(Question, 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(Question, 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(Question, pk=pk) ids_param = request.GET.get("ids") ids = ids_param.split(",") if ids_param else [] return render( request, "shorts/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(Question, 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() # At the moment only the series owner / author can delete findings but anyone can create # TODO: duplicated from atlas "create_series_findings" def create_findings(request): # posts = Post.objects.all() response_data = {} if request.POST.get("action") == "post": question_finding_id = int(request.POST.get("question_finding_id")) question_id = request.POST.get("question") findings_ids = json.loads(request.POST.get("findings")) structure_ids = json.loads(request.POST.get("structures")) condition_ids = json.loads(request.POST.get("conditions")) description = request.POST.get("description") annotation_json = request.POST.get("annotation_json") viewport_json = request.POST.get("viewport_json") current_image_id_index = request.POST.get("current_image_id_index") question = Question.objects.get(pk=question_id) findings = Finding.objects.filter(pk__in=findings_ids) structures = Structure.objects.filter(pk__in=structure_ids) conditions = Condition.objects.filter(pk__in=condition_ids) if question_finding_id > 0: sf = get_object_or_404(QuestionFinding, pk=question_finding_id) sf.question = question sf.description = description sf.annotation_json = annotation_json sf.viewport_json = viewport_json else: sf = QuestionFinding.objects.create( question=question, description=description, annotation_json=annotation_json, viewport_json=viewport_json, ) sf.findings.set(findings) sf.structures.set(structures) sf.conditions.set(conditions) sf.current_image_id_index = current_image_id_index sf.save() return JsonResponse({"success": True}) return JsonResponse({"success": False}) return render(request, "create_post.html", {"posts": posts}) # TODO fix permissions class FindingDelete(RevisionMixin, DeleteView): # PermissionRequiredMixin, #permission_required = "atlas.delete_seriesfinding" model = QuestionFinding template_name = "confirm_delete.html" # success_url = reverse_lazy("atlas:case_view") def get_object(self): question_finding = super().get_object() question = question_finding.question if question.check_user_can_edit(self.request.user): return question_finding else: raise PermissionDenied def get_success_url(self): pk = self.kwargs["pk"] question_id = get_object_or_404(QuestionFinding, pk=pk).question.id return reverse("shorts:question_detail", kwargs={"pk": question_id}) def question_findings(request, question_id, finding_pk=None): question = get_object_or_404(Question, pk=question_id) #can_edit = series.check_user_can_edit(request.user) can_edit = True editing_finding = -1 if finding_pk is not None: finding = get_object_or_404(QuestionFinding, pk=finding_pk) question_finding_form = QuestionFindingForm(None, instance=finding) editing_finding = finding.pk else: question_finding_form = QuestionFindingForm(question_id=question.id) # logging.debug(atlas.subspecialty.first().name.all()) return render( request, "shorts/question_findings.html", { "question": question, "question_finding_form": question_finding_form, "editing_finding": editing_finding, "can_edit": can_edit, }, ) @require_POST def combine_images_side_by_side(request, question_id): image_ids = request.POST.getlist('combine-image-checkbox') if len(image_ids) != 2: return render(request, "shorts/combine_error.html", {"error": "Please select exactly two images."}) question = get_object_or_404(Question, pk=question_id) image = question.combine_images_side_by_side(*image_ids) return render(request, "shorts/combine_result.html", {"image": image, "remote_url": settings.REMOTE_URL, })