419 lines
14 KiB
Python
419 lines
14 KiB
Python
from dataclasses import asdict, dataclass
|
|
from enum import Enum
|
|
from typing import List, Set
|
|
|
|
from rest_framework.decorators import api_view
|
|
from rest_framework.exceptions import PermissionDenied
|
|
from rest_framework.response import Response
|
|
|
|
from vbv_lernwelt.assignment.models import (
|
|
AssignmentCompletion,
|
|
AssignmentCompletionStatus,
|
|
)
|
|
from vbv_lernwelt.core.models import User
|
|
from vbv_lernwelt.course.models import (
|
|
CourseConfiguration,
|
|
CourseSession,
|
|
CourseSessionUser,
|
|
)
|
|
from vbv_lernwelt.course.views import logger
|
|
from vbv_lernwelt.course_session_group.models import CourseSessionGroup
|
|
from vbv_lernwelt.learning_mentor.models import LearningMentor
|
|
from vbv_lernwelt.self_evaluation_feedback.models import SelfEvaluationFeedback
|
|
|
|
|
|
class WidgetType(Enum):
|
|
PROGRESS_WIDGET = "ProgressWidget"
|
|
COMPETENCE_WIDGET = "CompetenceWidget"
|
|
MENTOR_TASKS_WIDGET = "MentorTasksWidget"
|
|
MENTOR_PERSON_WIDGET = "MentorPersonWidget"
|
|
MENTOR_COMPETENCE_WIDGET = "MentorCompetenceWidget"
|
|
COMPETENCE_CERTIFICATE_WIDGET = "CompetenceCertificateWidget"
|
|
UK_STATISTICS_WIDGET = "UKStatisticsWidget"
|
|
|
|
|
|
class RoleKeyType(Enum):
|
|
MEMBER = "Member"
|
|
MENTOR_VV = "MentorVV"
|
|
MENTOR_UK = "MentorUK"
|
|
SUPERVISOR = "Supervisor"
|
|
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
|
|
course_slug: str
|
|
course_title: str
|
|
role_key: str
|
|
is_uk: bool
|
|
is_vv: bool
|
|
is_mentor: bool
|
|
widgets: List[str]
|
|
has_preview: bool
|
|
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:
|
|
return (
|
|
"SUPERVISOR"
|
|
if "SUPERVISOR" in roles
|
|
else ("EXPERT" if "EXPERT" in roles else "MEMBER")
|
|
)
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_dashboard_persons(request):
|
|
try:
|
|
course_sessions = get_course_sessions_with_roles_for_user(request.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
|
|
)
|
|
my_role = user_role(cs.roles)
|
|
for csu in course_session_users:
|
|
result_persons[csu.user.id] = {
|
|
"user_id": csu.user.id,
|
|
"first_name": csu.user.first_name,
|
|
"last_name": csu.user.last_name,
|
|
"email": csu.user.email,
|
|
"course_sessions": [
|
|
{
|
|
"id": cs.id,
|
|
"session_title": cs.title,
|
|
"course_id": cs.course.id,
|
|
"course_title": cs.course.title,
|
|
"course_slug": cs.course.slug,
|
|
"user_role": csu.role,
|
|
"my_role": my_role,
|
|
"is_uk": cs.course.configuration.is_uk,
|
|
"is_vv": cs.course.configuration.is_vv,
|
|
}
|
|
],
|
|
}
|
|
|
|
# add persons where request.user is mentor
|
|
for cs in course_sessions:
|
|
if "LEARNING_MENTOR" in cs.roles:
|
|
lm = LearningMentor.objects.filter(
|
|
mentor=request.user, course_session=cs.id
|
|
).first()
|
|
|
|
for participant in lm.participants.all():
|
|
course_session_entry = {
|
|
"id": cs.id,
|
|
"session_title": cs.title,
|
|
"course_id": cs.course.id,
|
|
"course_title": cs.course.title,
|
|
"course_slug": cs.course.slug,
|
|
"user_role": "LEARNING_MENTEE",
|
|
"my_role": "LEARNING_MENTOR",
|
|
"is_uk": cs.course.configuration.is_uk,
|
|
"is_vv": cs.course.configuration.is_vv,
|
|
}
|
|
|
|
if participant.user.id not in result_persons:
|
|
result_persons[participant.user.id] = {
|
|
"user_id": participant.user.id,
|
|
"first_name": participant.user.first_name,
|
|
"last_name": participant.user.last_name,
|
|
"email": participant.user.email,
|
|
"course_sessions": [course_session_entry],
|
|
}
|
|
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=request.user
|
|
).prefetch_related("mentor", "course_session")
|
|
for mentor_relation in mentor_relation_qs:
|
|
cs = mentor_relation.course_session
|
|
course_session_entry = {
|
|
"id": cs.id,
|
|
"session_title": cs.title,
|
|
"course_id": cs.course.id,
|
|
"course_title": cs.course.title,
|
|
"course_slug": cs.course.slug,
|
|
"user_role": "LEARNING_MENTOR",
|
|
"my_role": "LEARNING_MENTEE",
|
|
"is_uk": cs.course.configuration.is_uk,
|
|
"is_vv": cs.course.configuration.is_vv,
|
|
}
|
|
|
|
if mentor_relation.mentor.id not in result_persons:
|
|
result_persons[mentor_relation.mentor.id] = {
|
|
"user_id": mentor_relation.mentor.id,
|
|
"first_name": mentor_relation.mentor.first_name,
|
|
"last_name": mentor_relation.mentor.last_name,
|
|
"email": mentor_relation.mentor.email,
|
|
"course_sessions": [course_session_entry],
|
|
}
|
|
else:
|
|
# user is already in result_persons
|
|
result_persons[mentor_relation.mentor.id]["course_sessions"].append(
|
|
course_session_entry
|
|
)
|
|
|
|
return Response(
|
|
status=200,
|
|
data=list(result_persons.values()),
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
def get_widgets_for_course(
|
|
role_key: RoleKeyType, is_uk: bool, is_vv: bool, is_mentor: bool
|
|
) -> List[str]:
|
|
widgets = []
|
|
|
|
if role_key == RoleKeyType.MEMBER:
|
|
widgets.append(WidgetType.PROGRESS_WIDGET.value)
|
|
widgets.append(WidgetType.COMPETENCE_WIDGET.value)
|
|
if is_uk:
|
|
widgets.append(WidgetType.COMPETENCE_CERTIFICATE_WIDGET.value)
|
|
|
|
if role_key in [RoleKeyType.SUPERVISOR, RoleKeyType.TRAINER] and is_uk:
|
|
widgets.append(WidgetType.UK_STATISTICS_WIDGET.value)
|
|
|
|
if is_mentor:
|
|
widgets.append(WidgetType.MENTOR_PERSON_WIDGET.value)
|
|
if is_vv:
|
|
widgets.append(WidgetType.MENTOR_TASKS_WIDGET.value)
|
|
if is_uk:
|
|
widgets.append(WidgetType.MENTOR_COMPETENCE_WIDGET.value)
|
|
|
|
return widgets
|
|
|
|
|
|
def get_role_key_and_mentor(
|
|
course_sessions: List[CourseSessionWithRoles], is_uk: bool, is_vv: bool
|
|
) -> tuple[RoleKeyType, bool]:
|
|
roles = set()
|
|
role = None
|
|
for cs in course_sessions:
|
|
roles.update(cs.roles)
|
|
|
|
if "SUPERVISOR" in roles:
|
|
role = RoleKeyType.SUPERVISOR
|
|
elif "EXPERT" in roles:
|
|
role = RoleKeyType.TRAINER
|
|
elif "MEMBER" in roles:
|
|
role = RoleKeyType.MEMBER
|
|
elif "LEARNING_MENTOR" in roles:
|
|
if is_uk:
|
|
role = RoleKeyType.MENTOR_UK
|
|
elif is_vv:
|
|
role = RoleKeyType.MENTOR_VV
|
|
|
|
is_mentor = "LEARNING_MENTOR" in roles
|
|
return role, is_mentor
|
|
|
|
|
|
def collect_course_sessions_by_course(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> dict:
|
|
course_sessions_by_course = {}
|
|
for cs in course_sessions:
|
|
if cs.course.id not in course_sessions_by_course:
|
|
course_sessions_by_course[cs.course.id] = []
|
|
course_sessions_by_course[cs.course.id].append(cs)
|
|
return course_sessions_by_course
|
|
|
|
|
|
def has_preview(role_key: RoleKeyType) -> bool:
|
|
return (
|
|
role_key in [RoleKeyType.MENTOR_VV, RoleKeyType.MENTOR_UK]
|
|
and not role_key == RoleKeyType.MEMBER
|
|
)
|
|
|
|
|
|
def get_newest_cs(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> CourseSessionWithRoles | None:
|
|
newest: CourseSessionWithRoles | None = None
|
|
|
|
for cs in course_sessions:
|
|
generation_newest = newest.generation if newest else None
|
|
if generation_newest is None or cs.generation > generation_newest:
|
|
newest = cs
|
|
|
|
return newest
|
|
|
|
|
|
def get_course_config(
|
|
course_sessions: List[CourseSessionWithRoles],
|
|
) -> List[CourseConfig]:
|
|
course_configs = []
|
|
cs_by_course = collect_course_sessions_by_course(course_sessions)
|
|
for _id, cs_in_course in cs_by_course.items():
|
|
is_uk = cs_in_course[0].course.configuration.is_uk
|
|
is_vv = cs_in_course[0].course.configuration.is_vv
|
|
role_key, is_mentor = get_role_key_and_mentor(cs_in_course, is_uk, is_vv)
|
|
session_to_continue = get_newest_cs(cs_in_course)
|
|
course_configs.append(
|
|
CourseConfig(
|
|
course_id=str(cs_in_course[0].course.id),
|
|
course_slug=cs_in_course[0].course.slug,
|
|
course_title=cs_in_course[0].course.title,
|
|
role_key=role_key.value,
|
|
is_uk=is_uk,
|
|
is_vv=is_vv,
|
|
is_mentor=is_mentor,
|
|
widgets=get_widgets_for_course(role_key, is_uk, is_vv, is_mentor),
|
|
has_preview=has_preview(role_key),
|
|
session_to_continue_id=str(session_to_continue.id)
|
|
if session_to_continue
|
|
else None,
|
|
)
|
|
)
|
|
|
|
return course_configs
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_dashboard_config(request):
|
|
try:
|
|
course_sessions = get_course_sessions_with_roles_for_user(request.user) # noqa
|
|
course_configs = get_course_config(course_sessions)
|
|
|
|
return Response(
|
|
status=200,
|
|
data=[asdict(cc) for cc in course_configs],
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_mentee_count(request, course_id: str):
|
|
try:
|
|
return Response(
|
|
status=200,
|
|
data={"mentee_count": _get_mentee_count(course_id, request.user)}, # noqa
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
def _get_mentee_count(course_id: str, mentor: User) -> int:
|
|
return CourseSessionUser.objects.filter(
|
|
participants__mentor=mentor, course_session__course__id=course_id
|
|
).count()
|
|
|
|
|
|
@api_view(["GET"])
|
|
def get_mentor_open_tasks_count(request, course_id: str):
|
|
try:
|
|
return Response(
|
|
status=200,
|
|
data={
|
|
"open_task_count": _get_mentor_open_tasks_count(
|
|
course_id, request.user
|
|
) # noqa
|
|
},
|
|
)
|
|
except PermissionDenied as e:
|
|
raise e
|
|
except Exception as e:
|
|
logger.error(e, exc_info=True)
|
|
return Response({"error": str(e)}, status=404)
|
|
|
|
|
|
def _get_mentor_open_tasks_count(course_id: str, mentor: User) -> int:
|
|
open_assigment_count = 0
|
|
open_feedback_count = 0
|
|
|
|
course_configuration = CourseConfiguration.objects.get(course_id=course_id)
|
|
|
|
if course_configuration.is_vv:
|
|
open_assigment_count = AssignmentCompletion.objects.filter(
|
|
course_session__course__id=course_id,
|
|
completion_status=AssignmentCompletionStatus.SUBMITTED.value,
|
|
evaluation_user=mentor, # noqa
|
|
assignment_user__coursesessionuser__participants__mentor=mentor,
|
|
).count()
|
|
|
|
open_feedback_count = SelfEvaluationFeedback.objects.filter(
|
|
feedback_provider_user=mentor, # noqa
|
|
feedback_requester_user__coursesessionuser__participants__mentor=mentor,
|
|
feedback_submitted=False,
|
|
).count()
|
|
|
|
return open_assigment_count + open_feedback_count
|