From fb1d48ab581f102d8ab1a1553c214e4e6646b210 Mon Sep 17 00:00:00 2001 From: codey Date: Sat, 9 Aug 2025 07:48:07 -0400 Subject: [PATCH] formatting / CORS changes --- .gitignore | 4 ++ base.py | 8 +-- endpoints/meme.py | 4 +- endpoints/rip.py | 32 ++++++--- lyric_search/utils.py | 8 ++- utils/hifi_wrapper.py | 155 ++++++++++++++++++++++-------------------- 6 files changed, 121 insertions(+), 90 deletions(-) diff --git a/.gitignore b/.gitignore index bb506e1..de92564 100644 --- a/.gitignore +++ b/.gitignore @@ -21,5 +21,9 @@ get_next_track.py endpoints/radio.py utils/radio_util.py redis_playlist.py +endpoints/auth.py endpoints/radio2 endpoints/radio2/** +hash_password.py +**/auth/* +.gitignore diff --git a/base.py b/base.py index 249b248..f827ea1 100644 --- a/base.py +++ b/base.py @@ -38,10 +38,11 @@ app.add_middleware( CORSMiddleware, # type: ignore allow_origins=origins, allow_credentials=True, - allow_methods=["POST", "GET", "HEAD"], + allow_methods=["POST", "GET", "HEAD", "OPTIONS"], allow_headers=["*"], ) # type: ignore + """ Blacklisted routes """ @@ -98,9 +99,8 @@ routes: dict = { app, util, constants, loop ), "meme": importlib.import_module("endpoints.meme").Meme(app, util, constants), - "trip": importlib.import_module("endpoints.rip").RIP( - app, util, constants - ), + "trip": importlib.import_module("endpoints.rip").RIP(app, util, constants), + "auth": importlib.import_module("endpoints.auth").Auth(app), } # Misc endpoint depends on radio endpoint instance diff --git a/endpoints/meme.py b/endpoints/meme.py index 009334f..c716e95 100644 --- a/endpoints/meme.py +++ b/endpoints/meme.py @@ -23,7 +23,9 @@ class Meme(FastAPI): for endpoint, handler in self.endpoints.items(): dependencies = None if endpoint == "memes/list_memes": - dependencies = [Depends(RateLimiter(times=10, seconds=2))] # Do not rate limit image retrievals (cached) + dependencies = [ + Depends(RateLimiter(times=10, seconds=2)) + ] # Do not rate limit image retrievals (cached) app.add_api_route( f"/{endpoint}", handler, diff --git a/endpoints/rip.py b/endpoints/rip.py index 949136a..5465043 100644 --- a/endpoints/rip.py +++ b/endpoints/rip.py @@ -3,9 +3,11 @@ from fastapi import FastAPI, Request, Response, Depends from fastapi_throttle import RateLimiter from fastapi.responses import JSONResponse from utils.hifi_wrapper import HifiUtil +from auth.deps import get_current_user logging.getLogger().setLevel(logging.INFO) + class RIP(FastAPI): """ Ripping Endpoints @@ -25,7 +27,9 @@ class RIP(FastAPI): } for endpoint, handler in self.endpoints.items(): - dependencies = [Depends(RateLimiter(times=8, seconds=2))] # Do not rate limit image retrievals (cached) + dependencies = [ + Depends(RateLimiter(times=8, seconds=2)) + ] # Do not rate limit image retrievals (cached) app.add_api_route( f"/{endpoint}", handler, @@ -34,36 +38,46 @@ class RIP(FastAPI): dependencies=dependencies, ) - async def artists_by_name_handler(self, artist: str, request: Request) -> Response: + async def artists_by_name_handler( + self, artist: str, request: Request, user=Depends(get_current_user) + ) -> Response: """Get artists by name""" artists = await self.trip_util.get_artists_by_name(artist) if not artists: return Response(status_code=404, content="Not found") return JSONResponse(content=artists) - - async def albums_by_artist_id_handler(self, artist_id: int, request: Request) -> Response: + + async def albums_by_artist_id_handler( + self, artist_id: int, request: Request, user=Depends(get_current_user) + ) -> Response: """Get albums by artist ID""" albums = await self.trip_util.get_albums_by_artist_id(artist_id) if not albums: return Response(status_code=404, content="Not found") return JSONResponse(content=albums) - - async def tracks_by_album_id_handler(self, album_id: int, request: Request) -> Response: + + async def tracks_by_album_id_handler( + self, album_id: int, request: Request, user=Depends(get_current_user) + ) -> Response: """Get tracks by album id""" tracks = await self.trip_util.get_tracks_by_album_id(album_id) if not tracks: return Response(status_code=404, content="Not Found") return JSONResponse(content=tracks) - async def tracks_by_artist_song_handler(self, artist: str, song: str, request: Request) -> Response: + async def tracks_by_artist_song_handler( + self, artist: str, song: str, request: Request, user=Depends(get_current_user) + ) -> Response: """Get tracks by artist and song name""" logging.critical("Searching for tracks by artist: %s, song: %s", artist, song) tracks = await self.trip_util.get_tracks_by_artist_song(artist, song) if not tracks: return Response(status_code=404, content="Not found") return JSONResponse(content=tracks) - - async def track_by_id_handler(self, track_id: int, request: Request) -> Response: + + async def track_by_id_handler( + self, track_id: int, request: Request, user=Depends(get_current_user) + ) -> Response: """Get track by ID""" track = await self.trip_util.get_stream_url_by_track_id(track_id) if not track: diff --git a/lyric_search/utils.py b/lyric_search/utils.py index 4329547..a46e2bb 100644 --- a/lyric_search/utils.py +++ b/lyric_search/utils.py @@ -111,8 +111,9 @@ class DataUtils: """ def __init__(self) -> None: - self.lrc_regex = regex.compile( # capture mm:ss and optional .xxx, then the lyric text - r""" + self.lrc_regex = ( + regex.compile( # capture mm:ss and optional .xxx, then the lyric text + r""" \[ # literal “[” ( # 1st (and only) capture group: [0-9]{2} # two-digit minutes @@ -123,7 +124,8 @@ class DataUtils: \s* # optional whitespace (.*) # capture the rest of the line as words """, - regex.VERBOSE, + regex.VERBOSE, + ) ) self.scrub_regex_1: Pattern = regex.compile(r"(\[.*?\])(\s){0,}(\:){0,1}") self.scrub_regex_2: Pattern = regex.compile( diff --git a/utils/hifi_wrapper.py b/utils/hifi_wrapper.py index 14bcde8..2acc744 100644 --- a/utils/hifi_wrapper.py +++ b/utils/hifi_wrapper.py @@ -3,38 +3,40 @@ from typing import Optional from urllib.parse import urlencode, quote import logging + class HifiUtil: """ HiFi API Utility Class """ + def __init__(self) -> None: """Initialize HiFi API utility with base URLs.""" - self.hifi_api_url: str = 'http://127.0.0.1:8000' + self.hifi_api_url: str = "http://127.0.0.1:8000" self.hifi_search_url: str = f"{self.hifi_api_url}/search" - def dedupe_by_key(self, - key: str, - entries: list[dict]) -> list[dict]: + def dedupe_by_key(self, key: str, entries: list[dict]) -> list[dict]: deduped = {} for entry in entries: norm = entry[key].strip().lower() if norm not in deduped: deduped[norm] = entry return list(deduped.values()) - + def format_duration(self, seconds): if not seconds: return None m, s = divmod(seconds, 60) return f"{m}:{s:02}" - async def search(self, - artist: str, - song: str = "", - album: str = "", - video_name: str = "", - playlist_name: str = "") -> Optional[list[dict]]: - """Search HiFi API + async def search( + self, + artist: str, + song: str = "", + album: str = "", + video_name: str = "", + playlist_name: str = "", + ) -> Optional[list[dict]]: + """Search HiFi API Args: artist (str, required) song (str, optional) @@ -44,7 +46,7 @@ class HifiUtil: Returns: Optional[dict]: Returns the first result from the HiFi API search or None if no results found. """ - async with ClientSession(timeout=ClientTimeout(total=30)) as session: + async with ClientSession(timeout=ClientTimeout(total=30)) as session: params: dict = { "a": artist, "s": song, @@ -52,7 +54,7 @@ class HifiUtil: "v": video_name, "p": playlist_name, } - query: str = urlencode(params, quote_via=quote) + query: str = urlencode(params, quote_via=quote) built_url: str = f"{self.hifi_search_url}?{query}" async with session.get(built_url) as response: json_response: dict = await response.json() @@ -70,11 +72,10 @@ class HifiUtil: logging.info("No results found.") return None return items - - async def _retrieve(self, - req_type: str, - id: int, - quality: str = "LOSSLESS") -> Optional[list|dict]: + + async def _retrieve( + self, req_type: str, id: int, quality: str = "LOSSLESS" + ) -> Optional[list | dict]: """Retrieve a specific item by type and ID from the HiFi API. Args: type (str): The type of item (e.g., 'song', 'album'). @@ -85,15 +86,17 @@ class HifiUtil: """ async with ClientSession(timeout=ClientTimeout(total=10)) as session: params: dict = { - 'id': id, - 'quality': quality, + "id": id, + "quality": quality, } if req_type not in ["track", "artist", "album", "playlist", "video"]: logging.error("Invalid type: %s", type) return None if req_type in ["artist"]: - params.pop('id') - params['f'] = id # For non-track types, use 'f' instead of 'id' for full API output + params.pop("id") + params["f"] = ( + id # For non-track types, use 'f' instead of 'id' for full API output + ) query: str = urlencode(params, quote_via=quote) built_url: str = f"{self.hifi_api_url}/{req_type}/?{query}" logging.info("Built URL: %s", built_url) @@ -101,29 +104,34 @@ class HifiUtil: if response.status != 200: logging.warning("Item not found: %s %s", req_type, id) return None - response_json: list|dict = await response.json() + response_json: list | dict = await response.json() match req_type: case "artist": response_json = response_json[0] if not isinstance(response_json, dict): - logging.error("Expected a dict but got: %s", type(response_json)) + logging.error( + "Expected a dict but got: %s", type(response_json) + ) return None - response_json_rows = response_json.get('rows') + response_json_rows = response_json.get("rows") if not isinstance(response_json_rows, list): - logging.error("Expected a list but got: %s", type(response_json_rows)) + logging.error( + "Expected a list but got: %s", type(response_json_rows) + ) return None - response_json = response_json_rows[0].get('modules')[0].get('pagedList') + response_json = ( + response_json_rows[0].get("modules")[0].get("pagedList") + ) case "album": - return response_json[1].get('items', []) + return response_json[1].get("items", []) case "track": return response_json if not isinstance(response_json, dict): logging.error("Expected a list but got: %s", type(response_json)) return None - return response_json.get('items') - - async def get_artists_by_name(self, - artist_name: str) -> Optional[list]: + return response_json.get("items") + + async def get_artists_by_name(self, artist_name: str) -> Optional[list]: """Get artist(s) by name from HiFi API. Args: artist_name (str): The name of the artist. @@ -137,15 +145,16 @@ class HifiUtil: return None artists_out = [ { - 'artist': res['name'], - 'id': res['id'], - } for res in artists if 'name' in res and 'id' in res + "artist": res["name"], + "id": res["id"], + } + for res in artists + if "name" in res and "id" in res ] - artists_out = self.dedupe_by_key('artist', artists_out) # Remove duplicates + artists_out = self.dedupe_by_key("artist", artists_out) # Remove duplicates return artists_out - - async def get_albums_by_artist_id(self, - artist_id: int) -> Optional[list|dict]: + + async def get_albums_by_artist_id(self, artist_id: int) -> Optional[list | dict]: """Get albums by artist ID from HiFi API. Args: artist_id (int): The ID of the artist. @@ -159,18 +168,19 @@ class HifiUtil: return None albums_out = [ { - 'artist': ", ".join(artist['name'] for artist in album['artists']), - 'album': album['title'], - 'id': album['id'], - 'release_date': album.get('releaseDate', 'Unknown') - } for album in albums if 'title' in album and 'id' in album and 'artists' in album - ] + "artist": ", ".join(artist["name"] for artist in album["artists"]), + "album": album["title"], + "id": album["id"], + "release_date": album.get("releaseDate", "Unknown"), + } + for album in albums + if "title" in album and "id" in album and "artists" in album + ] logging.info("Retrieved albums: %s", albums_out) return albums_out - - async def get_tracks_by_album_id(self, - album_id: int) -> Optional[list|dict]: + + async def get_tracks_by_album_id(self, album_id: int) -> Optional[list | dict]: """Get tracks by album ID from HiFi API. Args: album_id (int): The ID of the album. @@ -183,20 +193,18 @@ class HifiUtil: return None tracks_out: list[dict] = [ { - 'id': track.get('item').get('id'), - 'artist': track.get('item').get('artist').get('name'), - 'title': track.get('item').get('title'), - 'duration': self.format_duration(track.get('item').get('duration', 0)), - 'version': track.get('item').get('version'), - 'audioQuality': track.get('item').get('audioQuality'), - } for track in track_list + "id": track.get("item").get("id"), + "artist": track.get("item").get("artist").get("name"), + "title": track.get("item").get("title"), + "duration": self.format_duration(track.get("item").get("duration", 0)), + "version": track.get("item").get("version"), + "audioQuality": track.get("item").get("audioQuality"), + } + for track in track_list ] return tracks_out - - async def get_tracks_by_artist_song(self, - artist: str, - song: str) -> Optional[list]: + async def get_tracks_by_artist_song(self, artist: str, song: str) -> Optional[list]: """Get track by artist and song name from HiFi API. Args: artist (str): The name of the artist. @@ -211,22 +219,24 @@ class HifiUtil: return None tracks_out = [ { - 'artist': ", ".join(artist['name'] for artist in track['artists']), - 'song': track['title'], - 'id': track['id'], - 'album': track.get('album', {}).get('title', 'Unknown'), - 'duration': track.get('duration', 0) - } for track in tracks if 'title' in track and 'id' in track and 'artists' in track + "artist": ", ".join(artist["name"] for artist in track["artists"]), + "song": track["title"], + "id": track["id"], + "album": track.get("album", {}).get("title", "Unknown"), + "duration": track.get("duration", 0), + } + for track in tracks + if "title" in track and "id" in track and "artists" in track ] if not tracks_out: logging.warning("No valid tracks found after processing.") return None logging.info("Retrieved tracks: %s", tracks_out) - return tracks_out + return tracks_out - async def get_stream_url_by_track_id(self, - track_id: int, - quality: str = "LOSSLESS") -> Optional[str]: + async def get_stream_url_by_track_id( + self, track_id: int, quality: str = "LOSSLESS" + ) -> Optional[str]: """Get stream URL by track ID from HiFi API. Args: track_id (int): The ID of the track. @@ -235,13 +245,12 @@ class HifiUtil: Optional[str]: The stream URL or None if not found. """ track = await self._retrieve("track", track_id, quality) - if not isinstance(track, list) : + if not isinstance(track, list): logging.warning("No track found for ID: %s", track_id) return None - stream_url = track[2].get('OriginalTrackUrl') + stream_url = track[2].get("OriginalTrackUrl") if not stream_url: logging.warning("No stream URL found for track ID: %s", track_id) return None logging.info("Retrieved stream URL: %s", stream_url) return stream_url - \ No newline at end of file