api/endpoints/radio.py

#!/usr/bin/env python3.12

import logging
import traceback
import time
import random
import asyncio
from . import radio_util
from .constructors import ValidRadioNextRequest, ValidRadioReshuffleRequest, ValidRadioQueueShiftRequest,\
    ValidRadioQueueRemovalRequest, ValidRadioSongRequest, 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
# pylint: disable=bare-except, broad-exception-caught, invalid-name


"""
TODO: 
    - Radio request typeahead
"""
    
class Radio(FastAPI):
    """Radio Endpoints"""
    def __init__(self, app: FastAPI,
                 my_util, constants) -> None:  # pylint: disable=super-init-not-called
        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/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_out: list[dict] = []
        for x, item in enumerate(self.radio_util.active_playlist[0:limit]):
            queue_out.append({
                'pos': x,
                'id': item.get('id'),
                'uuid': item.get('uuid'),
                'artist': item.get('artist'),
                'song': item.get('song'),
                '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
            })