tubearchivist/backend/download/views.py

"""all download API views"""

from common.serializers import (
    AsyncTaskResponseSerializer,
    ErrorResponseSerializer,
)
from common.views_base import AdminOnly, ApiBaseView
from download.serializers import (
    AddToDownloadListSerializer,
    AddToDownloadQuerySerializer,
    BulkUpdateDowloadDataSerializer,
    BulkUpdateDowloadQuerySerializer,
    DownloadAggsSerializer,
    DownloadItemSerializer,
    DownloadListQuerySerializer,
    DownloadListQueueDeleteQuerySerializer,
    DownloadListSerializer,
    DownloadQueueItemUpdateSerializer,
)
from download.src.queue_interact import PendingInteract
from drf_spectacular.utils import OpenApiResponse, extend_schema
from rest_framework.response import Response
from task.tasks import download_pending, extrac_dl


class DownloadApiListView(ApiBaseView):
    """resolves to /api/download/
    GET: returns latest videos in the download queue
    POST: add a list of videos to download queue
    DELETE: remove items based on query filter
    """

    search_base = "ta_download/_search/"
    valid_filter = ["pending", "ignore"]
    permission_classes = [AdminOnly]

    @extend_schema(
        responses={
            200: OpenApiResponse(DownloadListSerializer()),
        },
        parameters=[DownloadListQuerySerializer()],
    )
    def get(self, request):
        """get download queue list"""
        query_filter = request.GET.get("filter", False)
        self.data.update(
            {
                "sort": [
                    {"auto_start": {"order": "desc"}},
                    {"timestamp": {"order": "asc"}},
                ],
            }
        )

        serializer = DownloadListQuerySerializer(data=request.query_params)
        serializer.is_valid(raise_exception=True)
        validated_data = serializer.validated_data

        must_list = []
        query_filter = validated_data.get("filter")
        if query_filter:
            must_list.append({"term": {"status": {"value": query_filter}}})

        filter_channel = validated_data.get("channel")
        if filter_channel:
            must_list.append(
                {"term": {"channel_id": {"value": filter_channel}}}
            )

        vid_type_filter = validated_data.get("vid_type")
        if vid_type_filter:
            must_list.append(
                {"term": {"vid_type": {"value": vid_type_filter}}}
            )

        search_query = validated_data.get("q")
        if search_query:
            must_list.append({"match_phrase_prefix": {"title": search_query}})

        if validated_data.get("error") is not None:
            operator = "must" if validated_data["error"] else "must_not"
            must_list.append(
                {"bool": {operator: [{"exists": {"field": "message"}}]}}
            )

        self.data["query"] = {"bool": {"must": must_list}}

        self.get_document_list(request)
        serializer = DownloadListSerializer(self.response)

        return Response(serializer.data)

    @staticmethod
    @extend_schema(
        request=AddToDownloadListSerializer(),
        parameters=[AddToDownloadQuerySerializer()],
        responses={
            200: OpenApiResponse(
                AsyncTaskResponseSerializer(),
                description="New async task started",
            ),
            400: OpenApiResponse(
                ErrorResponseSerializer(), description="Bad request"
            ),
        },
    )
    def post(request):
        """add list of videos to download queue"""
        data_serializer = AddToDownloadListSerializer(data=request.data)
        data_serializer.is_valid(raise_exception=True)
        validated_data = data_serializer.validated_data

        query_serializer = AddToDownloadQuerySerializer(
            data=request.query_params
        )
        query_serializer.is_valid(raise_exception=True)
        validated_query = query_serializer.validated_data

        auto_start = validated_query.get("autostart")
        flat = validated_query.get("flat", False)
        force = validated_query.get("force", False)
        print(f"auto_start: {auto_start}, flat: {flat}, force: {force}")
        to_add = validated_data["data"]

        pending = [i["youtube_id"] for i in to_add if i["status"] == "pending"]
        url_str = " ".join(pending)
        print(f"url_str: {url_str}")
        task = extrac_dl.delay(
            url_str, auto_start=auto_start, flat=flat, force=force
        )

        message = {
            "message": "add to queue task started",
            "task_id": task.id,
        }
        response_serializer = AsyncTaskResponseSerializer(message)

        return Response(response_serializer.data)

    @staticmethod
    @extend_schema(
        request=BulkUpdateDowloadDataSerializer(),
        parameters=[BulkUpdateDowloadQuerySerializer()],
        responses={204: OpenApiResponse(description="Status updated")},
    )
    def patch(request):
        """bulk update status"""
        data_serializer = BulkUpdateDowloadDataSerializer(data=request.data)
        data_serializer.is_valid(raise_exception=True)
        validated_data = data_serializer.validated_data

        new_status = validated_data["status"]

        query_serializer = BulkUpdateDowloadQuerySerializer(
            data=request.query_params
        )
        query_serializer.is_valid(raise_exception=True)
        validated_query = query_serializer.validated_data
        status_filter = validated_query.get("filter")

        PendingInteract(status=status_filter).update_bulk(
            channel_id=validated_query.get("channel"),
            vid_type=validated_query.get("vid_type"),
            new_status=validated_data["status"],
            error=validated_query.get("error"),
        )

        if new_status == "priority":
            download_pending.delay(auto_only=True)

        return Response(status=204)

    @extend_schema(
        parameters=[DownloadListQueueDeleteQuerySerializer()],
        responses={
            204: OpenApiResponse(description="Download items deleted"),
            400: OpenApiResponse(
                ErrorResponseSerializer(), description="Bad request"
            ),
        },
    )
    def delete(self, request):
        """bulk delete download queue items by filter"""
        serializer = DownloadListQueueDeleteQuerySerializer(
            data=request.query_params
        )
        serializer.is_valid(raise_exception=True)
        validated_query = serializer.validated_data

        query_filter = validated_query["filter"]
        channel = validated_query.get("channel")
        vid_type = validated_query.get("vid_type")
        message = f"delete queue by status: {query_filter}"
        if channel:
            message += f" - filter by channel: {channel}"
        if vid_type:
            message += f" - filter by vid_type: {vid_type}"

        print(message)
        PendingInteract(status=query_filter).delete_bulk(
            channel_id=channel, vid_type=vid_type
        )

        return Response(status=204)


