121 lines
3.5 KiB
Python
121 lines
3.5 KiB
Python
#!/usr/bin/env python3.12
|
|
|
|
import importlib
|
|
from fastapi import FastAPI
|
|
from pydantic import BaseModel
|
|
|
|
class ValidArtistSearchRequest(BaseModel):
|
|
"""
|
|
- **a**: artist name
|
|
"""
|
|
|
|
a: str
|
|
|
|
class ValidAlbumDetailRequest(BaseModel):
|
|
"""
|
|
- **a**: artist name
|
|
- **a2**: album/release name (as sourced from here/LastFM)
|
|
"""
|
|
|
|
a: str
|
|
a2: str
|
|
|
|
class LastFM(FastAPI):
|
|
"""Last.FM Endpoints"""
|
|
def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called
|
|
self.app = app
|
|
self.util = util
|
|
self.constants = constants
|
|
self.glob_state = glob_state
|
|
self.lastfm = importlib.import_module("lastfm_wrapper").LastFM()
|
|
|
|
self.endpoints = {
|
|
"get_artist_by_name": self.artist_by_name_handler,
|
|
"get_artist_albums": self.artist_album_handler,
|
|
"get_release": self.release_detail_handler,
|
|
#tbd
|
|
}
|
|
|
|
for endpoint, handler in self.endpoints.items():
|
|
app.add_api_route(f"/{endpoint}/", handler, methods=["POST"])
|
|
|
|
async def artist_by_name_handler(self, data: ValidArtistSearchRequest):
|
|
"""
|
|
/get_artist_by_name/
|
|
Get artist info
|
|
"""
|
|
artist = data.a.strip()
|
|
if not artist:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'No artist specified'
|
|
}
|
|
|
|
artist_result = await self.lastfm.search_artist(artist=artist)
|
|
if not artist_result or "err" in artist_result.keys():
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Search failed (no results?)'
|
|
}
|
|
|
|
return {
|
|
'success': True,
|
|
'result': artist_result
|
|
}
|
|
|
|
async def artist_album_handler(self, data: ValidArtistSearchRequest):
|
|
"""
|
|
/get_artist_albums/
|
|
Get artist's albums/releases
|
|
"""
|
|
artist = data.a.strip()
|
|
if not artist:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'No artist specified'
|
|
}
|
|
|
|
album_result = await self.lastfm.get_artist_albums(artist=artist)
|
|
album_result_out = []
|
|
seen_release_titles = []
|
|
|
|
for release in album_result:
|
|
release_title = release.get('title')
|
|
if release_title.lower() in seen_release_titles:
|
|
continue
|
|
seen_release_titles.append(release_title.lower())
|
|
album_result_out.append(release)
|
|
|
|
return {
|
|
'success': True,
|
|
'result': album_result_out
|
|
}
|
|
|
|
async def release_detail_handler(self, data: ValidAlbumDetailRequest):
|
|
"""
|
|
/get_release/
|
|
Get details of a particular release by an artist
|
|
"""
|
|
artist = data.a.strip()
|
|
release = data.a2.strip()
|
|
|
|
if not artist or not release:
|
|
return {
|
|
'err': True,
|
|
'errorText': 'Invalid request'
|
|
}
|
|
|
|
release_result = await self.lastfm.get_release(artist=artist, album=release)
|
|
ret_obj = {
|
|
'id': release_result.get('id'),
|
|
'artists': release_result.get('artists'),
|
|
'title': release_result.get('title'),
|
|
'summary': release_result.get('summary'),
|
|
'tracks': release_result.get('tracks')
|
|
}
|
|
|
|
return {
|
|
'success': True,
|
|
'result': ret_obj
|
|
}
|