dfdf
This commit is contained in:
68
API/hifi.py
Normal file
68
API/hifi.py
Normal file
@@ -0,0 +1,68 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any, Dict, Optional
|
||||
|
||||
from .HTTP import HTTPClient
|
||||
|
||||
DEFAULT_BASE_URL = "https://tidal-api.binimum.org"
|
||||
|
||||
|
||||
class HifiApiError(Exception):
|
||||
"""Raised when the HiFi API returns an error or malformed response."""
|
||||
|
||||
|
||||
class HifiApiClient:
|
||||
"""Lightweight client for the hifi-api endpoints.
|
||||
|
||||
Supported endpoints:
|
||||
- GET /search/ with exactly one of s, a, v, p
|
||||
- GET /track/ with id (and optional quality)
|
||||
- GET /info/ with id
|
||||
"""
|
||||
|
||||
def __init__(self, base_url: str = DEFAULT_BASE_URL, *, timeout: float = 10.0) -> None:
|
||||
self.base_url = str(base_url or DEFAULT_BASE_URL).rstrip("/")
|
||||
self.timeout = float(timeout)
|
||||
|
||||
def _get_json(self, path: str, params: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||||
url = f"{self.base_url}/{str(path or '').lstrip('/')}"
|
||||
with HTTPClient(timeout=self.timeout) as client:
|
||||
response = client.get(url, params=params, allow_redirects=True)
|
||||
response.raise_for_status()
|
||||
try:
|
||||
return response.json()
|
||||
except Exception as exc: # pragma: no cover - defensive
|
||||
raise HifiApiError(f"Invalid JSON response from {url}: {exc}") from exc
|
||||
|
||||
def search(self, params: Dict[str, str]) -> Dict[str, Any]:
|
||||
usable = {k: v for k, v in (params or {}).items() if v}
|
||||
search_keys = [key for key in ("s", "a", "v", "p") if usable.get(key)]
|
||||
if not search_keys:
|
||||
raise HifiApiError("One of s/a/v/p is required for /search/")
|
||||
if len(search_keys) > 1:
|
||||
first = search_keys[0]
|
||||
usable = {first: usable[first]}
|
||||
return self._get_json("search/", params=usable)
|
||||
|
||||
def track(self, track_id: int, *, quality: Optional[str] = None) -> Dict[str, Any]:
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise HifiApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
if track_int <= 0:
|
||||
raise HifiApiError("track_id must be positive")
|
||||
|
||||
params: Dict[str, Any] = {"id": track_int}
|
||||
if quality:
|
||||
params["quality"] = str(quality)
|
||||
return self._get_json("track/", params=params)
|
||||
|
||||
def info(self, track_id: int) -> Dict[str, Any]:
|
||||
try:
|
||||
track_int = int(track_id)
|
||||
except Exception as exc:
|
||||
raise HifiApiError(f"track_id must be int-compatible: {exc}") from exc
|
||||
if track_int <= 0:
|
||||
raise HifiApiError("track_id must be positive")
|
||||
|
||||
return self._get_json("info/", params={"id": track_int})
|
||||
Reference in New Issue
Block a user