141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
from copy import deepcopy
|
|
from typing import Type
|
|
|
|
from django.utils import timezone
|
|
from rest_framework import serializers
|
|
|
|
from vbv_lernwelt.assignment.models import (
|
|
Assignment,
|
|
AssignmentCompletion,
|
|
AssignmentCompletionAuditLog,
|
|
AssignmentCompletionStatus,
|
|
)
|
|
from vbv_lernwelt.core.models import User
|
|
from vbv_lernwelt.core.utils import find_first
|
|
from vbv_lernwelt.course.models import CourseSession
|
|
|
|
|
|
def update_assignment_completion(
|
|
assignment_user: User,
|
|
assignment: Assignment,
|
|
course_session: CourseSession,
|
|
completion_data=None,
|
|
completion_status: Type[AssignmentCompletionStatus] = "in_progress",
|
|
grading_user: User | None = None,
|
|
validate_completion_status_change: bool = True,
|
|
copy_task_data: bool = False,
|
|
) -> AssignmentCompletion:
|
|
"""
|
|
:param completion_data: should have the following structure:
|
|
{
|
|
"<user_text_input:uuid>": {"user_data": {"text": "some text from user"}},
|
|
"<user_confirmation:uuid>": {"user_data": {"confirmation": true}},
|
|
}
|
|
every input field has the data stored in sub dict of the question uuid
|
|
|
|
it can also contain "trainer_input" when the trainer has entered grading data
|
|
{
|
|
"<user_text_input:uuid>": {
|
|
"expert_data": {"points": 4, "text": "Gute Antwort"}
|
|
},
|
|
}
|
|
:param copy_task_data: if true, the task data will be copied to the completion data
|
|
used for "submitted" and "graded" status, so that we don't lose the question
|
|
context
|
|
:return: AssignmentCompletion
|
|
"""
|
|
if completion_data is None:
|
|
completion_data = {}
|
|
|
|
ac, created = AssignmentCompletion.objects.get_or_create(
|
|
assignment_user_id=assignment_user.id,
|
|
assignment_id=assignment.id,
|
|
course_session_id=course_session.id,
|
|
)
|
|
|
|
if validate_completion_status_change:
|
|
# TODO: check time?
|
|
if completion_status == "submitted":
|
|
if ac.completion_status in ["submitted", "grading_in_progress", "graded"]:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"completion_status": f"Cannot update completion status from {ac.completion_status} to submitted"
|
|
}
|
|
)
|
|
elif completion_status == "graded":
|
|
if ac.completion_status == "graded":
|
|
raise serializers.ValidationError(
|
|
{
|
|
"completion_status": f"Cannot update completion status from {ac.completion_status} to graded"
|
|
}
|
|
)
|
|
|
|
if completion_status in ["graded", "grading_in_progress"]:
|
|
if grading_user is None:
|
|
raise serializers.ValidationError(
|
|
{"grading_user": "grading_user is required for graded status"}
|
|
)
|
|
ac.grading_user = grading_user
|
|
|
|
if completion_status == "submitted":
|
|
ac.submitted_at = timezone.now()
|
|
elif completion_status == "graded":
|
|
ac.graded_at = timezone.now()
|
|
|
|
ac.completion_status = completion_status
|
|
|
|
# TODO: make more validation of the provided input -> maybe with graphql
|
|
completion_data = _remove_unknown_entries(assignment, completion_data)
|
|
for key, value in completion_data.items():
|
|
# retain already stored data
|
|
stored_entry = ac.completion_data.get(key, {})
|
|
stored_entry.update(value)
|
|
ac.completion_data[key] = stored_entry
|
|
|
|
if copy_task_data:
|
|
# copy over the question data, so that we don't lose the context
|
|
substasks = assignment.filter_user_subtasks()
|
|
for key, value in ac.completion_data.items():
|
|
task_data = find_first(substasks, pred=lambda x: x["id"] == key)
|
|
ac.completion_data[key].update(task_data)
|
|
|
|
ac.save()
|
|
|
|
if completion_status in ["graded", "submitted"]:
|
|
acl = AssignmentCompletionAuditLog.objects.create(
|
|
assignment_user=assignment_user,
|
|
assignment=assignment,
|
|
course_session=course_session,
|
|
grading_user=grading_user,
|
|
completion_status=completion_status,
|
|
assignment_user_email=assignment_user.email,
|
|
assignment_slug=assignment.slug,
|
|
completion_data=deepcopy(ac.completion_data),
|
|
)
|
|
if grading_user:
|
|
acl.grading_user_email = grading_user.email
|
|
|
|
# copy over the question data, so that we don't lose the context
|
|
substasks = assignment.filter_user_subtasks()
|
|
for key, value in acl.completion_data.items():
|
|
task_data = find_first(substasks, pred=lambda x: x["id"] == key)
|
|
acl.completion_data[key].update(task_data)
|
|
acl.save()
|
|
|
|
return ac
|
|
|
|
|
|
def _remove_unknown_entries(assignment, completion_data):
|
|
"""
|
|
Removes all entries from completion_data which are not known to the assignment
|
|
"""
|
|
possible_subtask_uuids = [
|
|
subtask["id"] for subtask in assignment.filter_user_subtasks()
|
|
]
|
|
filtered_completion_data = {
|
|
key: value
|
|
for key, value in completion_data.items()
|
|
if key in possible_subtask_uuids
|
|
}
|
|
return filtered_completion_data
|