diff --git a/base.py b/base.py index a5e5cea..b514bff 100644 --- a/base.py +++ b/base.py @@ -94,7 +94,6 @@ routes: dict = { "lyrics": importlib.import_module("endpoints.lyric_search").LyricSearch( app, util, constants ), - "lastfm": importlib.import_module("endpoints.lastfm").LastFM(app, util, constants), "yt": importlib.import_module("endpoints.yt").YT(app, util, constants), "radio": importlib.import_module("endpoints.radio").Radio( app, util, constants, loop diff --git a/endpoints/constructors.py b/endpoints/constructors.py index fa667ba..0334040 100644 --- a/endpoints/constructors.py +++ b/endpoints/constructors.py @@ -4,84 +4,6 @@ from pydantic import BaseModel Station = Literal["main", "rock", "rap", "electronic", "pop"] -""" -LastFM -""" - - -class LastFMException(Exception): - pass - - -class ValidArtistSearchRequest(BaseModel): - """ - Request model for searching an artist by name. - - Attributes: - - **a** (str): Artist name. - """ - - a: str - - model_config = { - "json_schema_extra": { - "examples": [ - { - "a": "eminem", - } - ] - } - } - - -class ValidAlbumDetailRequest(BaseModel): - """ - Request model for album details. - - Attributes: - - **a** (str): Artist name. - - **release** (str): Album/release name. - """ - - a: str - release: str - - model_config = { - "json_schema_extra": { - "examples": [ - { - "a": "eminem", - "release": "houdini", - } - ] - } - } - - -class ValidTrackInfoRequest(BaseModel): - """ - Request model for track info. - - Attributes: - - **a** (str): Artist name. - - **t** (str): Track name. - """ - - a: str - t: str - - model_config = { - "json_schema_extra": { - "examples": [ - { - "a": "eminem", - "t": "rap god", - } - ] - } - } - - """ Rand Msg """ diff --git a/endpoints/lastfm.py b/endpoints/lastfm.py deleted file mode 100644 index dd1a9c0..0000000 --- a/endpoints/lastfm.py +++ /dev/null @@ -1,255 +0,0 @@ -import importlib -import logging -import traceback -from typing import Optional, Union -from fastapi import FastAPI, Depends -from fastapi_throttle import RateLimiter -from fastapi.responses import JSONResponse -from .constructors import ( - ValidArtistSearchRequest, - ValidAlbumDetailRequest, - ValidTrackInfoRequest, - LastFMException, -) - - -class LastFM(FastAPI): - """Last.FM Endpoints""" - - def __init__(self, app: FastAPI, util, constants) -> None: - """Initialize LastFM endpoints.""" - self.app: FastAPI = app - self.util = util - self.constants = constants - self.lastfm = importlib.import_module("utils.lastfm_wrapper").LastFM() - - self.endpoints: dict = { - "lastfm/get_artist_by_name": self.artist_by_name_handler, - "lastfm/get_artist_albums": self.artist_album_handler, - "lastfm/get_release": self.release_detail_handler, - "lastfm/get_release_tracklist": self.release_tracklist_handler, - "lastfm/get_track_info": self.track_info_handler, - # tbd - } - - for endpoint, handler in self.endpoints.items(): - app.add_api_route( - f"/{endpoint}", - handler, - methods=["POST"], - include_in_schema=True, - dependencies=[Depends(RateLimiter(times=2, seconds=2))], - ) - - async def artist_by_name_handler( - self, data: ValidArtistSearchRequest - ) -> JSONResponse: - """ - Get artist information by name. - - Parameters: - - **data** (ValidArtistSearchRequest): Request containing artist name. - - Returns: - - **JSONResponse**: Contains artist information or an error message. - """ - artist: Optional[str] = data.a.strip() - if not artist: - return JSONResponse( - content={ - "err": True, - "errorText": "No artist specified", - } - ) - - artist_result = await self.lastfm.search_artist(artist=artist) - if ( - not artist_result - or not artist_result.get("bio") - or "err" in artist_result.keys() - ): - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "Search failed (no results?)", - }, - ) - - return JSONResponse( - content={ - "success": True, - "result": artist_result, - } - ) - - async def artist_album_handler( - self, data: ValidArtistSearchRequest - ) -> JSONResponse: - """ - Get artist's albums/releases. - - Parameters: - - **data** (ValidArtistSearchRequest): Request containing artist name. - - Returns: - - **JSONResponse**: Contains a list of albums or an error message. - """ - artist: str = data.a.strip() - if not artist: - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "Invalid request: No artist specified", - }, - ) - - album_result: Union[dict, list[dict]] = await self.lastfm.get_artist_albums( - artist=artist - ) - if isinstance(album_result, dict): - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "General failure.", - }, - ) - album_result_out: list = [] - seen_release_titles: list = [] - - for release in album_result: - release_title: str = release.get("title", "Unknown") - if release_title.lower() in seen_release_titles: - continue - seen_release_titles.append(release_title.lower()) - album_result_out.append(release) - - return JSONResponse(content={"success": True, "result": album_result_out}) - - async def release_detail_handler( - self, data: ValidAlbumDetailRequest - ) -> JSONResponse: - """ - Get details of a particular release by an artist. - - Parameters: - - **data** (ValidAlbumDetailRequest): Request containing artist and release name. - - Returns: - - **JSONResponse**: Release details or error. - """ - artist: str = data.a.strip() - release: str = data.release.strip() - - if not artist or not release: - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "Invalid request", - }, - ) - - release_result = await self.lastfm.get_release(artist=artist, album=release) - ret_obj = { - "id": release_result.get("id"), - "artists": release_result.get("artists"), - "title": release_result.get("title"), - "summary": release_result.get("summary"), - "tracks": release_result.get("tracks"), - } - - return JSONResponse( - content={ - "success": True, - "result": ret_obj, - } - ) - - async def release_tracklist_handler( - self, data: ValidAlbumDetailRequest - ) -> JSONResponse: - """ - Get track list for a particular release by an artist. - - Parameters: - - **data** (ValidAlbumDetailRequest): Request containing artist and release name. - - Returns: - - **JSONResponse**: Track list or error. - """ - artist: str = data.a.strip() - release: str = data.release.strip() - - if not artist or not release: - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "Invalid request", - }, - ) - - tracklist_result: dict = await self.lastfm.get_album_tracklist( - artist=artist, album=release - ) - return JSONResponse( - content={ - "success": True, - "id": tracklist_result.get("id"), - "artists": tracklist_result.get("artists"), - "title": tracklist_result.get("title"), - "summary": tracklist_result.get("summary"), - "tracks": tracklist_result.get("tracks"), - } - ) - - async def track_info_handler(self, data: ValidTrackInfoRequest) -> JSONResponse: - """ - Get track info from Last.FM given an artist/track. - - Parameters: - - **data** (ValidTrackInfoRequest): Request containing artist and track name. - - Returns: - - **JSONResponse**: Track info or error. - """ - try: - artist: str = data.a - track: str = data.t - - if not artist or not track: - return JSONResponse( - status_code=500, - content={"err": True, "errorText": "Invalid request"}, - ) - - track_info_result: Optional[dict] = await self.lastfm.get_track_info( - artist=artist, track=track - ) - if not track_info_result: - return JSONResponse( - status_code=200, - content={ - "err": True, - "errorText": "Not found.", - }, - ) - if "err" in track_info_result: - raise LastFMException( - "Unknown error occurred: %s", - track_info_result.get("errorText", "??"), - ) - return JSONResponse(content={"success": True, "result": track_info_result}) - except Exception as e: - logging.debug("Exception: %s", str(e)) - traceback.print_exc() - return JSONResponse( - status_code=500, - content={ - "err": True, - "errorText": "General error", - }, - ) diff --git a/endpoints/lyric_search.py b/endpoints/lyric_search.py index f123dd3..c36d2c7 100644 --- a/endpoints/lyric_search.py +++ b/endpoints/lyric_search.py @@ -81,6 +81,8 @@ class LyricSearch(FastAPI): ) for endpoint, handler in self.endpoints.items(): + times: int = 20 + seconds: int = 2 rate_limit: tuple[int, int] = (2, 3) # Default; (Times, Seconds) _schema_include = endpoint in ["lyric/search"] diff --git a/endpoints/rip.py b/endpoints/rip.py index 5a82949..02a68b6 100644 --- a/endpoints/rip.py +++ b/endpoints/rip.py @@ -5,7 +5,7 @@ from fastapi.responses import JSONResponse from utils.sr_wrapper import SRUtil from auth.deps import get_current_user from redis import Redis -from rq import Queue, Retry +from rq import Queue from rq.job import Job from rq.job import JobStatus from rq.registry import ( diff --git a/test/test_search_track.py b/test/test_search_track.py deleted file mode 100644 index 36aebcc..0000000 --- a/test/test_search_track.py +++ /dev/null @@ -1,32 +0,0 @@ -import asyncio -import logging -import sys - -sys.path.insert(0, "..") -from utils.sr_wrapper import SRUtil - -# logging.getLogger("sr_wrapper").propagate = False -logger = logging.getLogger() -logger.setLevel(logging.CRITICAL) - - -async def main(): - sr = SRUtil() - artist_search = await sr.get_artists_by_name("Ren") - # logging.critical("Artist search: %s", artist_search) - res = [ - dict(x) - for x in artist_search - if x.get("popularity", 0) and x.get("artist").lower() == "ren" - ] - logging.critical("Results: %s", res) - # search_res = await sr.get_album_by_name(artist[:8], album) - # logging.critical("Search result: %s", search_res) - # album = search_res - # _cover = await sr.get_cover_by_album_id(album.get('id'), 640) - # # cover = sr._get_tidal_cover_url(album.get('cover'), 640) - # logging.critical("Result: %s, Cover: %s", album, _cover) - return - - -asyncio.run(main()) diff --git a/util.py b/util.py index 48ef321..2f946cd 100644 --- a/util.py +++ b/util.py @@ -2,7 +2,7 @@ import logging from typing import Optional -from fastapi import FastAPI, Response, HTTPException +from fastapi import FastAPI, HTTPException from fastapi.responses import RedirectResponse diff --git a/utils/constructors.py b/utils/constructors.py index 3e05c50..e69de29 100644 --- a/utils/constructors.py +++ b/utils/constructors.py @@ -1,7 +0,0 @@ -""" -LastFM -""" - - -class InvalidLastFMResponseException(Exception): - pass diff --git a/utils/lastfm_wrapper.py b/utils/lastfm_wrapper.py deleted file mode 100644 index 93f7bdc..0000000 --- a/utils/lastfm_wrapper.py +++ /dev/null @@ -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(" 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("