#!/usr/bin/env python3.12 # pylint: disable=bare-except, broad-exception-caught, invalid-name import traceback import logging from typing import Union import regex from aiohttp import ClientSession, ClientTimeout from constants import Constants class LastFM: """LastFM Endpoints""" def __init__(self, noInit: Union[None, bool] = False): # pylint: disable=unused-argument self.creds = Constants().LFM_CREDS self.api_base_url = "https://ws.audioscrobbler.com/2.0/?method=" async def search_artist(self, artist=None): """Search LastFM for an artist""" try: if artist is None: return { 'err': 'No artist specified.' } 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", timeout=ClientTimeout(connect=3, sock_read=8)) as request: assert request.status in [200, 204] data = await request.json() data = data.get('artist') logging.debug("Using data:\n%s", data) # return data.get('results') retObj = { 'id': data.get('mbid'), 'touring': data.get('ontour'), 'name': data.get('name'), 'bio': data.get('bio').get('summary').strip().split(" dict: """Get Track Info from LastFM""" try: if artist is None or track is None: logging.info("inv request") return { 'err': 'Invalid/No artist or track specified' } 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", timeout=ClientTimeout(connect=3, sock_read=8)) as request: assert request.status in [200, 204] data = await request.json() data = data.get('track') retObj = { 'artist_mbid': data.get('artist').get('mbid'), 'album': data.get('album').get('title') } logging.debug("Returning:\n%s", retObj) return retObj except: traceback.print_exc() return { 'err': 'General Failure' } async def get_album_tracklist(self, artist=None, album=None) -> dict: """Get Album Tracklist""" try: if artist is None or album is None: logging.info("inv request") return { 'err': 'No artist or album specified' } tracks = await self.get_release(artist=artist, album=album) tracks = tracks.get('tracks') retObj = { 'tracks': tracks } logging.debug("Returning:\n%s", retObj) return retObj except: traceback.print_exc() return { 'err': 'General Failure' } async def get_artist_albums(self, artist=None) -> dict|list[dict]: """Get Artists Albums from LastFM""" try: if artist is None: return { 'err': 'No artist specified.' } 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", timeout=ClientTimeout(connect=3, sock_read=8)) as request: assert request.status in [200, 204] data = await request.json() data = data.get('topalbums').get('album') retObj = [ { 'title': item.get('name') } for item in data if not(item.get('name').lower() == "(null)") and int(item.get('playcount')) >= 50 ] return retObj except: traceback.print_exc() return { 'err': 'Failed' } async def get_artist_id(self, artist=None): """Get Artist ID from LastFM""" try: if artist is None: return { 'err': 'No artist specified.' } artist_search = await self.search_artist(artist=artist) if artist_search is None or len(artist_search) < 1: logging.debug("[get_artist_id] Throwing no result error") return { 'err': 'No results.' } artist_id = artist_search[0].get('id') return artist_id except: traceback.print_exc() return { 'err': 'Failed' } async def get_artist_info_by_id(self, artist_id=None): """Get Artist info by ID from LastFM""" try: if artist_id is None or not str(artist_id).isnumeric(): return { 'err': 'Invalid/no artist_id specified.' } 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')}", timeout=ClientTimeout(connect=3, sock_read=8)) as request: assert request.status in [200, 204] data = await request.json() retObj = { 'id': data.get('id'), 'name': data.get('name'), 'profile': regex.sub(r"(\[(\/{0,})(u|b|i)])", "", data.get('profile')), 'members': data.get('members') } return retObj except: traceback.print_exc() return { 'err': 'Failed' } async def get_artist_info(self, artist=None): """Get Artist Info from LastFM""" try: if artist is None: return { 'err': 'No artist specified.' } artist_id = await self.get_artist_id(artist=artist) if artist_id is None: return { 'err': 'Failed', } artist_info = await self.get_artist_info_by_id(artist_id=artist_id) if artist_info is None: return { 'err': 'Failed', } return artist_info except: traceback.print_exc() return { 'err': 'Failed' } async def get_release(self, artist=None, album=None): """Get Release info from LastFM""" try: if artist is None or album is None: return { 'err': 'Invalid artist/album pair' } 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", timeout=ClientTimeout(connect=3, sock_read=8)) as request: assert request.status in [200, 204] data = await request.json() data = data.get('album') retObj = { 'id': data.get('mbid'), 'artists': data.get('artist'), 'tags': data.get('tags'), 'title': data.get('name'), 'summary': data.get('wiki').get('summary').split("