494 lines
17 KiB
Python
494 lines
17 KiB
Python
from copy import deepcopy
|
|
from dataclasses import dataclass
|
|
from io import BytesIO
|
|
|
|
import structlog
|
|
from django.utils import timezone
|
|
from openpyxl import Workbook
|
|
from rest_framework import serializers
|
|
from wagtail.models import Page
|
|
|
|
from vbv_lernwelt.assignment.models import (
|
|
Assignment,
|
|
AssignmentCompletion,
|
|
AssignmentCompletionAuditLog,
|
|
AssignmentCompletionStatus,
|
|
AssignmentType,
|
|
is_valid_assignment_completion_status,
|
|
recalculate_assignment_passed,
|
|
)
|
|
from vbv_lernwelt.core.models import User
|
|
from vbv_lernwelt.core.utils import find_first
|
|
from vbv_lernwelt.course.models import (
|
|
CourseCompletionStatus,
|
|
CourseSession,
|
|
CourseSessionUser,
|
|
)
|
|
from vbv_lernwelt.course.services import mark_course_completion
|
|
from vbv_lernwelt.course_session.models import (
|
|
CourseSessionAssignment,
|
|
CourseSessionEdoniqTest,
|
|
)
|
|
from vbv_lernwelt.course_session.services.export import (
|
|
add_user_export_data,
|
|
add_user_headers,
|
|
get_ordered_csus_by_course_session,
|
|
group_by_session_title,
|
|
make_export_filename,
|
|
sanitize_sheet_name,
|
|
)
|
|
from vbv_lernwelt.duedate.models import DueDate
|
|
from vbv_lernwelt.learnpath.models import LearningContent
|
|
from vbv_lernwelt.notify.services import NotificationService
|
|
|
|
logger = structlog.get_logger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class CompetenceCertificateElement:
|
|
assignment: Assignment
|
|
date: DueDate
|
|
learning_content: LearningContent
|
|
course_session: CourseSession
|
|
|
|
|
|
def update_assignment_completion(
|
|
assignment_user: User,
|
|
assignment: Assignment,
|
|
course_session: CourseSession,
|
|
learning_content_page: Page | None = None,
|
|
completion_data=None,
|
|
completion_status: AssignmentCompletionStatus = AssignmentCompletionStatus.IN_PROGRESS,
|
|
evaluation_user: User | None = None,
|
|
evaluation_points: float | None = None,
|
|
evaluation_passed: bool | None = None,
|
|
evaluation_max_points: float | None = None,
|
|
validate_completion_status_change: bool = True,
|
|
copy_task_data: bool = False,
|
|
initialize_completion: bool = False,
|
|
additional_json_data: dict | None = None,
|
|
edoniq_extended_time_flag: bool = False,
|
|
validate_submission_update: bool = True,
|
|
) -> tuple[AssignmentCompletion, bool]:
|
|
"""
|
|
: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
|
|
:param initialize_completion: if true, the completion will be created, but not updated
|
|
used as a workaround for initial work with the object on the frontend
|
|
"""
|
|
if completion_data is None:
|
|
completion_data = {}
|
|
|
|
if additional_json_data is None:
|
|
additional_json_data = {}
|
|
|
|
ac, created = AssignmentCompletion.objects.get_or_create(
|
|
assignment_user_id=assignment_user.id,
|
|
assignment_id=assignment.id,
|
|
course_session_id=course_session.id,
|
|
learning_content_page_id=(
|
|
learning_content_page.id if learning_content_page else None
|
|
),
|
|
)
|
|
|
|
if initialize_completion:
|
|
# ignore further updates
|
|
return ac, True
|
|
|
|
if not is_valid_assignment_completion_status(completion_status):
|
|
raise serializers.ValidationError(
|
|
{
|
|
"completion_status": f"Invalid completion status {completion_status.value}"
|
|
}
|
|
)
|
|
|
|
if validate_completion_status_change:
|
|
if completion_status == AssignmentCompletionStatus.SUBMITTED:
|
|
completion_status_to_check = [
|
|
"EVALUATION_IN_PROGRESS",
|
|
"EVALUATION_SUBMITTED",
|
|
]
|
|
if validate_submission_update:
|
|
completion_status_to_check.append("SUBMITTED")
|
|
if ac.completion_status in completion_status_to_check:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"completion_status": f"Cannot update completion status from {ac.completion_status} to SUBMITTED"
|
|
}
|
|
)
|
|
elif (
|
|
completion_status == AssignmentCompletionStatus.EVALUATION_SUBMITTED
|
|
and validate_submission_update
|
|
):
|
|
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 == AssignmentCompletionStatus.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 [
|
|
AssignmentCompletionStatus.EVALUATION_SUBMITTED,
|
|
AssignmentCompletionStatus.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 == AssignmentCompletionStatus.EVALUATION_SUBMITTED:
|
|
if evaluation_points is None:
|
|
raise serializers.ValidationError(
|
|
{
|
|
"evaluation_points": "evaluation_points is required for EVALUATION_SUBMITTED status"
|
|
}
|
|
)
|
|
|
|
if evaluation_max_points is None:
|
|
ac.evaluation_max_points = assignment.get_max_points()
|
|
else:
|
|
ac.evaluation_max_points = evaluation_max_points
|
|
ac.evaluation_points = evaluation_points
|
|
|
|
# if no evaluation_passed is provided, we calculate it from the points
|
|
if evaluation_passed is None and ac.evaluation_max_points > 0:
|
|
recalculate_assignment_passed(ac)
|
|
else:
|
|
ac.evaluation_passed = evaluation_passed
|
|
|
|
if completion_status == AssignmentCompletionStatus.SUBMITTED:
|
|
ac.submitted_at = timezone.now()
|
|
if evaluation_user:
|
|
ac.evaluation_user = evaluation_user
|
|
NotificationService.send_assignment_submitted_notification(
|
|
recipient=evaluation_user,
|
|
sender=ac.assignment_user,
|
|
assignment_completion=ac,
|
|
)
|
|
elif completion_status == AssignmentCompletionStatus.EVALUATION_SUBMITTED:
|
|
ac.evaluation_submitted_at = timezone.now()
|
|
learning_content_assignment = (
|
|
learning_content_page
|
|
if learning_content_page
|
|
else assignment.find_attached_learning_content()
|
|
)
|
|
if learning_content_assignment:
|
|
assignment_frontend_url = learning_content_assignment.get_frontend_url()
|
|
NotificationService.send_assignment_evaluated_notification(
|
|
recipient=ac.assignment_user,
|
|
sender=evaluation_user,
|
|
assignment_completion=ac,
|
|
target_url=assignment_frontend_url,
|
|
)
|
|
|
|
ac.completion_status = completion_status.value
|
|
ac.edoniq_extended_time_flag = edoniq_extended_time_flag
|
|
ac.additional_json_data = ac.additional_json_data | additional_json_data
|
|
|
|
task_ids = [task.id for task in assignment.tasks]
|
|
|
|
for key, value in completion_data.items():
|
|
if key in task_ids:
|
|
stored_entry = ac.completion_data.get(key, {})
|
|
stored_entry.update(value)
|
|
ac.completion_data[key] = stored_entry
|
|
|
|
# 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
|
|
sub_tasks = assignment.get_input_tasks()
|
|
for key, value in ac.completion_data.items():
|
|
task_data = find_first(sub_tasks, pred=lambda x: x["id"] == key)
|
|
if task_data:
|
|
ac.completion_data[key].update(task_data)
|
|
|
|
ac.save()
|
|
|
|
if (
|
|
completion_status
|
|
in [
|
|
AssignmentCompletionStatus.EVALUATION_SUBMITTED,
|
|
AssignmentCompletionStatus.SUBMITTED,
|
|
]
|
|
and assignment.assignment_type != AssignmentType.EDONIQ_TEST.value
|
|
):
|
|
acl = AssignmentCompletionAuditLog.objects.create(
|
|
assignment_user=assignment_user,
|
|
assignment=assignment,
|
|
course_session=course_session,
|
|
evaluation_user=evaluation_user,
|
|
completion_status=completion_status.value,
|
|
assignment_user_email=assignment_user.email,
|
|
assignment_slug=assignment.slug,
|
|
completion_data=deepcopy(ac.completion_data),
|
|
evaluation_points=evaluation_points,
|
|
evaluation_max_points=ac.evaluation_max_points,
|
|
evaluation_passed=ac.evaluation_passed,
|
|
)
|
|
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()
|
|
|
|
if completion_status in [
|
|
AssignmentCompletionStatus.EVALUATION_SUBMITTED,
|
|
AssignmentCompletionStatus.EVALUATION_IN_PROGRESS,
|
|
AssignmentCompletionStatus.SUBMITTED,
|
|
]:
|
|
learning_content = (
|
|
learning_content_page
|
|
if learning_content_page
|
|
else assignment.find_attached_learning_content()
|
|
)
|
|
if learning_content:
|
|
mark_course_completion(
|
|
user=assignment_user,
|
|
page=learning_content,
|
|
course_session=course_session,
|
|
completion_status=CourseCompletionStatus.SUCCESS.value,
|
|
)
|
|
|
|
return ac, created
|
|
|
|
|
|
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
|
|
|
|
|
|
def export_competence_certificates(
|
|
course_session_ids: list[str],
|
|
circle_ids: list[int] = None,
|
|
save_as_file: bool = False,
|
|
):
|
|
if len(course_session_ids) == 0:
|
|
return
|
|
|
|
COMPETENCE_ASSIGNMENT_TYPES = [
|
|
AssignmentType.CASEWORK.value,
|
|
AssignmentType.EDONIQ_TEST.value,
|
|
]
|
|
|
|
wb = Workbook()
|
|
# remove the first sheet is just easier than keeping track of the active sheet
|
|
wb.remove(wb.active)
|
|
|
|
competence_certificate_elements = _get_competence_certificate_elements(
|
|
course_session_ids
|
|
)
|
|
|
|
assignemnt_completions = AssignmentCompletion.objects.filter(
|
|
course_session_id__in=course_session_ids,
|
|
assignment__assignment_type__in=COMPETENCE_ASSIGNMENT_TYPES,
|
|
).order_by("course_session", "assignment")
|
|
|
|
# group all by the sessions title {session_id1: [...], session_id2: [...], ...}
|
|
grouped_cs_users = get_ordered_csus_by_course_session(course_session_ids)
|
|
grouped_cce = group_by_session_title(competence_certificate_elements)
|
|
grouped_ac = group_by_session_title(assignemnt_completions)
|
|
|
|
# create a sheet for each course session
|
|
for course_session, cs_users in grouped_cs_users.items():
|
|
logger.debug(
|
|
"export_assignment_completion",
|
|
data={
|
|
"course_session": course_session,
|
|
},
|
|
label="assignment_export",
|
|
)
|
|
_create_sheet(
|
|
wb,
|
|
course_session,
|
|
cs_users,
|
|
grouped_cce[course_session],
|
|
grouped_ac[course_session],
|
|
circle_ids,
|
|
)
|
|
|
|
if save_as_file:
|
|
wb.save(make_export_filename(name="competence_certificate_export"))
|
|
else:
|
|
output = BytesIO()
|
|
wb.save(output)
|
|
|
|
output.seek(0)
|
|
return output.getvalue()
|
|
|
|
|
|
def _create_sheet(
|
|
wb: Workbook,
|
|
title: str,
|
|
users: list[CourseSessionUser],
|
|
competence_certificate_element: list[CompetenceCertificateElement],
|
|
assignment_completions: list[AssignmentCompletion],
|
|
circle_ids: list[int],
|
|
):
|
|
sheet = wb.create_sheet(title=sanitize_sheet_name(title))
|
|
|
|
if len(users) == 0:
|
|
return sheet
|
|
|
|
# headers
|
|
# common user headers, Circle <title> <learningcontenttitle> bestanden, Circle <title> <learningcontenttitle> Resultat, ...
|
|
col_idx = add_user_headers(sheet)
|
|
|
|
ordered_assignement_ids = (
|
|
[]
|
|
) # keep track of the order of the columns when adding the rows
|
|
for cse in competence_certificate_element:
|
|
circle = cse.learning_content.get_circle()
|
|
|
|
if circle_ids and circle.id not in circle_ids:
|
|
continue
|
|
|
|
col_prefix = f'Circle "{circle.title}" {cse.learning_content.title} '
|
|
|
|
sheet.cell(
|
|
row=1,
|
|
column=col_idx,
|
|
value=f"{col_prefix} bestanden",
|
|
)
|
|
|
|
sheet.cell(
|
|
row=1,
|
|
column=col_idx + 1,
|
|
value=f"{col_prefix} Resultat %",
|
|
)
|
|
|
|
ordered_assignement_ids.append(cse.assignment.id)
|
|
|
|
col_idx += 2
|
|
|
|
# add rows with user results
|
|
_add_rows(sheet, users, ordered_assignement_ids, assignment_completions)
|
|
|
|
return sheet
|
|
|
|
|
|
def _add_rows(
|
|
sheet,
|
|
users: list[CourseSessionUser],
|
|
ordered_assignement_ids,
|
|
assignment_completions,
|
|
):
|
|
for row_idx, user in enumerate(users, start=2):
|
|
col_idx = add_user_export_data(sheet, user, row_idx)
|
|
|
|
for assignment_id in ordered_assignement_ids:
|
|
# get the completion for the user and the assignment
|
|
user_acs = [
|
|
ac
|
|
for ac in assignment_completions
|
|
if ac.assignment_id == assignment_id and ac.assignment_user == user.user
|
|
]
|
|
user_ac = user_acs[0] if user_acs else None
|
|
|
|
if user_ac:
|
|
status_text = (
|
|
"Bestanden" if user_ac.evaluation_passed else "Nicht bestanden"
|
|
)
|
|
sheet.cell(row=row_idx, column=col_idx, value=status_text)
|
|
try:
|
|
sheet.cell(
|
|
row=row_idx,
|
|
column=col_idx + 1,
|
|
value=round(
|
|
100
|
|
* user_ac.evaluation_points
|
|
/ user_ac.evaluation_max_points
|
|
),
|
|
)
|
|
except (ZeroDivisionError, TypeError):
|
|
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
|
|
|
|
else:
|
|
sheet.cell(row=row_idx, column=col_idx, value="Keine Daten")
|
|
sheet.cell(row=row_idx, column=col_idx + 1, value="Keine Daten")
|
|
|
|
col_idx += 2
|
|
|
|
|
|
def _get_competence_certificate_elements(
|
|
course_session_ids: list[str],
|
|
) -> list[CompetenceCertificateElement]:
|
|
course_session_assignments = CourseSessionAssignment.objects.filter(
|
|
course_session__id__in=course_session_ids,
|
|
learning_content__content_assignment__competence_certificate__isnull=False,
|
|
).order_by("course_session", "submission_deadline__start")
|
|
|
|
course_session_edoniqtests = CourseSessionEdoniqTest.objects.filter(
|
|
course_session__id__in=course_session_ids,
|
|
learning_content__content_assignment__competence_certificate__isnull=False,
|
|
).order_by("course_session", "deadline__start")
|
|
|
|
cse = [
|
|
CompetenceCertificateElement(
|
|
assignment=csa.learning_content.content_assignment,
|
|
date=csa.submission_deadline,
|
|
learning_content=csa.learning_content,
|
|
course_session=csa.course_session,
|
|
)
|
|
for csa in course_session_assignments
|
|
]
|
|
|
|
cse += [
|
|
CompetenceCertificateElement(
|
|
assignment=cset.learning_content.content_assignment,
|
|
date=cset.deadline,
|
|
learning_content=cset.learning_content,
|
|
course_session=cset.course_session,
|
|
)
|
|
for cset in course_session_edoniqtests
|
|
]
|
|
|
|
# order by course_session and submission_deadline
|
|
cse.sort(key=lambda x: (x.course_session.title, x.date.start))
|
|
|
|
return cse
|