mirror of
https://gitlab.science.ru.nl/technicie/MarietjeDjango.git
synced 2025-12-10 09:12:23 +01:00
362 lines
14 KiB
Python
362 lines
14 KiB
Python
from oauth2_provider.views.mixins import ClientProtectedResourceMixin
|
|
from rest_framework import status, mixins
|
|
from rest_framework.generics import (
|
|
ListAPIView,
|
|
RetrieveAPIView,
|
|
get_object_or_404,
|
|
CreateAPIView,
|
|
DestroyAPIView,
|
|
UpdateAPIView,
|
|
)
|
|
from rest_framework.views import APIView
|
|
from rest_framework.response import Response
|
|
|
|
from marietje.api.openapi import CustomAutoSchema
|
|
from marietje.api.permissions import IsAuthenticatedOrTokenHasScopeForMethod
|
|
from django.http import Http404
|
|
|
|
from queues.api.v1.serializers import (
|
|
PlaylistSerializer,
|
|
QueueSerializer,
|
|
PlaylistSongSerializer,
|
|
QueueCommandSerializer,
|
|
)
|
|
from queues.exceptions import RequestException
|
|
from queues.models import Playlist, PlaylistSong, QueueCommand, Queue
|
|
from queues.services import get_user_or_default_queue
|
|
from songs.counters import request_counter
|
|
from songs.models import Song
|
|
|
|
|
|
class PlaylistListAPIView(ListAPIView):
|
|
serializer_class = PlaylistSerializer
|
|
queryset = Playlist.objects.all()
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"GET": ["read"]}
|
|
|
|
|
|
class PlaylistRetrieveAPIView(RetrieveAPIView):
|
|
serializer_class = PlaylistSerializer
|
|
queryset = Playlist.objects.all()
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"GET": ["read"]}
|
|
|
|
|
|
class QueueAPIView(APIView):
|
|
serializer_class = QueueSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"GET": ["read"]}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def get_object(self):
|
|
queue = get_user_or_default_queue(self.request)
|
|
if queue is None:
|
|
raise Http404()
|
|
return queue
|
|
|
|
def get(self, request, **kwargs):
|
|
queue = self.get_object()
|
|
return Response(status=200, data=self.serializer_class(queue).data)
|
|
|
|
|
|
class QueueUpdateAPIView(ClientProtectedResourceMixin, UpdateAPIView):
|
|
serializer_class = QueueSerializer
|
|
queryset = Queue.objects.all()
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"PUT": ["write"], "PATCH": ["write"]}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def put(self, request, **kwargs):
|
|
queue = self.get_object()
|
|
if queue.oauth_client is None or queue.oauth_client != request.auth.application:
|
|
return Response(
|
|
{"detail": "Unauthorized"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
else:
|
|
return super(QueueUpdateAPIView, self).put(request, **kwargs)
|
|
|
|
def patch(self, request, **kwargs):
|
|
queue = self.get_object()
|
|
if queue.oauth_client is None or queue.oauth_client != request.auth.application:
|
|
return Response(
|
|
{"detail": "Unauthorized"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
else:
|
|
return super(QueueUpdateAPIView, self).patch(request, **kwargs)
|
|
|
|
|
|
class QueueSkipAPIView(APIView):
|
|
serializer_class = QueueSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"POST": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def post(self, request, **kwargs):
|
|
queue = get_user_or_default_queue(request)
|
|
if queue is None:
|
|
return Response(status=404)
|
|
|
|
playlist_song = queue.current_song()
|
|
if request.user is not None:
|
|
if playlist_song.user != request.user and not request.user.has_perm("queues.can_skip"):
|
|
return Response(status=403)
|
|
elif request.auth is not None:
|
|
if queue.oauth_client is None or request.auth.application != queue.oauth_client:
|
|
return Response(status=403)
|
|
else:
|
|
return Response(status=403)
|
|
|
|
playlist_song.state = 2
|
|
playlist_song.save()
|
|
|
|
return Response(status=200, data=QueueSerializer(queue).data)
|
|
|
|
|
|
class PlaylistSongMoveDownAPIView(APIView):
|
|
serializer_class = PlaylistSongSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"PATCH": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(response_schema={"$ref": "#/components/schemas/PlaylistSong"})
|
|
|
|
def patch(self, request, **kwargs):
|
|
playlist_song_id = kwargs.get("id")
|
|
playlist_song = get_object_or_404(PlaylistSong, id=playlist_song_id)
|
|
if (
|
|
request.user is not None
|
|
and playlist_song.user != request.user
|
|
and not request.user.has_perm("queues.can_move")
|
|
):
|
|
return Response(status=403)
|
|
playlist_song.move_down()
|
|
return Response(status=200, data=self.serializer_class(playlist_song).data)
|
|
|
|
|
|
class PlaylistSongCancelAPIView(DestroyAPIView):
|
|
serializer_class = PlaylistSongSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"DELETE": ["write"],
|
|
}
|
|
|
|
def delete(self, request, **kwargs):
|
|
playlist_song_id = kwargs.get("id")
|
|
playlist_song = get_object_or_404(PlaylistSong, id=playlist_song_id)
|
|
if (
|
|
request.user is not None
|
|
and playlist_song.user != request.user
|
|
and not request.user.has_perm("queues.can_cancel")
|
|
):
|
|
return Response(status=403)
|
|
playlist_song.delete()
|
|
return Response(status=200, data=self.serializer_class(playlist_song).data)
|
|
|
|
|
|
class QueueRequestAPIView(CreateAPIView):
|
|
serializer_class = PlaylistSongSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"POST": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(
|
|
request_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"song": {"type": "int", "example": 1},
|
|
},
|
|
}
|
|
)
|
|
|
|
def post(self, request, **kwargs):
|
|
queue = get_user_or_default_queue(request)
|
|
if queue is None:
|
|
return Response(status=404)
|
|
song_id = request.data.get("song", None)
|
|
if song_id is None:
|
|
return Response(status=400, data={"success": False, "errorMessage": "Song ID not set."})
|
|
|
|
song = get_object_or_404(Song, id=song_id, deleted=False)
|
|
|
|
try:
|
|
playlist_song = queue.request(song, request.user)
|
|
except RequestException as e:
|
|
return Response(data={"success": False, "errorMessage": str(e)})
|
|
|
|
request_counter.labels(queue=queue.name).inc()
|
|
return Response(status=200, data=self.serializer_class(playlist_song).data)
|
|
|
|
|
|
class QueueVolumeDownAPIView(APIView):
|
|
serializer_class = QueueSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"POST": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def post(self, request, **kwargs):
|
|
queue = get_user_or_default_queue(request)
|
|
if queue is None:
|
|
return Response(status=404)
|
|
if request.user is not None and not request.user.has_perm("queues.can_control_volume"):
|
|
return Response(status=403)
|
|
QueueCommand.objects.create(queue=queue, command="volume_down")
|
|
return Response(status=200, data=self.serializer_class(queue).data)
|
|
|
|
|
|
class QueueVolumeUpAPIView(APIView):
|
|
serializer_class = QueueSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"POST": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def post(self, request, **kwargs):
|
|
queue = get_user_or_default_queue(request)
|
|
if queue is None:
|
|
return Response(status=404)
|
|
if request.user is not None and not request.user.has_perm("queues.can_control_volume"):
|
|
return Response(status=403)
|
|
QueueCommand.objects.create(queue=queue, command="volume_up")
|
|
return Response(status=200, data=self.serializer_class(queue).data)
|
|
|
|
|
|
class QueueMuteAPIView(APIView):
|
|
serializer_class = QueueSerializer
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {
|
|
"POST": ["write"],
|
|
}
|
|
schema = CustomAutoSchema(
|
|
response_schema={
|
|
"type": "object",
|
|
"properties": {
|
|
"id": {"type": "int", "example": 1},
|
|
"name": {"type": "string", "example": "string"},
|
|
"playlist": {"type": "int", "example": 1},
|
|
"random_playlist": {"type": "int", "example": 1},
|
|
"current_song": {"$ref": "#/components/schemas/PlaylistSong"},
|
|
"queue": {"type": "array", "items": {"$ref": "#/components/schemas/PlaylistSong"}},
|
|
"started_at": {"type": "string", "format": "date-time", "nullable": True},
|
|
},
|
|
}
|
|
)
|
|
|
|
def post(self, request, **kwargs):
|
|
queue = get_user_or_default_queue(request)
|
|
if queue is None:
|
|
return Response(status=404)
|
|
if request.user is not None and not request.user.has_perm("queues.can_control_volume"):
|
|
return Response(status=403)
|
|
QueueCommand.objects.create(queue=queue, command="mute")
|
|
return Response(status=200, data=self.serializer_class(queue).data)
|
|
|
|
|
|
class QueueCommandListAPIView(ClientProtectedResourceMixin, ListAPIView):
|
|
serializer_class = QueueCommandSerializer
|
|
queryset = QueueCommand.objects.all()
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"GET": ["read"]}
|
|
|
|
def get_queryset(self):
|
|
queue = get_object_or_404(Queue, pk=self.kwargs.get("pk"))
|
|
self.queryset.filter(queue=queue)
|
|
|
|
def get(self, request, **kwargs):
|
|
queue = get_object_or_404(Queue, pk=kwargs.get("pk"))
|
|
if queue.oauth_client is None or queue.oauth_client != request.auth.application:
|
|
return Response(
|
|
{"detail": "Unauthorized"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
else:
|
|
return super(QueueCommandListAPIView, self).get(request, **kwargs)
|
|
|
|
|
|
class QueueCommandDestroyAPIView(ClientProtectedResourceMixin, DestroyAPIView):
|
|
serializer_class = QueueCommandSerializer
|
|
queryset = QueueCommand.objects.all()
|
|
permission_classes = [IsAuthenticatedOrTokenHasScopeForMethod]
|
|
required_scopes_for_method = {"DELETE": ["write"]}
|
|
|
|
def delete(self, request, **kwargs):
|
|
queue_command = self.get_object()
|
|
if queue_command.queue.oauth_client is None or queue_command.queue.oauth_client != request.auth.application:
|
|
return Response(
|
|
{"detail": "Unauthorized"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
else:
|
|
return super(QueueCommandDestroyAPIView, self).delete(request, **kwargs)
|