Files
penracourses/sbas/views.py
T

686 lines
24 KiB
Python

from reversion.views import RevisionMixin
from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
from django.views.generic.detail import DetailView
from generic.models import CidUser
from sbas.forms import UserAnswerForm, ExamAuthorForm, ExamForm
from django.shortcuts import render, get_object_or_404, redirect
from django.views.decorators.csrf import csrf_exempt
from django import forms
from django.utils import timezone
from django.forms.models import model_to_dict
# from django.contrib.auth.models import User
from django.contrib.auth.decorators import login_required, user_passes_test
from django.contrib.auth.models import User
from django.contrib.auth.mixins import LoginRequiredMixin
from django.views.generic.edit import CreateView, UpdateView, DeleteView
from django.views.generic import ListView
from django.db.models.functions import Lower
from django.core.cache import cache
from django.urls import reverse_lazy, reverse
from django.http import Http404, HttpResponseBadRequest, JsonResponse
from django.http import HttpResponseRedirect, HttpResponse
from .models import Question, Category, Exam, UserAnswer
from django.views.decorators.http import require_http_methods
from django.db import transaction
import re
import logging
logger = logging.getLogger(__name__)
from loguru import logger
try:
import jsonschema
except Exception:
jsonschema = None
class LLMImportForm(forms.Form):
file = forms.FileField(required=False, help_text="Upload a JSON or JSONL file matching the SBA import schema")
raw = forms.CharField(
required=False,
widget=forms.Textarea(attrs={"rows": 10, "cols": 80}),
help_text="Or paste JSON, JSONL (newline-delimited), or multiple JSON documents separated by blank lines",
)
def _resolve_m2m(model_class, value):
"""Resolve an item which may be an int (pk) or a string (name/slug).
Returns a queryset or empty list of matching objects.
"""
if value is None:
return []
if isinstance(value, int):
try:
return [model_class.objects.get(pk=value)]
except model_class.DoesNotExist:
return []
if isinstance(value, str):
# try exact name field, then case-insensitive contains
qs = model_class.objects.filter(name__iexact=value)
if not qs.exists():
qs = model_class.objects.filter(name__icontains=value)
return list(qs[:5])
return []
@login_required
@user_passes_test(lambda u: u.is_superuser)
@require_http_methods(["GET", "POST"])
def import_llm_questions(request):
"""Upload a JSON/JSONL file of LLM-produced questions and import them.
Only superusers may use this view. The view validates each object against
the agreed JSON schema (draft-07) and attempts to resolve M2M references
by numeric id or by name. Returns a JSON report of created items and errors.
"""
schema = {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "SBA Question",
"type": "object",
"additionalProperties": False,
"required": [
"title",
"stem",
"a_answer",
"b_answer",
"c_answer",
"d_answer",
"e_answer",
"best_answer",
"category",
],
"properties": {
"title": {"type": "string"},
"stem": {"type": "string"},
"a_answer": {"type": "string"},
"a_feedback": {"type": "string"},
"b_answer": {"type": "string"},
"b_feedback": {"type": "string"},
"c_answer": {"type": "string"},
"c_feedback": {"type": "string"},
"d_answer": {"type": "string"},
"d_feedback": {"type": "string"},
"e_answer": {"type": "string"},
"e_feedback": {"type": "string"},
"feedback": {"type": "string"},
"best_answer": {"type": "string", "enum": ["a", "b", "c", "d", "e"]},
"category": {"oneOf": [{"type": "string"}]},
"finding": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"structure": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"condition": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"presentation": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"subspecialty": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
},
}
if request.method == "GET":
form = LLMImportForm()
return render(request, "sbas/import_llm_questions.html", {"form": form})
# POST
form = LLMImportForm(request.POST, request.FILES)
if not form.is_valid():
return JsonResponse({"ok": False, "errors": ["No file uploaded or invalid form"]}, status=400)
raw_text = None
if form.cleaned_data.get("raw"):
raw_text = form.cleaned_data.get("raw")
elif form.cleaned_data.get("file"):
f = form.cleaned_data["file"]
raw_text = f.read().decode("utf-8")
else:
return JsonResponse({"ok": False, "errors": ["No file uploaded or text provided"]}, status=400)
# support: single JSON object, JSON array, JSONL (one JSON object per line),
# or multiple JSON documents separated by one or more blank lines.
candidates = []
# sanitize raw_text by escaping backslashes that are not part of a valid
# JSON escape sequence. This handles inputs containing LaTeX-style
# sequences such as "\ge" which are invalid JSON (Invalid \escape).
def _sanitize_backslashes(s: str):
# valid escapes after backslash in JSON: \" \\ \/ \b \f \n \r \t and \uXXXX
# This regex finds backslashes not followed by ", \\ , /, b, f, n, r, t, or u
pattern = re.compile(r"\\(?!(?:[\"\\/bfnrtu]))")
new_s, n = pattern.subn(r"\\\\", s)
return new_s, n
raw_text_stripped = raw_text.strip()
raw_text, sanitized_count = _sanitize_backslashes(raw_text_stripped)
logger.debug("import_llm_questions: sanitized input length %d (escaped %d backslashes)", len(raw_text), sanitized_count)
# Try whole-text JSON first (object or array)
parse_error = None
try:
parsed = json.loads(raw_text)
if isinstance(parsed, list):
candidates = parsed
elif isinstance(parsed, dict):
candidates = [parsed]
except Exception as e_outer:
parse_error = e_outer
# Try JSONL: each non-empty line is a JSON object
for i, line in enumerate(raw_text.splitlines()):
line = line.strip()
if not line:
continue
try:
candidates.append(json.loads(line))
except Exception:
# not JSONL or some lines are multi-line JSON; fallthrough
candidates = []
break
# If JSONL didn't produce candidates, try splitting by blank lines
if not candidates:
blocks = [b.strip() for b in re.split(r"\n\s*\n", raw_text) if b.strip()]
if len(blocks) > 1:
for i, block in enumerate(blocks):
try:
candidates.append(json.loads(block))
except Exception as e_block:
return JsonResponse({"ok": False, "errors": [f"Invalid JSON in block {i+1}: {e_block}"]}, status=400)
if not candidates:
# If still empty, return the original parse error if present
msg = str(parse_error) if parse_error is not None else "No JSON objects found"
logger.debug(f"Failed to parse input: {msg}")
return JsonResponse({"ok": False, "errors": [f"Failed to parse input: {msg}"]}, status=400)
report = {"total": len(candidates), "created": 0, "errors": [], "sanitized_backslashes": sanitized_count}
if jsonschema is None:
return JsonResponse({"ok": False, "errors": ["jsonschema library not available in environment"]}, status=500)
validator = jsonschema.Draft7Validator(schema)
from atlas.models import Finding, Structure, Condition, Presentation, Subspecialty
for idx, payload in enumerate(candidates):
errors = []
for err in validator.iter_errors(payload):
errors.append(err.message)
if errors:
logger.debug(f"Validation errors for item {idx}: {errors}")
report["errors"].append({"index": idx, "errors": errors})
continue
# create the Question inside a transaction; partial failures should not leave half-written objects
try:
with transaction.atomic():
q = Question()
q.stem = payload.get("stem", "").strip()
# map title -> not present in model; use as stem prefix or store in feedback
title = payload.get("title")
if title:
q.stem = f"<strong>{title}</strong>\n" + q.stem
q.a_answer = payload.get("a_answer", "").strip()
q.a_feedback = payload.get("a_feedback", "")
q.b_answer = payload.get("b_answer", "").strip()
q.b_feedback = payload.get("b_feedback", "")
q.c_answer = payload.get("c_answer", "").strip()
q.c_feedback = payload.get("c_feedback", "")
q.d_answer = payload.get("d_answer", "").strip()
q.d_feedback = payload.get("d_feedback", "")
q.e_answer = payload.get("e_answer", "").strip()
q.e_feedback = payload.get("e_feedback", "")
q.feedback = payload.get("feedback", "")
q.best_answer = payload.get("best_answer")
# category resolve by name
cat_val = payload.get("category")
if isinstance(cat_val, str):
cat_obj, created = Category.objects.get_or_create(category=cat_val)
q.category = cat_obj
q.save()
# Resolve M2Ms
for key, model_cls in (
("finding", Finding),
("structure", Structure),
("condition", Condition),
("presentation", Presentation),
("subspecialty", Subspecialty),
):
vals = payload.get(key) or []
resolved = []
for v in vals:
if isinstance(v, int):
try:
resolved.append(model_cls.objects.get(pk=v))
except model_cls.DoesNotExist:
# skip unknown
continue
elif isinstance(v, str):
qs = model_cls.objects.filter(name__iexact=v)
if not qs.exists():
qs = model_cls.objects.filter(name__icontains=v)
if qs.exists():
resolved.extend(list(qs[:5]))
if resolved:
getattr(q, key).add(*[o.pk for o in resolved])
report["created"] += 1
except Exception as e:
report["errors"].append({"index": idx, "errors": [str(e)]})
return JsonResponse({"ok": True, "report": report})
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
from collections import defaultdict
import os
import base64
import mimetypes
import json
import statistics
import plotly.express as px
from generic.views import (
AuthorRequiredMixin,
ExamCloneMixin,
ExamCreateBase,
ExamDeleteBase,
ExamGroupsUpdateBase,
ExamUpdateBase,
ExamViews,
GenericViewBase,
UpdateQuestionMixin,
exam_inactive,
)
from django.core.exceptions import PermissionDenied
from .tables import QuestionTable, UserAnswerTable
from .filters import QuestionFilter, UserAnswerFilter
from .forms import (
ExamGroupsForm,
ExamMarkerForm,
QuestionForm,
ExamForm,
)
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if self.request.user.groups.filter(name="sbas_checker").exists():
return obj
if self.request.user not in obj.author.all():
raise PermissionDenied() # or Http404
return obj
# def question_view(request):
# questions = Question.objects.all()
# return render(request, "sbas/question_view.html", {"questions": questions})
@login_required
def question_detail(request, pk):
question = get_object_or_404(Question, pk=pk)
return render(request, "sbas/question_detail.html", {"question": question})
def active_exams(request):
exams = Exam.objects.all()
print(exams)
active_exams = []
for exam in exams:
if exam.active:
active_exams.append(exam)
return render(request, "sbas/available_exam_list.html", {"exams": active_exams})
def exam_start(request, pk):
exam = get_object_or_404(Exam, pk=pk)
if not exam.active:
return exam_inactive(request, context={"exam": exam})
return render(
request,
"sbas/exam_start.html",
{"exam": exam, "valid_user": exam.check_logged_in_user(request)},
)
def exam_complete(request, pk, cid=None, passcode=None):
exam = get_object_or_404(Exam, pk=pk)
exam.check_user_can_take(cid, passcode, request.user)
cid_user_exam = exam.get_or_create_cid_user_exam(cid=cid, user_user=request.user)
cid_user_exam.complete_exam()
return HttpResponse("<div role='alert' class='alert alert-info'>Exam completed</div>")
def exam_take_overview(request, pk, cid=None, passcode=None):
exam = get_object_or_404(Exam, pk=pk)
if not exam.active:
return exam_inactive(request, context={"exam": exam})
exam.check_user_can_take(cid, passcode, request.user)
questions = exam.get_questions()
if cid is not None:
answers = UserAnswer.objects.filter(cid=cid, exam=exam)
else:
answers = UserAnswer.objects.filter(user=request.user, exam=exam)
answer_question_map = {}
for ans in answers:
answer_question_map[ans.question] = ans
question_answer_tuples = []
answer_count = 0
for q in questions:
if q in answer_question_map and answer_question_map[q].answer:
question_answer_tuples.append((q, answer_question_map[q]))
answer_count += 1
else:
question_answer_tuples.append((q, None))
cid_user_exam = exam.get_or_create_cid_user_exam(cid=cid, user_user=request.user)
return render(
request,
"sbas/exam_take_overview.html",
{
"exam": exam,
"cid": cid,
"question_answer_tuples": question_answer_tuples,
"answer_count": answer_count,
"exam_length": len(questions),
"passcode": passcode,
"cid_user_exam": cid_user_exam,
},
)
def exam_take(request, pk: int, sk: int, cid: int = None, passcode: str = None):
exam = get_object_or_404(Exam, pk=pk)
if not exam.active:
return exam_inactive(request, context={"exam": exam})
exam.check_user_can_take(cid, passcode, request.user)
cid_user_exam = exam.get_or_create_cid_user_exam(cid = cid, user_user=request.user)
question = exam.get_questions()[sk]
exam_length = len(exam.exam_questions.all())
pos = exam.get_question_index(question)
if cid is not None:
answer = question.cid_user_answers.filter(cid=cid, exam=exam).first()
else:
answer = question.cid_user_answers.filter(user=request.user, exam=exam).first()
previous = -1
if sk > 0:
previous = sk - 1
next = sk + 1
if sk == exam_length - 1:
next = False
if request.method == "POST":
if answer:
form = UserAnswerForm(request.POST, instance=answer)
else:
form = UserAnswerForm(request.POST)
if form.is_valid() and not exam.publish_results and not cid_user_exam.completed:
answer = form.save(commit=False)
if cid is not None:
answer.cid = cid
else:
answer.user = request.user
answer.question = question
answer.exam = exam
# answer.published_date = timezone.now()
answer.save()
cid_user_exam.end_time = timezone.now()
cid_user_exam.save()
if cid is not None:
kwargs = {"pk": pk, "cid": cid, "passcode": passcode}
take_url = "sbas:exam_take"
finish_url = "sbas:exam_take_overview"
else:
kwargs = {"pk": pk}
take_url = "sbas:exam_take_user"
finish_url = "sbas:exam_take_overview_user"
if "next" in request.POST:
if not next:
return HttpResponseBadRequest("Invalid request")
return redirect(take_url, sk=pos + 1, **kwargs)
elif "previous" in request.POST:
return redirect(take_url, sk=pos - 1, **kwargs)
elif "finish" in request.POST:
return redirect(finish_url, **kwargs)
elif "goto" in request.POST:
return redirect(take_url, sk=int(request.POST.get("goto")), **kwargs)
else:
form = UserAnswerForm(instance=answer)
saved_answer = False
if answer is not None:
saved_answer = answer.answer
return render(
request,
"sbas/exam_take.html",
{
"form": form,
"cid": cid,
"exam": exam,
"question": question,
"next": next,
"previous": previous,
"exam_length": exam_length,
"pos": pos,
"saved_answer": saved_answer,
"passcode": passcode,
"cid_user_exam": cid_user_exam,
},
)
GenericExamViews = ExamViews(Exam, Question, None, UserAnswer, "sbas", "sbas")
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"] = "sbas"
return context
class QuestionCreateBase(RevisionMixin, LoginRequiredMixin, CreateView):
model = Question
form_class = QuestionForm
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["answer_formset"] = AnswerFormSet(
# self.request.POST, self.request.FILES
# )
# else:
# 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)
response = super().form_valid(form)
# 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("sbas:question_clone", pk=self.object.pk)
class QuestionCreate(QuestionCreateBase):
# initial = {'laterality': AnatomyQuestion.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
class QuestionUpdate(UpdateQuestionMixin):
model = Question
form_class = QuestionForm
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)
response = super().form_valid(form)
return response
class QuestionClone(QuestionCreateBase):
"""Clones a existing 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"])
# We don't want to clone the exam details
# exams = old_object.exams.all().values_list("id", flat=True)
# initial_data["exams"] = list(exams)
return initial_data
class UserAnswerView(LoginRequiredMixin, DetailView):
model = UserAnswer
class UserAnswerTableView(LoginRequiredMixin, SingleTableMixin, FilterView):
model = UserAnswer
table_class = UserAnswerTable
template_name = "sbas/user_answer_question_view.html"
filterset_class = UserAnswerFilter
class UserAnswerDelete(SuperuserRequiredMixin, DeleteView):
model = UserAnswer
template_name = "user_answer_delete.html"
success_url = reverse_lazy("sbas:user_answer_table_view")
# @user_passes_test(lambda u: u.is_superuser)
# def user_answer_delete_multiple(request):
# 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:
# UserAnswer.objects.get(pk=id).delete()
# return JsonResponse({"status": "success"})
GenericViews = GenericViewBase("sbas", Question, UserAnswer, Exam)
class QuestionDelete(AuthorOrCheckerRequiredMixin, DeleteView):
model = Question
success_url = reverse_lazy("sbas:question_view")
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 ExamClone(ExamCloneMixin, ExamCreate):
"""Clone exam view"""
def exam_clone2(request, exam_id):
exam = get_object_or_404(Exam, pk=exam_id)
new_exam = exam.clone_model()
return redirect("sbas:exam_update", pk=new_exam.id)
def llm_prompt_view(request):
return render(request, "sbas/llm_prompt_view.html", {})