schon/core/views.py

import logging
import mimetypes
import os
import traceback

import requests
from django.contrib.sitemaps.views import index as _sitemap_index_view
from django.contrib.sitemaps.views import sitemap as _sitemap_detail_view
from django.core.cache import cache
from django.core.exceptions import BadRequest
from django.http import FileResponse, Http404, HttpRequest, HttpResponse, HttpResponseRedirect, JsonResponse
from django.shortcuts import redirect
from django.utils.decorators import method_decorator
from django.utils.http import urlsafe_base64_decode
from django.utils.translation import gettext_lazy as _
from django.views.decorators.cache import cache_page
from django.views.decorators.vary import vary_on_headers
from django_ratelimit.decorators import ratelimit
from djangorestframework_camel_case.render import CamelCaseJSONRenderer
from djangorestframework_camel_case.util import camelize
from drf_spectacular.utils import extend_schema_view
from drf_spectacular.views import SpectacularRedocView, SpectacularSwaggerView
from graphene_file_upload.django import FileUploadGraphQLView
from rest_framework import status
from rest_framework.permissions import AllowAny
from rest_framework.renderers import MultiPartRenderer
from rest_framework.request import Request
from rest_framework.response import Response
from rest_framework.views import APIView
from rest_framework_xml.renderers import XMLRenderer
from rest_framework_yaml.renderers import YAMLRenderer
from sentry_sdk import capture_exception

from core.docs.drf.views import (
    BUY_AS_BUSINESS_SCHEMA,
    CACHE_SCHEMA,
    CONTACT_US_SCHEMA,
    LANGUAGE_SCHEMA,
    PARAMETERS_SCHEMA,
    REQUEST_CURSED_URL_SCHEMA,
    SEARCH_SCHEMA,
)
from core.elasticsearch import process_query
from core.models import DigitalAssetDownload, Order
from core.serializers import (
    BuyAsBusinessOrderSerializer,
    CacheOperatorSerializer,
    ContactUsSerializer,
    LanguageSerializer,
)
from core.utils import get_project_parameters, is_url_safe
from core.utils.caching import web_cache
from core.utils.emailing import contact_us_email
from core.utils.languages import get_flag_by_language
from evibes import settings
from evibes.settings import LANGUAGES
from payments.serializers import TransactionProcessSerializer

logger = logging.getLogger("django")


@cache_page(60 * 60 * 12)
@vary_on_headers("Host")
def sitemap_index(request, *args, **kwargs):
    response = _sitemap_index_view(request, *args, **kwargs)
    response["Content-Type"] = "application/xml; charset=utf-8"
    return response


# noinspection PyTypeChecker
sitemap_index.__doc__ = _(  # type: ignore [assignment]
    "Handles the request for the sitemap index and returns an XML response. "
    "It ensures the response includes the appropriate content type header for XML."
)


@cache_page(60 * 60 * 24)
@vary_on_headers("Host")
def sitemap_detail(request, *args, **kwargs):
    response = _sitemap_detail_view(request, *args, **kwargs)
    response["Content-Type"] = "application/xml; charset=utf-8"
    return response


# noinspection PyTypeChecker
sitemap_detail.__doc__ = _(  # type: ignore [assignment]
    "Handles the detailed view response for a sitemap. "
    "This function processes the request, fetches the appropriate "
    "sitemap detail response, and sets the Content-Type header for XML."
)


class CustomGraphQLView(FileUploadGraphQLView):
    def get_context(self, request):
        return request


class CustomSwaggerView(SpectacularSwaggerView):
    def get_context_data(self, **kwargs):
        # noinspection PyUnresolvedReferences
        context = super().get_context_data(**kwargs)
        context["script_url"] = self.request.build_absolute_uri()
        return context


class CustomRedocView(SpectacularRedocView):
    def get_context_data(self, **kwargs):
        # noinspection PyUnresolvedReferences
        context = super().get_context_data(**kwargs)
        context["script_url"] = self.request.build_absolute_uri()
        return context


@extend_schema_view(**LANGUAGE_SCHEMA)
class SupportedLanguagesView(APIView):
    __doc__ = _("Returns a list of supported languages and their corresponding information.")  # type: ignore [assignment]

    serializer_class = LanguageSerializer
    permission_classes = [
        AllowAny,
    ]
    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    def get(self, request: Request, *args, **kwargs) -> Response:
        return Response(
            data=self.serializer_class(
                [
                    {
                        "code": lang[0],
                        "name": lang[1],
                        "flag": get_flag_by_language(lang[0]),
                    }
                    for lang in LANGUAGES
                ],
                many=True,
            ).data,
            status=status.HTTP_200_OK,
        )


