diff --git a/server/vbv_lernwelt/dashboard/person_export.py b/server/vbv_lernwelt/dashboard/person_export.py new file mode 100644 index 00000000..e2167ca9 --- /dev/null +++ b/server/vbv_lernwelt/dashboard/person_export.py @@ -0,0 +1,101 @@ +from io import BytesIO + +import structlog +from django.utils.translation import gettext_lazy as _ +from openpyxl import Workbook + +from vbv_lernwelt.core.models import User +from vbv_lernwelt.course.models import CourseSession +from vbv_lernwelt.course_session.services.export_attendance import ( + add_user_export_data, + add_user_headers, + make_export_filename, + sanitize_sheet_name, +) +from vbv_lernwelt.dashboard.utils import create_person_list_with_roles + +PERSONS_EXPORT_FILENAME = _("export_personen") + +logger = structlog.get_logger(__name__) + + +def export_persons( + user: User, + course_session_ids: list[str], + save_as_file: bool = False, +): + if len(course_session_ids) == 0: + return + + wb = Workbook() + # remove the first sheet is just easier than keeping track of the active sheet + wb.remove(wb.active) + + user_with_roles = create_person_list_with_roles(user, course_session_ids) + course_sessions = CourseSession.objects.filter(id__in=course_session_ids) + + for cs in course_sessions: + _create_sheet( + wb, + cs.title, + cs.id, + user_with_roles, + ) + + if save_as_file: + wb.save(make_export_filename(PERSONS_EXPORT_FILENAME)) + else: + output = BytesIO() + wb.save(output) + + output.seek(0) + return output.getvalue() + + +def _create_sheet( + wb: Workbook, + title: str, + cs_id: int, + user_with_roles, +): + sheet = wb.create_sheet(title=sanitize_sheet_name(title)) + + if len(user_with_roles) == 0: + return sheet + + # headers + # common user headers, Circle <learningcontenttitle> bestanden, Circle <title> <learningcontenttitle> Resultat, ... + col_idx = add_user_headers(sheet) + sheet.cell(row=1, column=col_idx, value=str(_("Telefon"))) + sheet.cell(row=1, column=col_idx + 1, value=str(_("Rolle"))) + + _add_rows(sheet, user_with_roles, cs_id) + + return sheet + + +def _add_rows( + sheet, + users, + course_session_id, +): + for row_idx, user in enumerate(users, start=2): + + def get_user_cs_by_id(user_cs, cs_id): + return next((cs for cs in user_cs if cs.get("course_id") == cs_id), None) + + col_idx = add_user_export_data(sheet, user, row_idx) + sheet.cell( + row=row_idx, + column=col_idx, + value=user.user.additional_json_data.get("phone", ""), + ) + sheet.cell( + row=row_idx, + column=col_idx + 1, + value=get_user_cs_by_id(user.course_sessions, course_session_id).get( + "user_role" + ), + ) + + col_idx += 2 diff --git a/server/vbv_lernwelt/dashboard/utils.py b/server/vbv_lernwelt/dashboard/utils.py new file mode 100644 index 00000000..0002004c --- /dev/null +++ b/server/vbv_lernwelt/dashboard/utils.py @@ -0,0 +1,170 @@ +from dataclasses import dataclass +from typing import List, Set + +from vbv_lernwelt.core.models import User +from vbv_lernwelt.course.models import CourseSession, CourseSessionUser +from vbv_lernwelt.course_session_group.models import CourseSessionGroup +from vbv_lernwelt.learning_mentor.models import LearningMentor + + +@dataclass(frozen=True) +class CourseSessionWithRoles: + _original: CourseSession + roles: Set[str] + + def __getattr__(self, name: str): + # Delegate attribute access to the _original CourseSession object + return getattr(self._original, name) + + def save(self, *args, **kwargs): + raise NotImplementedError("This proxy object cannot be saved.") + + +def get_course_sessions_with_roles_for_user(user: User) -> List[CourseSessionWithRoles]: + result_course_sessions = {} + + # participant/member/expert course sessions + csu_qs = CourseSessionUser.objects.filter(user=user).prefetch_related( + "course_session", "course_session__course" + ) + for csu in csu_qs: + cs = csu.course_session + # member/expert is mutually exclusive... + cs.roles = {csu.role} + result_course_sessions[cs.id] = cs + + # enrich with supervisor course sessions + csg_qs = CourseSessionGroup.objects.filter(supervisor=user).prefetch_related( + "course_session", "course_session__course" + ) + for csg in csg_qs: + for cs in csg.course_session.all(): + cs.roles = set() + cs = result_course_sessions.get(cs.id, cs) + + cs.roles.add("SUPERVISOR") + result_course_sessions[cs.id] = cs + + # enrich with mentor course sessions + lm_qs = LearningMentor.objects.filter(mentor=user).prefetch_related( + "course_session", "course_session__course" + ) + for lm in lm_qs: + cs = lm.course_session + cs.roles = set() + cs = result_course_sessions.get(cs.id, cs) + + cs.roles.add("LEARNING_MENTOR") + result_course_sessions[cs.id] = cs + + return [ + CourseSessionWithRoles(cs, cs.roles) for cs in result_course_sessions.values() + ] + + +def has_cs_role(roles: Set[str]) -> bool: + return bool(roles & {"SUPERVISOR", "EXPERT", "MEMBER"}) + + +def user_role(roles: Set[str]) -> str: + if "SUPERVISOR" in roles: + return "SUPERVISOR" + if "EXPERT" in roles: + return "EXPERT" + if "MEMBER" in roles: + return "MEMBER" + return "LEARNING_MENTOR" + + +def create_course_session_dict(course_session_object, my_role, user_role): + return { + "id": str(course_session_object.id), + "session_title": course_session_object.title, + "course_id": str(course_session_object.course.id), + "course_title": course_session_object.course.title, + "course_slug": course_session_object.course.slug, + "region": course_session_object.region, + "generation": course_session_object.generation, + "my_role": my_role, + "user_role": user_role, + "is_uk": course_session_object.course.configuration.is_uk, + "is_vv": course_session_object.course.configuration.is_vv, + } + + +def create_person_list_with_roles(user, course_session_ids=None): + def create_user_dict(user_object): + return { + "user_id": user_object.id, + "first_name": user_object.first_name, + "last_name": user_object.last_name, + "email": user_object.email, + "avatar_url_small": user_object.avatar_url_small, + "avatar_url": user_object.avatar_url, + "course_sessions": [], + } + + course_sessions = get_course_sessions_with_roles_for_user(user) + + result_persons = {} + for cs in course_sessions: + if has_cs_role(cs.roles) and cs.course.configuration.is_uk: + course_session_users = CourseSessionUser.objects.filter( + course_session=cs.id + ).select_related("user") + my_role = user_role(cs.roles) + for csu in course_session_users: + person_data = result_persons.get( + csu.user.id, create_user_dict(csu.user) + ) + person_data["course_sessions"].append( + create_course_session_dict(cs, my_role, csu.role) + ) + result_persons[csu.user.id] = person_data + + # add persons where request.user is mentor + for cs in course_sessions: + if "LEARNING_MENTOR" in cs.roles: + lm = LearningMentor.objects.filter( + mentor=user, course_session=cs.id + ).first() + + for participant in lm.participants.all(): + course_session_entry = create_course_session_dict( + cs, + "LEARNING_MENTOR", + "LEARNING_MENTEE", + ) + + if participant.user.id not in result_persons: + person_data = create_user_dict(participant.user) + person_data["course_sessions"] = [course_session_entry] + result_persons[participant.user.id] = person_data + else: + # user is already in result_persons + result_persons[participant.user.id]["course_sessions"].append( + course_session_entry + ) + + # add persons where request.user is mentee + mentor_relation_qs = LearningMentor.objects.filter( + participants__user=user + ).prefetch_related("mentor", "course_session") + for mentor_relation in mentor_relation_qs: + cs = mentor_relation.course_session + course_session_entry = create_course_session_dict( + cs, + "LEARNING_MENTEE", + "LEARNING_MENTOR", + ) + + if mentor_relation.mentor.id not in result_persons: + person_data = create_user_dict(mentor_relation.mentor) + person_data["course_sessions"] = [course_session_entry] + result_persons[mentor_relation.mentor.id] = person_data + else: + # user is already in result_persons + result_persons[mentor_relation.mentor.id]["course_sessions"].append( + course_session_entry + ) + return result_persons.values() diff --git a/server/vbv_lernwelt/dashboard/views.py b/server/vbv_lernwelt/dashboard/views.py index 2e64c804..8616a2bf 100644 --- a/server/vbv_lernwelt/dashboard/views.py +++ b/server/vbv_lernwelt/dashboard/views.py @@ -2,7 +2,7 @@ import base64 from dataclasses import asdict, dataclass from datetime import date from enum import Enum -from typing import List, Set, Tuple +from typing import List, Tuple from django.db.models import Q from django.http import HttpResponse @@ -24,25 +24,23 @@ from vbv_lernwelt.competence.services import ( query_competence_course_session_edoniq_tests, ) from vbv_lernwelt.core.models import User -from vbv_lernwelt.course.models import ( - CourseConfiguration, - CourseSession, - CourseSessionUser, -) +from vbv_lernwelt.course.models import CourseConfiguration, CourseSessionUser from vbv_lernwelt.course.views import logger from vbv_lernwelt.course_session.services.export_attendance import ( ATTENDANCE_EXPORT_FILENAME, export_attendance, make_export_filename, ) -from vbv_lernwelt.course_session_group.models import CourseSessionGroup +from vbv_lernwelt.dashboard.person_export import export_persons +from vbv_lernwelt.dashboard.utils import ( + CourseSessionWithRoles, + create_course_session_dict, + create_person_list_with_roles, + get_course_sessions_with_roles_for_user, + user_role, +) from vbv_lernwelt.duedate.models import DueDate from vbv_lernwelt.duedate.serializers import DueDateSerializer -from vbv_lernwelt.feedback.export import ( - export_feedback_with_circle_restriction, - FEEDBACK_EXPORT_FILE_NAME, -) -from vbv_lernwelt.learning_mentor.models import LearningMentor from vbv_lernwelt.learnpath.models import Circle from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback @@ -65,19 +63,6 @@ class RoleKeyType(Enum): TRAINER = "Trainer" -@dataclass(frozen=True) -class CourseSessionWithRoles: - _original: CourseSession - roles: Set[str] - - def __getattr__(self, name: str): - # Delegate attribute access to the _original CourseSession object - return getattr(self._original, name) - - def save(self, *args, **kwargs): - raise NotImplementedError("This proxy object cannot be saved.") - - @dataclass(frozen=True) class CourseConfig: course_id: str @@ -92,156 +77,6 @@ class CourseConfig: session_to_continue_id: str | None -def get_course_sessions_with_roles_for_user(user: User) -> List[CourseSessionWithRoles]: - result_course_sessions = {} - - # participant/member/expert course sessions - csu_qs = CourseSessionUser.objects.filter(user=user).prefetch_related( - "course_session", "course_session__course" - ) - for csu in csu_qs: - cs = csu.course_session - # member/expert is mutually exclusive... - cs.roles = {csu.role} - result_course_sessions[cs.id] = cs - - # enrich with supervisor course sessions - csg_qs = CourseSessionGroup.objects.filter(supervisor=user).prefetch_related( - "course_session", "course_session__course" - ) - for csg in csg_qs: - for cs in csg.course_session.all(): - cs.roles = set() - cs = result_course_sessions.get(cs.id, cs) - - cs.roles.add("SUPERVISOR") - result_course_sessions[cs.id] = cs - - # enrich with mentor course sessions - lm_qs = LearningMentor.objects.filter(mentor=user).prefetch_related( - "course_session", "course_session__course" - ) - for lm in lm_qs: - cs = lm.course_session - cs.roles = set() - cs = result_course_sessions.get(cs.id, cs) - - cs.roles.add("LEARNING_MENTOR") - result_course_sessions[cs.id] = cs - - return [ - CourseSessionWithRoles(cs, cs.roles) for cs in result_course_sessions.values() - ] - - -def has_cs_role(roles: Set[str]) -> bool: - return bool(roles & {"SUPERVISOR", "EXPERT", "MEMBER"}) - - -def user_role(roles: Set[str]) -> str: - if "SUPERVISOR" in roles: - return "SUPERVISOR" - if "EXPERT" in roles: - return "EXPERT" - if "MEMBER" in roles: - return "MEMBER" - return "LEARNING_MENTOR" - - -def _create_course_session_dict(course_session_object, my_role, user_role): - return { - "id": str(course_session_object.id), - "session_title": course_session_object.title, - "course_id": str(course_session_object.course.id), - "course_title": course_session_object.course.title, - "course_slug": course_session_object.course.slug, - "region": course_session_object.region, - "generation": course_session_object.generation, - "my_role": my_role, - "user_role": user_role, - "is_uk": course_session_object.course.configuration.is_uk, - "is_vv": course_session_object.course.configuration.is_vv, - } - - -def _create_person_list_with_roles(user): - def create_user_dict(user_object): - return { - "user_id": user_object.id, - "first_name": user_object.first_name, - "last_name": user_object.last_name, - "email": user_object.email, - "avatar_url_small": user_object.avatar_url_small, - "avatar_url": user_object.avatar_url, - "course_sessions": [], - } - - course_sessions = get_course_sessions_with_roles_for_user(user) - - result_persons = {} - for cs in course_sessions: - if has_cs_role(cs.roles) and cs.course.configuration.is_uk: - course_session_users = CourseSessionUser.objects.filter( - course_session=cs.id - ).select_related("user") - my_role = user_role(cs.roles) - for csu in course_session_users: - person_data = result_persons.get( - csu.user.id, create_user_dict(csu.user) - ) - person_data["course_sessions"].append( - _create_course_session_dict(cs, my_role, csu.role) - ) - result_persons[csu.user.id] = person_data - - # add persons where request.user is mentor - for cs in course_sessions: - if "LEARNING_MENTOR" in cs.roles: - lm = LearningMentor.objects.filter( - mentor=user, course_session=cs.id - ).first() - - for participant in lm.participants.all(): - course_session_entry = _create_course_session_dict( - cs, - "LEARNING_MENTOR", - "LEARNING_MENTEE", - ) - - if participant.user.id not in result_persons: - person_data = create_user_dict(participant.user) - person_data["course_sessions"] = [course_session_entry] - result_persons[participant.user.id] = person_data - else: - # user is already in result_persons - result_persons[participant.user.id]["course_sessions"].append( - course_session_entry - ) - - # add persons where request.user is mentee - mentor_relation_qs = LearningMentor.objects.filter( - participants__user=user - ).prefetch_related("mentor", "course_session") - for mentor_relation in mentor_relation_qs: - cs = mentor_relation.course_session - course_session_entry = _create_course_session_dict( - cs, - "LEARNING_MENTEE", - "LEARNING_MENTOR", - ) - - if mentor_relation.mentor.id not in result_persons: - person_data = create_user_dict(mentor_relation.mentor) - person_data["course_sessions"] = [course_session_entry] - result_persons[mentor_relation.mentor.id] = person_data - else: - # user is already in result_persons - result_persons[mentor_relation.mentor.id]["course_sessions"].append( - course_session_entry - ) - return result_persons.values() - - def _persons_list_add_competence_metrics(persons): course_session_ids = {cs["id"] for p in persons for cs in p["course_sessions"]} competence_assignments = query_competence_course_session_assignments( @@ -284,7 +119,7 @@ def _persons_list_add_competence_metrics(persons): @api_view(["GET"]) def get_dashboard_persons(request): try: - persons = list(_create_person_list_with_roles(request.user)) + persons = list(create_person_list_with_roles(request.user)) if request.GET.get("with_competence_metrics", "") == "true": persons = _persons_list_add_competence_metrics(persons) @@ -322,7 +157,7 @@ def get_dashboard_due_dates(request): cs = course_session_map.get(due_date.course_session_id) if cs: - data["course_session"] = _create_course_session_dict( + data["course_session"] = create_course_session_dict( cs, my_role=user_role(cs.roles), user_role="" ) result_due_dates.append(data) @@ -561,20 +396,15 @@ def export_competence_elements_as_xsl(request): @api_view(["POST"]) def export_feedback_as_xsl(request): - circle_ids = request.data.get("circleIds", None) requested_course_session_ids = request.data.get("courseSessionIds", []) course_sessions_with_roles = _get_permitted_courses_sessions_for_user( request.user, requested_course_session_ids ) # noqa - allowed_circles = _get_permitted_circles_ids_for_user_and_course_session( - request.user, - course_sessions_with_roles, - circle_ids, - ) # noqa - - data = export_feedback_with_circle_restriction(allowed_circles, False) - return _make_excel_response(data, FEEDBACK_EXPORT_FILE_NAME) + data = export_persons( + [cswr.id for cswr in course_sessions_with_roles], + ) + return _make_excel_response(data, COMPETENCE_ELEMENT_EXPORT_FILE_NAME) def _get_permitted_courses_sessions_for_user( @@ -608,7 +438,7 @@ def _get_course_sessions_with_roles_for_user( csr for csr in get_course_sessions_with_roles_for_user(user) if any(role in allowed_roles for role in csr.roles) - and csr.id in requested_cs_ids + and csr.id in requested_cs_ids ] # noqa return all_cs_roles_for_user