147 lines
5.1 KiB
Python
147 lines
5.1 KiB
Python
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_sequence_ids():
|
|
distinct_sequence_ids = LearningContentEdoniqTest.objects.values_list(
|
|
"edoniq_sequence_id", flat=True
|
|
).distinct()
|
|
|
|
distinct_sequence_extended_ids = LearningContentEdoniqTest.objects.values_list(
|
|
"edoniq_extended_sequence_id", flat=True
|
|
).distinct()
|
|
|
|
# filter out empty values
|
|
return {
|
|
seq_id
|
|
for seq_id in list(distinct_sequence_ids) + list(distinct_sequence_extended_ids)
|
|
if seq_id
|
|
}
|
|
|
|
|
|
def filter_relevant_rows(csv_data):
|
|
sequence_ids = get_distinct_sequence_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("Sequenz ID") in sequence_ids]
|
|
|
|
return relevant_rows
|
|
|
|
|
|
@dataclass
|
|
class EdoniqCsvTestResult:
|
|
user_id: str
|
|
user_points: float
|
|
max_points: float
|
|
passed: bool
|
|
edoniq_sequence_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_sequence_id = edoniq_csv_row_dict["Sequenz ID"]
|
|
|
|
return EdoniqCsvTestResult(
|
|
user_id=user_id,
|
|
user_points=user_points,
|
|
max_points=max_points,
|
|
passed=passed,
|
|
edoniq_sequence_id=edoniq_sequence_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_sequence_id=edoniq_csv_test_result.edoniq_sequence_id)
|
|
| Q(edoniq_extended_sequence_id=edoniq_csv_test_result.edoniq_sequence_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)
|
|
)
|
|
|
|
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
|
|
|
|
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_sequence_id": edoniq_csv_test_result.edoniq_sequence_id,
|
|
},
|
|
validate_submission_update=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,
|
|
sequence_id=edoniq_csv_test_result.edoniq_sequence_id,
|
|
exc_info=True,
|
|
)
|
|
if not fail_silently:
|
|
raise e
|
|
return None, False
|