ross/penracourses
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Add LLM question import functionality with enhanced validation and M2M handling

Browse Source
This commit is contained in:
Ross
2025-10-20 10:47:11 +01:00
parent 93f0bb8927
commit 06b7239daa
1 changed files with 308 additions and 230 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sbas/views.py
+308 -230
View File
@@ -44,235 +44,6 @@ except Exception:
jsonschema = None
class LLMImportForm(forms.Form):
file = forms.FileField(required=False, help_text="Upload a JSON or JSONL file matching the SBA import schema")
raw = forms.CharField(
required=False,
widget=forms.Textarea(attrs={"rows": 10, "cols": 80}),
help_text="Or paste JSON, JSONL (newline-delimited), or multiple JSON documents separated by blank lines",
)
def _resolve_m2m(model_class, value):
"""Resolve an item which may be an int (pk) or a string (name/slug).
Returns a queryset or empty list of matching objects.
"""
if value is None:
return []
if isinstance(value, int):
try:
return [model_class.objects.get(pk=value)]
except model_class.DoesNotExist:
return []
if isinstance(value, str):
# try exact name field, then case-insensitive contains
qs = model_class.objects.filter(name__iexact=value)
if not qs.exists():
qs = model_class.objects.filter(name__icontains=value)
return list(qs[:5])
return []
@login_required
@user_passes_test(lambda u: u.is_superuser)
@require_http_methods(["GET", "POST"])
def import_llm_questions(request):
"""Upload a JSON/JSONL file of LLM-produced questions and import them.
Only superusers may use this view. The view validates each object against
the agreed JSON schema (draft-07) and attempts to resolve M2M references
by numeric id or by name. Returns a JSON report of created items and errors.
"""
schema = {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "SBA Question",
"type": "object",
"additionalProperties": False,
"required": [
"title",
"stem",
"a_answer",
"b_answer",
"c_answer",
"d_answer",
"e_answer",
"best_answer",
"category",
],
"properties": {
"title": {"type": "string"},
"stem": {"type": "string"},
"a_answer": {"type": "string"},
"a_feedback": {"type": "string"},
"b_answer": {"type": "string"},
"b_feedback": {"type": "string"},
"c_answer": {"type": "string"},
"c_feedback": {"type": "string"},
"d_answer": {"type": "string"},
"d_feedback": {"type": "string"},
"e_answer": {"type": "string"},
"e_feedback": {"type": "string"},
"feedback": {"type": "string"},
"best_answer": {"type": "string", "enum": ["a", "b", "c", "d", "e"]},
"category": {"oneOf": [{"type": "string"}]},
"finding": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"structure": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"condition": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"presentation": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"subspecialty": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
},
}
if request.method == "GET":
form = LLMImportForm()
return render(request, "sbas/import_llm_questions.html", {"form": form})
# POST
form = LLMImportForm(request.POST, request.FILES)
if not form.is_valid():
return JsonResponse({"ok": False, "errors": ["No file uploaded or invalid form"]}, status=400)
raw_text = None
if form.cleaned_data.get("raw"):
raw_text = form.cleaned_data.get("raw")
elif form.cleaned_data.get("file"):
f = form.cleaned_data["file"]
raw_text = f.read().decode("utf-8")
else:
return JsonResponse({"ok": False, "errors": ["No file uploaded or text provided"]}, status=400)
# support: single JSON object, JSON array, JSONL (one JSON object per line),
# or multiple JSON documents separated by one or more blank lines.
candidates = []
# sanitize raw_text by escaping backslashes that are not part of a valid
# JSON escape sequence. This handles inputs containing LaTeX-style
# sequences such as "\ge" which are invalid JSON (Invalid \escape).
def _sanitize_backslashes(s: str):
# valid escapes after backslash in JSON: \" \\ \/ \b \f \n \r \t and \uXXXX
# This regex finds backslashes not followed by ", \\ , /, b, f, n, r, t, or u
pattern = re.compile(r"\\(?!(?:[\"\\/bfnrtu]))")
new_s, n = pattern.subn(r"\\\\", s)
return new_s, n
raw_text_stripped = raw_text.strip()
raw_text, sanitized_count = _sanitize_backslashes(raw_text_stripped)
logger.debug("import_llm_questions: sanitized input length %d (escaped %d backslashes)", len(raw_text), sanitized_count)
# Try whole-text JSON first (object or array)
parse_error = None
try:
parsed = json.loads(raw_text)
if isinstance(parsed, list):
candidates = parsed
elif isinstance(parsed, dict):
candidates = [parsed]
except Exception as e_outer:
parse_error = e_outer
# Try JSONL: each non-empty line is a JSON object
for i, line in enumerate(raw_text.splitlines()):
line = line.strip()
if not line:
continue
try:
candidates.append(json.loads(line))
except Exception:
# not JSONL or some lines are multi-line JSON; fallthrough
candidates = []
break
# If JSONL didn't produce candidates, try splitting by blank lines
if not candidates:
blocks = [b.strip() for b in re.split(r"\n\s*\n", raw_text) if b.strip()]
if len(blocks) > 1:
for i, block in enumerate(blocks):
try:
candidates.append(json.loads(block))
except Exception as e_block:
return JsonResponse({"ok": False, "errors": [f"Invalid JSON in block {i+1}: {e_block}"]}, status=400)
if not candidates:
# If still empty, return the original parse error if present
msg = str(parse_error) if parse_error is not None else "No JSON objects found"
logger.debug(f"Failed to parse input: {msg}")
return JsonResponse({"ok": False, "errors": [f"Failed to parse input: {msg}"]}, status=400)
report = {"total": len(candidates), "created": 0, "errors": [], "sanitized_backslashes": sanitized_count}
if jsonschema is None:
return JsonResponse({"ok": False, "errors": ["jsonschema library not available in environment"]}, status=500)
validator = jsonschema.Draft7Validator(schema)
from atlas.models import Finding, Structure, Condition, Presentation, Subspecialty
for idx, payload in enumerate(candidates):
errors = []
for err in validator.iter_errors(payload):
errors.append(err.message)
if errors:
logger.debug(f"Validation errors for item {idx}: {errors}")
report["errors"].append({"index": idx, "errors": errors})
continue
# create the Question inside a transaction; partial failures should not leave half-written objects
try:
with transaction.atomic():
q = Question()
q.stem = payload.get("stem", "").strip()
# map title -> not present in model; use as stem prefix or store in feedback
title = payload.get("title")
if title:
q.stem = f"<strong>{title}</strong>\n" + q.stem
q.a_answer = payload.get("a_answer", "").strip()
q.a_feedback = payload.get("a_feedback", "")
q.b_answer = payload.get("b_answer", "").strip()
q.b_feedback = payload.get("b_feedback", "")
q.c_answer = payload.get("c_answer", "").strip()
q.c_feedback = payload.get("c_feedback", "")
q.d_answer = payload.get("d_answer", "").strip()
q.d_feedback = payload.get("d_feedback", "")
q.e_answer = payload.get("e_answer", "").strip()
q.e_feedback = payload.get("e_feedback", "")
q.feedback = payload.get("feedback", "")
q.best_answer = payload.get("best_answer")
# category resolve by name
cat_val = payload.get("category")
if isinstance(cat_val, str):
cat_obj, created = Category.objects.get_or_create(category=cat_val)
q.category = cat_obj
q.save()
# Resolve M2Ms
for key, model_cls in (
("finding", Finding),
("structure", Structure),
("condition", Condition),
("presentation", Presentation),
("subspecialty", Subspecialty),
):
vals = payload.get(key) or []
resolved = []
for v in vals:
if isinstance(v, int):
try:
resolved.append(model_cls.objects.get(pk=v))
except model_cls.DoesNotExist:
# skip unknown
continue
elif isinstance(v, str):
qs = model_cls.objects.filter(name__iexact=v)
if not qs.exists():
qs = model_cls.objects.filter(name__icontains=v)
if qs.exists():
resolved.extend(list(qs[:5]))
if resolved:
getattr(q, key).add(*[o.pk for o in resolved])
report["created"] += 1
except Exception as e:
report["errors"].append({"index": idx, "errors": [str(e)]})
return JsonResponse({"ok": True, "report": report})
from django_tables2 import SingleTableView, SingleTableMixin
from django_filters.views import FilterView
@@ -683,4 +454,311 @@ def exam_clone2(request, exam_id):
def llm_prompt_view(request):
return render(request, "sbas/llm_prompt_view.html", {})
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 []
@login_required
@user_passes_test(lambda u: u.is_superuser)
@require_http_methods(["GET", "POST"])
def import_llm_questions(request):
"""Upload a JSON/JSONL file of LLM-produced questions and import them.
Only superusers may use this view. The view validates each object against
the agreed JSON schema (draft-07) and attempts to resolve M2M references
by numeric id or by name. Returns a JSON report of created items and errors.
"""
schema = {
"$schema": "http://json-schema.org/draft-07/schema#",
"title": "SBA Question",
"type": "object",
"additionalProperties": False,
"required": [
"title",
"stem",
"a_answer",
"b_answer",
"c_answer",
"d_answer",
"e_answer",
"best_answer",
"category",
],
"properties": {
"title": {"type": "string"},
"stem": {"type": "string"},
"a_answer": {"type": "string"},
"a_feedback": {"type": "string"},
"b_answer": {"type": "string"},
"b_feedback": {"type": "string"},
"c_answer": {"type": "string"},
"c_feedback": {"type": "string"},
"d_answer": {"type": "string"},
"d_feedback": {"type": "string"},
"e_answer": {"type": "string"},
"e_feedback": {"type": "string"},
"feedback": {"type": "string"},
"best_answer": {"type": "string", "enum": ["a", "b", "c", "d", "e"]},
"category": {"oneOf": [{"type": "string"}]},
"finding": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"structure": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"condition": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"presentation": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
"subspecialty": {"type": "array", "items": {"oneOf": [{"type": "integer"}, {"type": "string"}]}, "uniqueItems": True},
},
}
if request.method == "GET":
form = LLMImportForm()
return render(request, "sbas/import_llm_questions.html", {"form": form})
# POST
form = LLMImportForm(request.POST, request.FILES)
if not form.is_valid():
return JsonResponse({"ok": False, "errors": ["No file uploaded or invalid form"]}, status=400)
raw_text = None
if form.cleaned_data.get("raw"):
raw_text = form.cleaned_data.get("raw")
elif form.cleaned_data.get("file"):
f = form.cleaned_data["file"]
raw_text = f.read().decode("utf-8")
else:
return JsonResponse({"ok": False, "errors": ["No file uploaded or text provided"]}, status=400)
# support: single JSON object, JSON array, JSONL (one JSON object per line),
# or multiple JSON documents separated by one or more blank lines.
candidates = []
# sanitize raw_text by escaping backslashes that are not part of a valid
# JSON escape sequence. This handles inputs containing LaTeX-style
# sequences such as "\ge" which are invalid JSON (Invalid \escape).
def _sanitize_backslashes(s: str):
# valid escapes after backslash in JSON: \" \\ \/ \b \f \n \r \t and \uXXXX
# This regex finds backslashes not followed by ", \\ , /, b, f, n, r, t, or u
pattern = re.compile(r"\\(?!(?:[\"\\/bfnrtu]))")
new_s, n = pattern.subn(r"\\\\", s)
return new_s, n
raw_text_stripped = raw_text.strip()
raw_text, sanitized_count = _sanitize_backslashes(raw_text_stripped)
logger.debug("import_llm_questions: sanitized input length %d (escaped %d backslashes)", len(raw_text), sanitized_count)
# Try whole-text JSON first (object or array)
parse_error = None
try:
parsed = json.loads(raw_text)
if isinstance(parsed, list):
candidates = parsed
elif isinstance(parsed, dict):
candidates = [parsed]
except Exception as e_outer:
parse_error = e_outer
# Try JSONL: each non-empty line is a JSON object
for i, line in enumerate(raw_text.splitlines()):
line = line.strip()
if not line:
continue
try:
candidates.append(json.loads(line))
except Exception:
# not JSONL or some lines are multi-line JSON; fallthrough
candidates = []
break
# If JSONL didn't produce candidates, try splitting by blank lines
if not candidates:
blocks = [b.strip() for b in re.split(r"\n\s*\n", raw_text) if b.strip()]
if len(blocks) > 1:
for i, block in enumerate(blocks):
try:
candidates.append(json.loads(block))
except Exception as e_block:
return JsonResponse({"ok": False, "errors": [f"Invalid JSON in block {i+1}: {e_block}"]}, status=400)
if not candidates:
# If still empty, return the original parse error if present
msg = str(parse_error) if parse_error is not None else "No JSON objects found"
logger.debug(f"Failed to parse input: {msg}")
return JsonResponse({"ok": False, "errors": [f"Failed to parse input: {msg}"]}, status=400)
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": {}}
for err in validator.iter_errors(payload):
item_report["errors"].append(err.message)
if item_report["errors"]:
logger.debug(f"Validation errors for item {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]
if missing:
item_report["missing_m2m"].extend([{"model": model_cls.__name__, "values": missing}])
for v in missing:
aggregated_would_create[model_cls.__name__].add(v)
# 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()
title = payload.get("title")
if title:
q.stem = f"<strong>{title}</strong>\n" + q.stem
q.a_answer = payload.get("a_answer", "").strip()
q.a_feedback = payload.get("a_feedback", "")
q.b_answer = payload.get("b_answer", "").strip()
q.b_feedback = payload.get("b_feedback", "")
q.c_answer = payload.get("c_answer", "").strip()
q.c_feedback = payload.get("c_feedback", "")
q.d_answer = payload.get("d_answer", "").strip()
q.d_feedback = payload.get("d_feedback", "")
q.e_answer = payload.get("e_answer", "").strip()
q.e_feedback = payload.get("e_feedback", "")
q.feedback = payload.get("feedback", "")
q.best_answer = payload.get("best_answer")
if cat_obj:
q.category = cat_obj
q.save()
# 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
return JsonResponse({"ok": True, "report": report})