feat: implement import exam answers feature with preview and confirmation steps
This commit is contained in:
Ross
@@ -29,6 +29,8 @@
|
||||
<ul class="dropdown-menu" aria-labelledby="navCids">
|
||||
<li><a class="dropdown-item" href="{% url 'generic:manage_cids' %}"><i class="bi bi-file-earmark-text me-1"></i> All Cids</a></li>
|
||||
<li><a class="dropdown-item" href="{% url 'generic:manage_cid_exams' %}"><i class="bi bi-mortarboard me-1"></i> Cids (exams)</a></li>
|
||||
<li><hr class="dropdown-divider"></li>
|
||||
<li><a class="dropdown-item" href="{% url 'generic:import_exam_answers' %}"><i class="bi bi-upload me-1"></i> Import Exam Answers</a></li>
|
||||
</ul>
|
||||
</li>
|
||||
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
{% extends 'generic/base.html' %}
|
||||
|
||||
{% block content %}
|
||||
<div class="container py-4">
|
||||
<div class="row justify-content-center">
|
||||
<div class="col-12 col-xl-10">
|
||||
<div class="card shadow">
|
||||
<div class="card-header bg-warning text-dark d-flex justify-content-between align-items-center">
|
||||
<h3 class="card-title mb-0">Preview Import Exam Answers</h3>
|
||||
<span class="badge bg-dark text-white fs-6">{{ answers_count }} Answers in file</span>
|
||||
</div>
|
||||
<div class="card-body">
|
||||
<div class="row mb-4">
|
||||
<div class="col-md-6">
|
||||
<strong>Exam:</strong> <span class="text-primary">{{ eid_str }}</span>
|
||||
</div>
|
||||
<div class="col-md-6">
|
||||
<strong>Candidate/User ID:</strong> <span class="text-primary">{{ cid_str }}</span>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<p class="text-muted">
|
||||
Review the changes below before confirming the import. Existing database answers will be overwritten.
|
||||
</p>
|
||||
|
||||
<div class="table-responsive" style="max-height: 400px; overflow-y: auto; border: 1px solid #ddd; border-radius: 4px;">
|
||||
<table class="table table-striped table-hover mb-0">
|
||||
<thead class="table-dark sticky-top">
|
||||
<tr>
|
||||
<th>Question ID</th>
|
||||
<th>Part (qidn)</th>
|
||||
<th>File Answer (To Import)</th>
|
||||
<th>Database Answer (To Overwrite)</th>
|
||||
<th>Action</th>
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for item in preview_items %}
|
||||
<tr>
|
||||
<td>{{ item.qid }}</td>
|
||||
<td>{{ item.qidn }}</td>
|
||||
<td><code class="text-success">{{ item.posted }}</code></td>
|
||||
<td>
|
||||
{% if item.existing %}
|
||||
<code class="text-danger">{{ item.existing }}</code>
|
||||
{% else %}
|
||||
<span class="text-muted italic">- None -</span>
|
||||
{% endif %}
|
||||
</td>
|
||||
<td>
|
||||
{% if item.action == "Update" %}
|
||||
<span class="badge bg-warning text-dark">Overwrite (Update)</span>
|
||||
{% else %}
|
||||
<span class="badge bg-success">New (Create)</span>
|
||||
{% endif %}
|
||||
</td>
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
|
||||
<form method="post" class="mt-4">
|
||||
{% csrf_token %}
|
||||
<input type="hidden" name="confirm" value="1">
|
||||
<div class="d-flex justify-content-between align-items-center">
|
||||
<a href="{% url 'generic:import_exam_answers' %}" class="btn btn-secondary">Cancel</a>
|
||||
<button type="submit" class="btn btn-success px-4 fw-bold">Confirm Import</button>
|
||||
</div>
|
||||
</form>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -0,0 +1,50 @@
|
||||
{% extends 'generic/base.html' %}
|
||||
|
||||
{% block content %}
|
||||
<div class="container py-4">
|
||||
<div class="row justify-content-center">
|
||||
<div class="col-md-8 col-lg-6">
|
||||
<div class="card shadow border-success">
|
||||
<div class="card-header bg-success text-white">
|
||||
<h3 class="card-title mb-0">Import Successful</h3>
|
||||
</div>
|
||||
<div class="card-body text-center">
|
||||
<div class="mb-4">
|
||||
<i class="bi bi-check-circle-fill text-success" style="font-size: 4rem;"></i>
|
||||
</div>
|
||||
|
||||
<h4 class="mb-3">Answers Imported Successfully</h4>
|
||||
|
||||
<div class="bg-light p-3 rounded mb-4 text-start">
|
||||
<div class="row mb-2">
|
||||
<div class="col-6 text-muted">Exam:</div>
|
||||
<div class="col-6 fw-bold">{{ eid_str }}</div>
|
||||
</div>
|
||||
<div class="row mb-2">
|
||||
<div class="col-6 text-muted">Candidate/User ID:</div>
|
||||
<div class="col-6 fw-bold">{{ cid_str }}</div>
|
||||
</div>
|
||||
<div class="row mb-2 border-top pt-2">
|
||||
<div class="col-6 text-muted">Created (New):</div>
|
||||
<div class="col-6 fw-bold text-success">{{ created_count }}</div>
|
||||
</div>
|
||||
<div class="row mb-2">
|
||||
<div class="col-6 text-muted">Updated (Overwritten):</div>
|
||||
<div class="col-6 fw-bold text-warning">{{ updated_count }}</div>
|
||||
</div>
|
||||
<div class="row border-top pt-2">
|
||||
<div class="col-6 text-muted">Total Processed:</div>
|
||||
<div class="col-6 fw-bold text-primary">{{ total_count }}</div>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div class="d-grid gap-2">
|
||||
<a href="{% url 'generic:manage_cids' %}" class="btn btn-primary">Go to Candidate Management</a>
|
||||
<a href="{% url 'generic:import_exam_answers' %}" class="btn btn-outline-secondary">Import Another File</a>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{% endblock %}
|
||||
@@ -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"
|
||||
|
||||
+320
-192
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user