diff --git a/API/hifi.py b/API/hifi.py new file mode 100644 index 0000000..93f8df1 --- /dev/null +++ b/API/hifi.py @@ -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}) diff --git a/MPV/portable_config/script-opts/medeia.conf b/MPV/portable_config/script-opts/medeia.conf index 89db4c9..4077249 100644 --- a/MPV/portable_config/script-opts/medeia.conf +++ b/MPV/portable_config/script-opts/medeia.conf @@ -1,2 +1,2 @@ # Medeia MPV script options -store=video +store=tutorial diff --git a/Provider/HIFI.py b/Provider/HIFI.py index af4578e..8c150c5 100644 --- a/Provider/HIFI.py +++ b/Provider/HIFI.py @@ -4,8 +4,7 @@ import re import sys from typing import Any, Dict, List, Optional, Tuple -import httpx - +from API.hifi import HifiApiClient from ProviderCore.base import Provider, SearchResult from SYS.logger import log @@ -38,6 +37,11 @@ class HIFI(Provider): def __init__(self, config: Optional[Dict[str, Any]] = None) -> None: super().__init__(config) self.api_urls = self._resolve_api_urls() + try: + self.api_timeout = float(self.config.get("timeout", 10.0)) + except Exception: + self.api_timeout = 10.0 + self.api_clients = [HifiApiClient(base_url=url, timeout=self.api_timeout) for url in self.api_urls] def validate(self) -> bool: return bool(self.api_urls) @@ -59,10 +63,10 @@ class HIFI(Provider): for base in self.api_urls: endpoint = f"{base.rstrip('/')}/search/" try: - resp = httpx.get(endpoint, params=params, timeout=10.0) - resp.raise_for_status() - payload = resp.json() - break + client = self._get_api_client_for_base(base) + payload = client.search(params) if client else None + if payload is not None: + break except Exception as exc: log(f"[hifi] Search failed for {endpoint}: {exc}", file=sys.stderr) continue @@ -71,7 +75,7 @@ class HIFI(Provider): return [] data = payload.get("data") or {} - items = data.get("items") or [] + items = self._extract_track_items(data) results: List[SearchResult] = [] for item in items: if limit and len(results) >= limit: @@ -82,6 +86,57 @@ class HIFI(Provider): return results[:limit] + def _get_api_client_for_base(self, base_url: str) -> Optional[HifiApiClient]: + base = base_url.rstrip("/") + for client in self.api_clients: + if getattr(client, "base_url", "").rstrip("/") == base: + return client + return None + + def _extract_track_items(self, data: Any) -> List[Dict[str, Any]]: + if isinstance(data, list): + return [item for item in data if isinstance(item, dict)] + if not isinstance(data, dict): + return [] + + items: List[Dict[str, Any]] = [] + direct = data.get("items") + if isinstance(direct, list): + items.extend(item for item in direct if isinstance(item, dict)) + + tracks_section = data.get("tracks") + if isinstance(tracks_section, dict): + track_items = tracks_section.get("items") + if isinstance(track_items, list): + items.extend(item for item in track_items if isinstance(item, dict)) + + top_hits = data.get("topHits") + if isinstance(top_hits, list): + for hit in top_hits: + if not isinstance(hit, dict): + continue + hit_type = str(hit.get("type") or "").upper() + if hit_type != "TRACKS": + continue + value = hit.get("value") + if isinstance(value, dict): + items.append(value) + + seen: set[int] = set() + deduped: List[Dict[str, Any]] = [] + for item in items: + track_id = item.get("id") or item.get("trackId") + try: + track_int = int(track_id) + except Exception: + track_int = None + if track_int is None or track_int in seen: + continue + seen.add(track_int) + deduped.append(item) + + return deduped + def _resolve_api_urls(self) -> List[str]: urls: List[str] = [] raw = self.config.get("api_urls") @@ -210,10 +265,12 @@ class HIFI(Provider): detail = " | ".join(detail_parts) columns: List[tuple[str, str]] = [] - if artist_display: - columns.append(("Artist", artist_display)) + if title: + columns.append(("Title", title)) if album_title: columns.append(("Album", album_title)) + if artist_display: + columns.append(("Artist", artist_display)) duration_text = self._format_duration(item.get("duration")) if duration_text: columns.append(("Duration", duration_text)) @@ -311,14 +368,12 @@ class HIFI(Provider): def _fetch_track_details(self, track_id: int) -> Optional[Dict[str, Any]]: if track_id <= 0: return None - params = {"id": str(track_id)} for base in self.api_urls: endpoint = f"{base.rstrip('/')}/track/" try: - resp = httpx.get(endpoint, params=params, timeout=10.0) - resp.raise_for_status() - payload = resp.json() - data = payload.get("data") + client = self._get_api_client_for_base(base) + payload = client.track(track_id) if client else None + data = payload.get("data") if isinstance(payload, dict) else None if isinstance(data, dict): return data except Exception as exc: