import traceback import logging from typing import Optional, Union import regex from aiohttp import ClientSession, ClientTimeout from constants import Constants from .constructors import InvalidLastFMResponseException class LastFM: """LastFM Endpoints""" def __init__(self, noInit: Optional[bool] = False) -> None: self.creds = Constants().LFM_CREDS self.api_base_url: str = "https://ws.audioscrobbler.com/2.0" async def search_artist(self, artist: Optional[str] = None) -> dict: """Search LastFM for an artist""" try: if not artist: return { 'err': 'No artist specified.', } request_params: list[tuple] = [ ("method", "artist.getInfo"), ("artist", artist), ("api_key", self.creds.get('key')), ("autocorrect", "1"), ("format", "json"), ] async with ClientSession() as session: async with await session.get(self.api_base_url, params=request_params, timeout=ClientTimeout(connect=3, sock_read=8)) as request: request.raise_for_status() data: dict = await request.json() data = data.get('artist', 'N/A') ret_obj: dict = { 'id': data.get('mbid'), 'touring': data.get('ontour'), 'name': data.get('name'), 'bio': data.get('bio', None).get('summary').strip()\ .split(" Optional[dict]: """ Get Track Info from LastFM Args: artist (Optional[str]) track (Optional[str]) Returns: dict """ try: if not artist or not track: logging.info("inv request") return { 'err': 'Invalid/No artist or track specified', } request_params: list[tuple] = [ ("method", "track.getInfo"), ("api_key", self.creds.get('key')), ("autocorrect", "1"), ("artist", artist), ("track", track), ("format", "json"), ] async with ClientSession() as session: async with await session.get(self.api_base_url, params=request_params, timeout=ClientTimeout(connect=3, sock_read=8)) as request: request.raise_for_status() data: dict = await request.json() data = data.get('track', None) if not isinstance(data.get('artist'), dict): return None artist_mbid: int = data.get('artist', None).get('mbid') album: str = data.get('album', None).get('title') ret_obj: dict = { 'artist_mbid': artist_mbid, 'album': album, } return ret_obj except: traceback.print_exc() return { 'err': 'General Failure', } async def get_album_tracklist(self, artist: Optional[str] = None, album: Optional[str] = None) -> dict: """ Get Album Tracklist Args: artist (str) album (str) Returns: dict """ try: if not artist or not album: return { 'err': 'No artist or album specified', } tracks: dict = await self.get_release(artist=artist, album=album) tracks = tracks.get('tracks', None) ret_obj: dict = { 'tracks': tracks, } return ret_obj except: traceback.print_exc() return { 'err': 'General Failure', } async def get_artist_albums(self, artist: Optional[str] = None) -> Union[dict, list[dict]]: """ Get Artists Albums from LastFM Args: artist (Optional[str]) Returns: Union[dict, list[dict]] """ try: if not artist: return { 'err': 'No artist specified.', } request_params: list[tuple] = [ ("method", "artist.gettopalbums"), ("artist", artist), ("api_key", self.creds.get('key')), ("autocorrect", "1"), ("format", "json"), ] async with ClientSession() as session: async with await session.get(self.api_base_url, params=request_params, timeout=ClientTimeout(connect=3, sock_read=8)) as request: request.raise_for_status() json_data: dict = await request.json() data: dict = json_data.get('topalbums', None).get('album') ret_obj: list = [ { 'title': item.get('name') } for item in data if not(item.get('name').lower() == "(null)")\ and int(item.get('playcount')) >= 50 ] return ret_obj except: traceback.print_exc() return { 'err': 'Failed', } async def get_artist_id(self, artist: Optional[str] = None) -> int: """ Get Artist ID from LastFM Args: artist (Optional[str]) Returns: int """ try: if not artist: return -1 artist_search: dict = await self.search_artist(artist=artist) if not artist_search: logging.debug("[get_artist_id] Throwing no result error") return -1 artist_id: int = int(artist_search[0].get('id', 0)) return artist_id except: traceback.print_exc() return -1 async def get_artist_info_by_id(self, artist_id: Optional[int] = None) -> dict: """ Get Artist info by ID from LastFM Args: artist_id (Optional[int]) Returns: dict """ try: if not artist_id or not str(artist_id).isnumeric(): return { 'err': 'Invalid/no artist_id specified.', } req_url: str = f"{self.api_base_url}/artists/{artist_id}" request_params: list[tuple] = [ ("key", self.creds.get('key')), ("secret", self.creds.get('secret')), ] async with ClientSession() as session: async with await session.get(req_url, params=request_params, timeout=ClientTimeout(connect=3, sock_read=8)) as request: request.raise_for_status() data: dict = await request.json() if not data.get('profile'): raise InvalidLastFMResponseException("Data did not contain 'profile' key.") _id: int = data.get('id', None) name: str = data.get('name', None) profile: str = data.get('profile', '') profile = regex.sub(r"(\[(\/{0,})(u|b|i)])", "", profile) members: list = data.get('members', None) ret_obj: dict = { 'id': _id, 'name': name, 'profile': profile, 'members': members, } return ret_obj except: traceback.print_exc() return { 'err': 'Failed', } async def get_artist_info(self, artist: Optional[str] = None) -> dict: """ Get Artist Info from LastFM Args: artist (Optional[str]) Returns: dict """ try: if not artist: return { 'err': 'No artist specified.', } artist_id: Optional[int] = await self.get_artist_id(artist=artist) if not artist_id: return { 'err': 'Failed', } artist_info: Optional[dict] = await self.get_artist_info_by_id(artist_id=artist_id) if not artist_info: return { 'err': 'Failed', } return artist_info except: traceback.print_exc() return { 'err': 'Failed', } async def get_release(self, artist: Optional[str] = None, album: Optional[str] = None) -> dict: """ Get Release info from LastFM Args: artist (Optional[str]) album (Optional[str]) Returns: dict """ try: if not artist or not album: return { 'err': 'Invalid artist/album pair', } request_params: list[tuple] = [ ("method", "album.getinfo"), ("artist", artist), ("album", album), ("api_key", self.creds.get('key')), ("autocorrect", "1"), ("format", "json"), ] async with ClientSession() as session: async with await session.get(self.api_base_url, params=request_params, timeout=ClientTimeout(connect=3, sock_read=8)) as request: request.raise_for_status() json_data: dict = await request.json() data: dict = json_data.get('album', None) ret_obj: dict = { 'id': data.get('mbid'), 'artists': data.get('artist'), 'tags': data.get('tags'), 'title': data.get('name'), 'summary': data.get('wiki', None).get('summary').split("