vbv/server/vbv_lernwelt/edoniq_test/views.py

import csv
from datetime import date
from itertools import chain
from typing import List

import structlog
from django.contrib.admin.views.decorators import staff_member_required
from django.core.exceptions import PermissionDenied
from django.db.models import CharField, Q, Value
from django.http import HttpResponse, JsonResponse
from django.shortcuts import get_object_or_404
from rest_framework.decorators import api_view

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
from vbv_lernwelt.edoniq_test.edoniq_sso import create_token
from vbv_lernwelt.iam.permissions import has_course_access_by_page_request
from vbv_lernwelt.learnpath.models import LearningContentEdoniqTest

logger = structlog.get_logger(__name__)

UK_COURSE_IDS = [COURSE_UK, COURSE_UK_FR, COURSE_UK_IT]
DEFAULT_EXCLUDED_DOMAINS = ["eiger-versicherung.ch", "assurance.ch", "example.com"]


@api_view(["POST"])
def get_edoniq_token_redirect(request):
    page_id = request.data.get("learning_content_id", "")
    extended_time_test = request.data.get("extended_time_test", False)

    logger.debug(
        "get_edoniq_test_redirect",
        label="edoniq_test_api",
        page_id=page_id,
        user_id=request.user.id,
        extended_time_test=extended_time_test,
    )

    if not page_id:
        return JsonResponse({"error": "learning_content_id is required"}, status=400)

    try:
        test_page = get_object_or_404(LearningContentEdoniqTest, id=page_id)

        if not has_course_access_by_page_request(request, test_page):
            logger.debug(
                "get_edoniq_test_redirect_permission_denied",
                label="edoniq_test_api",
                page_id=page_id,
                page_slug=test_page.slug,
                page_title=test_page.title,
                user_id=request.user.id,
            )
            raise PermissionDenied()

        url = (
            test_page.test_url
            if not extended_time_test
            else test_page.extended_time_test_url
        )
        token = create_token(
            str(request.user.id),
        )
        redirect_url = f"{url}&ssoToken={token}"
        response_data = {"redirect_url": redirect_url}

        logger.debug(
            "get_edoniq_test_redirect_permission_success",
            label="edoniq_test_api",
            page_id=page_id,
            page_slug=test_page.slug,
            page_title=test_page.title,
            user_id=request.user.id,
        )
        return JsonResponse(response_data, status=200)

    except Exception as e:
        logger.error(e, exc_info=True)
        return JsonResponse({"error": str(e)}, status=404)


@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; charset=utf-8")
    response["Content-Disposition"] = (
        f"attachment; filename=edoniq_user_export_{date.today().strftime('%Y%m%d')}.csv"
    )

    response.write("\ufeff".encode("utf8"))  # UTF-8 BOM

    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.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