goyimnose/Medios-Macina
1
0
Fork 0
Code Issues Pull Requests Packages Releases Wiki Activity

f

Browse Source
This commit is contained in:
Nose
2026-01-11 14:46:56 -08:00
parent 275f18cb31
commit 27646a8c3e
1 changed files with 114 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

279
Provider/Tidal.py
View File

@@ -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: