diff --git a/sbas/views.py b/sbas/views.py
index 10b381b8..5f42d70c 100644
--- a/sbas/views.py
+++ b/sbas/views.py
@@ -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"{title}\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", {})
\ No newline at end of file
+ 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"{title}\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})
\ No newline at end of file