wip: Add functions for person export [skip ci]

This commit is contained in:
Christian Cueni 2024-07-15 15:10:33 +02:00
parent a024a2f244
commit fd2cbb96bc
3 changed files with 288 additions and 187 deletions

View File

@ -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 <title> <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

View File

@ -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()

View File

@ -2,7 +2,7 @@ import base64
from dataclasses import asdict, dataclass from dataclasses import asdict, dataclass
from datetime import date from datetime import date
from enum import Enum from enum import Enum
from typing import List, Set, Tuple from typing import List, Tuple
from django.db.models import Q from django.db.models import Q
from django.http import HttpResponse from django.http import HttpResponse
@ -24,25 +24,23 @@ from vbv_lernwelt.competence.services import (
query_competence_course_session_edoniq_tests, query_competence_course_session_edoniq_tests,
) )
from vbv_lernwelt.core.models import User from vbv_lernwelt.core.models import User
from vbv_lernwelt.course.models import ( from vbv_lernwelt.course.models import CourseConfiguration, CourseSessionUser
CourseConfiguration,
CourseSession,
CourseSessionUser,
)
from vbv_lernwelt.course.views import logger from vbv_lernwelt.course.views import logger
from vbv_lernwelt.course_session.services.export_attendance import ( from vbv_lernwelt.course_session.services.export_attendance import (
ATTENDANCE_EXPORT_FILENAME, ATTENDANCE_EXPORT_FILENAME,
export_attendance, export_attendance,
make_export_filename, 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.models import DueDate
from vbv_lernwelt.duedate.serializers import DueDateSerializer 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.learnpath.models import Circle
from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback
@ -65,19 +63,6 @@ class RoleKeyType(Enum):
TRAINER = "Trainer" 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) @dataclass(frozen=True)
class CourseConfig: class CourseConfig:
course_id: str course_id: str
@ -92,156 +77,6 @@ class CourseConfig:
session_to_continue_id: str | None 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): def _persons_list_add_competence_metrics(persons):
course_session_ids = {cs["id"] for p in persons for cs in p["course_sessions"]} course_session_ids = {cs["id"] for p in persons for cs in p["course_sessions"]}
competence_assignments = query_competence_course_session_assignments( competence_assignments = query_competence_course_session_assignments(
@ -284,7 +119,7 @@ def _persons_list_add_competence_metrics(persons):
@api_view(["GET"]) @api_view(["GET"])
def get_dashboard_persons(request): def get_dashboard_persons(request):
try: 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": if request.GET.get("with_competence_metrics", "") == "true":
persons = _persons_list_add_competence_metrics(persons) 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) cs = course_session_map.get(due_date.course_session_id)
if cs: 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="" cs, my_role=user_role(cs.roles), user_role=""
) )
result_due_dates.append(data) result_due_dates.append(data)
@ -561,20 +396,15 @@ def export_competence_elements_as_xsl(request):
@api_view(["POST"]) @api_view(["POST"])
def export_feedback_as_xsl(request): def export_feedback_as_xsl(request):
circle_ids = request.data.get("circleIds", None)
requested_course_session_ids = request.data.get("courseSessionIds", []) requested_course_session_ids = request.data.get("courseSessionIds", [])
course_sessions_with_roles = _get_permitted_courses_sessions_for_user( course_sessions_with_roles = _get_permitted_courses_sessions_for_user(
request.user, requested_course_session_ids request.user, requested_course_session_ids
) # noqa ) # noqa
allowed_circles = _get_permitted_circles_ids_for_user_and_course_session( data = export_persons(
request.user, [cswr.id for cswr in course_sessions_with_roles],
course_sessions_with_roles, )
circle_ids, return _make_excel_response(data, COMPETENCE_ELEMENT_EXPORT_FILE_NAME)
) # noqa
data = export_feedback_with_circle_restriction(allowed_circles, False)
return _make_excel_response(data, FEEDBACK_EXPORT_FILE_NAME)
def _get_permitted_courses_sessions_for_user( def _get_permitted_courses_sessions_for_user(
@ -608,7 +438,7 @@ def _get_course_sessions_with_roles_for_user(
csr csr
for csr in get_course_sessions_with_roles_for_user(user) for csr in get_course_sessions_with_roles_for_user(user)
if any(role in allowed_roles for role in csr.roles) 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 ] # noqa
return all_cs_roles_for_user return all_cs_roles_for_user