class DownloadApiView(ApiBaseView):
    """resolves to /api/download/<video_id>/
    GET: returns metadata dict of an item in the download queue
    POST: update status of item to pending or ignore
    DELETE: forget from download queue
    """

    search_base = "ta_download/_doc/"
    valid_status = ["pending", "ignore", "ignore-force", "priority"]
    permission_classes = [AdminOnly]

    @extend_schema(
        responses={
            200: OpenApiResponse(DownloadItemSerializer()),
            404: OpenApiResponse(
                ErrorResponseSerializer(),
                description="Download item not found",
            ),
        },
    )
    def get(self, request, video_id):
        # pylint: disable=unused-argument
        """get download queue item"""
        self.get_document(video_id)
        if not self.response:
            error = ErrorResponseSerializer(
                {"error": "Download item not found"}
            )
            return Response(error.data, status=404)

        response_serializer = DownloadItemSerializer(self.response)

        return Response(response_serializer.data, status=self.status_code)

    @extend_schema(
        request=DownloadQueueItemUpdateSerializer(),
        responses={
            200: OpenApiResponse(
                DownloadQueueItemUpdateSerializer(),
                description="Download item update",
            ),
            400: OpenApiResponse(
                ErrorResponseSerializer(), description="Bad request"
            ),
            404: OpenApiResponse(
                ErrorResponseSerializer(),
                description="Download item not found",
            ),
        },
    )
    def post(self, request, video_id):
        """post to video to change status"""
        data_serializer = DownloadQueueItemUpdateSerializer(data=request.data)
        data_serializer.is_valid(raise_exception=True)
        validated_data = data_serializer.validated_data
        item_status = validated_data["status"]

        if item_status == "ignore-force":
            extrac_dl.delay(video_id, status="ignore")
            return Response(data_serializer.data)

        _, status_code = PendingInteract(video_id).get_item()
        if status_code == 404:
            error = ErrorResponseSerializer(
                {"error": "Download item not found"}
            )
            return Response(error.data, status=404)

        print(f"{video_id}: change status to {item_status}")
        PendingInteract(video_id, item_status).update_status()
        if item_status == "priority":
            download_pending.delay(auto_only=True)

        return Response(data_serializer.data)

    @staticmethod
    @extend_schema(
        responses={
            204: OpenApiResponse(description="delete download item"),
            404: OpenApiResponse(
                ErrorResponseSerializer(),
                description="Download item not found",
            ),
        },
    )
    def delete(request, video_id):
        # pylint: disable=unused-argument
        """delete single video from queue"""
        print(f"{video_id}: delete from queue")
        PendingInteract(video_id).delete_item()

        return Response(status=204)


class DownloadAggsApiView(ApiBaseView):
    """resolves to /api/download/aggs/
    GET: get download aggregations
    """

    search_base = "ta_download/_search"
    valid_filter_view = ["ignore", "pending"]

    @extend_schema(
        parameters=[DownloadListQueueDeleteQuerySerializer()],
        responses={
            200: OpenApiResponse(DownloadAggsSerializer()),
            400: OpenApiResponse(
                ErrorResponseSerializer(), description="bad request"
            ),
        },
    )
    def get(self, request):
        """get aggs"""
        serializer = DownloadListQueueDeleteQuerySerializer(
            data=request.query_params
        )
        serializer.is_valid(raise_exception=True)
        validated_query = serializer.validated_data

        filter_view = validated_query.get("filter")
        if filter_view:
            if filter_view not in self.valid_filter_view:
                message = f"invalid filter: {filter_view}"
                return Response({"message": message}, status=400)

            self.data.update(
                {
                    "query": {"term": {"status": {"value": filter_view}}},
                }
            )

        self.data.update(
            {
                "aggs": {
                    "channel_downloads": {
                        "multi_terms": {
                            "size": 30,
                            "terms": [
                                {"field": "channel_name.keyword"},
                                {"field": "channel_id"},
                            ],
                            "order": {"_count": "desc"},
                        }
                    }
                }
            }
        )
        self.get_aggs()
        serializer = DownloadAggsSerializer(self.response["channel_downloads"])

        return Response(serializer.data)