codey/api
1
0
Fork 0
Code Issues 1 Pull Requests Activity

cleanup

Browse Source
This commit is contained in:
codey
2025-01-11 20:59:10 -05:00
parent 85a0d6bc62
commit 3c57f13557
18 changed files with 464 additions and 365 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
lastfm_wrapper.py
View File

@@ -1,20 +1,24 @@
#!/usr/bin/env python3.11
#!/usr/bin/env python3.12
# pylint: disable=bare-except, broad-exception-caught, invalid-name
from aiohttp import ClientSession, ClientTimeout
import json
import regex
import traceback
import logging
from typing import Union
import regex
from aiohttp import ClientSession, ClientTimeout
from constants import Constants
from typing import Union, Any
class LastFM:
def __init__(self, noInit: Union[None, bool] = False):
"""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 {
@@ -22,12 +26,13 @@ class LastFM:
}
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:
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}")
logging.debug("Using data:\n%s", data)
# return data.get('results')
retObj = {
'id': data.get('mbid'),
@@ -37,21 +42,23 @@ class LastFM:
}
return retObj
except:
print(traceback.format_exc())
traceback.print_exc()
return {
'err': 'Failed'
}
async def get_track_info(self, artist=None, track=None):
"""Get Track Info from LastFM"""
try:
if artist is None or track is None:
print("inv request")
logging.info("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:
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')
@@ -59,18 +66,19 @@ class LastFM:
'artist_mbid': data.get('artist').get('mbid'),
'album': data.get('album').get('title')
}
print(f"Returning:\n{retObj}")
logging.debug("Returning:\n%s", retObj)
return retObj
except:
print(traceback.format_exc())
traceback.print_exc()
return {
'err': 'General Failure'
}
async def get_album_tracklist(self, artist=None, album=None):
"""Get Album Tracklist"""
try:
if artist is None or album is None:
print("inv request")
logging.info("inv request")
return {
'err': 'No artist or album specified'
}
@@ -81,16 +89,17 @@ class LastFM:
'tracks': tracks
}
print(f"Returning:\n{retObj}")
logging.debug("Returning:\n%s", retObj)
return retObj
except:
print(traceback.format_exc())
traceback.print_exc()
return {
'err': 'General Failure'
}
async def get_artist_albums(self, artist=None):
"""Get Artists Albums from LastFM"""
try:
if artist is None:
return {
@@ -98,7 +107,8 @@ class LastFM:
}
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:
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()
@@ -107,17 +117,18 @@ class LastFM:
retObj = [
{
'title': item.get('name')
} for item in data if not(item.get('name').lower() == "(null)") and not(int(item.get('playcount')) < 50)
} for item in data if not(item.get('name').lower() == "(null)") and int(item.get('playcount')) >= 50
]
# # print(f"Keys: {data[0].keys()}")
return retObj
except:
print(traceback.format_exc())
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 {
@@ -125,27 +136,29 @@ class LastFM:
}
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")
logging.debug("[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())
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()):
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:
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()
@@ -157,35 +170,37 @@ class LastFM:
}
return retObj
except:
print(traceback.format_exc())
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_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
'err': 'Failed',
}
return artist_info
except:
print(traceback.format_exc())
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 {
@@ -193,7 +208,8 @@ class LastFM:
}
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:
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')
@@ -206,9 +222,10 @@ class LastFM:
}
try:
track_key = data.get('tracks').get('track')
except: track_key = []
if type(track_key) == list:
print(f"Track key: {track_key}")
except:
track_key = []
if isinstance(track_key, list):
logging.debug("Track key: %s", track_key)
retObj['tracks'] = [
{
'duration': item.get('duration', 'N/A'),
@@ -223,7 +240,7 @@ class LastFM:
]
return retObj
except:
print(traceback.format_exc())
traceback.print_exc()
return {
'err': 'Failed'
}