From 5ce4a67b3121c5f92a4804399a1f792fb9b04e66 Mon Sep 17 00:00:00 2001 From: Ross Date: Sun, 5 Jul 2026 22:47:50 +0100 Subject: [PATCH] feat: implement import exam answers feature with preview and confirmation steps --- generic/templates/generic/base.html | 2 + .../generic/import_exam_answers_preview.html | 76 +++ .../generic/import_exam_answers_success.html | 50 ++ generic/tests/test_exam_submit_security.py | 16 +- generic/views.py | 512 +++++++++++------- 5 files changed, 462 insertions(+), 194 deletions(-) create mode 100644 generic/templates/generic/import_exam_answers_preview.html create mode 100644 generic/templates/generic/import_exam_answers_success.html diff --git a/generic/templates/generic/base.html b/generic/templates/generic/base.html index 88b8cac6..65e3277e 100755 --- a/generic/templates/generic/base.html +++ b/generic/templates/generic/base.html @@ -29,6 +29,8 @@ diff --git a/generic/templates/generic/import_exam_answers_preview.html b/generic/templates/generic/import_exam_answers_preview.html new file mode 100644 index 00000000..079580e6 --- /dev/null +++ b/generic/templates/generic/import_exam_answers_preview.html @@ -0,0 +1,76 @@ +{% extends 'generic/base.html' %} + +{% block content %} +
+
+
+
+
+

Preview Import Exam Answers

+ {{ answers_count }} Answers in file +
+
+
+
+ Exam: {{ eid_str }} +
+
+ Candidate/User ID: {{ cid_str }} +
+
+ +

+ Review the changes below before confirming the import. Existing database answers will be overwritten. +

+ +
+ + + + + + + + + + + + {% for item in preview_items %} + + + + + + + + {% endfor %} + +
Question IDPart (qidn)File Answer (To Import)Database Answer (To Overwrite)Action
{{ item.qid }}{{ item.qidn }}{{ item.posted }} + {% if item.existing %} + {{ item.existing }} + {% else %} + - None - + {% endif %} + + {% if item.action == "Update" %} + Overwrite (Update) + {% else %} + New (Create) + {% endif %} +
+
+ +
+ {% csrf_token %} + +
+ Cancel + +
+
+
+
+
+
+
+{% endblock %} diff --git a/generic/templates/generic/import_exam_answers_success.html b/generic/templates/generic/import_exam_answers_success.html new file mode 100644 index 00000000..025630a3 --- /dev/null +++ b/generic/templates/generic/import_exam_answers_success.html @@ -0,0 +1,50 @@ +{% extends 'generic/base.html' %} + +{% block content %} +
+
+
+
+
+

Import Successful

+
+
+
+ +
+ +

Answers Imported Successfully

