codey/api
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

lastfm cleanup

Browse Source
This commit is contained in:
codey 2025-02-12 19:33:51 -05:00
parent 236aab6bd2
commit 00af36703a
2 changed files with 150 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
endpoints/lastfm.py
View File

@ -10,7 +10,7 @@ from .constructors import ValidArtistSearchRequest, ValidAlbumDetailRequest,\
class LastFM(FastAPI): class LastFM(FastAPI):
"""Last.FM Endpoints""" """Last.FM Endpoints"""
def __init__(self, app: FastAPI, util, constants, glob_state): # pylint: disable=super-init-not-called def __init__(self, app: FastAPI, util, constants, glob_state) -> None: # pylint: disable=super-init-not-called
self.app = app self.app = app
self.util = util self.util = util
self.constants = constants self.constants = constants

250
lastfm_wrapper.py
View File

@ -3,7 +3,7 @@
import traceback import traceback
import logging import logging
from typing import Union from typing import Optional
import regex import regex
from aiohttp import ClientSession, ClientTimeout from aiohttp import ClientSession, ClientTimeout
from constants import Constants from constants import Constants
@ -12,180 +12,220 @@ from constants import Constants
class LastFM: class LastFM:
"""LastFM Endpoints""" """LastFM Endpoints"""
def __init__(self, noInit: Union[None, bool] = False): # pylint: disable=unused-argument def __init__(self, noInit: Optional[bool] = False) -> None: # pylint: disable=unused-argument
self.creds = Constants().LFM_CREDS self.creds = Constants().LFM_CREDS
self.api_base_url = "https://ws.audioscrobbler.com/2.0/?method=" self.api_base_url: str = "https://ws.audioscrobbler.com/2.0/?method="
async def search_artist(self, artist=None): async def search_artist(self, artist: Optional[str] = None) -> dict:
"""Search LastFM for an artist""" """Search LastFM for an artist"""
try: try:
if artist is None: if artist is None:
return { return {
'err': 'No artist specified.' 'err': 'No artist specified.',
} }
async with ClientSession() as session: async with ClientSession() as session:
async with await session.get(f"{self.api_base_url}artist.getinfo&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json", async with await session.get(f"{self.api_base_url}artist.getinfo&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json",
timeout=ClientTimeout(connect=3, sock_read=8)) as request: timeout=ClientTimeout(connect=3, sock_read=8)) as request:
assert request.status in [200, 204] request.raise_for_status()
data = await request.json() data: dict = await request.json()
data = data.get('artist') data = data.get('artist')
logging.debug("Using data:\n%s", data) ret_obj: dict = {
# return data.get('results') 'id': data.get('mbid'),
retObj = { 'touring': data.get('ontour'),
'id': data.get('mbid'), 'name': data.get('name'),
'touring': data.get('ontour'), 'bio': data.get('bio').get('summary').strip()\
'name': data.get('name'), .split("<a href")[0],
'bio': data.get('bio').get('summary').strip().split("<a href")[0]
} }
return retObj return ret_obj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }
async def get_track_info(self, artist=None, track=None) -> dict: async def get_track_info(self, artist: Optional[str] = None,
"""Get Track Info from LastFM""" track: Optional[str] = None) -> dict:
"""
Get Track Info from LastFM
Args:
artist (Optional[str])
track (Optional[str])
Returns:
dict
"""
try: try:
if artist is None or track is None: if not artist or not track:
logging.info("inv request") logging.info("inv request")
return { return {
'err': 'Invalid/No artist or track specified' 'err': 'Invalid/No artist or track specified',
} }
async with ClientSession() as session: async with ClientSession() as session:
async with await session.get(f"{self.api_base_url}track.getInfo&api_key={self.creds.get('key')}&autocorrect=1&artist={artist}&track={track}&format=json", async with await session.get(f"{self.api_base_url}track.getInfo&api_key={self.creds.get('key')}&autocorrect=1&artist={artist}&track={track}&format=json",
timeout=ClientTimeout(connect=3, sock_read=8)) as request: timeout=ClientTimeout(connect=3, sock_read=8)) as request:
assert request.status in [200, 204] request.raise_for_status()
data = await request.json() data: dict = await request.json()
data = data.get('track') data = data.get('track')
retObj = { ret_obj: dict = {
'artist_mbid': data.get('artist').get('mbid'), 'artist_mbid': data.get('artist').get('mbid'),
'album': data.get('album').get('title') 'album': data.get('album').get('title'),
} }
logging.debug("Returning:\n%s", retObj) return ret_obj
return retObj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'General Failure' 'err': 'General Failure',
} }
async def get_album_tracklist(self, artist=None, album=None) -> dict: async def get_album_tracklist(self, artist: Optional[str] = None,
"""Get Album Tracklist""" album: Optional[str] = None) -> dict:
"""
Get Album Tracklist
Args:
artist (str)
album (str)
Returns:
dict
"""
try: try:
if artist is None or album is None: if artist is None or album is None:
logging.info("inv request") logging.info("inv request")
return { return {
'err': 'No artist or album specified' 'err': 'No artist or album specified',
} }
tracks = await self.get_release(artist=artist, album=album) tracks: list|dict = await self.get_release(artist=artist, album=album)
tracks = tracks.get('tracks') tracks: list|dict = tracks.get('tracks')
retObj = { ret_obj: dict = {
'tracks': tracks 'tracks': tracks,
} }
logging.debug("Returning:\n%s", retObj) return ret_obj
return retObj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'General Failure' 'err': 'General Failure',
} }
async def get_artist_albums(self, artist=None) -> dict|list[dict]: async def get_artist_albums(self, artist: Optional[str] = None) -> dict|list[dict]:
"""Get Artists Albums from LastFM""" """
Get Artists Albums from LastFM
Args:
artist (Optional[str])
Returns:
dict|list[dict]
"""
try: try:
if artist is None: if artist is None:
return { return {
'err': 'No artist specified.' 'err': 'No artist specified.',
} }
async with ClientSession() as session: async with ClientSession() as session:
async with await session.get(f"{self.api_base_url}artist.gettopalbums&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json", async with await session.get(f"{self.api_base_url}artist.gettopalbums&artist={artist}&api_key={self.creds.get('key')}&autocorrect=1&format=json",
timeout=ClientTimeout(connect=3, sock_read=8)) as request: timeout=ClientTimeout(connect=3, sock_read=8)) as request:
assert request.status in [200, 204] request.raise_for_status()
data = await request.json() data: dict = await request.json()
data = data.get('topalbums').get('album') data: str = data.get('topalbums').get('album')
retObj = [ ret_obj: dict = [
{ {
'title': item.get('name') 'title': item.get('name')
} for item in data if not(item.get('name').lower() == "(null)") and int(item.get('playcount')) >= 50 } for item in data if not(item.get('name').lower() == "(null)") and int(item.get('playcount')) >= 50
] ]
return retObj return ret_obj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }
async def get_artist_id(self, artist=None): async def get_artist_id(self, artist: Optional[str] = None) -> int|dict:
"""Get Artist ID from LastFM""" """
Get Artist ID from LastFM
Args:
artist (Optional[str])
Returns:
int|dict
"""
try: try:
if artist is None: if artist is None:
return { return {
'err': 'No artist specified.' 'err': 'No artist specified.',
} }
artist_search = await self.search_artist(artist=artist) artist_search: dict = await self.search_artist(artist=artist)
if artist_search is None or len(artist_search) < 1: if not artist_search:
logging.debug("[get_artist_id] Throwing no result error") logging.debug("[get_artist_id] Throwing no result error")
return { return {
'err': 'No results.' 'err': 'No results.',
} }
artist_id = artist_search[0].get('id') artist_id: int = int(artist_search[0].get('id', 0))
return artist_id return artist_id
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }
async def get_artist_info_by_id(self, artist_id=None): async def get_artist_info_by_id(self, artist_id: Optional[int] = None) -> dict:
"""Get Artist info by ID from LastFM""" """
Get Artist info by ID from LastFM
Args:
artist_id (Optional[int])
Returns:
dict
"""
try: try:
if artist_id is None or not str(artist_id).isnumeric(): if not artist_id or not str(artist_id).isnumeric():
return { return {
'err': 'Invalid/no artist_id specified.' 'err': 'Invalid/no artist_id specified.',
} }
req_url: str = f"{self.api_base_url}artists/{artist_id}?key={self.creds.get('key')}\
&secret={self.creds.get('secret')}"
async with ClientSession() as session: async with ClientSession() as session:
async with await session.get(f"{self.api_base_url}artists/{artist_id}?key={self.creds.get('key')}&secret={self.creds.get('secret')}", async with await session.get(req_url,
timeout=ClientTimeout(connect=3, sock_read=8)) as request: timeout=ClientTimeout(connect=3, sock_read=8)) as request:
assert request.status in [200, 204] request.raise_for_status()
data = await request.json() data: dict = await request.json()
retObj = { ret_obj: dict = {
'id': data.get('id'), 'id': data.get('id'),
'name': data.get('name'), 'name': data.get('name'),
'profile': regex.sub(r"(\[(\/{0,})(u|b|i)])", "", data.get('profile')), 'profile': regex.sub(r"(\[(\/{0,})(u|b|i)])", "", data.get('profile')),
'members': data.get('members') 'members': data.get('members'),
} }
return retObj return ret_obj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }
async def get_artist_info(self, artist=None): async def get_artist_info(self, artist: Optional[str] = None) -> dict:
"""Get Artist Info from LastFM""" """
Get Artist Info from LastFM
Args:
artist (Optional[str])
Returns:
dict
"""
try: try:
if artist is None: if not artist:
return { return {
'err': 'No artist specified.' 'err': 'No artist specified.',
} }
artist_id = await self.get_artist_id(artist=artist) artist_id: int = await self.get_artist_id(artist=artist)
if artist_id is None: if not artist_id:
return { return {
'err': 'Failed', 'err': 'Failed',
} }
artist_info = await self.get_artist_info_by_id(artist_id=artist_id) artist_info: dict = await self.get_artist_info_by_id(artist_id=artist_id)
if artist_info is None: if not artist_info:
return { return {
'err': 'Failed', 'err': 'Failed',
} }
@ -193,56 +233,64 @@ class LastFM:
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }
async def get_release(self, artist=None, album=None): async def get_release(self, artist: Optional[str] = None,
"""Get Release info from LastFM""" album: Optional[str] = None) -> dict:
"""
Get Release info from LastFM
Args:
artist (Optional[str])
album (Optioanl[str])
Returns:
dict
"""
try: try:
if artist is None or album is None: if not artist or not album:
return { return {
'err': 'Invalid artist/album pair' 'err': 'Invalid artist/album pair',
} }
req_url: str = f"{self.api_base_url}album.getinfo&artist={artist}&album={album}\
&api_key={self.creds.get('key')}&autocorrect=1&format=json"
async with ClientSession() as session: async with ClientSession() as session:
async with await session.get(f"{self.api_base_url}album.getinfo&artist={artist}&album={album}&api_key={self.creds.get('key')}&autocorrect=1&format=json", async with await session.get(req_url,
timeout=ClientTimeout(connect=3, sock_read=8)) as request: timeout=ClientTimeout(connect=3, sock_read=8)) as request:
assert request.status in [200, 204] request.raise_for_status()
data = await request.json() data: dict = await request.json()
data = data.get('album') data: dict = data.get('album')
retObj = { ret_obj: dict = {
'id': data.get('mbid'), 'id': data.get('mbid'),
'artists': data.get('artist'), 'artists': data.get('artist'),
'tags': data.get('tags'), 'tags': data.get('tags'),
'title': data.get('name'), 'title': data.get('name'),
'summary': data.get('wiki').get('summary').split("<a href")[0] if "wiki" in data.keys() else "No summary available for this release.", 'summary': data.get('wiki').get('summary').split("<a href")[0]\
if "wiki" in data.keys()\
else "No summary available for this release.",
} }
try: try:
track_key = data.get('tracks').get('track') track_key: list = data.get('tracks').get('track')
except: except:
track_key = [] track_key: list = []
if isinstance(track_key, list): if isinstance(track_key, list):
logging.debug("Track key: %s", track_key) ret_obj['tracks'] = [
retObj['tracks'] = [
{ {
'duration': item.get('duration', 'N/A'), 'duration': item.get('duration', 'N/A'),
'title': item.get('name') 'title': item.get('name'),
} for item in track_key] } for item in track_key]
else: else:
retObj['tracks'] = [ ret_obj['tracks'] = [
{ {
'duration': data.get('tracks').get('track').get('duration'), 'duration': data.get('tracks').get('track')\
'title': data.get('tracks').get('track').get('name') .get('duration'),
'title': data.get('tracks').get('track')\
.get('name'),
} }
] ]
return retObj return ret_obj
except: except:
traceback.print_exc() traceback.print_exc()
return { return {
'err': 'Failed' 'err': 'Failed',
} }