vbv/server/vbv_lernwelt/assignment/services.py

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,
    is_valid_assignment_completion_status,
)
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",
    evaluation_user: User | None = None,
    evaluation_grade: float | None = None,
    evaluation_points: float | 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 "EVALUATION_SUBMITTED" 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 not is_valid_assignment_completion_status(completion_status):
        raise serializers.ValidationError(
            {"completion_status": f"Invalid completion status {completion_status}"}
        )

    if validate_completion_status_change:
        # TODO: check time?
        if completion_status == "SUBMITTED":
            if ac.completion_status in [
                "SUBMITTED",
                "EVALUATION_IN_PROGRESS",
                "EVALUATION_SUBMITTED",
            ]:
                raise serializers.ValidationError(
                    {
                        "completion_status": f"Cannot update completion status from {ac.completion_status} to SUBMITTED"
                    }
                )
        elif completion_status == "EVALUATION_SUBMITTED":
            if ac.completion_status == "EVALUATION_SUBMITTED":
                raise serializers.ValidationError(
                    {
                        "completion_status": f"Cannot update completion status from {ac.completion_status} to EVALUATION_SUBMITTED"
                    }
                )

        if completion_status == "IN_PROGRESS" and ac.completion_status != "IN_PROGRESS":
            raise serializers.ValidationError(
                {
                    "completion_status": f"Cannot set completion status to IN_PROGRESS when it is {ac.completion_status}"
                }
            )

    if completion_status in ["EVALUATION_SUBMITTED", "EVALUATION_IN_PROGRESS"]:
        if evaluation_user is None:
            raise serializers.ValidationError(
                {
                    "evaluation_user": "evaluation_user is required for EVALUATION_SUBMITTED status"
                }
            )
        ac.evaluation_user = evaluation_user

    if completion_status == "EVALUATION_SUBMITTED":
        if evaluation_grade is None:
            raise serializers.ValidationError(
                {
                    "evaluation_grade": "evaluation_grade is required for EVALUATION_SUBMITTED status"
                }
            )
        if evaluation_points is None:
            raise serializers.ValidationError(
                {
                    "evaluation_points": "evaluation_points is required for EVALUATION_SUBMITTED status"
                }
            )

    ac.evaluation_grade = evaluation_grade
    ac.evaluation_points = evaluation_points

    if completion_status == "SUBMITTED":
        ac.submitted_at = timezone.now()
    elif completion_status == "EVALUATION_SUBMITTED":
        ac.evaluation_submitted_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.get_input_tasks()
        for key, value in ac.completion_data.items():
            task_data = find_first(substasks, pred=lambda x: x["id"] == key)
            if task_data:
                ac.completion_data[key].update(task_data)

    ac.save()

    if completion_status in ["EVALUATION_SUBMITTED", "SUBMITTED"]:
        acl = AssignmentCompletionAuditLog.objects.create(
            assignment_user=assignment_user,
            assignment=assignment,
            course_session=course_session,
            evaluation_user=evaluation_user,
            completion_status=completion_status,
            assignment_user_email=assignment_user.email,
            assignment_slug=assignment.slug,
            completion_data=deepcopy(ac.completion_data),
        )
        if evaluation_user:
            acl.evaluation_user_email = evaluation_user.email

        # copy over the question data, so that we don't lose the context
        subtasks = assignment.get_input_tasks()
        for key, value in acl.completion_data.items():
            task_data = find_first(subtasks, pred=lambda x: x["id"] == key)
            if task_data:
                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
    """
    input_task_ids = [task["id"] for task in assignment.get_input_tasks()]
    filtered_completion_data = {
        key: value for key, value in completion_data.items() if key in input_task_ids
    }
    return filtered_completion_data