+ +
+
+
Exam:
+
{{ eid_str }}
+
+
+
Candidate/User ID:
+
{{ cid_str }}
+
+
+
Created (New):
+
{{ created_count }}
+
+
+
Updated (Overwritten):
+
{{ updated_count }}
+
+
+
Total Processed:
+
{{ total_count }}
+
+
+ +
+ Go to Candidate Management + Import Another File +
+
+
+
+
+
+{% endblock %} diff --git a/generic/tests/test_exam_submit_security.py b/generic/tests/test_exam_submit_security.py index 38d01494..8445e84c 100644 --- a/generic/tests/test_exam_submit_security.py +++ b/generic/tests/test_exam_submit_security.py @@ -223,11 +223,23 @@ def test_import_exam_answers_post_success(cid_manager_client, exam_data, logged_ answers_file.name = "backup.json" url = reverse("generic:import_exam_answers") + + # 1. First stage: Upload file -> renders preview response = cid_manager_client.post(url, {"answers_file": answers_file}, format="multipart") + assert response.status_code == 200 + assert b"Preview Import" in response.content + assert b"Imported Test Answer" in response.content - assert response.status_code == 302 # redirects to manage_cids on success + # Check session has pending data + assert cid_manager_client.session["pending_import_data"] == import_data - # Verify answers were saved + # 2. Second stage: Confirm import + response2 = cid_manager_client.post(url, {"confirm": "1"}) + assert response2.status_code == 200 + assert b"Import Successful" in response2.content + assert b"1" in response2.content # created_count count + + # Verify answers were saved to DB answers = AnatomyUserAnswer.objects.filter(exam=exam, user=logged_in_user) assert answers.count() == 1 assert answers.first().answer == "Imported Test Answer" diff --git a/generic/views.py b/generic/views.py index 25cdc6b5..c82ffa9a 100644 --- a/generic/views.py +++ b/generic/views.py @@ -4905,209 +4905,337 @@ def cid_group_view(request): def import_exam_answers(request): """ Imports manual JSON exam answers files exported from RTS when direct API submission failed. + Supports a preview step comparing file answers with existing database entries. """ from django.contrib import messages + if request.method == "POST": - json_file = request.FILES.get("answers_file") - if not json_file: - messages.error(request, "No file uploaded") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) + confirm = request.POST.get("confirm") == "1" - try: - data = json.load(json_file) - except Exception as e: - messages.error(request, f"Failed to parse JSON: {e}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - eid_str = data.get("eid") - cid_str = data.get("cid") - start_time_val = data.get("start_time") - answers_data = data.get("answers") - - if not eid_str or not cid_str or answers_data is None: - messages.error(request, "Uploaded JSON is missing required fields (eid, cid, answers)") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - try: - exam_type, exam_id_str = eid_str.split("/") - eid = int(exam_id_str) - except Exception: - messages.error(request, f"Invalid eid format: {eid_str}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - EXAM_IMPORT_MAP = { - "physics": (PhysicsExam, PhysicsUserAnswer, "physics"), - "anatomy": (AnatomyExam, AnatomyUserAnswer, "anatomy"), - "rapid": (RapidsExam, RapidsUserAnswer, "rapids"), - "rapids": (RapidsExam, RapidsUserAnswer, "rapids"), - "short": (ShortsExam, ShortsUserAnswer, "shorts"), - "shorts": (ShortsExam, ShortsUserAnswer, "shorts"), - "long": (LongsExam, LongsUserAnswer, "longs"), - "longs": (LongsExam, LongsUserAnswer, "longs"), - "sba": (SbasExam, SbasUserAnswer, "sbas"), - "sbas": (SbasExam, SbasUserAnswer, "sbas"), - } - - mapping = EXAM_IMPORT_MAP.get(exam_type.lower()) - if not mapping: - messages.error(request, f"Unsupported exam type: {exam_type}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - ExamModel, UserAnswerModel, app_name = mapping - - # Check CID/UID type - uid = False - cid = False - if str(cid_str).startswith("u-"): + if confirm: + data = request.session.get("pending_import_data") + if not data: + messages.error(request, "Import session expired. Please upload the file again.") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + eid_str = data.get("eid") + cid_str = data.get("cid") + start_time_val = data.get("start_time") + answers_data = data.get("answers") + try: + exam_type, exam_id_str = eid_str.split("/") + eid = int(exam_id_str) + except Exception: + messages.error(request, "Invalid exam ID format in stored session.") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + EXAM_IMPORT_MAP = { + "physics": (PhysicsExam, PhysicsUserAnswer, "physics"), + "anatomy": (AnatomyExam, AnatomyUserAnswer, "anatomy"), + "rapid": (RapidsExam, RapidsUserAnswer, "rapids"), + "rapids": (RapidsExam, RapidsUserAnswer, "rapids"), + "short": (ShortsExam, ShortsUserAnswer, "shorts"), + "shorts": (ShortsExam, ShortsUserAnswer, "shorts"), + "long": (LongsExam, LongsUserAnswer, "longs"), + "longs": (LongsExam, LongsUserAnswer, "longs"), + "sba": (SbasExam, SbasUserAnswer, "sbas"), + "sbas": (SbasExam, SbasUserAnswer, "sbas"), + } + + mapping = EXAM_IMPORT_MAP.get(exam_type.lower()) + if not mapping: + messages.error(request, f"Unsupported exam type: {exam_type}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + ExamModel, UserAnswerModel, app_name = mapping + + uid = False + cid = False + if str(cid_str).startswith("u-"): uid = int(cid_str[2:]) - except ValueError: - messages.error(request, f"Invalid user CID format: {cid_str}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - else: - try: - cid = int(cid_str) - except ValueError: - messages.error(request, f"Invalid candidate CID format: {cid_str}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - try: - exam = ExamModel.objects.get(pk=eid) - except ExamModel.DoesNotExist: - messages.error(request, f"Exam {eid} of type {exam_type} does not exist") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - # Parse answers (can be string or array) - if isinstance(answers_data, str): - try: - answers_list = json.loads(answers_data) - except Exception as e: - messages.error(request, f"Failed to parse answers string JSON: {e}") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - elif isinstance(answers_data, list): - answers_list = answers_data - else: - messages.error(request, "answers field must be a list or a JSON string representation of a list") - return HttpResponseRedirect(reverse("generic:import_exam_answers")) - - # Import each answer - n = 0 - for answer in answers_list: - try: - qid = int(answer["qid"]) - posted_answer = answer["ans"] - qidn = str(answer.get("qidn", "1")) - except (KeyError, ValueError, TypeError): - continue - - if uid: - existing_answers = UserAnswerModel.objects.filter( - question__id=qid, exam__id=eid, user__id=uid - ) else: - existing_answers = UserAnswerModel.objects.filter( - question__id=qid, exam__id=eid, cid=cid - ) + cid = int(cid_str) - ans = None - if app_name == "rapids": - normal = (posted_answer == "Normal") - if not existing_answers: - if uid: - ans = UserAnswerModel(answer=posted_answer, normal=normal, user=User.objects.get(id=uid)) - else: - ans = UserAnswerModel(answer=posted_answer, normal=normal, cid=cid) - ans.question_id = qid - ans.exam_id = eid - else: - ans = existing_answers[0] - if qidn == "1": - ans.normal = normal - if normal: - ans.answer = "" - elif qidn == "2" and not ans.normal: - ans.answer = posted_answer - elif app_name == "shorts": - if not existing_answers: - if uid: - ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) - else: - ans = UserAnswerModel(answer=posted_answer, cid=cid) - ans.question_id = qid - ans.exam_id = eid - ans.score = None - else: - ans = existing_answers[0] - ans.answer = posted_answer - ans.score = None - elif app_name == "anatomy": - from anatomy.models import Answer as AnatomyAnswer - if not existing_answers: - if uid: - ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) - else: - ans = UserAnswerModel(answer=posted_answer, cid=cid) - ans.question_id = qid - ans.exam_id = eid - ans.score = AnatomyAnswer.MarkOptions.UNMARKED - else: - ans = existing_answers[0] - ans.answer = posted_answer - ans.score = AnatomyAnswer.MarkOptions.UNMARKED - elif app_name == "longs": - if not existing_answers: - if uid: - ans = UserAnswerModel(user=User.objects.get(id=uid)) - else: - ans = UserAnswerModel(cid=cid) - ans.question_id = qid - ans.exam_id = eid - else: - ans = existing_answers[0] + try: + exam = ExamModel.objects.get(pk=eid) + except ExamModel.DoesNotExist: + messages.error(request, f"Exam {eid} does not exist.") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) - if qidn == "1": - ans.answer_observations = posted_answer - elif qidn == "2": - ans.answer_interpretation = posted_answer - elif qidn == "3": - ans.answer_principle_diagnosis = posted_answer - elif qidn == "4": - ans.answer_differential_diagnosis = posted_answer - elif qidn == "5": - ans.answer_management = posted_answer - ans.score = ans.ScoreOptions.UNMARKED + if isinstance(answers_data, str): + answers_list = json.loads(answers_data) + else: + answers_list = answers_data - if ans: - time_answered_str = answer.get("time_answered") - if time_answered_str: - from django.utils.dateparse import parse_datetime - parsed_time = parse_datetime(time_answered_str) - if parsed_time: - if timezone.is_naive(parsed_time): - parsed_time = timezone.make_aware(parsed_time) - ans.time_answered = parsed_time - try: - ans.full_clean() - ans.save() - n += 1 - except Exception as e: - messages.warning(request, f"Failed to save answer for question {qid}: {e}") - - # Also get or create CidUserExam - try: - if uid: - t = timezone.now() - if start_time_val: - t = timezone.datetime.fromtimestamp(start_time_val, tz=timezone.utc) - exam.get_or_create_cid_user_exam( - user_user=User.objects.get(id=uid), start_time=t - ) - except Exception: - pass + created_count = 0 + updated_count = 0 - messages.success(request, f"Successfully imported {n} answers for CID/UID {cid_str} on exam {eid_str}") - return HttpResponseRedirect(reverse("generic:manage_cids")) - + for answer in answers_list: + try: + qid = int(answer["qid"]) + posted_answer = answer["ans"] + qidn = str(answer.get("qidn", "1")) + except (KeyError, ValueError, TypeError): + continue + + if uid: + existing_answers = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, user__id=uid + ) + else: + existing_answers = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, cid=cid + ) + + ans = None + is_new = len(existing_answers) == 0 + + if app_name == "rapids": + normal = (posted_answer == "Normal") + if is_new: + if uid: + ans = UserAnswerModel(answer=posted_answer, normal=normal, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, normal=normal, cid=cid) + ans.question_id = qid + ans.exam_id = eid + else: + ans = existing_answers[0] + if qidn == "1": + ans.normal = normal + if normal: + ans.answer = "" + elif qidn == "2" and not ans.normal: + ans.answer = posted_answer + elif app_name == "shorts": + if is_new: + if uid: + ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, cid=cid) + ans.question_id = qid + ans.exam_id = eid + ans.score = None + else: + ans = existing_answers[0] + ans.answer = posted_answer + ans.score = None + elif app_name == "anatomy": + from anatomy.models import Answer as AnatomyAnswer + if is_new: + if uid: + ans = UserAnswerModel(answer=posted_answer, user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(answer=posted_answer, cid=cid) + ans.question_id = qid + ans.exam_id = eid + ans.score = AnatomyAnswer.MarkOptions.UNMARKED + else: + ans = existing_answers[0] + ans.answer = posted_answer + ans.score = AnatomyAnswer.MarkOptions.UNMARKED + elif app_name == "longs": + if is_new: + if uid: + ans = UserAnswerModel(user=User.objects.get(id=uid)) + else: + ans = UserAnswerModel(cid=cid) + ans.question_id = qid + ans.exam_id = eid + else: + ans = existing_answers[0] + + if qidn == "1": + ans.answer_observations = posted_answer + elif qidn == "2": + ans.answer_interpretation = posted_answer + elif qidn == "3": + ans.answer_principle_diagnosis = posted_answer + elif qidn == "4": + ans.answer_differential_diagnosis = posted_answer + elif qidn == "5": + ans.answer_management = posted_answer + ans.score = ans.ScoreOptions.UNMARKED + + if ans: + time_answered_str = answer.get("time_answered") + if time_answered_str: + from django.utils.dateparse import parse_datetime + parsed_time = parse_datetime(time_answered_str) + if parsed_time: + if timezone.is_naive(parsed_time): + parsed_time = timezone.make_aware(parsed_time) + ans.time_answered = parsed_time + try: + ans.full_clean() + ans.save() + if is_new: + created_count += 1 + else: + updated_count += 1 + except Exception as e: + messages.warning(request, f"Failed to save answer for question {qid}: {e}") + + # Create CidUserExam if user attempt + try: + if uid: + t = timezone.now() + if start_time_val: + t = timezone.datetime.fromtimestamp(start_time_val, tz=timezone.utc) + exam.get_or_create_cid_user_exam( + user_user=User.objects.get(id=uid), start_time=t + ) + except Exception: + pass + + del request.session["pending_import_data"] + + return render(request, "generic/import_exam_answers_success.html", { + "created_count": created_count, + "updated_count": updated_count, + "total_count": created_count + updated_count, + "cid_str": cid_str, + "eid_str": eid_str, + }) + + else: + # First stage: parse and preview + json_file = request.FILES.get("answers_file") + if not json_file: + messages.error(request, "No file uploaded") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + data = json.load(json_file) + except Exception as e: + messages.error(request, f"Failed to parse JSON: {e}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + eid_str = data.get("eid") + cid_str = data.get("cid") + answers_data = data.get("answers") + + if not eid_str or not cid_str or answers_data is None: + messages.error(request, "Uploaded JSON is missing required fields (eid, cid, answers)") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + exam_type, exam_id_str = eid_str.split("/") + eid = int(exam_id_str) + except Exception: + messages.error(request, f"Invalid eid format: {eid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + EXAM_IMPORT_MAP = { + "physics": (PhysicsExam, PhysicsUserAnswer, "physics"), + "anatomy": (AnatomyExam, AnatomyUserAnswer, "anatomy"), + "rapid": (RapidsExam, RapidsUserAnswer, "rapids"), + "rapids": (RapidsExam, RapidsUserAnswer, "rapids"), + "short": (ShortsExam, ShortsUserAnswer, "shorts"), + "shorts": (ShortsExam, ShortsUserAnswer, "shorts"), + "long": (LongsExam, LongsUserAnswer, "longs"), + "longs": (LongsExam, LongsUserAnswer, "longs"), + "sba": (SbasExam, SbasUserAnswer, "sbas"), + "sbas": (SbasExam, SbasUserAnswer, "sbas"), + } + + mapping = EXAM_IMPORT_MAP.get(exam_type.lower()) + if not mapping: + messages.error(request, f"Unsupported exam type: {exam_type}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + ExamModel, UserAnswerModel, app_name = mapping + + uid = False + cid = False + if str(cid_str).startswith("u-"): + try: + uid = int(cid_str[2:]) + except ValueError: + messages.error(request, f"Invalid user CID format: {cid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + else: + try: + cid = int(cid_str) + except ValueError: + messages.error(request, f"Invalid candidate CID format: {cid_str}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + try: + exam = ExamModel.objects.get(pk=eid) + except ExamModel.DoesNotExist: + messages.error(request, f"Exam {eid} of type {exam_type} does not exist") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + if isinstance(answers_data, str): + try: + answers_list = json.loads(answers_data) + except Exception as e: + messages.error(request, f"Failed to parse answers string JSON: {e}") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + elif isinstance(answers_data, list): + answers_list = answers_data + else: + messages.error(request, "answers field must be a list or a JSON string representation of a list") + return HttpResponseRedirect(reverse("generic:import_exam_answers")) + + request.session["pending_import_data"] = data + + preview_items = [] + for answer in answers_list: + try: + qid = int(answer["qid"]) + posted_answer = answer["ans"] + qidn = str(answer.get("qidn", "1")) + except (KeyError, ValueError, TypeError): + continue + + if uid: + existing = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, user__id=uid + ).first() + else: + existing = UserAnswerModel.objects.filter( + question__id=qid, exam__id=eid, cid=cid + ).first() + + existing_str = "" + action = "Create" + if existing: + action = "Update" + if app_name == "longs": + if qidn == "1": + existing_str = existing.answer_observations + elif qidn == "2": + existing_str = existing.answer_interpretation + elif qidn == "3": + existing_str = existing.answer_principle_diagnosis + elif qidn == "4": + existing_str = existing.answer_differential_diagnosis + elif qidn == "5": + existing_str = existing.answer_management + else: + existing_str = existing.answer + + preview_items.append({ + "qid": qid, + "qidn": qidn, + "posted": posted_answer, + "existing": existing_str, + "action": action, + }) + + return render(request, "generic/import_exam_answers_preview.html", { + "eid_str": eid_str, + "cid_str": cid_str, + "preview_items": preview_items, + "answers_count": len(preview_items), + }) + + # GET request + if "pending_import_data" in request.session: + del request.session["pending_import_data"] return render(request, "generic/import_exam_answers.html") @user_is_cid_user_manager