from aiohttp import ClientSession, ClientTimeout 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_search_url: str = f"{self.hifi_api_url}/search" 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 Args: artist (str, required) song (str, optional) album (str, optional) video_name (str, optional) playlist_name (str, optional) 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: params: dict = { "a": artist, "s": song, "al": album, "v": video_name, "p": playlist_name, } 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() if isinstance(json_response, list): json_response = json_response[0] key = next(iter(json_response), None) if not key: logging.error("No matching key found in JSON response.") return None logging.info("Key: %s", key) if key not in ["limit"]: json_response = json_response[key] items: list[dict] = json_response.get("items", []) if not items: logging.info("No results found.") return None return items 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'). id (int): The ID of the item. quality (str): The quality of the item, default is "LOSSLESS". Other options: HIGH, LOW Returns: Optional[dict]: The item details or None if not found. """ async with ClientSession(timeout=ClientTimeout(total=10)) as session: params: dict = { '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 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) async with session.get(built_url) as response: if response.status != 200: logging.warning("Item not found: %s %s", req_type, id) return None 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)) return None 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)) return None response_json = response_json_rows[0].get('modules')[0].get('pagedList') case "album": 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]: """Get artist(s) by name from HiFi API. Args: artist_name (str): The name of the artist. Returns: Optional[dict]: The artist details or None if not found. """ artists_out: list[dict] = [] artists = await self.search(artist=artist_name) if not artists: logging.warning("No artist found for name: %s", artist_name) return None artists_out = [ { '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 return artists_out 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. Returns: Optional[list[dict]]: List of albums or None if not found. """ albums_out: list[dict] = [] albums = await self._retrieve("artist", artist_id) if not albums: logging.warning("No albums found for artist ID: %s", artist_id) 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 ] logging.info("Retrieved albums: %s", albums_out) return albums_out 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. Returns: Optional[list[dict]]: List of tracks or None if not found. """ track_list = await self._retrieve("album", album_id) if not track_list: logging.warning("No tracks found for album ID: %s", album_id) 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 ] return tracks_out 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. song (str): The name of the song. Returns: Optional[dict]: The track details or None if not found. """ tracks_out: list[dict] = [] tracks = await self.search(artist=artist, song=song) if not tracks: logging.warning("No track found for artist: %s, song: %s", artist, song) 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 ] if not tracks_out: logging.warning("No valid tracks found after processing.") return None logging.info("Retrieved tracks: %s", tracks_out) return tracks_out 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. quality (str): The quality of the stream, default is "LOSSLESS". Other options: HIGH, LOW Returns: Optional[str]: The stream URL or None if not found. """ track = await self._retrieve("track", track_id, quality) if not isinstance(track, list) : logging.warning("No track found for ID: %s", track_id) return None 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