api/endpoints/radio.py

import logging
import traceback
import time
import random
import asyncio
from utils import radio_util
from .constructors import (
    ValidRadioNextRequest,
    ValidRadioReshuffleRequest,
    ValidRadioQueueShiftRequest,
    ValidRadioQueueRemovalRequest,
    ValidRadioSongRequest,
    ValidRadioTypeaheadRequest,
    RadioException,
)

from uuid import uuid4 as uuid
from typing import Optional
from fastapi import FastAPI, BackgroundTasks, Request, Response, HTTPException
from fastapi.responses import RedirectResponse, JSONResponse


class Radio(FastAPI):
    """Radio Endpoints"""

    def __init__(self, app: FastAPI, my_util, constants) -> None:
        self.app: FastAPI = app
        self.util = my_util
        self.constants = constants
        self.radio_util = radio_util.RadioUtil(self.constants)

        self.endpoints: dict = {
            "radio/np": self.radio_now_playing,
            "radio/request": self.radio_request,
            "radio/typeahead": self.radio_typeahead,
            "radio/get_queue": self.radio_get_queue,
            "radio/skip": self.radio_skip,
            "radio/queue_shift": self.radio_queue_shift,
            "radio/reshuffle": self.radio_reshuffle,
            "radio/queue_remove": self.radio_queue_remove,
            "radio/ls._next_": self.radio_get_next,
        }

        for endpoint, handler in self.endpoints.items():
            app.add_api_route(
                f"/{endpoint}", handler, methods=["POST"], include_in_schema=True
            )

        # NOTE: Not in loop because method is GET for this endpoint
        app.add_api_route(
            "/radio/album_art",
            self.album_art_handler,
            methods=["GET"],
            include_in_schema=True,
        )

        asyncio.get_event_loop().run_until_complete(self.radio_util.load_playlist())
        asyncio.get_event_loop().run_until_complete(self.radio_util._ls_skip())

    async def radio_skip(
        self, data: ValidRadioNextRequest, request: Request
    ) -> JSONResponse:
        """
        Skip to the next track in the queue, or to uuid specified in skipTo if provided
        - **key**: API key
        - **skipTo**: Optional UUID to skip to
        """
        try:
            if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
                raise HTTPException(status_code=403, detail="Unauthorized")
            if data.skipTo:
                queue_item = self.radio_util.get_queue_item_by_uuid(data.skipTo)
                if not queue_item:
                    return JSONResponse(
                        status_code=500,
                        content={
                            "err": True,
                            "errorText": "No such queue item.",
                        },
                    )
                self.radio_util.active_playlist = self.radio_util.active_playlist[
                    queue_item[0] :
                ]
                if not self.radio_util.active_playlist:
                    await self.radio_util.load_playlist()
            skip_result: bool = await self.radio_util._ls_skip()
            status_code = 200 if skip_result else 500
            return JSONResponse(
                status_code=status_code,
                content={
                    "success": skip_result,
                },
            )
        except Exception as e:
            traceback.print_exc()
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "General failure.",
                },
            )

    async def radio_reshuffle(
        self, data: ValidRadioReshuffleRequest, request: Request
    ) -> JSONResponse:
        """
        Reshuffle the play queue
        - **key**: API key
        """
        if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
            raise HTTPException(status_code=403, detail="Unauthorized")

        random.shuffle(self.radio_util.active_playlist)
        return JSONResponse(content={"ok": True})

    async def radio_get_queue(
        self, request: Request, limit: Optional[int] = 15_000
    ) -> JSONResponse:
        """
        Get current play queue, up to limit [default: 15k]
        - **limit**: Number of queue items to return, default 15k
        """
        queue: list = self.radio_util.active_playlist[0:limit]
        queue_out: list[dict] = []
        for x, item in enumerate(queue):
            queue_out.append(
                {
                    "pos": x,
                    "id": item.get("id"),
                    "uuid": item.get("uuid"),
                    "artist": item.get("artist"),
                    "song": item.get("song"),
                    "album": item.get("album", "N/A"),
                    "genre": item.get("genre", "N/A"),
                    "artistsong": item.get("artistsong"),
                    "duration": item.get("duration"),
                }
            )
        return JSONResponse(content={"items": queue_out})

    async def radio_queue_shift(
        self, data: ValidRadioQueueShiftRequest, request: Request
    ) -> JSONResponse:
        """
        Shift position of a UUID within the queue
        [currently limited to playing next or immediately]
        - **key**: API key
        - **uuid**: UUID to shift
        - **next**: Play track next? If False, skips to the track
        """
        if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
            raise HTTPException(status_code=403, detail="Unauthorized")

        queue_item = self.radio_util.get_queue_item_by_uuid(data.uuid)
        if not queue_item:
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Queue item not found.",
                },
            )
        (x, item) = queue_item
        self.radio_util.active_playlist.pop(x)
        self.radio_util.active_playlist.insert(0, item)
        if not data.next:
            await self.radio_util._ls_skip()
        return JSONResponse(
            content={
                "ok": True,
            }
        )

    async def radio_queue_remove(
        self, data: ValidRadioQueueRemovalRequest, request: Request
    ) -> JSONResponse:
        """
        Remove an item from the current play queue
        - **key**: API key
        - **uuid**: UUID of queue item to remove
        """
        if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
            raise HTTPException(status_code=403, detail="Unauthorized")

        queue_item = self.radio_util.get_queue_item_by_uuid(data.uuid)
        if not queue_item:
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Queue item not found.",
                },
            )
        self.radio_util.active_playlist.pop(queue_item[0])
        return JSONResponse(
            content={
                "ok": True,
            }
        )

    async def album_art_handler(
        self, request: Request, track_id: Optional[int] = None
    ) -> Response:
        """
        Get album art, optional parameter track_id may be specified.
        Otherwise, current track album art will be pulled.
        - **track_id**: Optional, if provided, will attempt to retrieve the album art of this track_id. Current track used otherwise.
        """
        try:
            if not track_id:
                track_id = self.radio_util.now_playing.get("id")
            logging.debug("Seeking album art with trackId: %s", track_id)
            album_art: Optional[bytes] = await self.radio_util.get_album_art(
                track_id=track_id
            )
            if not album_art:
                return RedirectResponse(
                    url="https://codey.lol/images/radio_art_default.jpg",
                    status_code=302,
                )
            return Response(content=album_art, media_type="image/png")
        except Exception as e:
            traceback.print_exc()
            return RedirectResponse(
                url="https://codey.lol/images/radio_art_default.jpg", status_code=302
            )

    async def radio_now_playing(self, request: Request) -> JSONResponse:
        """
        Get currently playing track info
        """
        ret_obj: dict = {**self.radio_util.now_playing}
        try:
            ret_obj["elapsed"] = int(time.time()) - ret_obj["start"]
        except KeyError:
            traceback.print_exc()
            ret_obj["elapsed"] = 0
        ret_obj.pop("file_path")
        return JSONResponse(content=ret_obj)

    async def radio_get_next(
        self,
        data: ValidRadioNextRequest,
        request: Request,
        background_tasks: BackgroundTasks,
    ) -> JSONResponse:
        """
        Get next track
        Track will be removed from the queue in the process.
        - **key**: API key
        - **skipTo**: Optional UUID to skip to
        """
        if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
            raise HTTPException(status_code=403, detail="Unauthorized")
        if (
            not isinstance(self.radio_util.active_playlist, list)
            or not self.radio_util.active_playlist
        ):
            await self.radio_util.load_playlist()
            await self.radio_util._ls_skip()
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "General failure occurred, prompting playlist reload.",
                },
            )
        next = self.radio_util.active_playlist.pop(0)
        if not isinstance(next, dict):
            logging.critical("next is of type: %s, reloading playlist...", type(next))
            await self.radio_util.load_playlist()
            await self.radio_util._ls_skip()
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "General failure occurred, prompting playlist reload.",
                },
            )

        duration: int = next["duration"]
        time_started: int = int(time.time())
        time_ends: int = int(time_started + duration)

        if len(self.radio_util.active_playlist) > 1:
            self.radio_util.active_playlist.append(next)  # Push to end of playlist
        else:
            await self.radio_util.load_playlist()

        self.radio_util.now_playing = next
        next["start"] = time_started
        next["end"] = time_ends
        try:
            background_tasks.add_task(self.radio_util.webhook_song_change, next)
        except Exception as e:
            traceback.print_exc()
        try:
            if not await self.radio_util.get_album_art(file_path=next["file_path"]):
                album_art = await self.radio_util.get_album_art(
                    file_path=next["file_path"]
                )
                if album_art:
                    await self.radio_util.cache_album_art(next["id"], album_art)
                else:
                    logging.debug("Could not read album art for %s", next["file_path"])
        except:
            traceback.print_exc()
        return JSONResponse(content=next)

    async def radio_request(
        self, data: ValidRadioSongRequest, request: Request
    ) -> JSONResponse:
        """
        Song request handler
        - **key**: API key
        - **artist**: Artist to search
        - **song**: Song to search
        - **artistsong**: Optional "Artist - Song" pair to search, in place of artist/song
        - **alsoSkip**: If True, skips to the track; otherwise, track will be placed next up in queue
        """
        if not self.util.check_key(path=request.url.path, req_type=4, key=data.key):
            raise HTTPException(status_code=403, detail="Unauthorized")
        artistsong: Optional[str] = data.artistsong
        artist: Optional[str] = data.artist
        song: Optional[str] = data.song
        if artistsong and (artist or song):
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Invalid request",
                },
            )
        if not artistsong and (not artist or not song):
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Invalid request",
                },
            )

        search: bool = await self.radio_util.search_playlist(
            artistsong=artistsong, artist=artist, song=song
        )
        if data.alsoSkip:
            await self.radio_util._ls_skip()
        return JSONResponse(content={"result": search})

    async def radio_typeahead(
        self, data: ValidRadioTypeaheadRequest, request: Request
    ) -> JSONResponse:
        """
        Radio typeahead handler
        - **query**: Typeahead query
        """
        if not isinstance(data.query, str):
            return JSONResponse(
                status_code=500,
                content={
                    "err": True,
                    "errorText": "Invalid request.",
                },
            )
        typeahead: Optional[list[str]] = await self.radio_util.trackdb_typeahead(
            data.query
        )
        if not typeahead:
            return JSONResponse(content=[])
        return JSONResponse(content=typeahead)