261 lines
8.2 KiB
Python
261 lines
8.2 KiB
Python
"""all channel API views"""
|
|
|
|
from channel.serializers import (
|
|
ChannelListQuerySerializer,
|
|
ChannelListSerializer,
|
|
ChannelNavSerializer,
|
|
ChannelSearchQuerySerializer,
|
|
ChannelSerializer,
|
|
ChannelUpdateSerializer,
|
|
)
|
|
from channel.src.index import YoutubeChannel, channel_overwrites
|
|
from channel.src.nav import ChannelNav
|
|
from common.serializers import ErrorResponseSerializer
|
|
from common.src.urlparser import Parser
|
|
from common.views_base import AdminWriteOnly, ApiBaseView
|
|
from drf_spectacular.utils import (
|
|
OpenApiParameter,
|
|
OpenApiResponse,
|
|
extend_schema,
|
|
)
|
|
from rest_framework.response import Response
|
|
from task.tasks import index_channel_playlists, subscribe_to
|
|
|
|
|
|
class ChannelApiListView(ApiBaseView):
|
|
"""resolves to /api/channel/
|
|
GET: returns list of channels
|
|
POST: edit a list of channels
|
|
"""
|
|
|
|
search_base = "ta_channel/_search/"
|
|
valid_filter = ["subscribed"]
|
|
permission_classes = [AdminWriteOnly]
|
|
|
|
sort_field_map = {
|
|
"name": "channel_name.keyword",
|
|
"subscribers": "channel_subs",
|
|
"video_count": "channel_video_count",
|
|
"duration": "channel_media_duration",
|
|
"media_size": "channel_media_size",
|
|
"last_refresh": "channel_last_refresh",
|
|
}
|
|
|
|
@extend_schema(
|
|
responses={
|
|
200: OpenApiResponse(ChannelListSerializer()),
|
|
},
|
|
parameters=[ChannelListQuerySerializer()],
|
|
)
|
|
def get(self, request):
|
|
"""get request"""
|
|
serializer = ChannelListQuerySerializer(data=request.query_params)
|
|
serializer.is_valid(raise_exception=True)
|
|
validated_data = serializer.validated_data
|
|
|
|
sort_by = validated_data.get("sort", "name")
|
|
sort_order = validated_data.get("order", "asc")
|
|
es_sort_field = self.sort_field_map[sort_by]
|
|
self.data["sort"] = [{es_sort_field: {"order": sort_order}}]
|
|
|
|
must_list = []
|
|
query_filter = validated_data.get("filter")
|
|
if query_filter is not None:
|
|
channel_subscribed = query_filter == "subscribed"
|
|
must_list.append(
|
|
{"term": {"channel_subscribed": {"value": channel_subscribed}}}
|
|
)
|
|
|
|
self.data["query"] = {"bool": {"must": must_list}}
|
|
self.get_document_list(request)
|
|
serializer = ChannelListSerializer(self.response)
|
|
|
|
return Response(serializer.data)
|
|
|
|
def post(self, request):
|
|
"""subscribe/unsubscribe to list of channels"""
|
|
data = request.data
|
|
try:
|
|
to_add = data["data"]
|
|
except KeyError:
|
|
message = "missing expected data key"
|
|
print(message)
|
|
return Response({"message": message}, status=400)
|
|
|
|
pending = []
|
|
for channel_item in to_add:
|
|
channel_id = channel_item["channel_id"]
|
|
if channel_item["channel_subscribed"]:
|
|
pending.append(channel_id)
|
|
else:
|
|
self._unsubscribe(channel_id)
|
|
|
|
if pending:
|
|
url_str = " ".join(pending)
|
|
subscribe_to.delay(url_str, expected_type="channel")
|
|
|
|
return Response(data)
|
|
|
|
@staticmethod
|
|
def _unsubscribe(channel_id: str):
|
|
"""unsubscribe"""
|
|
print(f"[{channel_id}] unsubscribe from channel")
|
|
YoutubeChannel(channel_id).change_subscribe(new_subscribe_state=False)
|
|
|
|
|
|
class ChannelApiView(ApiBaseView):
|
|
"""resolves to /api/channel/<channel_id>/
|
|
GET: returns metadata dict of channel
|
|
"""
|
|
|
|
search_base = "ta_channel/_doc/"
|
|
permission_classes = [AdminWriteOnly]
|
|
|
|
@extend_schema(
|
|
responses={
|
|
200: OpenApiResponse(ChannelSerializer()),
|
|
404: OpenApiResponse(
|
|
ErrorResponseSerializer(), description="Channel not found"
|
|
),
|
|
}
|
|
)
|
|
def get(self, request, channel_id):
|
|
# pylint: disable=unused-argument
|
|
"""get channel detail"""
|
|
self.get_document(channel_id)
|
|
if not self.response:
|
|
error = ErrorResponseSerializer({"error": "channel not found"})
|
|
return Response(error.data, status=404)
|
|
|
|
response_serializer = ChannelSerializer(self.response)
|
|
return Response(response_serializer.data, status=self.status_code)
|
|
|
|
@extend_schema(
|
|
request=ChannelUpdateSerializer(),
|
|
responses={
|
|
200: OpenApiResponse(ChannelUpdateSerializer()),
|
|
400: OpenApiResponse(
|
|
ErrorResponseSerializer(), description="Bad request"
|
|
),
|
|
404: OpenApiResponse(
|
|
ErrorResponseSerializer(), description="Channel not found"
|
|
),
|
|
},
|
|
)
|
|
def post(self, request, channel_id):
|
|
"""modify channel"""
|
|
self.get_document(channel_id)
|
|
if not self.response:
|
|
error = ErrorResponseSerializer({"error": "channel not found"})
|
|
return Response(error.data, status=404)
|
|
|
|
serializer = ChannelUpdateSerializer(data=request.data)
|
|
serializer.is_valid(raise_exception=True)
|
|
validated_data = serializer.validated_data
|
|
|
|
subscribed = validated_data.get("channel_subscribed")
|
|
if subscribed is not None:
|
|
YoutubeChannel(channel_id).change_subscribe(
|
|
new_subscribe_state=subscribed
|
|
)
|
|
|
|
overwrites = validated_data.get("channel_overwrites")
|
|
if overwrites:
|
|
channel_overwrites(channel_id, overwrites)
|
|
if overwrites.get("index_playlists"):
|
|
index_channel_playlists.delay(channel_id)
|
|
|
|
return Response(serializer.data, status=200)
|
|
|
|
@extend_schema(
|
|
responses={
|
|
204: OpenApiResponse(description="Channel deleted"),
|
|
404: OpenApiResponse(
|
|
ErrorResponseSerializer(), description="Channel not found"
|
|
),
|
|
},
|
|
)
|
|
def delete(self, request, channel_id):
|
|
# pylint: disable=unused-argument
|
|
"""delete channel"""
|
|
try:
|
|
YoutubeChannel(channel_id).delete_channel()
|
|
return Response(status=204)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
error = ErrorResponseSerializer({"error": "channel not found"})
|
|
return Response(error.data, status=404)
|
|
|
|
|
|
class ChannelNavApiView(ApiBaseView):
|
|
"""resolves to /api/channel/<channel_id>/nav/
|
|
GET: get channel nav
|
|
"""
|
|
|
|
@extend_schema(
|
|
responses={
|
|
200: OpenApiResponse(ChannelNavSerializer()),
|
|
},
|
|
)
|
|
def get(self, request, channel_id):
|
|
"""get navigation"""
|
|
|
|
nav = ChannelNav(channel_id).get_nav()
|
|
serializer = ChannelNavSerializer(nav)
|
|
return Response(serializer.data)
|
|
|
|
|
|
class ChannelApiSearchView(ApiBaseView):
|
|
"""resolves to /api/channel/search/
|
|
search for channel
|
|
"""
|
|
|
|
search_base = "ta_channel/_doc/"
|
|
|
|
@extend_schema(
|
|
responses={
|
|
200: OpenApiResponse(ChannelSerializer()),
|
|
400: OpenApiResponse(description="Bad Request"),
|
|
404: OpenApiResponse(
|
|
ErrorResponseSerializer(), description="Channel not found"
|
|
),
|
|
},
|
|
parameters=[
|
|
OpenApiParameter(
|
|
name="q",
|
|
description="Search query string",
|
|
required=True,
|
|
type=str,
|
|
),
|
|
],
|
|
)
|
|
def get(self, request):
|
|
"""search for local channel ID"""
|
|
|
|
serializer = ChannelSearchQuerySerializer(data=request.query_params)
|
|
serializer.is_valid(raise_exception=True)
|
|
validated_data = serializer.validated_data
|
|
|
|
query = validated_data.get("q")
|
|
if not query:
|
|
message = "missing expected q parameter"
|
|
return Response({"message": message, "data": False}, status=400)
|
|
|
|
try:
|
|
parsed = Parser(query).parse()[0]
|
|
except (ValueError, IndexError, AttributeError):
|
|
error = ErrorResponseSerializer(
|
|
{"error": f"channel not found: {query}"}
|
|
)
|
|
return Response(error.data, status=404)
|
|
|
|
if not parsed["type"] == "channel":
|
|
error = ErrorResponseSerializer({"error": "expected channel data"})
|
|
return Response(error.data, status=400)
|
|
|
|
self.get_document(parsed["url"])
|
|
serializer = ChannelSerializer(self.response)
|
|
|
|
return Response(serializer.data, status=self.status_code)
|