vbv/server/vbv_lernwelt/core/views.py

# Create your views here.
import glob
from pathlib import Path

import requests
import structlog
from django.conf import settings
from django.contrib.auth import authenticate, login
from django.core.management import call_command
from django.http import (
    HttpResponse,
    HttpResponseRedirect,
    JsonResponse,
    StreamingHttpResponse,
)
from django.shortcuts import render
from django.template import loader
from django.views.decorators.csrf import ensure_csrf_cookie
from django_ratelimit.decorators import ratelimit
from rest_framework import authentication
from rest_framework.decorators import (
    api_view,
    authentication_classes,
    permission_classes,
)
from rest_framework.permissions import IsAdminUser
from rest_framework.response import Response

from vbv_lernwelt.core.middleware.auth import django_view_authentication_exempt
from vbv_lernwelt.core.serializers import UserSerializer

logger = structlog.get_logger(__name__)


@django_view_authentication_exempt
@ensure_csrf_cookie
def vue_home(request, *args):
    if settings.IT_SERVE_VUE:
        from gunicorn.util import is_hoppish

        try:
            path = request.get_full_path()
            res = requests.get(f"{settings.IT_SERVE_VUE_URL}{path}", stream=True)
            response = StreamingHttpResponse(
                streaming_content=(chunk for chunk in res.iter_content(4096)),
                content_type=res.headers.get("Content-Type", "text/html"),
                status=res.status_code,
            )
            for name, value in res.headers.items():
                if not is_hoppish(name):
                    response[name] = value

            return response
        except Exception as e:
            return HttpResponse(
                f"Can not connect to vue dev server at {settings.IT_SERVE_VUE_URL}: {e}"
            )

    # render index.html from `npm run build`
    content = loader.render_to_string("vue/index.html", context={}, request=request)
    # inject Plausible tracking tag
    if settings.TRACKING_TAG:
        content = content.replace(
            "</head>",
            f"\n{settings.TRACKING_TAG}\n</head>",
        )
    return HttpResponse(content)


@api_view(["POST"])
@ensure_csrf_cookie
def vue_login(request):
    if settings.ALLOW_LOCAL_LOGIN:
        try:
            username = request.data.get("username")
            password = request.data.get("password")
            if username and password:
                user = authenticate(request, username=username, password=password)
                if user:
                    login(request, user)
                    logger.debug(
                        "login successful",
                        username=username,
                        email=user.email,
                        label="login",
                    )
                    return Response(UserSerializer(user).data)
        except Exception as e:
            logger.exception(e)

        logger.debug("login failed", username=username, label="login")
        return Response({"success": False}, status=401)
    else:
        return Response(
            {"success": False, "message": "ALLOW_LOCAL_LOGIN=false"}, status=403
        )


def permission_denied_view(request, exception):
    return render(request, "403.html", status=403)


def rate_limit_exceeded_view(request, exception):
    return render(request, "429.html", status=429)


@django_view_authentication_exempt
def server_json_error(request, *args, **kwargs):
    """
    Generic 500 error handler.
    """
    data = {
        "detail": "Server Error (500)",
        "status_code": 500,
    }
    return JsonResponse(data, status=500)


@ratelimit(key="ip", rate="5/m", block=True)
@django_view_authentication_exempt
def check_rate_limit(request):
    return HttpResponse(content=b"Hello")


@api_view(["POST"])
@authentication_classes((authentication.SessionAuthentication,))
@permission_classes((IsAdminUser,))
def cypress_reset_view(request):
    if not settings.APP_ENVIRONMENT.startswith("prod"):
        # Checking for the flags in the POST request
        options = {}

        create_assignment_completion = (
            request.data.get("create_assignment_completion") == "true"
        )
        if create_assignment_completion:
            options["create_assignment_completion"] = create_assignment_completion

        create_assignment_evaluation = (
            request.data.get("create_assignment_evaluation") == "true"
        )
        if create_assignment_evaluation:
            options["create_assignment_evaluation"] = create_assignment_evaluation
        # assignment evaluation scores
        assignment_evaluation_scores = request.data.get("assignment_evaluation_scores")
        if assignment_evaluation_scores:
            options["assignment_evaluation_scores"] = assignment_evaluation_scores
        options["assignment_points_deducted"] = float(
            request.data.get("assignment_points_deducted") or 0
        )

        options["create_feedback_responses"] = (
            request.data.get("create_feedback_responses") == "true"
        )

        # edoniq test results
        edoniq_test_user_points = request.data.get("edoniq_test_user_points")
        edoniq_test_max_points = request.data.get("edoniq_test_max_points")
        edoniq_points_deducted = request.data.get("edoniq_test_points_deducted") or 0
        if bool(edoniq_test_user_points and edoniq_test_max_points):
            options["create_edoniq_test_results"] = (
                int(edoniq_test_user_points),
                int(edoniq_test_max_points),
                float(edoniq_points_deducted),
            )

        options["create_course_completion_performance_criteria"] = (
            request.data.get("create_course_completion_performance_criteria") == "true"
        )

        options["create_attendance_days"] = (
            request.data.get("create_attendance_days") == "true"
        )

        options["create_learning_mentor"] = (
            request.data.get("create_learning_mentor") == "true"
        )

        call_command(
            "cypress_reset",
            **options,
        )

    return HttpResponseRedirect("/server/admin/")


@api_view(["POST"])
@authentication_classes((authentication.SessionAuthentication,))
@permission_classes((IsAdminUser,))
def iterativ_test_coursesessions_reset_view(request):
    call_command(
        "reset_iterativ_test_sessions",
    )

    return HttpResponseRedirect("/server/admin/")


@django_view_authentication_exempt
def generate_web_component_icons(request):
    svg_files = []
    for filepath in glob.iglob(f"{settings.APPS_DIR}/static/icons/*.svg"):
        with open(filepath, "r") as f:
            filename = Path(filepath).stem
            elementname = "it-" + filename
            svg_files.append(
                {
                    "filepath": filepath,
                    "content": f.read(),
                    "filename": filename,
                    "elementname": elementname,
                    "classname": filename.replace("-", "_"),
                }
            )
    return render(
        request,
        "core/icons.html",
        context={"svg_files": svg_files},
        content_type="application/javascript",
    )