339 lines
12 KiB
Python
339 lines
12 KiB
Python
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("<a href")[0],
|
|
}
|
|
return ret_obj
|
|
except:
|
|
traceback.print_exc()
|
|
return {
|
|
'err': 'Failed',
|
|
}
|
|
|
|
async def get_track_info(self, artist: Optional[str] = None,
|
|
track: Optional[str] = None) -> 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 (Optioanl[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("<a href")[0]\
|
|
if "wiki" in data.keys()\
|
|
else "No summary available for this release.",
|
|
}
|
|
try:
|
|
track_key: list = data.get('tracks', None).get('track')
|
|
except:
|
|
track_key = []
|
|
if isinstance(track_key, list):
|
|
ret_obj['tracks'] = [
|
|
{
|
|
'duration': item.get('duration', 'N/A'),
|
|
'title': item.get('name'),
|
|
} for item in track_key]
|
|
else:
|
|
ret_obj['tracks'] = [
|
|
{
|
|
'duration': data.get('tracks').get('track')\
|
|
.get('duration'),
|
|
'title': data.get('tracks').get('track')\
|
|
.get('name'),
|
|
}
|
|
]
|
|
return ret_obj
|
|
except:
|
|
traceback.print_exc()
|
|
return {
|
|
'err': 'Failed',
|
|
} |