@extend_schema_view(**PARAMETERS_SCHEMA)
class WebsiteParametersView(APIView):
    __doc__ = _("Returns the parameters of the website as a JSON object.")  # type: ignore [assignment]

    serializer_class = None
    permission_classes = [
        AllowAny,
    ]
    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    def get(self, request: Request, *args, **kwargs) -> Response:
        return Response(data=camelize(get_project_parameters()), status=status.HTTP_200_OK)


@extend_schema_view(**CACHE_SCHEMA)
class CacheOperatorView(APIView):
    __doc__ = _("Handles cache operations such as reading and setting cache data with a specified key and timeout.")  # type: ignore [assignment]

    serializer_class = CacheOperatorSerializer
    permission_classes = [
        AllowAny,
    ]
    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    def post(self, request: Request, *args, **kwargs) -> Response:
        return Response(
            data=web_cache(
                request,
                request.data.get("key"),  # type: ignore [arg-type]
                request.data.get("data", {}),
                request.data.get("timeout"),  # type: ignore [arg-type]
            ),
            status=status.HTTP_200_OK,
        )


@extend_schema_view(**CONTACT_US_SCHEMA)
class ContactUsView(APIView):
    __doc__ = _("Handles `contact us` form submissions.")  # type: ignore [assignment]

    serializer_class = ContactUsSerializer
    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    @method_decorator(ratelimit(key="ip", rate="2/h", method="POST", block=True))
    def post(self, request: Request, *args, **kwargs) -> Response:
        serializer = self.serializer_class(data=request.data)
        serializer.is_valid(raise_exception=True)
        contact_us_email.delay(serializer.validated_data)  # type: ignore [attr-defined]

        return Response(data=serializer.data, status=status.HTTP_200_OK)


@extend_schema_view(**REQUEST_CURSED_URL_SCHEMA)
class RequestCursedURLView(APIView):
    __doc__ = _("Handles requests for processing and validating URLs from incoming POST requests.")  # type: ignore [assignment]

    permission_classes = [
        AllowAny,
    ]
    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    @method_decorator(ratelimit(key="ip", rate="10/h"))
    def post(self, request: Request, *args, **kwargs) -> Response:
        url = request.data.get("url")
        if not is_url_safe(str(url)):
            return Response(
                data={"error": _("only URLs starting with http(s):// are allowed")},
                status=status.HTTP_400_BAD_REQUEST,
            )
        try:
            data = cache.get(url, None)
            if not data:
                response = requests.get(str(url), headers={"content-type": "application/json"})
                response.raise_for_status()
                data = camelize(response.json())
                cache.set(url, data, 86400)
            return Response(
                data=data,
                status=status.HTTP_200_OK,
            )
        except Exception as e:
            return Response(
                data={"error": str(e)},
                status=status.HTTP_400_BAD_REQUEST,
            )


@extend_schema_view(**SEARCH_SCHEMA)
class GlobalSearchView(APIView):
    __doc__ = _("Handles global search queries.")  # type: ignore [assignment]

    renderer_classes = [
        CamelCaseJSONRenderer,
        MultiPartRenderer,
        XMLRenderer,
        YAMLRenderer,
    ]

    def get(self, request: Request, *args, **kwargs) -> Response:
        return Response(camelize({"results": process_query(query=request.GET.get("q", "").strip(), request=request)}))


@extend_schema_view(**BUY_AS_BUSINESS_SCHEMA)
class BuyAsBusinessView(APIView):
    __doc__ = _("Handles the logic of buying as a business without registration.")  # type: ignore [assignment]

    # noinspection PyUnusedLocal
    @method_decorator(ratelimit(key="ip", rate="10/h" if not settings.DEBUG else "888/h"))
    def post(self, request: Request, *args, **kwargs) -> Response:
        serializer = BuyAsBusinessOrderSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        order = Order.objects.create(status="MOMENTAL")
        products = [product.get("product_uuid") for product in serializer.validated_data.get("products")]
        try:
            transaction = order.buy_without_registration(
                products=products,
                promocode_uuid=serializer.validated_data.get("promocode_uuid"),
                customer_name=serializer.validated_data.get("business_identificator"),
                customer_email=serializer.validated_data.get("business_email"),
                customer_phone_number=serializer.validated_data.get("business_phone_number"),
                billing_customer_address=serializer.validated_data.get("billing_business_address_uuid"),
                shipping_customer_address=serializer.validated_data.get("shipping_business_address_uuid"),
                payment_method=serializer.validated_data.get("payment_method"),
                is_business=True,
            )
            return Response(
                status=status.HTTP_201_CREATED,
                data=TransactionProcessSerializer(transaction).data,
            )
        except Exception as e:
            order.is_active = False
            order.save()
            return Response(
                status=status.HTTP_400_BAD_REQUEST,
                data={"detail": str(e)},
            )


class DownloadDigitalAssetView(APIView):
    __doc__ = _(  # type: ignore [assignment]
        "Handles the downloading of a digital asset associated with an order.\n"
        "This function attempts to serve the digital asset file located in the "
        "storage directory of the project. If the file is not found, an HTTP 404 "
        "error is raised to indicate the resource is unavailable."
    )

    def get(self, request: HttpRequest, *args, **kwargs) -> HttpResponse | FileResponse:
        try:
            op_uuid = str(kwargs.get("order_product_uuid"))
            if not op_uuid:
                raise BadRequest(_("order_product_uuid is required"))
            uuid = urlsafe_base64_decode(op_uuid).decode("utf-8")

            download = DigitalAssetDownload.objects.get(order_product__uuid=uuid)

            if download.num_downloads >= 1:
                raise BadRequest(_("you can only download the digital asset once"))

            if download.order_product.status != "FINISHED":
                raise BadRequest(_("the order must be paid before downloading the digital asset"))

            download.num_downloads += 1
            download.save()

            if not download.order_product.product:
                raise BadRequest(_("the order product does not have a product"))

            file_path = download.order_product.product.stocks.first().digital_asset.path  # type: ignore [union-attr]

            content_type, encoding = mimetypes.guess_type(file_path)
            if not content_type:
                content_type = "application/octet-stream"

            response = FileResponse(open(file_path, "rb"), content_type=content_type)
            filename = os.path.basename(file_path)
            response["Content-Disposition"] = f'attachment; filename="{filename}"'
            return response

        except BadRequest as e:
            return Response(data=camelize({"error": str(e)}), status=status.HTTP_400_BAD_REQUEST)

        except DigitalAssetDownload.DoesNotExist:
            return Response(data=camelize({"error": "Digital asset not found"}), status=status.HTTP_404_NOT_FOUND)

        except Exception as e:
            capture_exception(e)
            return Response(
                data=camelize(
                    {
                        "error": "An error occurred while trying to download the digital asset",
                        "traceback": traceback.format_exc() if settings.DEBUG else None,
                        "received": {"order_product_uuid": kwargs.get("order_product_uuid", "")},
                    }
                ),
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )


def favicon_view(request: HttpRequest, *args, **kwargs) -> HttpResponse | FileResponse:
    try:
        favicon_path = os.path.join(settings.BASE_DIR, "static/favicon.png")
        return FileResponse(open(favicon_path, "rb"), content_type="image/x-icon")
    except FileNotFoundError as fnfe:
        raise Http404(_("favicon not found")) from fnfe


# noinspection PyTypeChecker
favicon_view.__doc__ = _(  # type: ignore [assignment]
    "Handles requests for the favicon of a website.\n"
    "This function attempts to serve the favicon file located in the static directory of the project. "
    "If the favicon file is not found, an HTTP 404 error is raised to indicate the resource is unavailable."
)


def index(request: HttpRequest, *args, **kwargs) -> HttpResponse | HttpResponseRedirect:
    return redirect("admin:index")


# noinspection PyTypeChecker
index.__doc__ = _(  # type: ignore [assignment]
    "Redirects the request to the admin index page. "
    "The function handles incoming HTTP requests and redirects them to the Django "
    "admin interface index page. It uses Django's `redirect` function for handling "
    "the HTTP redirection."
)


def version(request: HttpRequest, *args, **kwargs) -> HttpResponse:
    return JsonResponse(camelize({"version": settings.EVIBES_VERSION}), status=200)


# noinspection PyTypeChecker
version.__doc__ = _(  # type: ignore [assignment]
    "Returns current version of the eVibes. "
)