309 lines
13 KiB
Python
309 lines
13 KiB
Python
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) |