Files
penracourses/generic/views.py
T
Ross 8d7f67b950 .
2021-09-25 20:57:35 +01:00

742 lines
26 KiB
Python

from django.contrib.contenttypes.models import ContentType
from django.forms.models import model_to_dict
from django.shortcuts import render, get_object_or_404, redirect
from django.contrib.auth.decorators import login_required, user_passes_test
from django.views.decorators.csrf import csrf_exempt
from django.core.exceptions import PermissionDenied
from django.http import Http404, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from django.contrib.auth.mixins import LoginRequiredMixin
from django.utils.decorators import method_decorator
from django.core.cache import cache
import json
import urllib
from django.views import View
from .forms import (
ExaminationForm,
)
from .models import Examination, QuestionNote
from rapids.models import Rapid as RapidQuestion
from rapids.models import Exam as RapidExam
from longs.models import Long as LongQuestion
from longs.models import Exam as LongExam
from anatomy.models import AnatomyQuestion as AnatomyQuestion
from anatomy.models import Exam as AnatomyExam
from sbas.models import Question as SbasQuestion
#from sbas.models import Exam as SbasExam
from physics.models import Question as PhysicsQuestion
#from physics.models import Exam as PhysicsExam
from django.db.models import Case, When
from django.conf import settings
from datetime import datetime
import os
#from rad.views import get_question_and_content_type
def get_question_and_content_type(question_type):
if question_type == "rapid":
question = RapidQuestion
content_type = ContentType.objects.get(model="rapid")
elif question_type == "anatomy":
question = AnatomyQuestion
content_type = ContentType.objects.get(model="anatomyquestion")
elif question_type == "long":
question = LongQuestion
content_type = ContentType.objects.get(model="long")
elif question_type == "sbas":
question = SbasQuestion
content_type = ContentType.objects.get(app_label="sbas", model="question")
elif question_type == "physics":
question = PhysicsQuestion
content_type = ContentType.objects.get(app_label="physics", model="question")
else:
raise PermissionError()
return question, content_type
# Create your views here.
@login_required
def create_examination(request):
form = ExaminationForm(request.POST or None)
if form.is_valid():
instance = form.save()
return HttpResponse(
'<script>opener.closePopup(window, "%s", "%s", "#id_examination");</script>'
% (instance.pk, instance)
)
return render(
request, "rapids/create_simple.html", {"form": form, "name": "Examination"}
)
@csrf_exempt
def get_examination_id(request):
if request.is_ajax():
examination_name = request.GET["examination_name"]
examination_id = Examination.objects.get(name=examination_name).id
data = {
"examination_id": examination_id,
}
return HttpResponse(json.dumps(data), content_type="application/json")
return HttpResponse("/")
# Generic view to dispatch to indivuals app views
@login_required
def generic_exam_json_edit(request):
if request.is_ajax() and request.method == "POST":
question_type = request.POST.get("type")
if question_type == "rapid":
Exam = RapidExam
Question = RapidQuestion
elif question_type == "long":
Exam = LongExam
Question = LongQuestion
elif question_type == "anatomy":
Exam = AnatomyExam
Question = AnatomyQuestion
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
exam = get_object_or_404(Exam, pk=request.POST.get("exam_id"))
if "add_exam_questions" in request.POST:
question_ids = json.loads(request.POST.get("add_exam_questions"))
question_objects = Question.objects.filter(pk__in=question_ids)
for i in question_ids:
exam.exam_questions.add(i)
#exam.exam_questions.add(question_objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
if "set_exam_order" in request.POST:
new_order = json.loads(request.POST.get("set_exam_order"))
preserved = Case(*[When(pk=pk, then=pos) for pos, pk in enumerate(new_order)])
objects = Question.objects.filter(pk__in=new_order).order_by(preserved)
if len(objects) != exam.exam_questions.count():
data = {"status": "error, count does not match"}
return JsonResponse(data, status=400)
exam.exam_questions.set(objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
# Only checkers can do the following
if question_type == "rapid" and not request.user.groups.filter(name='rapid_checker').exists():
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if question_type == "anatomy" and not request.user.groups.filter(name='anatomy_checker').exists():
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if question_type == "long" and not request.user.groups.filter(name='long_checker').exists():
data = {"status": "error, permission denied"}
return JsonResponse(data, status=400)
if "set_open_access" in request.POST:
if request.POST.get("set_open_access") == "true":
state = True
else:
state = False
questions_changed = []
for q in exam.exam_questions.all():
q.open_access = state
q.save()
questions_changed.append(q.pk)
data = {"status": "success", "questions": questions_changed, "state": state}
return JsonResponse(data, status=200)
data = {"status": "error, unknown request"}
return JsonResponse(data, status=400)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
class ExamViews(View, LoginRequiredMixin):
def __init__(self, exam, question, app, question_type, loadJsonAnswer):
self.Exam = exam
self.Question = question
self.app_name = app
self.question_type = question_type
self.loadJsonAnswer = loadJsonAnswer
# THis may be better than check_user_access below
#group_map = {"rapids" : "rapid_checker", "anatomy" : "anatomy_checker", "longs":"long_checker"}
#exam_group = group_map[self.app_name]
def check_user_access(self, user):
"""Check if a user should be able to access a view
Args:
user ([user]): user object
Returns:
[boolean]: True if the user has access
"""
if self.app_name == "rapids" and not user.groups.filter(name='rapid_checker').exists():
return False
if self.app_name == "anatomy" and not user.groups.filter(name='anatomy_checker').exists():
return False
if self.app_name == "longs" and not user.groups.filter(name__in=('long_checker', 'long_marker')).exists():
return False
if self.app_name == "physics" and not user.groups.filter(name='physic_checker').exists():
return False
if self.app_name == "sbas" and not user.groups.filter(name='sba_checker').exists():
return False
return True
def check_user_edit_access(self, user):
"""Check if a user should be able to access a view
Args:
user ([user]): user object
Returns:
[boolean]: True if the user has access
"""
if self.app_name == "rapids" and not user.groups.filter(name='rapid_checker').exists():
return False
if self.app_name == "anatomy" and not user.groups.filter(name='anatomy_checker').exists():
return False
if self.app_name == "longs" and not user.groups.filter(name__in=('long_checker')).exists():
return False
if self.app_name == "physics" and not user.groups.filter(name='physic_checker').exists():
return False
if self.app_name == "sbas" and not user.groups.filter(name='sba_checker').exists():
return False
return True
@method_decorator(login_required)
def exam_toggle_results_published(self, request, pk):
if request.is_ajax() and request.method == "POST":
exam = get_object_or_404(self.Exam, pk=pk)
exam.publish_results = (
True if request.POST.get("publish_results") == "true" else False
)
exam.save()
data = {"status": "success", "publish_results": exam.publish_results}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def exam_toggle_active(self, request, pk):
if request.is_ajax() and request.method == "POST":
exam = get_object_or_404(self.Exam, pk=pk)
exam.active = True if request.POST.get("active") == "true" else False
exam.save()
data = {"status": "success", "active": exam.active}
return JsonResponse(data, status=200)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def exam_json_recreate(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
exam.recreate_json = True
exam.save()
exam.get_exam_json()
return redirect("{}:exam_overview".format(self.app_name), pk=pk)
@method_decorator(login_required)
def question_json_recreate(self, request, pk):
question = get_object_or_404(self.Question, pk=pk)
question.recreate_json = True
question.save()
return redirect("{}:question_detail".format(self.app_name), pk=pk)
@method_decorator(login_required)
def exam_list_all(self, request):
return self.exam_list(request, all=True)
@method_decorator(login_required)
def exam_list(self, request, all=False):
if all:
exams = self.Exam.objects.all().order_by("name")
else:
exams = self.Exam.objects.filter(exam_mode=True, archive=False).order_by("name")
if not self.check_user_access(request.user):
raise PermissionDenied
return render(request, "{}/exam_list.html".format(self.app_name), {"exams": exams})
@method_decorator(login_required)
def exam_overview(self, request, pk):
#print(Exam.objects.all())
#exams = Exam.objects.all()
#print("test", Exam.objects.all().get(id=pk))
exam = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user):
raise PermissionDenied
can_edit = self.check_user_edit_access(request.user) | request.user.is_superuser
notes = []
if can_edit:
notes = self.exam_notes(request, pk)
questions = exam.exam_questions.all()
question_number = len(questions)
return render(
request,
"{}/exam_overview.html".format(self.app_name),
{"exam": exam, "questions": questions, "question_number": question_number, "can_edit": can_edit, "notes": notes},
)
#@method_decorator(login_required)
def exam_notes(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user):
raise PermissionDenied
q, content_type = get_question_and_content_type(self.question_type)
question_pks = exam.exam_questions.values_list("pk", flat=True)
# Get active notes for type
notes = QuestionNote.objects.filter(content_type=content_type, object_id__in=question_pks, complete=False)
return notes
@method_decorator(login_required)
def exam_json_edit(self, request, pk):
if request.is_ajax() and request.method == "POST":
if not self.check_user_edit_access(request.user):
data = {"status": "error, invalid permisions"}
return JsonResponse(data, status=400)
exam = get_object_or_404(self.Exam, pk=pk)
if "add_exam_questions" in request.POST:
question_ids = json.loads(request.POST.get("add_exam_questions"))
question_objects = self.Question.objects.filter(pk__in=question_ids)
exam.exam_questions.add(question_objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
if "set_exam_order" in request.POST:
new_order = json.loads(request.POST.get("set_exam_order"))
preserved = Case(*[When(pk=pk, then=pos) for pos, pk in enumerate(new_order)])
objects = self.Question.objects.filter(pk__in=new_order).order_by(preserved)
# We now allow deleting exam questions
#if len(objects) != exam.exam_questions.count():
# data = {"status": "error, count does not match"}
# return JsonResponse(data, status=400)
exam.exam_questions.set(objects)
exam.save()
data = {"status": "success"}
return JsonResponse(data, status=200)
if not self.check_user_access(request.user):
raise PermissionDenied
if "set_open_access" in request.POST:
if request.POST.get("set_open_access") == "true":
state = True
else:
state = False
questions_changed = []
for q in exam.exam_questions.all():
q.open_access = state
q.save()
questions_changed.append(q.pk)
data = {"status": "success", "questions": questions_changed, "state": state}
return JsonResponse(data, status=200)
data = {"status": "error, unknown request"}
return JsonResponse(data, status=400)
else:
data = {"status": "error"}
return JsonResponse(data, status=400)
@method_decorator(login_required)
def mark_overview(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user):
raise PermissionDenied
questions = exam.exam_questions.all()
# Long cases require some special info
try:
if exam.double_mark:
question_unmarked_map = []
for q in questions:
question_unmarked_map.append((q, int(q.get_unmarked_answer_count(exam_pk=exam.pk)), int(q.get_unmarked_answer_count(exam_pk=exam.pk, marker=request.user)) ))
return render(
request, "{}/mark_overview.html".format(self.app_name), {"exam": exam, "question_unmarked_map": question_unmarked_map}
)
except AttributeError:
pass
question_unmarked_map = []
for q in questions:
question_unmarked_map.append((q, int(q.get_unmarked_answer_count(exam_pk=exam.pk))))
return render(
request, "{}/mark_overview.html".format(self.app_name), {"exam": exam, "question_unmarked_map": question_unmarked_map}
)
@method_decorator(login_required)
def index(self, request):
exams = self.Exam.objects.filter(author__id=request.user.id)
if self.app_name == "rapids" and request.user.groups.filter(name='rapid_checker').exists():
exams = self.Exam.objects.all()
if self.app_name == "anatomy" and request.user.groups.filter(name='anatomy_checker').exists():
exams = self.Exam.objects.all()
if self.app_name == "longs" and request.user.groups.filter(name='long_checker').exists():
exams = self.Exam.objects.all()
return render(request, "{}/index.html".format(self.app_name), {"exams": exams})
@method_decorator(login_required)
def exam_question_detail(self, request, pk, sk):
exam = get_object_or_404(self.Exam, pk=pk)
if request.user not in exam.author.all():
if not self.check_user_access(request.user):
raise PermissionDenied
question = exam.exam_questions.all()[sk]
exam_length = len(exam.exam_questions.all())
pos = exam.get_question_index(question) + 1
previous = -1
if sk > 0:
previous = sk - 1
next = sk + 1
if sk == exam_length - 1:
next = False
return render(
request,
"{}/question_detail.html".format(self.app_name),
{
"question": question,
"exam": exam,
"next": next,
"previous": previous,
"exam_length": exam_length,
"pos": pos,
},
)
def active_exams(self, request, json=True, based=True):
exams = self.Exam.objects.filter(archive=False)
active_exams = {"exams": []}
for exam in exams:
if exam.active or self.check_user_access(request.user):
if exam.json_creation_time:
creation_time = exam.json_creation_time.isoformat()
else:
creation_time = "None"
url = request.build_absolute_uri(exam.get_json_url())
# hacky
if not based:
url = url + "/unbased"
obj = {
"name": exam.get_exam_name(),
"url": url,
"type": self.question_type,
"eid": "{}/{}".format(self.question_type, exam.pk),
"json_creation_time": creation_time,
"exam_json_id": exam.exam_json_id,
"exam_active": exam.active
}
if self.question_type == "long":
h = {}
for question in exam.exam_questions.all():
# Generate json if needed
if question.json_creation_time is None:
question.get_question_json()
h[question.pk] = question.question_json_id
obj["multi_question_json"] = h
active_exams["exams"].append( obj )
if json == False:
return active_exams["exams"]
return JsonResponse(active_exams)
@method_decorator(csrf_exempt)
def postExamAnswers(self, request):
if request.is_ajax and request.method == "POST":
n = 0
for answer in json.loads(request.POST.get("answers")):
ret_status, ret = self.loadJsonAnswer(answer)
#if ret is not True:
# return ret
if ret_status:
n = n + 1
else:
return ret
# print(UserAnswer.objects.filter(exam__id=q["eid"]))
# print(request.urlencode())
return JsonResponse({"success": True, "question_count": n})
return JsonResponse({"success": False, "error": "Invalid data"})
def exam_json_unbased(self, request, pk):
"""
No (file based) caching is enabled for unbased exams
"""
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.active and not self.check_user_access(request.user):
raise Http404("No available exam")
exam_json = exam.get_exam_json(based=False)
#exam_json["exam_active"] = False
return JsonResponse(exam_json)
def exam_json(self, request, pk):
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.active and not self.check_user_access(request.user):
raise Http404("No available exam")
#exam_json_cache = cache.get("{}_exam_json_{}".format(self.app_name, pk))
path= "{0}{1}/exam/{2}.json".format(settings.MEDIA_ROOT, self.app_name, pk)
url= "{0}{1}/exam/{2}.json".format(settings.MEDIA_URL, self.app_name, pk)
if os.path.isfile(path) and not exam.recreate_json:
#exam_json_cache["cached"] = True
# Make sure we pass on an argument if we are forcing a reload
if request.GET.get('_'):
query_string = urllib.parse.urlencode({'_': request.GET.get('_')})
url = '{}?{}'.format(url, query_string)
return redirect(url)
return redirect(url)
return JsonResponse(exam_json_cache)
time = datetime.now()
exam.exam_json_id += 1
with open(path, "w+") as f:
exam_json = exam.get_exam_json()
exam_json["generated"] = time.isoformat()
exam_json["exam_json_id"] = exam.exam_json_id
f.write(json.dumps(exam_json))
exam.recreate_json = False
exam.json_creation_time = time
exam.save()
# We also try to clear the cache of any associated questions (longs)
# see exam_question_json
#n = exam.exam_questions.count()
#question_keys = ["{}_exam_json_{}_{}".format(self.app_name, pk, i) for i in range(1, n)]
#cache.delete_many(question_keys)
#cache.set("{}_exam_json_{}".format(self.app_name, pk), exam_json, 3600)
return JsonResponse(exam_json)
def exam_question_json(self, request, pk, sk):
question = get_object_or_404(self.Question, pk=sk)
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.active and not self.check_user_access(request.user):
raise Http404("No available exam")
if request.GET.get('_'):
base_url = redirect("{}:question_json".format(self.app_name), pk=sk)
query_string = urllib.parse.urlencode({'_': request.GET.get('_')})
url = '{}?{}'.format(base_url, query_string)
return redirect(url)
return redirect("{}:question_json".format(self.app_name), pk=sk)
def exam_question_json_unbased(self, request, pk, sk):
question = get_object_or_404(self.Question, pk=sk)
exam = get_object_or_404(self.Exam, pk=pk)
if not exam.active and not self.check_user_access(request.user):
raise Http404("No available exam")
return redirect("{}:question_json_unbased".format(self.app_name), pk=sk)
redirect()
question_json_cache = cache.get("{}_question_json_{}".format(self.app_name, sk))
if question_json_cache is not None and not question.recreate_json:
question_json_cache["cached"] = True
return JsonResponse(question_json_cache)
question_json = exam.get_exam_question_json(sk)
cache.set("{}_question_json_{}".format(self.app_name, sk), question_json, 3600)
return JsonResponse(question_json)
class GenericViewBase:
def __init__(self, app_name, QuestionObject, CidUserAnswerObject, ExamObject):
self.question_object = QuestionObject
self.cid_user_answer_object = CidUserAnswerObject
self.exam_object = ExamObject
self.app_name = app_name
g = {
"rapids" : "rapid_checker",
"anatomy" : "anatomy_checker",
"longs" : "longs_checker",
"sbas" : "sbas_checker",
"physics" : "physics_checker",
}
self.checker_group = g[app_name]
def question_review(self, request, pk):
"""
Return a json representation of the question when the exam is active
"""
if request.is_ajax():
exam = get_object_or_404(self.exam_object, pk=pk)
print(request.POST["question_number"])
print(type(request.POST["question_number"]))
if not exam.publish_results:
raise Http404
question = exam.exam_questions.all()[int(request.POST["question_number"])]
question_json = question.get_question_json(based=False)
return JsonResponse(question_json)
return JsonResponse({"status": "error"})
@user_passes_test(lambda u: u.is_superuser)
def user_answer_delete_multiple(self, request):
if request.is_ajax():
print(request.POST["answer_ids"])
answer_ids = json.loads(request.POST["answer_ids"])
# We could probably delete them all at once....
for id in answer_ids:
self.cid_user_answer_object.objects.get(pk=id).delete()
return JsonResponse({"status": "success"})
return JsonResponse({"status": "error"})
#@login_required
def question_detail(self, request, pk):
question = get_object_or_404(self.question_object, pk=pk)
if not question.open_access:
if (
not request.user.groups.filter(name=self.checker_group).exists()
and request.user not in question.author.all()
):
raise PermissionDenied()
# if request.user not in rapid.author.all():
# raise PermissionDenied
view_feedback = False
if (
request.user.groups.filter(name=self.checker_group).exists()
or request.user.groups.filter(name="feedback_checker").exists()
or request.user in question.author.all()
):
view_feedback = True
# logging.debug(rapid.subspecialty.first().name.all())
return render(
request,
f"{self.app_name}/question_detail.html",
{"question": question, "view_feedback": view_feedback},
)
@login_required
def question_user_answers(self, request, pk):
question = get_object_or_404(self.question_object, pk=pk)
answers = self.cid_user_answer_object.objects.filter(question__id=rapid.pk)
return render(
request,
f"{self.app_name}/question_user_answers.html",
{"question": question, "answers": answers},
)