257 lines
10 KiB
Python
257 lines
10 KiB
Python
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
|