235 lines
9.0 KiB
Python
235 lines
9.0 KiB
Python
|
#!/usr/bin/env python3.11
|
||
|
|
|
||
|
|
from aiohttp import ClientSession, ClientTimeout
|
||
|
|
import json
|
||
|
|
import regex
|
||
|
|
import traceback
|
||
|
|
|
||
|
|
from constants import Constants
|
||
|
|
from typing import Union, Any
|
||
|
|
|
||
|
|
class LastFM:
|
||
|
|
def __init__(self, noInit: Union[None, bool] = False):
|
||
|
|
self.creds = Constants().LFM_CREDS
|
||
|
|
self.api_base_url = "https://ws.audioscrobbler.com/2.0/?method="
|
||
|
|
|
||
|
|
|
||
|
|
async def search_artist(self, artist=None):
|
||
|
|
try:
|
||
|
|
if artist is None:
|
||
|
|
return {
|
||
|
|
'err': 'No artist specified.'
|
||
|
|
}
|
||
|
|
|
||
|
|
async with ClientSession() as session:
|
||
|
|
async with 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')
|
||
|
|
|
||
|
|
print(f"Using data:\n{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("<a href")[0]
|
||
|
|
}
|
||
|
|
return retObj
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_track_info(self, artist=None, track=None):
|
||
|
|
try:
|
||
|
|
if artist is None or track is None:
|
||
|
|
print("inv request")
|
||
|
|
return {
|
||
|
|
'err': 'Invalid/No artist or track specified'
|
||
|
|
}
|
||
|
|
|
||
|
|
async with ClientSession() as session:
|
||
|
|
async with 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')
|
||
|
|
}
|
||
|
|
print(f"Returning:\n{retObj}")
|
||
|
|
return retObj
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'General Failure'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_album_tracklist(self, artist=None, album=None):
|
||
|
|
try:
|
||
|
|
if artist is None or album is None:
|
||
|
|
print("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
|
||
|
|
}
|
||
|
|
|
||
|
|
print(f"Returning:\n{retObj}")
|
||
|
|
return retObj
|
||
|
|
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'General Failure'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_artist_albums(self, artist=None):
|
||
|
|
try:
|
||
|
|
if artist is None:
|
||
|
|
return {
|
||
|
|
'err': 'No artist specified.'
|
||
|
|
}
|
||
|
|
|
||
|
|
async with ClientSession() as session:
|
||
|
|
async with 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]
|
||
|
|
# return request.text
|
||
|
|
data = await request.json()
|
||
|
|
data = data.get('topalbums').get('album')
|
||
|
|
# print(f"Data:\n{data}")
|
||
|
|
retObj = [
|
||
|
|
{
|
||
|
|
'title': item.get('name')
|
||
|
|
} for item in data if not(item.get('name').lower() == "(null)") and not(int(item.get('playcount')) < 50)
|
||
|
|
]
|
||
|
|
# # print(f"Keys: {data[0].keys()}")
|
||
|
|
return retObj
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_artist_id(self, artist=None):
|
||
|
|
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:
|
||
|
|
print("[get_artist_id] Throwing no result error")
|
||
|
|
return {
|
||
|
|
'err': 'No results.'
|
||
|
|
}
|
||
|
|
artist_id = artist_search[0].get('id')
|
||
|
|
return artist_id
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_artist_info_by_id(self, artist_id=None):
|
||
|
|
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 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:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_artist_info(self, artist=None):
|
||
|
|
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:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
async def get_release(self, artist=None, album=None):
|
||
|
|
try:
|
||
|
|
if artist is None or album is None:
|
||
|
|
return {
|
||
|
|
'err': 'Invalid artist/album pair'
|
||
|
|
}
|
||
|
|
|
||
|
|
async with ClientSession() as session:
|
||
|
|
async with 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("<a href")[0] if "wiki" in data.keys() else "No summary available for this release.",
|
||
|
|
}
|
||
|
|
try:
|
||
|
|
track_key = data.get('tracks').get('track')
|
||
|
|
except: track_key = []
|
||
|
|
if type(track_key) == list:
|
||
|
|
print(f"Track key: {track_key}")
|
||
|
|
retObj['tracks'] = [
|
||
|
|
{
|
||
|
|
'duration': item.get('duration', 'N/A'),
|
||
|
|
'title': item.get('name')
|
||
|
|
} for item in track_key]
|
||
|
|
else:
|
||
|
|
retObj['tracks'] = [
|
||
|
|
{
|
||
|
|
'duration': data.get('tracks').get('track').get('duration'),
|
||
|
|
'title': data.get('tracks').get('track').get('name')
|
||
|
|
}
|
||
|
|
]
|
||
|
|
return retObj
|
||
|
|
except:
|
||
|
|
print(traceback.format_exc())
|
||
|
|
return {
|
||
|
|
'err': 'Failed'
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
|