from reversion.views import RevisionMixin
from generic.mixins import CheckCanEditMixin, SuperuserRequiredMixin
from django.views.generic.detail import DetailView
from generic.models import CidUser, ExamUserStatus, CidUserExam
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, HttpResponseForbidden
from django.http import HttpResponseRedirect, HttpResponse
from .models import Question, Category, Exam, UserAnswer
from django.contrib.contenttypes.models import ContentType
from generic.models import QuestionReview
from django.http import HttpResponseBadRequest
import random
from django.views.decorators.http import require_http_methods
from django.db import transaction
import re
import logging
from generic.models import Flag
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,
)
@login_required
def exam_review_question_responses(request, pk: int, q_index: int):
"""Return the responses partial for a given exam question (HTMX-friendly).
pk: exam pk
q_index: question index within the exam
"""
exam = get_object_or_404(Exam, pk=pk)
questions = exam.get_questions()
try:
question = questions[q_index]
except Exception:
raise Http404("Question not found in exam")
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"app_name": "sbas",
}
# Prefilter user answers for this exam to avoid template-side filtering
try:
exam_user_answers = question.cid_user_answers.filter(exam=exam).select_related("user")
except Exception:
exam_user_answers = question.cid_user_answers.all()
# Optionally reveal actual respondent identities when requested (HTMX or query param).
# Default behaviour is anonymous listing; when `reveal=1` is present we'll include
# the real user names in the fragment.
reveal_flag = str(request.GET.get("reveal", "")).lower() in ("1", "true", "on")
context["reveal_respondents"] = reveal_flag
context["exam_user_answers"] = exam_user_answers
return render(request, "sbas/partials/exam_review_question_responses_fragment.html", context)
@login_required
def exam_review_question_summary(request, pk: int, q_index: int):
"""Return the aggregated response summary partial for a given exam question (HTMX-friendly).
Computes counts and percentages for choices a..e and returns the summary partial.
"""
exam = get_object_or_404(Exam, pk=pk)
questions = exam.get_questions()
try:
question = questions[q_index]
except Exception:
raise Http404("Question not found in exam")
# Aggregate user answers for this question & exam
ua_qs = question.cid_user_answers.filter(exam=exam)
total_responses = ua_qs.count()
# Track counts for the five canonical choices and aggregate any
# empty / unexpected answers into an 'unanswered' bucket so the
# template can show them explicitly and percentages use the same
# denominator (total_responses).
counts = {"a": 0, "b": 0, "c": 0, "d": 0, "e": 0}
other_count = 0
for ans in ua_qs.values_list("answer", flat=True):
# Normalize None/empty strings to be considered 'unanswered'
if not ans:
other_count += 1
continue
if ans in counts:
counts[ans] += 1
else:
# Unexpected answer labels — treat as 'other/unanswered'
other_count += 1
# Expose 'unanswered' so templates can report it
counts['unanswered'] = other_count
pcts = {}
if total_responses:
for k, v in counts.items():
pcts[k] = round(100.0 * v / total_responses, 1)
else:
for k in counts.keys():
pcts[k] = 0.0
correct_count = None
correct_pct = None
if hasattr(question, "best_answer") and question.best_answer:
correct_count = ua_qs.filter(answer=question.best_answer).count()
if total_responses:
correct_pct = round(100.0 * correct_count / total_responses, 1)
# Provide the actual answer text for each canonical choice so the
# summary template can display the full answer alongside the letter.
exam_response_texts = {
"a": (getattr(question, "a_answer", None) or "") ,
"b": (getattr(question, "b_answer", None) or "") ,
"c": (getattr(question, "c_answer", None) or "") ,
"d": (getattr(question, "d_answer", None) or "") ,
"e": (getattr(question, "e_answer", None) or "") ,
}
context = {
"exam": exam,
"question": question,
"q_index": q_index,
"app_name": "sbas",
"exam_response_counts": counts,
"exam_response_pcts": pcts,
"exam_response_total": total_responses,
"exam_response_correct_count": correct_count,
"exam_response_correct_pct": correct_pct,
"exam_response_texts": exam_response_texts,
}
return render(request, "sbas/partials/exam_review_question_summary_fragment.html", context)
class AuthorOrCheckerRequiredMixin(object):
def get_object(self, *args, **kwargs):
obj = super().get_object(*args, **kwargs)
if self.request.user.is_superuser:
return obj
if self.request.user.groups.filter(name="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})
@login_required
def active_exams(request):
"""List all active SBA exams (candidate-facing).
Scope: authenticated users only.
Functionality: returns all non-archived active exams as a list for candidates
to take.
"""
exams = Exam.objects.all()
active_exams = []
for exam in exams:
if exam.active:
active_exams.append(exam)
return render(request, "sbas/available_exam_list.html", {"exams": active_exams})
@login_required
def question_search_page(request):
"""Render the SBAs search page which contains the HTMX search input.
The actual results are returned by `question_search_partial` and swapped
into the results container via HTMX.
"""
return render(request, "sbas/sbas_search.html", {})
@login_required
def question_search_partial(request):
"""Return an HTMX partial listing questions matching `q`.
Searches several Question fields and related names. Uses `distinct()` to
avoid duplicates from joins. The template expects `questions` in context.
"""
q = request.GET.get("q", "") or request.GET.get("sbas-question-search-input", "")
q = q.strip()
qs = Question.objects.all()
if q:
from django.db.models import Q
qs = qs.filter(
Q(title__icontains=q)
| Q(stem__icontains=q)
| Q(a_answer__icontains=q)
| Q(b_answer__icontains=q)
| Q(c_answer__icontains=q)
| Q(d_answer__icontains=q)
| Q(e_answer__icontains=q)
| Q(category__category__icontains=q)
| Q(author__first_name__icontains=q)
| Q(author__last_name__icontains=q)
| Q(author__username__icontains=q)
| Q(condition__name__icontains=q)
| Q(presentation__name__icontains=q)
| Q(subspecialty__name__icontains=q)
).distinct()
# Prefetch related data used in the partial to avoid N+1 queries
qs = qs.select_related("category").prefetch_related("author", "condition", "presentation", "subspecialty")[:50]
return render(request, "sbas/partials/_question_search_results.html", {"questions": qs})
def exam_start(request, pk):
exam = get_object_or_404(Exam, pk=pk)
if not exam.active:
return exam_inactive(request, context={"exam": exam})
# Check if pre-survey is required and not yet taken
from survey.views import check_survey_interception
response = check_survey_interception(request, exam)
if response:
return response
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()
try:
ct = ContentType.objects.get_for_model(exam)
ExamUserStatus.objects.create(
content_type=ct,
object_id=exam.pk,
cid_user_exam=cid_user_exam,
status="completed",
extra="sbas",
)
except Exception:
pass
return HttpResponse("Exam completed
")
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)
# Check if pre-survey is required and not yet taken
from survey.views import check_survey_interception
response = check_survey_interception(request, exam, cid, passcode)
if response:
return response
can_edit = exam.check_user_can_edit(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)
# Check if post-survey is mandatory and not yet taken
if exam.post_survey and getattr(exam, "post_survey_mandatory", False) and cid_user_exam.completed:
from survey.models import SurveyResponse
user_param = request.user if request.user.is_authenticated else None
cid_param = int(cid) if cid and str(cid).isdigit() else None
ct = ContentType.objects.get_for_model(exam)
has_responded = SurveyResponse.objects.filter(
survey=exam.post_survey,
user=user_param,
cid=cid_param,
content_type=ct,
object_id=exam.pk,
pre_or_post=SurveyResponse.SurveyContext.POST
).exists()
if not has_responded:
from django.urls import reverse
survey_url = reverse("survey:survey_take", kwargs={"pk": exam.post_survey.pk})
params = f"?pre_or_post=POST&content_type_id={ct.id}&object_id={exam.pk}&next={request.get_full_path()}&mandatory=true"
if cid and passcode:
params += f"&cid={cid}&passcode={passcode}"
return redirect(f"{survey_url}{params}")
post_survey_url = None
if exam.post_survey and not getattr(exam, "post_survey_mandatory", False):
from survey.models import SurveyResponse
user_param = request.user if request.user.is_authenticated else None
cid_param = int(cid) if cid and str(cid).isdigit() else None
ct = ContentType.objects.get_for_model(exam)
has_responded = SurveyResponse.objects.filter(
survey=exam.post_survey,
user=user_param,
cid=cid_param,
content_type=ct,
object_id=exam.pk,
pre_or_post=SurveyResponse.SurveyContext.POST
).exists()
if not has_responded:
session_key = f"skipped_post_survey_{exam.pk}"
if not request.session.get(session_key):
from django.urls import reverse
survey_url = reverse("survey:survey_take", kwargs={"pk": exam.post_survey.pk})
params = f"?pre_or_post=POST&content_type_id={ct.id}&object_id={exam.pk}&next={request.get_full_path()}&mandatory=false&skip_key={session_key}"
if cid and passcode:
params += f"&cid={cid}&passcode={passcode}"
post_survey_url = f"{survey_url}{params}"
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,
"can_edit": can_edit,
"post_survey_url": post_survey_url,
},
)
def exam_take(request, pk: int, sk: int, cid: int = None, passcode: str = None):
logger.debug(f"Taking exam pk={pk} sk={sk} cid={cid} passcode={passcode} user={request.user}")
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)
# Check if pre-survey is required and not yet taken
from survey.views import check_survey_interception
response = check_survey_interception(request, exam, cid, passcode)
if response:
return response
# log start only when a new CidUserExam is created
content_type = ContentType.objects.get_for_model(exam)
existing = False
if cid is not None:
cid_user_obj = CidUser.objects.filter(cid=cid).first()
existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, cid_user=cid_user_obj).exists()
else:
existing = CidUserExam.objects.filter(content_type=content_type, object_id=exam.pk, user_user=request.user).exists()
cid_user_exam = exam.get_or_create_cid_user_exam(cid = cid, user_user=request.user)
if not existing:
try:
ExamUserStatus.objects.create(
content_type=content_type,
object_id=exam.pk,
cid_user_exam=cid_user_exam,
status="started",
extra="sbas",
)
except Exception:
pass
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"
# HTMX detection: header 'HX-Request' or META 'HTTP_HX_REQUEST'
is_htmx = request.headers.get("HX-Request") == "true" or request.META.get("HTTP_HX_REQUEST")
if "next" in request.POST:
if not next:
return HttpResponseBadRequest("Invalid request")
if is_htmx:
return exam_take_fragment(request, pk=pk, sk=pos + 1, cid=cid, passcode=passcode)
return redirect(take_url, sk=pos + 1, **kwargs)
elif "previous" in request.POST:
if is_htmx:
return exam_take_fragment(request, pk=pk, sk=pos - 1, cid=cid, passcode=passcode)
return redirect(take_url, sk=pos - 1, **kwargs)
elif "finish" in request.POST:
if is_htmx:
try:
url = reverse(finish_url, kwargs=kwargs)
except Exception:
resp = HttpResponse()
resp["HX-Redirect"] = ""
return resp
resp = HttpResponse()
resp["HX-Redirect"] = url
return resp
return redirect(finish_url, **kwargs)
elif "save" in request.POST:
if is_htmx:
resp = exam_take_fragment(request, pk=pk, sk=pos, cid=cid, passcode=passcode)
try:
resp['HX-Trigger'] = json.dumps({'saved': True})
except Exception:
resp['HX-Trigger'] = 'saved'
return resp
return redirect(take_url, sk=pos, **kwargs)
elif "goto" in request.POST:
dest = int(request.POST.get("goto"))
if is_htmx:
return exam_take_fragment(request, pk=pk, sk=dest, cid=cid, passcode=passcode)
return redirect(take_url, sk=dest, **kwargs)
if is_htmx:
return exam_take_fragment(request, pk=pk, sk=pos, cid=cid, passcode=passcode)
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,
},
)
def exam_take_fragment(request, pk: int, sk: int, cid: int = None, passcode: str = None):
"""Return an HTMX partial for SBAs question form and controls."""
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)
questions = list(exam.get_questions())
try:
index = int(sk)
except Exception:
raise Http404("Invalid question index")
if index < 0 or index >= len(questions):
raise Http404("Question not found in exam")
question = questions[index]
exam_length = len(questions)
pos = index
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()
if answer is not None:
form = UserAnswerForm(instance=answer)
saved_answer = answer.answer
else:
form = UserAnswerForm()
saved_answer = None
# compute flagged state
try:
ct = ContentType.objects.get_for_model(question)
flags_qs = Flag.objects.filter(content_type=ct, object_id=question.pk)
if cid is not None:
flags_qs = flags_qs.filter(cid_user__cid=cid)
else:
flags_qs = flags_qs.filter(user=request.user)
flagged = flags_qs.exists()
except Exception:
flagged = False
previous = -1
if sk > 0:
previous = sk - 1
next = sk + 1
if sk == exam_length - 1:
next = False
# bulk answered/flagged sets for menu
q_ids = [q.pk for q in questions]
if cid is not None:
frag_answered_set = set(UserAnswer.objects.filter(cid=cid, exam=exam, question_id__in=q_ids).values_list('question_id', flat=True))
frag_flagged_set = set(Flag.objects.filter(content_type=ContentType.objects.get_for_model(Question), object_id__in=q_ids, cid_user__cid=cid).values_list('object_id', flat=True))
else:
frag_answered_set = set(UserAnswer.objects.filter(user=request.user, exam=exam, question_id__in=q_ids).values_list('question_id', flat=True))
frag_flagged_set = set(Flag.objects.filter(content_type=ContentType.objects.get_for_model(Question), object_id__in=q_ids, user=request.user).values_list('object_id', flat=True))
return render(request, "sbas/partials/exam_take_fragment.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,
"flagged": flagged,
"answered_json": json.dumps([q.pk in frag_answered_set for q in questions]),
"flagged_json": json.dumps([q.pk in frag_flagged_set for q in questions]),
})
def exam_toggle_flag(request, pk: int, sk: int, cid: str | None = None, passcode: str | None = None):
"""Toggle the flagged state for the current user's answer to a question.
Returns the small flag-button partial so HTMX clients can swap it in-place.
"""
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)
# canonical questions list
questions = list(exam.get_questions())
try:
index = int(sk)
except Exception:
raise Http404("Invalid question index")
if index < 0 or index >= len(questions):
raise Http404("Question not found in exam")
question = questions[index]
# Determine actor: either a CidUser (for cid flows) or the logged-in user
cid_user_obj = None
if cid is not None:
try:
cid_user_obj = CidUser.objects.filter(cid=cid).first()
except Exception:
cid_user_obj = None
# Find existing flag for this question+actor
ct = ContentType.objects.get_for_model(question)
flags_qs = Flag.objects.filter(content_type=ct, object_id=question.pk)
if cid_user_obj is not None:
flags_qs = flags_qs.filter(cid_user=cid_user_obj)
else:
flags_qs = flags_qs.filter(user=request.user)
flagged = flags_qs.exists()
if request.method == "POST":
# set param explicitly controls state; otherwise toggle
set_val = request.POST.get("set")
if set_val is None:
# toggle
if flagged:
flags_qs.delete()
flagged = False
else:
Flag.objects.create(content_type=ct, object_id=question.pk, user=(None if cid_user_obj else request.user), cid_user=cid_user_obj)
flagged = True
else:
desired = str(set_val).lower() in ("1", "true", "yes")
if desired and not flagged:
Flag.objects.create(content_type=ct, object_id=question.pk, user=(None if cid_user_obj else request.user), cid_user=cid_user_obj)
flagged = True
elif not desired and flagged:
flags_qs.delete()
flagged = False
# Render the partial button for the current state
return render(request, "sbas/partials/exam_flag_button.html", {"flagged": flagged, "exam": exam, "pos": index, "cid": cid, "passcode": passcode})
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
@login_required
def create_personal_exam_with_name(request):
"""Create a personal SBA exam from filters, using a provided name.
POST parameters expected:
- name (optional): exam name to use
- category (optional)
- status (optional)
- count (optional): number of questions to include (0 or missing = all)
"""
if request.method != "POST":
return HttpResponseBadRequest("POST required")
name = request.POST.get("name") or None
# reuse existing create_personal_exam filtering logic but allow custom name
category = request.POST.get("category") or None
# For generated personal exams we only include accepted questions
status = QuestionReview.StatusChoices.ACCEPTED
# Optionally exclude questions the current user has already answered
exclude_answered = request.POST.get("exclude_answered") in ("on", "true", "True", "1")
try:
count = int(request.POST.get("count") or 0)
if count < 0:
count = 0
except Exception:
count = 0
qs = Question.objects.all().order_by("pk")
if category:
try:
qs = qs.filter(category__id=int(category))
except Exception:
pass
# If requested, remove questions the current user has already answered
if exclude_answered:
try:
answered_qs = UserAnswer.objects.filter(user=request.user).values_list("question_id", flat=True)
qs = qs.exclude(pk__in=list(answered_qs))
except Exception:
pass
# Permission filter: non-checkers see open_access or their own questions
try:
if not request.user.groups.filter(name="sbas_checker").exists():
from django.db.models import Q
qs = qs.filter(Q(open_access=True) | Q(author__id=request.user.id))
except Exception:
pass
matches = []
for question in qs:
try:
if question.authors_only and not (
request.user.is_superuser or request.user in question.author.all()
):
continue
except Exception:
pass
ct = ContentType.objects.get_for_model(question)
latest = (
QuestionReview.objects.filter(content_type=ct, object_id=question.pk)
.order_by("-created_on").first()
)
match = False
if status == "ANY":
match = True
elif status == "UNREVIEWED":
match = latest is None
else:
if latest is not None and latest.status == status:
match = True
if match:
matches.append(question)
if not matches:
return HttpResponseBadRequest("No matching questions found")
# Optionally randomize and limit
if count and count > 0 and count < len(matches):
random.shuffle(matches)
matches = matches[:count]
# Create the Exam
if name is None or name.strip() == "":
name = f"{request.user.get_full_name() or request.user.username} - Personal SBA Exam"
exam = Exam(name=name, active=True, candidates_only=True, exam_mode=True)
exam.save()
# Add author and allow this user to take the exam
try:
exam.author.add(request.user)
except Exception:
pass
try:
exam.valid_user_users.add(request.user)
except Exception:
pass
# Add questions
for q in matches:
exam.exam_questions.add(q)
exam.order_questions()
return redirect("sbas:exam_start", pk=exam.pk)
@login_required
def generate_exam(request):
"""Render a small form allowing the user to name and generate a personal exam from filters."""
# Prepare categories and status options for the template
categories = Category.objects.all().order_by("category")
# Build status options: ANY, UNREVIEWED, plus choices from QuestionReview
status_options = [("ANY", "Any")]
status_options.append(("UNREVIEWED", "Unreviewed"))
for v, label in QuestionReview.StatusChoices.choices:
status_options.append((v, label))
return render(request, "sbas/generate_exam.html", {"categories": categories, "status_options": status_options})
@login_required
def question_overview(request):
"""Show counts of questions grouped by category (default) with simple filters.
Default behaviour:
- group_by=category
- status=AC (Accepted)
Available GET params:
- status: AC, OD, ER, RJ, IP, ANY, UNREVIEWED
- category: category id to filter (optional)
- open_access: '1' or '0' to filter
"""
# Prepare status options as elsewhere
status_options = [("ANY", "Any"), ("UNREVIEWED", "Unreviewed")]
for v, label in QuestionReview.StatusChoices.choices:
status_options.append((v, label))
status = request.GET.get("status") or QuestionReview.StatusChoices.ACCEPTED
group_by = request.GET.get("group_by") or "category"
# Use django-filter to build the base filtered queryset. QuestionFilter
# already applies permission filters when given the request.
from .filters import QuestionFilter
qfilter = QuestionFilter(data=request.GET or None, queryset=Question.objects.all().order_by("pk"), request=request)
qs = qfilter.qs
# We'll still support a top-level status filter (review status) via a GET param
# and advanced atlas filters are provided by the QuestionFilter form.
# Preserve selected values for building links (use getlist for multi-selects)
category_filters = request.GET.getlist("category")
finding_filters = request.GET.getlist("finding")
structure_filters = request.GET.getlist("structure")
condition_filters = request.GET.getlist("condition")
subspecialty_filters = request.GET.getlist("subspecialty")
presentation_filters = request.GET.getlist("presentation")
open_access = request.GET.get("open_access")
# Optimize status filtering with a DB subquery to get latest review status per question
from django.contrib.contenttypes.models import ContentType
from django.db.models import OuterRef, Subquery, Count
ct = ContentType.objects.get_for_model(Question)
latest_status_subq = QuestionReview.objects.filter(content_type=ct, object_id=OuterRef("pk")).order_by("-created_on").values("status")[:1]
qs = qs.annotate(latest_review_status=Subquery(latest_status_subq))
if status == "ANY":
pass
elif status == "UNREVIEWED":
qs = qs.filter(latest_review_status__isnull=True)
else:
qs = qs.filter(latest_review_status=status)
# Respect authors_only: non-owners should be excluded
# We applied permission filter earlier; ensure authors_only exclusion
from django.db.models import Q
try:
if not request.user.is_superuser:
qs = qs.exclude(Q(authors_only=True) & ~Q(author__in=[request.user]))
except Exception:
pass
# Aggregate counts grouped by category (DB-side)
grouped = qs.values("category__id", "category__category").annotate(count=Count("pk")).order_by("-count", "category__category")
counts_rows = []
total = qs.count()
# Build per-row link (to question list) preserving current filters but swapping category
from urllib.parse import urlencode
from django.urls import reverse
# Convert selected filters to strings for template membership tests
selected_category = [str(x) for x in category_filters]
selected_findings = [str(x) for x in finding_filters]
selected_structures = [str(x) for x in structure_filters]
selected_conditions = [str(x) for x in condition_filters]
selected_subspecialties = [str(x) for x in subspecialty_filters]
selected_presentations = [str(x) for x in presentation_filters]
for row in grouped:
cat_id = row.get("category__id")
cat_name = row.get("category__category") or "Uncategorized"
count = row.get("count")
params = []
# category param points to this row
if cat_id:
params.append(("category", str(cat_id)))
# preserve other filters
if status:
params.append(("status", status))
if open_access is not None and open_access != "":
params.append(("open_access", open_access))
for fid in selected_findings:
params.append(("finding", fid))
for sid in selected_structures:
params.append(("structure", sid))
for cid in selected_conditions:
params.append(("condition", cid))
for ss in selected_subspecialties:
params.append(("subspecialty", ss))
for p in selected_presentations:
params.append(("presentation", p))
link = reverse("sbas:question_view") + ("?" + urlencode(params, doseq=True) if params else "")
counts_rows.append({"cat_id": cat_id or "", "cat_name": cat_name, "count": count, "link": link})
# Populate atlas filter lists
from atlas.models import Finding, Structure, Condition as AtlasCondition, Subspecialty, Presentation as AtlasPresentation
categories = Category.objects.all().order_by("category")
findings = Finding.objects.all().order_by("name")
structures = Structure.objects.all().order_by("name")
conditions = AtlasCondition.objects.all().order_by("name")
subspecialties = Subspecialty.objects.all().order_by("name")
presentations = AtlasPresentation.objects.all().order_by("name")
context = {
"status_options": status_options,
"selected_status": status,
"group_by": group_by,
"counts": counts_rows,
"total": total,
"categories": categories,
"selected_category": selected_category,
"selected_open_access": open_access,
"findings": findings,
"structures": structures,
"conditions": conditions,
"subspecialties": subspecialties,
"presentations": presentations,
"selected_findings": selected_findings,
"selected_structures": selected_structures,
"selected_conditions": selected_conditions,
"selected_subspecialties": selected_subspecialties,
"selected_presentations": selected_presentations,
"qfilter": qfilter,
}
return render(request, "sbas/question_overview.html", context)
@login_required
@require_http_methods(["POST", "GET"])
def toggle_frcr_appropriate(request, pk):
"""Toggle or set the Question.frcr_appropriate boolean via HTMX.
Accepts POST requests. If POST contains 'frcr' parameter its truthiness
will be applied; otherwise the value is toggled.
Returns the rendered partial `_frcr_toggle.html` so HTMX can swap it in.
"""
question = get_object_or_404(Question, pk=pk)
# Simple permission: allow staff or users with change permission
if question.is_author(request.user):
pass
elif request.user.groups.filter(name="sba_reviewer").exists():
pass
elif not (request.user.is_staff or request.user.has_perm('sbas.change_question')):
return HttpResponse("Forbidden")
# Toggle or set
if request.method == "POST":
val = request.POST.get("frcr")
if val is None:
question.frcr_appropriate = not question.frcr_appropriate
else:
question.frcr_appropriate = str(val).lower() in ("1", "true", "yes", "on")
question.save()
return render(request, "sbas/_frcr_toggle.html", {"question": question})
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(["GET", "POST"])
def merge_category(request):
"""Admin view: merge one Category into another.
GET: render form to choose source and target categories.
POST: expects 'source' and 'target' (category PKs) and optional 'delete_source' checkbox.
The view will reassign all Question.category==source -> target and optionally delete the source Category.
"""
categories = Category.objects.all().order_by('category')
if request.method == 'GET':
return render(request, 'sbas/merge_category.html', {'categories': categories})
# POST
source_id = request.POST.get('source')
target_id = request.POST.get('target')
delete_source = request.POST.get('delete_source') in ('on', 'true', '1', 'True')
if not source_id or not target_id:
return render(request, 'sbas/merge_category.html', {'categories': categories, 'error': 'Both source and target categories must be selected.'})
try:
source = Category.objects.get(pk=int(source_id))
target = Category.objects.get(pk=int(target_id))
except Category.DoesNotExist:
return render(request, 'sbas/merge_category.html', {'categories': categories, 'error': 'Invalid category selected.'})
if source.pk == target.pk:
return render(request, 'sbas/merge_category.html', {'categories': categories, 'error': 'Source and target must be different.'})
# perform update inside a transaction
from django.db import transaction
with transaction.atomic():
qs = Question.objects.filter(category=source)
count = qs.count()
qs.update(category=target)
deleted = False
if delete_source:
# safe to delete now (no FK references from Question)
source.delete()
deleted = True
msg = f"Moved {count} question{'s' if count != 1 else ''} from '{source}' to '{target}'."
if deleted:
msg += f" Deleted source category '{source}'."
return render(request, 'sbas/merge_category.html', {'categories': Category.objects.all().order_by('category'), 'success': msg})
@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_="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__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)})