feat: add import exam answers functionality with UI and backend handling
This commit is contained in:
Ross
@@ -4900,6 +4900,216 @@ def cid_group_view(request):
|
||||
{"groups": groups},
|
||||
)
|
||||
|
||||
|
||||
@user_is_cid_user_manager
|
||||
def import_exam_answers(request):
|
||||
"""
|
||||
Imports manual JSON exam answers files exported from RTS when direct API submission failed.
|
||||
"""
|
||||
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"))
|
||||
|
||||
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-"):
|
||||
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"))
|
||||
|
||||
# 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
|
||||
)
|
||||
|
||||
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]
|
||||
|
||||
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()
|
||||
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
|
||||
|
||||
messages.success(request, f"Successfully imported {n} answers for CID/UID {cid_str} on exam {eid_str}")
|
||||
return HttpResponseRedirect(reverse("generic:manage_cids"))
|
||||
|
||||
return render(request, "generic/import_exam_answers.html")
|
||||
|
||||
@user_is_cid_user_manager
|
||||
def user_group_add_by_email_user(request, group_id):
|
||||
if request.htmx:
|
||||
|
||||
Reference in New Issue
Block a user