124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
import csv
|
|
from datetime import date
|
|
from itertools import chain
|
|
from typing import List
|
|
|
|
from django.contrib.admin.views.decorators import staff_member_required
|
|
from django.db.models import CharField, Q, Value
|
|
from django.http import HttpResponse
|
|
|
|
from vbv_lernwelt.core.models import User
|
|
from vbv_lernwelt.course.consts import COURSE_UK, COURSE_UK_FR, COURSE_UK_IT
|
|
from vbv_lernwelt.course.models import CourseSessionUser
|
|
|
|
UK_COURSE_IDS = [COURSE_UK, COURSE_UK_FR, COURSE_UK_IT]
|
|
DEFAULT_EXCLUDED_DOMAINS = ["eiger-versicherung.ch", "assurance.ch", "example.com"]
|
|
|
|
|
|
@staff_member_required
|
|
def export_students(request):
|
|
course_session_users = fetch_course_session_users(UK_COURSE_IDS)
|
|
return generate_export_response(course_session_users)
|
|
|
|
|
|
@staff_member_required
|
|
def export_trainers(request):
|
|
course_session_users = fetch_course_session_users(
|
|
UK_COURSE_IDS, role=CourseSessionUser.Role.EXPERT
|
|
)
|
|
return generate_export_response(course_session_users)
|
|
|
|
|
|
@staff_member_required
|
|
def export_students_and_trainers(request):
|
|
course_session_users = fetch_course_session_all_users(UK_COURSE_IDS)
|
|
return generate_export_response(course_session_users)
|
|
|
|
|
|
def fetch_course_session_users(
|
|
courses: List[int], role=CourseSessionUser.Role.MEMBER, excluded_domains=None
|
|
):
|
|
if role == CourseSessionUser.Role.EXPERT:
|
|
edoniq_role = "Trainer"
|
|
else:
|
|
edoniq_role = "Lernende"
|
|
|
|
# exclude test users if required
|
|
if excluded_domains is None:
|
|
excluded_domains = DEFAULT_EXCLUDED_DOMAINS
|
|
|
|
exclude_q = Q()
|
|
for condition in excluded_domains:
|
|
exclude_q |= Q(email__contains=condition)
|
|
|
|
# if users should be exported per course session, remove the distinct() call
|
|
return (
|
|
User.objects.filter(
|
|
coursesessionuser__course_session__course__id__in=courses,
|
|
coursesessionuser__role=role,
|
|
)
|
|
.exclude(exclude_q)
|
|
.order_by("email")
|
|
.annotate(edoniq_role=Value(edoniq_role, output_field=CharField()))
|
|
.distinct()
|
|
)
|
|
|
|
|
|
def fetch_course_session_all_users(courses: List[int], excluded_domains=None):
|
|
course_session_users = fetch_course_session_users(
|
|
courses, excluded_domains=excluded_domains
|
|
)
|
|
course_session_trainers = fetch_course_session_users(
|
|
courses, role=CourseSessionUser.Role.EXPERT, excluded_domains=excluded_domains
|
|
)
|
|
combined_queryset = chain(course_session_users, course_session_trainers)
|
|
|
|
# use email as key to remove duplicates. Maybe we can use the sso id in the future, but it is not set for all users
|
|
unique_dict = {obj.email: obj for obj in combined_queryset}
|
|
return list(unique_dict.values())
|
|
|
|
|
|
def generate_export_response(cs_users: List[User]) -> HttpResponse:
|
|
response = HttpResponse(content_type="text/csv")
|
|
response[
|
|
"Content-Disposition"
|
|
] = f"attachment; filename=edoniq_user_export_{date.today().strftime('%Y%m%d')}.csv"
|
|
|
|
writer = csv.writer(response)
|
|
writer.writerow(
|
|
[
|
|
"login",
|
|
"firstname",
|
|
"lastname",
|
|
"gender",
|
|
"preferred_language",
|
|
"email",
|
|
"birthday",
|
|
"group",
|
|
"region",
|
|
"division",
|
|
"oe",
|
|
"superior_login",
|
|
]
|
|
)
|
|
|
|
for cs_user in cs_users:
|
|
writer.writerow(
|
|
[
|
|
cs_user.id,
|
|
cs_user.first_name,
|
|
cs_user.last_name,
|
|
"m",
|
|
cs_user.language,
|
|
cs_user.email,
|
|
cs_user.additional_json_data.get("Geburtsdatum", ""),
|
|
cs_user.edoniq_role,
|
|
cs_user.additional_json_data.get("Firmenname", ""),
|
|
cs_user.additional_json_data.get("Lehrvertragsnummer", ""),
|
|
cs_user.coursesessionuser_set.first().course_session.import_id,
|
|
"",
|
|
]
|
|
)
|
|
|
|
return response
|