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.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 .forms import ( LongForm, MarkLongQuestionForm, SeriesFormSet, NoteForm, ExaminationForm, ) from .models import ( Long, Note, Examination, Exam, CidUserAnswer, ) from .tables import LongTable from .filters import LongFilter from django_tables2 import SingleTableView, SingleTableMixin from django_filters.views import FilterView from .decorators import user_is_author_or_long_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 copy import deepcopy from django.forms.models import model_to_dict from longs.forms import LongCreationDefaultForm from longs.models import LongCreationDefault logger = logging.getLogger(__name__) class AuthorOrCheckerRequiredMixin(object): def get_object(self, *args, **kwargs): obj = super(UpdateView, self).get_object(*args, **kwargs) if self.request.user.groups.filter(name="long_checker").exists(): return obj if self.request.user not in obj.author.all(): raise PermissionDenied() # or Http404 return obj @login_required def index(request): exams = Exam.objects.all() return render(request, "longs/index.html", {"exams": exams}) # def index(request): # other_longs = Long.objects.exclude(author=request.user.pk) # user_longs = Long.objects.filter(author=request.user.pk) # return render( # request, # "longs/index.html", # {"other_longs": other_longs, "user_longs": user_longs}, # ) @login_required def long_detail(request, pk): long = get_object_or_404(Long, pk=pk) # if request.user not in long.author.all(): # raise PermissionDenied # logging.debug(long.long_notes.first()) # logging.debug(long.subspecialty.first().name.all()) return render(request, "longs/long_detail.html", {"question": long}) @login_required def long_split(request, pk): long = get_object_or_404(Long, pk=pk) images = long.images.all() old_abnormality = long.abnormality.all() old_region = long.region.all() old_examination = long.examination.all() old_site = long.site.all() old_author = long.author.all() if not images: raise Http404 for n in range(len(images) - 1): long.pk = None long.save() long.abnormality.set(old_abnormality) long.region.set(old_region) long.examination.set(old_examination) long.site.set(old_site) long.author.set(old_author) images[n].long = long images[n].save() # images[-1].long # if request.user not in long.author.all(): # raise PermissionDenied # logging.debug(long.long_notes.first()) # logging.debug(long.subspecialty.first().name.all()) return render(request, "longs/long_detail.html", {"question": long}) @login_required def author_detail(request, pk): # logging.debug(Author.objects.all()) # author = get_object_or_404(Author, pk=pk) author = User.objects.get(pk=pk) longs = Long.objects.filter(author=pk) return render( request, "longs/category_detail.html", {"category": author, "longs": longs} ) def author_list(request): authors = User.objects.all() return render(request, "longs/author_list.html", {"authors": authors}) class LongCreationDefaultView(LoginRequiredMixin, UpdateView): model = LongCreationDefault form_class = LongCreationDefaultForm # 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 `Long` instance exists # before going any further. # """ # self.pk = get_object_or_404(Long, pk=kwargs['pk']) # return super().dispatch(request, *args, **kwargs) def get_object(self, queryset=None): obj, create = LongCreationDefault.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) class AddNote(LoginRequiredMixin, CreateView): model = Note form_class = NoteForm # 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 `Long` instance exists before going any further. """ self.pk = get_object_or_404(Long, pk=kwargs["pk"]) return super().dispatch(request, *args, **kwargs) def form_valid(self, form): note = form.save(commit=False) note.long = self.pk note.author = self.request.user note.save() response = super().form_valid(form) return response # form.instance.author.add(self.request.user.id) @login_required def long_clone(request, pk): new_item = get_object_or_404(Long, pk=pk) new_item.pk = None # autogen a new pk (item_id) # new_item.name = "Copy of " + new_item.name #need to change uniques form = LongForm(request.POST or None, instance=new_item) series_formset = SeriesFormSet() if form.is_valid(): form.instance.author.add(request.user.id) # logger.debug(formset.is_valid()) if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = obj series_formset.save() return response else: return super().form_invalid(form) context = { "form": form, "series_formset": series_formset, # other context } return render(request, "longs/long_form.html", context) class LongCreateBase(LoginRequiredMixin, CreateView): model = Long form_class = LongForm def get_context_data(self, **kwargs): context = super(LongCreateBase, self).get_context_data(**kwargs) if self.request.POST: context["series_formset"] = SeriesFormSet( self.request.POST, self.request.FILES ) context["series_formset"].full_clean() else: context["series_formset"] = SeriesFormSet() 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) series_formset = context["series_formset"] if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = self.object series_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("longs:long_clone", pk=self.object.pk) else: return super().form_invalid(form) # @login_required class LongCreate(LongCreateBase): def get_initial(self): # There has to be a better way... try: s = (i.pk for i in self.request.user.long_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 LongUpdate(LoginRequiredMixin, AuthorOrCheckerRequiredMixin, UpdateView): model = Long form_class = LongForm # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_context_data(self, **kwargs): context = super(LongUpdate, self).get_context_data(**kwargs) if self.request.POST: context["series_formset"] = SeriesFormSet( self.request.POST, self.request.FILES, instance=self.object ) context["series_formset"].full_clean() else: context["series_formset"] = SeriesFormSet(instance=self.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) series_formset = context["series_formset"] # logger.debug(formset.is_valid()) if series_formset.is_valid(): response = super().form_valid(form) series_formset.instance = self.object series_formset.save() return response else: return super().form_invalid(form) class LongClone(LongCreateBase): """Clones a existing long""" # fields = '__all__' # #fields = [ 'condition' ] # #initial = {'date_of_death': '05/01/2018'} # exclude = [ 'created_date', 'published_date' ] def get_initial(self): # print(self.request) old_object = get_object_or_404(Long, pk=self.kwargs["pk"]) initial_data = model_to_dict(old_object, exclude=["id"]) return initial_data @login_required @user_is_author_or_long_checker def long_scrap(request, pk): try: long = Long.objects.get(pk=pk) except Long.DoesNotExist: raise Http404("Long does not exist") long.scrapped = not long.scrapped long.save() return HttpResponseRedirect(reverse("longs:long_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, "longs/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, "longs/create_simple.html", {"form": form, "name": "Abnormality"} ) @csrf_exempt def get_abnormality_id(request): if request.is_ajax(): abnormality_name = request.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, "longs/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("/") @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, "longs/create_simple.html", {"form": form, "name": "Region"} ) @csrf_exempt def get_region_id(request): if request.is_ajax(): region_name = request.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 LongView(SingleTableMixin, FilterView): model = Long table_class = LongTable template_name = "longs/view.html" filterset_class = LongFilter def loadJsonAnswer(answer): eid = int(answer["eid"].split("/")[1]) # As access is not restricted make sure the data appears valid if (not isinstance(answer["cid"], int)) or ( not isinstance(eid, int) or (not isinstance(answer["ans"], str)) ): return JsonResponse( {"success": False, "error": "cid or eid or answers not defined"} ) # The model should catch invalid data but this should be less intensive max_int = 999999999999999999999 if answer["cid"] > max_int or eid > max_int: return JsonResponse({"success": False, "error": "invalid cid"}) exam = get_object_or_404(Exam, pk=eid) if not exam.active: return JsonResponse( {"success": False, "error": "No active exam: {}".format(eid)} ) exiting_answers = CidUserAnswer.objects.filter( question__id=answer["qid"], exam__id=eid, cid=answer["cid"] ) posted_answer = answer["ans"] # Normal answers are just posted with the answer "Normal" normal = False if posted_answer == "Normal": normal = True posted_answer = "" # We can ignore the first ab elif answer["qidn"] == "1" and posted_answer == "Abnormal": return False if not exiting_answers: ans = CidUserAnswer(answer=posted_answer, normal=normal, cid=answer["cid"]) ans.question_id = answer["qid"] ans.exam_id = eid ans.full_clean() ans.save() else: # Update an existing answer # should never be more than one (famous last words) ans = exiting_answers[0] ans.normal = normal ans.answer = posted_answer ans.full_clean() ans.save() return True def exam_json_edit(request, pk): if request.is_ajax() and request.method == "POST": exam = get_object_or_404(Exam, pk=pk) 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) else: data = {"status": "error"} return JsonResponse(data, status=400) @csrf_exempt def postExamAnswers(request): if request.is_ajax and request.method == "POST": n = 0 for answer in json.loads(request.POST.get("answers")): ret = loadJsonAnswer(answer) #if ret is not True: # return ret if ret: n = n + 1 # print(UserAnswer.objects.filter(exam__id=q["eid"])) # print(request.urlencode()) return JsonResponse({"success": True, "question_count": n}) return JsonResponse({"success": False, "error": "Invalid data"}) # @user_passes_test(user_is_admin, login_url="/accounts/login") @login_required def mark_overview(request, pk): exam = get_object_or_404(Exam, pk=pk) questions = exam.exam_questions.all() return render( request, "longs/mark_overview.html", {"exam": exam, "questions": questions} ) # @user_passes_test(user_is_admin, login_url="/accounts/login") @login_required def mark(request, pk, sk): exam = get_object_or_404(Exam, pk=pk) questions = exam.exam_questions.all() n = sk question_details = { "total": len(questions), "current": n + 1, } try: question = questions[sk] except IndexError: raise Http404("Exam question does not exist") answers_dict = {} for ans in question.answers.all(): answers_dict[ans.answer_compare] = ans marked_answers_set = set(answers_dict.keys()) # correct_answers = [i.answer.lower() for i in question.answers.all()] # half_correct_answers = [ # i.answer.lower() for i in question.half_mark_answers.all() # ] # incorrect_answers = [ # i.answer.lower() for i in question.incorrect_answers.all() # ] unmarked_user_answers = question.GetUnmarkedAnswers() if request.method == "POST": form = MarkLongQuestionForm(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("longs:mark", pk=pk, sk=n + 1) cd = form.cleaned_data # correct = cd.get("correct") # half_correct = cd.get("half_correct") # incorrect = cd.get("incorrect") marked_answers = json.loads(cd.get("marked_answers")) ## This is mildly dangerous as we delete all answers # question.answers.all().delete() # question.half_mark_answers.all().delete() # question.incorrect_answers.all().delete() # This should probably use json (or something else...) # ajax..... for ans in marked_answers["correct"]: 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 = Answer.MarkOptions.CORRECT a.save() # for ans in half_correct.split("--//--"): for ans in marked_answers["half-correct"]: 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 = Answer.MarkOptions.HALF_MARK a.save() for ans in marked_answers["incorrect"]: 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 = Answer.MarkOptions.INCORRECT a.save() # answer = form.save(commit=False) # answer.user = request.user # answer.question = question # answer.published_date = timezone.now() # answer.save() if "next" in request.POST: return redirect("longs:mark", pk=pk, sk=n + 1) elif "previous" in request.POST: return redirect("longs:mark", pk=pk, sk=n - 1) # Reset user answers (relies on the functional javascript) unmarked_user_answers = set() else: form = MarkLongQuestionForm() 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) return render( request, "longs/mark.html", { "exam": exam, "form": form, "question": question, "question_details": question_details, "user_answers": unmarked_user_answers, "correct_answers": correct_answers, "half_mark_answers": half_mark_answers, "incorrect_answers": incorrect_answers, }, ) def exam_toggle_results_published(request, pk): if request.is_ajax() and request.method == "POST": exam = get_object_or_404(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} return JsonResponse(data, status=200) else: data = {"status": "error"} return JsonResponse(data, status=400) def exam_toggle_active(request, pk): if request.is_ajax() and request.method == "POST": exam = get_object_or_404(Exam, pk=pk) exam.active = True if request.POST.get("active") == "true" else False exam.save() data = {"status": "success", "active": exam.active} return JsonResponse(data, status=200) else: data = {"status": "error"} return JsonResponse(data, status=400) def active_exams(request, json=True): exams = Exam.objects.all() active_exams = {"exams": []} for exam in exams: if exam.active: active_exams["exams"].append( { "name": exam.get_exam_name(), "url": request.build_absolute_uri(exam.get_json_url()), "type": "long", } ) if json == False: return active_exams["exams"] return JsonResponse(active_exams) def exam_json(request, pk): exam = get_object_or_404(Exam, pk=pk) if not exam.active: raise Http404("No available exam") exam_json_cache = cache.get("longs_exam_json_{}".format(pk)) if exam_json_cache is not None and not exam.recreate_json: exam_json_cache["cached"] = True return JsonResponse(exam_json_cache) questions = exam.exam_questions.all() exam_questions = defaultdict(dict) exam_order = [] for q in questions: exam_order.append(q.id) # Loop through longimage associations images = [] feedback_images = [] for i in q.images.all(): if i.feedback_image == True: feedback_images.append(image_as_base64(i.image)) # feedback_images.append(i.image.url) else: images.append(image_as_base64(i.image)) #images.append(i.image.url) exam_questions[q.id] = { "images": images, #"feedback_image": [], #"annotations": [str(q.image_annotations)], "type": "long", } #if feedback_images: # exam_questions[q.id]["feedback_image"] = feedback_images exam_json = { "eid": "long/{}".format(exam.id), "cached": False, "exam_type": "long", "exam_name": exam.name, "exam_mode": True, "exam_order": exam_order, "questions": exam_questions, } if exam.time_limit: exam_json["exam_time"] = exam.time_limit exam.recreate_json = False exam.save() cache.set("longs_exam_json_{}".format(pk), exam_json, 3600) return JsonResponse(exam_json) @login_required def exam_json_recreate(request, pk): exam = get_object_or_404(Exam, pk=pk) exam.recreate_json = True exam.save() return redirect("longs:exam_overview", pk=pk) @login_required def exam_scores_cid(request, pk): exam = get_object_or_404(Exam, pk=pk) questions = exam.exam_questions.all() cids = ( CidUserAnswer.objects.filter(question__in=questions) .order_by("cid") .values_list("cid", flat=True) .distinct() ) user_answers_and_marks = defaultdict(list) user_answers_marks = defaultdict(list) user_answers = defaultdict(list) user_names = {} by_question = defaultdict(list) unmarked = set() # Loop through all candidates for cid in cids: # Convoluted (probably...) user_names[cid] = cid for q in questions: # Get user answer s = q.cid_user_answers.filter(cid=cid).first() if not s: # skip if no answer user_answers_marks[cid].append(0) user_answers[cid].append("") by_question[q].append(("", 0)) continue elif s.normal: ans = "Normal" else: ans = s.answer 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)) by_question[q].append((ans, answer_score)) user_scores = {} for user in user_answers_marks: user_scores[user] = sum( [i for i in user_answers_marks[user] if i != "unmarked"] ) user_scores_list = list(user_scores.values()) 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() max_score = len(questions) * 2 return render( request, "longs/exam_scores.html", { "cids": cids, "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_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, }, ) def exam_scores_cid_user(request, pk, sk): exam = get_object_or_404(Exam, pk=pk) # TODO:Need some kind of test for cid cid = sk questions = exam.exam_questions.all() answers_and_marks = [] answers_marks = [] answers = [] for q in questions: # Get user answer user_answer = q.cid_user_answers.filter(cid=cid).first() if not user_answer or user_answer is None: # skip if no answer answers_marks.append(0) #answers.append("") answer_score = 0 ans = "Not answered" else: if user_answer.normal: ans = "Normal" else: ans = user_answer.answer answer_score = user_answer.get_answer_score() correct_answer = q.GetPrimaryAnswer() if not exam.publish_results: correct_answer = "*****" answer_score = 0 answers.append(ans) answers_marks.append(answer_score) answers_and_marks.append((ans, answer_score, correct_answer)) if "unmarked" in answers_marks: answered = [i for i in answers_marks if type(i) == int] total_score = sum(answered) unmarked_number = len(answers_marks) - len(answered) total_score = "{} ({} unmarked)".format(total_score, unmarked_number) else: total_score = sum(answers_marks) max_score = len(questions) * 2 return render( request, "longs/exam_scores_user.html", { "exam": exam, "cid": cid, "questions": questions, "answers": answers, "answers_marks": answers_marks, "total_score": total_score, "max_score": max_score, "answers_and_marks": answers_and_marks, }, ) @login_required def exam_question_detail(request, pk, sk): exam = get_object_or_404(Exam, pk=pk) question = exam.exam_questions.all()[sk] exam_length = len(exam.exam_questions.all()) pos = exam.get_question_index(question) + 1 print(question.exams.through.sort_value) previous = -1 if sk > 0: previous = sk - 1 next = sk + 1 if sk == exam_length - 1: next = False return render( request, "longs/long_detail.html", { "question": question, "exam": exam, "next": next, "previous": previous, "exam_length": exam_length, "pos": pos, }, ) @login_required def exam_list(request): exams = Exam.objects.all() return render(request, "longs/exam_list.html", {"exams": exams}) @login_required def exam_overview(request, pk): print(type(pk)) #print(Exam.objects.all()) #exams = Exam.objects.all() #print("test", Exam.objects.all().get(id=pk)) exam = get_object_or_404(Exam, pk=pk) print("exam", exam) questions = exam.exam_questions.all() question_number = len(questions) return render( request, "longs/exam_overview.html", {"exam": exam, "questions": questions, "question_number": question_number}, )