vbv/server/vbv_lernwelt/edoniq_test/result_import/services.py

import csv
from dataclasses import dataclass
from io import StringIO

import structlog
from django.db.models import Q

from vbv_lernwelt.assignment.models import AssignmentCompletionStatus
from vbv_lernwelt.assignment.services import update_assignment_completion
from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.models import CourseSession, CourseSessionUser
from vbv_lernwelt.learnpath.models import LearningContentEdoniqTest

logger = structlog.get_logger(__name__)


def get_distinct_course_release_ids():
    distinct_course_release_ids = LearningContentEdoniqTest.objects.values_list(
        "edoniq_course_release_id", flat=True
    ).distinct()

    distinct_course_release_extended_ids = (
        LearningContentEdoniqTest.objects.values_list(
            "edoniq_extended_course_release_id", flat=True
        ).distinct()
    )

    # filter out empty values
    return {
        seq_id
        for seq_id in list(distinct_course_release_ids)
        + list(distinct_course_release_extended_ids)
        if seq_id
    }


def filter_relevant_rows(csv_data):
    release_ids = get_distinct_course_release_ids()
    csv_reader = csv.reader(StringIO(csv_data), delimiter=";")

    # read headers
    headers = next(csv_reader)
    headers = [header.strip() for header in headers]

    # Read the rest of the lines and create a list of dictionaries
    records = []
    for row in csv_reader:
        # strip all whitespace around the column entries
        row = [item.strip() for item in row]
        records.append(dict(zip(headers, row)))

    relevant_rows = [
        row for row in records if row.get("Kursfreigaben ID") in release_ids
    ]

    return relevant_rows


@dataclass
class EdoniqCsvTestResult:
    user_id: str
    user_points: float
    max_points: float
    passed: bool
    edoniq_course_release_id: str
    unfinished: bool = False


def create_edoniq_csv_test_result(edoniq_csv_row_dict) -> EdoniqCsvTestResult:
    user_id = edoniq_csv_row_dict["Benutzer Login"]
    user_points = float(edoniq_csv_row_dict["erreichte Punkte absolut"])
    max_points = float(edoniq_csv_row_dict["erreichbares Punktetotal absolut"])
    passed = edoniq_csv_row_dict["Bestanden-/Nicht-Bestanden Status"] == "Bestanden"
    unfinished = edoniq_csv_row_dict["Bestanden-/Nicht-Bestanden Status"] == "Begonnen"
    edoniq_course_release_id = edoniq_csv_row_dict["Kursfreigaben ID"]

    return EdoniqCsvTestResult(
        user_id=user_id,
        user_points=user_points,
        max_points=max_points,
        passed=passed,
        edoniq_course_release_id=edoniq_course_release_id,
        unfinished=unfinished,
    )


def upsert_edoniq_test_result(
    edoniq_csv_test_result: EdoniqCsvTestResult, fail_silently=False
):
    try:
        u = User.objects.get(id=edoniq_csv_test_result.user_id)

        # because the same edoniq test is used in multiple languages,
        # we have to consider multiple learning contents
        learning_content_qs = LearningContentEdoniqTest.objects.filter(
            Q(edoniq_course_release_id=edoniq_csv_test_result.edoniq_course_release_id)
            | Q(
                edoniq_extended_course_release_id=edoniq_csv_test_result.edoniq_course_release_id
            )
        )

        user_course_session_ids = set(
            CourseSessionUser.objects.filter(
                user_id=u.id,
                course_session__course_id__in=[
                    lc.get_course().id for lc in learning_content_qs
                ],
            ).values_list("course_session__id", flat=True)
        )

        learning_content = learning_content_qs.first()

        for course_session_id in user_course_session_ids:
            course_session = CourseSession.objects.get(id=course_session_id)

            # find the assignment which is attached to this language course
            assignment = (
                learning_content_qs.descendant_of(course_session.course.coursepage)
                .first()
                .content_assignment
            )

            completion_status = AssignmentCompletionStatus.EVALUATION_SUBMITTED
            if edoniq_csv_test_result.unfinished:
                completion_status = AssignmentCompletionStatus.SUBMITTED

            edoniq_extended_time_flag = False
            if (
                learning_content.edoniq_extended_course_release_id
                == edoniq_csv_test_result.edoniq_course_release_id
            ):
                edoniq_extended_time_flag = True

            return update_assignment_completion(
                assignment_user=u,
                assignment=assignment,
                course_session=course_session,
                learning_content_page=assignment.find_attached_learning_content(),
                completion_data={},
                completion_status=completion_status,
                evaluation_user=User.objects.get(username="admin"),
                evaluation_points=edoniq_csv_test_result.user_points,
                evaluation_passed=edoniq_csv_test_result.passed,
                evaluation_max_points=edoniq_csv_test_result.max_points,
                additional_json_data={
                    "edoniq_course_release_id": edoniq_csv_test_result.edoniq_course_release_id,
                },
                edoniq_extended_time_flag=edoniq_extended_time_flag,
                validate_submission_update=False,
            )
        return None, False

    except Exception as e:
        logger.warning(
            "edoniq_test_result_import_fail",
            label="edoniq_import_results",
            error="user or learning_content not found",
            user_id=edoniq_csv_test_result.user_id,
            edoniq_course_release_id=edoniq_csv_test_result.edoniq_course_release_id,
            exc_info=True,
        )
        if not fail_silently:
            raise e
        return None, False