#!/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("= 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("