retire lastfm endpoints/minor formatting
This commit is contained in:
@@ -1,7 +0,0 @@
|
||||
"""
|
||||
LastFM
|
||||
"""
|
||||
|
||||
|
||||
class InvalidLastFMResponseException(Exception):
|
||||
pass
|
||||
|
||||
@@ -1,378 +0,0 @@
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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()
|
||||
if not data:
|
||||
return None
|
||||
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)
|
||||
if not isinstance(album, dict):
|
||||
return None
|
||||
album = album.get("title")
|
||||
ret_obj: dict = {
|
||||
"artist_mbid": artist_mbid,
|
||||
"album": album,
|
||||
}
|
||||
return ret_obj
|
||||
except Exception as e:
|
||||
traceback.print_exc()
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
traceback.print_exc()
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 (Optional[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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
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 Exception as e:
|
||||
logging.debug("Exception: %s", str(e))
|
||||
traceback.print_exc()
|
||||
return {
|
||||
"err": "Failed",
|
||||
}
|
||||
Reference in New Issue
Block a user