diff --git a/Provider/Tidal.py b/Provider/Tidal.py index 3d99102..9501136 100644 --- a/Provider/Tidal.py +++ b/Provider/Tidal.py @@ -12,19 +12,18 @@ from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from urllib.parse import urlparse -from API.Tidal import Tidal as tidalApiClient -from ProviderCore.base import Provider, SearchResult, parse_inline_query_arguments from API.Tidal import ( + Tidal as TidalApiClient, build_track_tags, coerce_duration_seconds, extract_artists, stringify, ) +from ProviderCore.base import Provider, SearchResult, parse_inline_query_arguments from SYS import pipeline as pipeline_context from SYS.logger import debug, log URL_API = ( - "https://tidal-api.binimum.org", "https://triton.squid.wtf", "https://wolf.qqdl.site", "https://maus.qqdl.site", @@ -33,6 +32,7 @@ URL_API = ( "https://hund.qqdl.site", "https://tidal.kinoplus.online", "https://tidal-api.binimum.org", + "https://tidal-api.binimum.org", ) @@ -66,7 +66,6 @@ def _format_total_seconds(seconds: Any) -> str: class Tidal(Provider): PROVIDER_NAME = "tidal" - PROVIDER_ALIASES = ("tidal",) TABLE_AUTO_STAGES = { "tidal.track": ["download-file"], @@ -85,13 +84,18 @@ class Tidal(Provider): "listen.tidal.com", ) URL = URL_DOMAINS - """Provider that targets the Tidal-RestAPI (Tidal proxy) search endpoint. + """Provider that targets the Tidal search endpoint. The CLI can supply a list of fail-over URLs via ``provider.tidal.api_urls`` or ``provider.tidal.api_url`` in the config. When not configured, it defaults to https://tidal-api.binimum.org. """ + _stringify = staticmethod(stringify) + _extract_artists = staticmethod(extract_artists) + _build_track_tags = staticmethod(build_track_tags) + _coerce_duration_seconds = staticmethod(coerce_duration_seconds) + def __init__(self, config: Optional[Dict[str, Any]] = None) -> None: super().__init__(config) self.api_urls = self._resolve_api_urls() @@ -99,7 +103,7 @@ class Tidal(Provider): self.api_timeout = float(self.config.get("timeout", 10.0)) except Exception: self.api_timeout = 10.0 - self.api_clients = [tidalApiClient(base_url=url, timeout=self.api_timeout) for url in self.api_urls] + self.api_clients = [TidalApiClient(base_url=url, timeout=self.api_timeout) for url in self.api_urls] def extract_query_arguments(self, query: str) -> Tuple[str, Dict[str, Any]]: normalized, parsed = parse_inline_query_arguments(query) @@ -239,6 +243,16 @@ class Tidal(Provider): except Exception: return "", None + scheme = str(parsed.scheme or "").lower().strip() + if scheme == "tidal": + # Handle tidal://view/id + view = str(parsed.netloc or "").lower().strip() + path_parts = [p for p in (parsed.path or "").split("/") if p] + identifier = None + if path_parts: + identifier = self._parse_int(path_parts[0]) + return view, identifier + parts = [segment for segment in (parsed.path or "").split("/") if segment] if not parts: return "", None @@ -250,7 +264,7 @@ class Tidal(Provider): return "", None view = parts[idx].lower() - if view not in {"album", "track"}: + if view not in {"album", "track", "artist"}: return "", None for segment in parts[idx + 1:]: @@ -271,7 +285,7 @@ class Tidal(Provider): title = f"Track {track_id}" if isinstance(detail, dict): - title = self._stringify(detail.get("title")) or title + title = stringify(detail.get("title")) or title return SearchResult( table="tidal.track", @@ -281,6 +295,7 @@ class Tidal(Provider): annotations=["tidal", "track"], media_kind="audio", full_metadata=dict(detail) if isinstance(detail, dict) else {}, + selection_args=["-url", f"tidal://track/{track_id}"], ) def _extract_artist_selection_context(self, selected_items: List[Any]) -> List[Tuple[int, str]]: @@ -364,24 +379,24 @@ class Tidal(Provider): if not isinstance(meta, dict): meta = {} - album_title = self._stringify(payload.get("title") or meta.get("title") or meta.get("name")) + album_title = stringify(payload.get("title") or meta.get("title") or meta.get("name")) if not album_title: - album_title = self._stringify(meta.get("album") or meta.get("albumTitle")) + album_title = stringify(meta.get("album") or meta.get("albumTitle")) if not album_title: continue - artist_name = self._stringify(meta.get("_artist_name") or meta.get("artist") or meta.get("artistName")) + artist_name = stringify(meta.get("_artist_name") or meta.get("artist") or meta.get("artistName")) if not artist_name: # Some album payloads include nested artist objects. artist_obj = meta.get("artist") if isinstance(artist_obj, dict): - artist_name = self._stringify(artist_obj.get("name")) + artist_name = stringify(artist_obj.get("name")) # Prefer albumId when available; some payloads carry both id/albumId. album_id = self._parse_int(meta.get("albumId") or meta.get("id")) if not album_id: - raw_path = self._stringify(payload.get("path")) + raw_path = stringify(payload.get("path")) if raw_path: m = re.search(r"tidal:(?://)?album[\\/](\d+)", raw_path, flags=re.IGNORECASE) if m: @@ -428,7 +443,7 @@ class Tidal(Provider): # Fallback: string-match extracted display. if wanted: try: - names = [n.lower() for n in self._extract_artists(track)] + names = [n.lower() for n in extract_artists(track)] except Exception: names = [] return wanted in names @@ -470,7 +485,7 @@ class Tidal(Provider): continue # Prefer albumId when available; some payloads carry both id/albumId. album_id = self._parse_int(album.get("albumId") or album.get("id")) - title = self._stringify(album.get("title")) + title = stringify(album.get("title")) if not title: continue if album_id: @@ -614,7 +629,7 @@ class Tidal(Provider): if album_id and self._parse_int(album.get("albumId") or album.get("id")) == album_id: album_ok = True else: - at = self._stringify(album.get("title")).lower() + at = stringify(album.get("title")).lower() if at: if at == wanted_album: album_ok = True @@ -627,7 +642,7 @@ class Tidal(Provider): album_ok = True else: # If album is not a dict, fall back to string compare. - at = self._stringify(track.get("album")).lower() + at = stringify(track.get("album")).lower() if at: if at == wanted_album: album_ok = True @@ -664,9 +679,9 @@ class Tidal(Provider): album = track.get("album") if isinstance(album, dict): - at = self._stringify(album.get("title")).lower() + at = stringify(album.get("title")).lower() else: - at = self._stringify(track.get("album")).lower() + at = stringify(track.get("album")).lower() if not at: continue @@ -765,7 +780,7 @@ class Tidal(Provider): def _album_item_to_result(self, album: Dict[str, Any], *, artist_name: str) -> Optional[SearchResult]: if not isinstance(album, dict): return None - title = self._stringify(album.get("title")) + title = stringify(album.get("title")) if not title: return None # Prefer albumId when available; some payloads carry both id/albumId. @@ -784,7 +799,7 @@ class Tidal(Provider): if total_time: columns.append(("Total", total_time)) - release_date = self._stringify(album.get("releaseDate") or album.get("release_date") or album.get("date")) + release_date = stringify(album.get("releaseDate") or album.get("release_date") or album.get("date")) if release_date: columns.append(("Release", release_date)) @@ -804,6 +819,11 @@ class Tidal(Provider): full_metadata=md, ) + @staticmethod + def url_patterns() -> List[str]: + """Return URL prefixes handled by this provider.""" + return ["tidal://", "tidal.com"] + @staticmethod def _find_ffmpeg() -> Optional[str]: exe = shutil.which("ffmpeg") @@ -936,7 +956,7 @@ class Tidal(Provider): dur = int(duration_seconds) if duration_seconds is not None else None except Exception: dur = None - if not dur or dur <= 0: + if not dur or dur <= 0: return None qual = str(audio_quality or "").strip().lower() @@ -1115,34 +1135,28 @@ class Tidal(Provider): if isinstance(getattr(result, "full_metadata", None), dict): md = dict(getattr(result, "full_metadata") or {}) - if not md.get("manifest"): - track_id = self._extract_track_id_from_result(result) - if track_id: - detail = self._fetch_track_details(track_id) - if isinstance(detail, dict) and detail: - try: - md.update(detail) - except Exception: - md = detail + track_id = self._extract_track_id_from_result(result) + if track_id: + # Multi-part enrichment from API: metadata, tags, and lyrics. + full_data = self._fetch_all_track_data(track_id) + if isinstance(full_data, dict): + # 1. Update metadata + api_md = full_data.get("metadata") + if isinstance(api_md, dict): + md.update(api_md) + + # 2. Update tags (re-sync result.tag so cmdlet sees them) + api_tags = full_data.get("tags") + if isinstance(api_tags, list) and api_tags: + result.tag = set(api_tags) - # Best-effort: fetch synced lyric subtitles for MPV (LRC). - try: - track_id_for_lyrics = self._extract_track_id_from_result(result) - except Exception: - track_id_for_lyrics = None - if track_id_for_lyrics and not md.get("_tidal_lyrics_subtitles"): - lyr = self._fetch_track_lyrics(track_id_for_lyrics) - if isinstance(lyr, dict) and lyr: - try: - md.setdefault("lyrics", lyr) - except Exception: - pass - try: - subtitles = lyr.get("subtitles") + # 3. Handle lyrics + lyrics = full_data.get("lyrics") + if isinstance(lyrics, dict) and lyrics: + md.setdefault("lyrics", lyrics) + subtitles = lyrics.get("subtitles") if isinstance(subtitles, str) and subtitles.strip(): md["_tidal_lyrics_subtitles"] = subtitles.strip() - except Exception: - pass # Ensure downstream cmdlets see our enriched metadata. try: @@ -1190,7 +1204,7 @@ class Tidal(Provider): output_path=out_file, progress=self.config.get("_pipeline_progress") if isinstance(self.config, dict) else None, transfer_label=title_part or getattr(result, "title", None), - duration_seconds=self._coerce_duration_seconds(md), + duration_seconds=coerce_duration_seconds(md), audio_quality=md.get("audioQuality") if isinstance(md, dict) else None, ) if materialized is not None: @@ -1223,7 +1237,7 @@ class Tidal(Provider): output_path=out_file, progress=self.config.get("_pipeline_progress") if isinstance(self.config, dict) else None, transfer_label=title_part or getattr(result, "title", None), - duration_seconds=self._coerce_duration_seconds(md), + duration_seconds=coerce_duration_seconds(md), audio_quality=md.get("audioQuality") if isinstance(md, dict) else None, ) if materialized is not None: @@ -1248,7 +1262,7 @@ class Tidal(Provider): output_path=out_file, progress=self.config.get("_pipeline_progress") if isinstance(self.config, dict) else None, transfer_label=title_part or getattr(result, "title", None), - duration_seconds=self._coerce_duration_seconds(md), + duration_seconds=coerce_duration_seconds(md), audio_quality=md.get("audioQuality") if isinstance(md, dict) else None, ) return materialized @@ -1300,10 +1314,10 @@ class Tidal(Provider): if isinstance(metadata, dict): album_obj = metadata.get("album") if isinstance(album_obj, dict): - album_title = self._stringify(album_obj.get("title")) + album_title = stringify(album_obj.get("title")) else: - album_title = self._stringify(album_obj or metadata.get("album")) - artists = self._extract_artists(metadata) + album_title = stringify(album_obj or metadata.get("album")) + artists = extract_artists(metadata) if artists: artist_name = artists[0] @@ -1320,7 +1334,7 @@ class Tidal(Provider): return False, None - def _get_api_client_for_base(self, base_url: str) -> Optional[tidalApiClient]: + def _get_api_client_for_base(self, base_url: str) -> Optional[TidalApiClient]: base = base_url.rstrip("/") for client in self.api_clients: if getattr(client, "base_url", "").rstrip("/") == base: @@ -1551,18 +1565,18 @@ class Tidal(Provider): except (TypeError, ValueError): return None - path = f"tidal://artist/{artist_id}" + path = f"hifi://artist/{artist_id}" columns: List[tuple[str, str]] = [("Artist", name), ("Artist ID", str(artist_id))] - popularity = self._stringify(item.get("popularity")) + popularity = stringify(item.get("popularity")) if popularity: columns.append(("Popularity", popularity)) return SearchResult( - table="tidal.artist", + table="hifi.artist", title=name, path=path, - detail="tidal.artist", + detail="hifi.artist", annotations=["tidal", "artist"], media_kind="audio", columns=columns, @@ -1580,18 +1594,6 @@ class Tidal(Provider): minutes, secs = divmod(total, 60) return f"{minutes}:{secs:02d}" - @staticmethod - def _coerce_duration_seconds(value: Any) -> Optional[int]: - return coerce_duration_seconds(value) - - @staticmethod - def _stringify(value: Any) -> str: - return stringify(value) - - @staticmethod - def _extract_artists(item: Dict[str, Any]) -> List[str]: - return extract_artists(item) - def _item_to_result(self, item: Dict[str, Any]) -> Optional[SearchResult]: if not isinstance(item, dict): return None @@ -1609,9 +1611,9 @@ class Tidal(Provider): return None # Avoid tidal.com URLs entirely; selection will resolve to a decoded MPD. - path = f"tidal://track/{track_id}" + path = f"hifi://track/{track_id}" - artists = self._extract_artists(item) + artists = extract_artists(item) artist_display = ", ".join(artists) album = item.get("album") @@ -1629,8 +1631,8 @@ class Tidal(Provider): columns: List[tuple[str, str]] = [] if title: columns.append(("Title", title)) - disc_no = self._stringify(item.get("volumeNumber") or item.get("discNumber") or item.get("disc_number")) - track_no = self._stringify(item.get("trackNumber") or item.get("track_number")) + disc_no = stringify(item.get("volumeNumber") or item.get("discNumber") or item.get("disc_number")) + track_no = stringify(item.get("trackNumber") or item.get("track_number")) if disc_no: columns.append(("Disc #", disc_no)) if track_no: @@ -1651,22 +1653,23 @@ class Tidal(Provider): # manifest path/URL. If multiple results share the same dict reference, # they can incorrectly collapse to a single playable target. full_md: Dict[str, Any] = dict(item) - url_value = self._stringify(full_md.get("url")) + url_value = stringify(full_md.get("url")) if url_value: full_md["url"] = url_value - tags = self._build_track_tags(full_md) + tags = build_track_tags(full_md) result = SearchResult( - table="tidal.track", + table="hifi.track", title=title, path=path, - detail="tidal.track", + detail="hifi.track", annotations=["tidal", "track"], media_kind="audio", tag=tags, columns=columns, full_metadata=full_md, + selection_args=["-url", path], ) if url_value: try: @@ -1735,91 +1738,52 @@ class Tidal(Provider): path = ( payload.get("path") or payload.get("url") - or f"tidal://track/{track_id}" + or f"hifi://track/{track_id}" ) contexts.append((track_id, str(title).strip(), str(path).strip())) return contexts def _fetch_track_details(self, track_id: int) -> Optional[Dict[str, Any]]: - if track_id <= 0: - return None + """Legacy wrapper returning just metadata from the consolidated API call.""" + res = self._fetch_all_track_data(track_id) + return res.get("metadata") if res else None - info_data = self._fetch_track_info(track_id) - - for base in self.api_urls: - endpoint = f"{base.rstrip('/')}/track/" - try: - 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): - merged: Dict[str, Any] = {} - if isinstance(info_data, dict): - merged.update(info_data) - merged.update(data) - return merged - except Exception as exc: - log(f"[tidal] Track lookup failed for {endpoint}: {exc}", file=sys.stderr) - continue - return None - - def _fetch_track_info(self, track_id: int) -> Optional[Dict[str, Any]]: + def _fetch_all_track_data(self, track_id: int) -> Optional[Dict[str, Any]]: + """Fetch full track details including metadata, tags, and lyrics from the API.""" if track_id <= 0: return None for base in self.api_urls: - endpoint = f"{base.rstrip('/')}/info/" try: client = self._get_api_client_for_base(base) - payload = client.info(track_id) if client else None - data = payload.get("data") if isinstance(payload, dict) else None - if isinstance(data, dict): - return data + if not client: + continue + # This method in the API client handles merging info+track and building tags. + return client.get_full_track_metadata(track_id) except Exception as exc: - debug(f"[tidal] Info lookup failed for {endpoint}: {exc}") + debug(f"[tidal] Full track fetch failed for {base}: {exc}") continue return None def _fetch_track_lyrics(self, track_id: int) -> Optional[Dict[str, Any]]: - if track_id <= 0: - return None - for base in self.api_urls: - endpoint = f"{base.rstrip('/')}/lyrics/" - try: - client = self._get_api_client_for_base(base) - payload = client.lyrics(track_id) if client else None - if not isinstance(payload, dict): - continue - - lyrics_obj = payload.get("lyrics") - if isinstance(lyrics_obj, dict) and lyrics_obj: - return lyrics_obj - - data_obj = payload.get("data") - if isinstance(data_obj, dict) and data_obj: - return data_obj - except Exception as exc: - debug(f"[tidal] Lyrics lookup failed for {endpoint}: {exc}") - continue - return None + """Legacy wrapper returning just lyrics from the consolidated API call.""" + res = self._fetch_all_track_data(track_id) + return res.get("lyrics") if res else None def _build_track_columns(self, detail: Dict[str, Any], track_id: int) -> List[Tuple[str, str]]: values: List[Tuple[str, str]] = [ ("Track ID", str(track_id)), - ("Quality", self._stringify(detail.get("audioQuality"))), - ("Mode", self._stringify(detail.get("audioMode"))), - ("Asset", self._stringify(detail.get("assetPresentation"))), - ("Manifest Type", self._stringify(detail.get("manifestMimeType"))), - ("Manifest Hash", self._stringify(detail.get("manifestHash"))), - ("Bit Depth", self._stringify(detail.get("bitDepth"))), - ("Sample Rate", self._stringify(detail.get("sampleRate"))), + ("Quality", stringify(detail.get("audioQuality"))), + ("Mode", stringify(detail.get("audioMode"))), + ("Asset", stringify(detail.get("assetPresentation"))), + ("Manifest Type", stringify(detail.get("manifestMimeType"))), + ("Manifest Hash", stringify(detail.get("manifestHash"))), + ("Bit Depth", stringify(detail.get("bitDepth"))), + ("Sample Rate", stringify(detail.get("sampleRate"))), ] return [(name, value) for name, value in values if value] - def _build_track_tags(self, metadata: Dict[str, Any]) -> set[str]: - return build_track_tags(metadata) - + @staticmethod def selection_auto_stage( - self, table_type: str, stage_args: Optional[Sequence[str]] = None, ) -> Optional[List[str]]: @@ -2006,17 +1970,11 @@ class Tidal(Provider): return True + # Optimization: If we are selecting tracks, do NOT force a "Detail View" (resolving manifest) here. + # This allows batch selection to flow immediately to `download-file` (via TABLE_AUTO_STAGES) + # or other downstream cmdlets. The download logic (tidal.download) handles manifest resolution locally. if table_type == "tidal.track" or (is_generic_tidal and any(str(get_field(i, "path")).startswith("tidal://track/") for i in selected_items)): - try: - meta = ( - current_table.get_table_metadata() - if current_table is not None and hasattr(current_table, "get_table_metadata") - else {} - ) - except Exception: - meta = {} - if isinstance(meta, dict) and meta.get("resolved_manifest"): - return False + return False contexts = self._extract_track_selection_context(selected_items) try: @@ -2049,34 +2007,24 @@ class Tidal(Provider): pass results_payload: List[Dict[str, Any]] = [] for track_id, title, path, detail in track_details: - # Decode the DASH MPD manifest to a local file and use it as the selectable/playable path. - try: - from cmdlet._shared import resolve_tidal_manifest_path + resolved_path = f"tidal://track/{track_id}" - manifest_path = resolve_tidal_manifest_path( - {"full_metadata": detail, "path": f"tidal://track/{track_id}"} - ) - except Exception: - manifest_path = None - - resolved_path = str(manifest_path) if manifest_path else f"tidal://track/{track_id}" - - artists = self._extract_artists(detail) + artists = extract_artists(detail) artist_display = ", ".join(artists) if artists else "" columns = self._build_track_columns(detail, track_id) if artist_display: columns.insert(1, ("Artist", artist_display)) album = detail.get("album") if isinstance(album, dict): - album_title = self._stringify(album.get("title")) + album_title = stringify(album.get("title")) else: - album_title = self._stringify(detail.get("album")) + album_title = stringify(detail.get("album")) if album_title: insert_pos = 2 if artist_display else 1 columns.insert(insert_pos, ("Album", album_title)) - tags = self._build_track_tags(detail) - url_value = self._stringify(detail.get("url")) + tags = build_track_tags(detail) + url_value = stringify(detail.get("url")) result = SearchResult( table="tidal.track", @@ -2088,6 +2036,7 @@ class Tidal(Provider): columns=columns, full_metadata=detail, tag=tags, + selection_args=["-url", resolved_path], ) if url_value: try: