Ross
1048 lines
41 KiB
Python
1048 lines
41 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__)
|
|
import uuid
|
|
|
|
from loguru import logger
|
|
|
|
try:
|
|
import jsonschema
|
|
except Exception:
|
|
jsonschema = None
|
|
|
|
|
|
|
|
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", {})
|
|
|
|
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",
|
|
)
|
|
dry_run = forms.BooleanField(required=False, initial=True, help_text="Validate and preview only; do not save to database")
|
|
allow_create_m2m = forms.BooleanField(required=False, initial=True, help_text="Automatically create missing many-to-many items when importing (if not dry-run)")
|
|
|
|
|
|
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 []
|
|
|
|
def _strip_numeric_refs_in_payload(p):
|
|
"""Return a copy of payload with inline numeric refs removed from any '*_feedback' or 'feedback' fields.
|
|
Also return a list of keys that were modified.
|
|
"""
|
|
if not isinstance(p, dict):
|
|
return p, []
|
|
cleaned = dict(p)
|
|
changed = []
|
|
# pattern matches [1], [1,2,3], or ranges like [6-9], or mixtures like [1,3-5,7]
|
|
pattern = re.compile(r"\[\s*(?:\d+(?:\s*-\s*\d+)?)(?:\s*,\s*(?:\d+(?:\s*-\s*\d+)?))*\s*\]")
|
|
for k, v in list(p.items()):
|
|
if not isinstance(v, str):
|
|
continue
|
|
if k.endswith("_feedback") or k == "feedback":
|
|
newv = pattern.sub("", v).strip()
|
|
# collapse multiple spaces
|
|
newv = re.sub(r"\s{2,}", " ", newv)
|
|
if newv != v:
|
|
cleaned[k] = newv
|
|
changed.append(k)
|
|
return cleaned, changed
|
|
|
|
@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},
|
|
"sources": {"type": "array", "items": {"type": "string"}},
|
|
},
|
|
}
|
|
|
|
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)
|
|
|
|
dry_run = bool(form.cleaned_data.get("dry_run"))
|
|
allow_create_m2m = bool(form.cleaned_data.get("allow_create_m2m"))
|
|
|
|
report = {
|
|
"total": len(candidates),
|
|
"created": 0,
|
|
"errors": [],
|
|
"sanitized_backslashes": sanitized_count,
|
|
"per_item": [],
|
|
"would_create_m2m": {},
|
|
"actually_created_m2m": {},
|
|
}
|
|
|
|
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
|
|
|
|
# Aggregators for M2M creation suggestions
|
|
aggregated_would_create = defaultdict(set)
|
|
aggregated_actually_created = defaultdict(list)
|
|
|
|
for idx, payload in enumerate(candidates):
|
|
item_report = {"index": idx, "status": None, "errors": [], "missing_m2m": [], "resolved_m2m": {}, "payload": payload}
|
|
|
|
# Strip numeric bracket references from feedback fields (e.g., [1], [1,2]) and record which fields were changed
|
|
cleaned_payload, stripped = _strip_numeric_refs_in_payload(payload)
|
|
if stripped:
|
|
item_report['stripped_fields'] = stripped
|
|
payload = cleaned_payload
|
|
|
|
# Collect validation errors
|
|
validation_errors = list(validator.iter_errors(payload))
|
|
if validation_errors:
|
|
# Separate additionalProperties errors from others
|
|
addl_errors = [e for e in validation_errors if getattr(e, 'validator', '') == 'additionalProperties']
|
|
other_errors = [e for e in validation_errors if getattr(e, 'validator', '') != 'additionalProperties']
|
|
|
|
if addl_errors and not other_errors:
|
|
# If the only problems are extra fields, drop them and re-validate the cleaned payload
|
|
allowed_keys = set(schema.get('properties', {}).keys())
|
|
extra_keys = [k for k in payload.keys() if k not in allowed_keys]
|
|
if extra_keys:
|
|
logger.debug("Validation: item %s has extra keys %s, dropping them for import", idx, extra_keys)
|
|
# Record dropped fields in the report for user visibility
|
|
item_report.setdefault('dropped_fields', []).extend(extra_keys)
|
|
# Create a cleaned payload without extra keys and use it from here on
|
|
cleaned_payload = {k: v for k, v in payload.items() if k in allowed_keys}
|
|
# Re-run validation on cleaned payload to be safe
|
|
cleaned_errors = list(validator.iter_errors(cleaned_payload))
|
|
if cleaned_errors:
|
|
# If cleaning still leaves errors, mark invalid
|
|
item_report['errors'] = [e.message for e in cleaned_errors]
|
|
logger.debug("Validation errors after cleaning for item %s: %s", idx, item_report['errors'])
|
|
report['errors'].append({"index": idx, "errors": item_report['errors']})
|
|
item_report['status'] = 'invalid'
|
|
report['per_item'].append(item_report)
|
|
continue
|
|
else:
|
|
# Replace payload used downstream with cleaned version
|
|
payload = cleaned_payload
|
|
item_report['status'] = 'cleaned'
|
|
# do not append here; allow the normal processing path to append the full item_report later
|
|
else:
|
|
# no extra keys found, treat as invalid to be safe
|
|
item_report['errors'] = [e.message for e in validation_errors]
|
|
logger.debug("Validation errors for item %s: %s", idx, item_report['errors'])
|
|
report['errors'].append({"index": idx, "errors": item_report['errors']})
|
|
item_report['status'] = 'invalid'
|
|
report['per_item'].append(item_report)
|
|
continue
|
|
else:
|
|
# Other validation problems exist; report and skip
|
|
item_report['errors'] = [e.message for e in validation_errors]
|
|
logger.debug("Validation errors for item %s: %s", idx, item_report['errors'])
|
|
report['errors'].append({"index": idx, "errors": item_report['errors']})
|
|
item_report['status'] = 'invalid'
|
|
report['per_item'].append(item_report)
|
|
continue
|
|
|
|
try:
|
|
# Prepare category resolution
|
|
cat_val = payload.get("category")
|
|
cat_obj = None
|
|
if isinstance(cat_val, str):
|
|
if not dry_run:
|
|
cat_obj, _ = Category.objects.get_or_create(category=cat_val)
|
|
else:
|
|
# dry-run preview: show that category would be used/created
|
|
item_report.setdefault("category_preview", cat_val)
|
|
|
|
# M2M resolution and missing detection
|
|
m2m_map = {}
|
|
for key, model_cls in (
|
|
("finding", Finding),
|
|
("structure", Structure),
|
|
("condition", Condition),
|
|
("presentation", Presentation),
|
|
("subspecialty", Subspecialty),
|
|
):
|
|
vals = payload.get(key) or []
|
|
resolved = []
|
|
missing = []
|
|
for v in vals:
|
|
if isinstance(v, int):
|
|
try:
|
|
resolved.append(model_cls.objects.get(pk=v))
|
|
except model_cls.DoesNotExist:
|
|
missing.append(v)
|
|
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]))
|
|
else:
|
|
missing.append(v)
|
|
|
|
m2m_map[key] = {"resolved": resolved, "missing": missing}
|
|
item_report["resolved_m2m"][key] = [o.name for o in resolved]
|
|
# Always include an entry for this M2M (values may be empty)
|
|
item_report["missing_m2m"].append({"model": model_cls.__name__, "values": missing})
|
|
if missing:
|
|
for v in missing:
|
|
aggregated_would_create[model_cls.__name__].add(v)
|
|
|
|
# Build a simple map of missing values per model (stringified) for robust template rendering
|
|
# Provide both lowercase and capitalized keys so templates referencing either form will work
|
|
# missing_map keys are lowercase (matching m2m_map keys)
|
|
missing_map = {k: [str(x) for x in v["missing"]] for k, v in m2m_map.items()}
|
|
item_report["missing_map"] = missing_map
|
|
|
|
# Debug: log missing_map and resolved_m2m for troubleshooting
|
|
logger.debug(
|
|
"import_llm_questions: item %s missing_map_keys=%s resolved_keys=%s",
|
|
idx,
|
|
list(missing_map.keys()),
|
|
list((item_report.get("resolved_m2m") or {}).keys()),
|
|
)
|
|
|
|
# Ensure the item report payload reflects any cleaning (dropped fields / stripped refs)
|
|
item_report['payload'] = payload
|
|
|
|
# If dry-run, do not save; just report what would happen
|
|
if dry_run:
|
|
item_report["status"] = "would_create"
|
|
report["per_item"].append(item_report)
|
|
continue
|
|
|
|
# Not dry-run: create Question and resolve/possibly create missing M2M
|
|
with transaction.atomic():
|
|
q = Question()
|
|
q.stem = payload.get("stem", "").strip()
|
|
q.title = payload.get("title")
|
|
|
|
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")
|
|
if cat_obj:
|
|
q.category = cat_obj
|
|
# populate sources JSONField if provided (accept list or single string)
|
|
srcs = payload.get("sources")
|
|
if isinstance(srcs, list):
|
|
q.sources = srcs
|
|
elif srcs:
|
|
q.sources = [str(srcs)]
|
|
q.save()
|
|
# Add current user as author of the imported question
|
|
try:
|
|
q.author.add(request.user)
|
|
except Exception:
|
|
logger.exception("Failed to add author for imported question idx=%s", idx)
|
|
|
|
# For each M2M, create missing items if allowed and attach resolved
|
|
for key, model_cls in (
|
|
("finding", Finding),
|
|
("structure", Structure),
|
|
("condition", Condition),
|
|
("presentation", Presentation),
|
|
("subspecialty", Subspecialty),
|
|
):
|
|
resolved = m2m_map[key]["resolved"]
|
|
missing_vals = m2m_map[key]["missing"]
|
|
# create missing if allowed
|
|
created_objs = []
|
|
if missing_vals and allow_create_m2m:
|
|
for v in missing_vals:
|
|
# create with name=v (use get_or_create to avoid races)
|
|
obj, created = model_cls.objects.get_or_create(name=v)
|
|
created_objs.append(obj)
|
|
aggregated_actually_created[model_cls.__name__].append(obj.pk)
|
|
|
|
# final list to attach
|
|
final_objs = resolved + created_objs
|
|
if final_objs:
|
|
getattr(q, key).add(*[o.pk for o in final_objs])
|
|
|
|
report["created"] += 1
|
|
item_report["status"] = "created"
|
|
item_report["created_pk"] = q.pk
|
|
report["per_item"].append(item_report)
|
|
|
|
except Exception as e:
|
|
logger.exception("Failed to import item %s", idx)
|
|
item_report["status"] = "error"
|
|
item_report["errors"].append(str(e))
|
|
report["errors"].append({"index": idx, "errors": [str(e)]})
|
|
report["per_item"].append(item_report)
|
|
|
|
# Flatten aggregated_would_create sets to lists
|
|
for k, v in aggregated_would_create.items():
|
|
report["would_create_m2m"][k] = sorted(list(v))
|
|
for k, v in aggregated_actually_created.items():
|
|
report["actually_created_m2m"][k] = v
|
|
|
|
# If this was a dry-run and the request came from HTMX, render a preview partial and store the parsed candidates in session
|
|
if dry_run and request.headers.get("HX-Request"):
|
|
session_key = f"llm_import_{uuid.uuid4().hex}"
|
|
# store minimal necessary data in session (the raw payloads)
|
|
request.session[session_key] = candidates
|
|
request.session.modified = True
|
|
preview_ctx = {
|
|
"report": report,
|
|
"items": report["per_item"],
|
|
"payloads": candidates,
|
|
"session_key": session_key,
|
|
"allow_create_m2m": allow_create_m2m,
|
|
}
|
|
return render(request, "sbas/partials/import_preview.html", preview_ctx)
|
|
|
|
# Non-HTMX callers get JSON
|
|
return JsonResponse({"ok": True, "report": report})
|
|
|
|
|
|
@login_required
|
|
@user_passes_test(lambda u: u.is_superuser)
|
|
@require_http_methods(["POST"])
|
|
def import_llm_confirm(request):
|
|
"""Confirm import of previously-previewed candidates stored in session.
|
|
|
|
Expects POST with 'session_key' and 'selected' (comma-separated indices) and optional allow_create_m2m.
|
|
Returns an HTML fragment suitable for HTMX replacement summarising the import results.
|
|
"""
|
|
session_key = request.POST.get("session_key")
|
|
# support both a comma-separated 'selected' string or multiple 'selected' values
|
|
selected_raw = request.POST.get("selected")
|
|
selected_list = request.POST.getlist("selected")
|
|
allow_create_m2m = request.POST.get("allow_create_m2m") in ("on", "true", "True", "1")
|
|
if selected_list:
|
|
# selected_list contains strings of indices
|
|
selected = ",".join(selected_list)
|
|
else:
|
|
selected = selected_raw
|
|
if not session_key or session_key not in request.session:
|
|
return render(request, "sbas/partials/import_error.html", {"errors": ["Session expired or invalid; please re-run preview."]}, status=400)
|
|
|
|
candidates = request.session.get(session_key, [])
|
|
indices = []
|
|
if selected:
|
|
for part in selected.split(','):
|
|
try:
|
|
indices.append(int(part))
|
|
except Exception:
|
|
continue
|
|
|
|
# default: import all if none specified
|
|
if not indices:
|
|
indices = list(range(len(candidates)))
|
|
|
|
# perform actual import for selected indices
|
|
from atlas.models import Finding, Structure, Condition, Presentation, Subspecialty
|
|
|
|
# Parse exclude inputs of the form exclude_<idx>="Model:::value" (may be multiple per idx)
|
|
excluded_map = defaultdict(lambda: defaultdict(set))
|
|
for k, vlist in request.POST.lists():
|
|
if not k.startswith("exclude_"):
|
|
continue
|
|
try:
|
|
idx_str = k.split("exclude_")[1]
|
|
idx_key = int(idx_str)
|
|
except Exception:
|
|
continue
|
|
for entry in vlist:
|
|
# entry format: Model:::value
|
|
try:
|
|
model_name, val = entry.split(":::", 1)
|
|
excluded_map[idx_key][model_name].add(val)
|
|
except Exception:
|
|
continue
|
|
# Debug: log exclusions parsed from POST
|
|
try:
|
|
# convert sets to lists for logging
|
|
log_map = {i: {m: list(vals) for m, vals in models.items()} for i, models in excluded_map.items()}
|
|
logger.debug("import_llm_confirm: excluded_map=%s", log_map)
|
|
except Exception:
|
|
logger.exception("Failed to log excluded_map")
|
|
|
|
created = 0
|
|
errors = []
|
|
actually_created = defaultdict(list)
|
|
for idx in indices:
|
|
try:
|
|
payload = candidates[idx]
|
|
except Exception:
|
|
errors.append({"index": idx, "errors": ["Missing candidate"]})
|
|
continue
|
|
# Overlay any edited fields posted from the preview form (names like payload_<idx>_stem)
|
|
overlay = {}
|
|
prefix = f"payload_{idx}_"
|
|
for k, v in request.POST.items():
|
|
if k.startswith(prefix):
|
|
field = k[len(prefix):]
|
|
overlay[field] = v
|
|
if overlay:
|
|
# apply overlay onto a shallow copy
|
|
new_payload = dict(payload)
|
|
for fk, fv in overlay.items():
|
|
# convert list-like fields and numeric conversions are not performed here; keep simple strings/lists
|
|
new_payload[fk] = fv
|
|
payload = new_payload
|
|
# Strip numeric bracket references here as well, before confirm import
|
|
payload, stripped = _strip_numeric_refs_in_payload(payload)
|
|
if stripped:
|
|
logger.debug("import_llm_confirm: stripped numeric refs for idx=%s fields=%s", idx, stripped)
|
|
try:
|
|
with transaction.atomic():
|
|
q = Question()
|
|
q.stem = payload.get("stem", "").strip()
|
|
q.title = payload.get("title")
|
|
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")
|
|
cat_val = payload.get("category")
|
|
if isinstance(cat_val, str):
|
|
cat_obj, _ = Category.objects.get_or_create(category=cat_val)
|
|
q.category = cat_obj
|
|
# populate sources JSONField if provided (accept list or single string)
|
|
srcs = payload.get("sources")
|
|
if isinstance(srcs, list):
|
|
q.sources = srcs
|
|
elif srcs:
|
|
q.sources = [str(srcs)]
|
|
q.save()
|
|
# Add current user as author of the imported question
|
|
try:
|
|
q.author.add(request.user)
|
|
except Exception:
|
|
logger.exception("Failed to add author for confirmed import idx=%s", idx)
|
|
|
|
for key, model_cls in (("finding", Finding), ("structure", Structure), ("condition", Condition), ("presentation", Presentation), ("subspecialty", Subspecialty)):
|
|
vals = payload.get(key) or []
|
|
final_objs = []
|
|
for v in vals:
|
|
if isinstance(v, int):
|
|
try:
|
|
final_objs.append(model_cls.objects.get(pk=v))
|
|
except model_cls.DoesNotExist:
|
|
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():
|
|
# add resolved objects unless excluded for this item
|
|
for obj in list(qs[:5]):
|
|
if obj.name in excluded_map.get(idx, {}).get(model_cls.__name__, set()):
|
|
logger.debug("Skipping resolved attach for idx=%s model=%s name=%s due to exclusion", idx, model_cls.__name__, obj.name)
|
|
continue
|
|
final_objs.append(obj)
|
|
else:
|
|
if allow_create_m2m:
|
|
# skip creation if excluded for this item
|
|
if v in excluded_map.get(idx, {}).get(model_cls.__name__, set()):
|
|
logger.debug("Skipping creation for idx=%s model=%s name=%s due to exclusion", idx, model_cls.__name__, v)
|
|
continue
|
|
obj, created_flag = model_cls.objects.get_or_create(name=v)
|
|
final_objs.append(obj)
|
|
actually_created[model_cls.__name__].append(obj.pk)
|
|
if final_objs:
|
|
getattr(q, key).add(*[o.pk for o in final_objs])
|
|
|
|
created += 1
|
|
except Exception as e:
|
|
errors.append({"index": idx, "errors": [str(e)]})
|
|
|
|
# cleanup session
|
|
try:
|
|
del request.session[session_key]
|
|
request.session.modified = True
|
|
except Exception:
|
|
pass
|
|
|
|
return render(request, "sbas/partials/import_result.html", {"created": created, "errors": errors, "actually_created": dict(actually_created)}) |