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