9fc3628062
Introduce a boot-phase guard so gunicorn worker import never blocks on Spotify, Qobuz, Deezer, or Tidal network validation. Network auth checks run only after module initialization completes.
1052 lines
42 KiB
Python
1052 lines
42 KiB
Python
"""
|
|
Tidal Download Client
|
|
Alternative music download source using tidalapi.
|
|
|
|
This client provides:
|
|
- Tidal search with metadata
|
|
- Device flow authentication (link.tidal.com)
|
|
- HiRes/Lossless/High quality audio downloads via Tidal v2 trackManifests endpoint
|
|
- Drop-in replacement compatible with Soulseek interface
|
|
"""
|
|
|
|
import os
|
|
import re
|
|
import asyncio
|
|
import uuid
|
|
import time
|
|
import shutil
|
|
import subprocess
|
|
from typing import List, Optional, Dict, Any, Tuple
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
from urllib.parse import urljoin
|
|
|
|
try:
|
|
import tidalapi
|
|
except ImportError:
|
|
tidalapi = None
|
|
|
|
import requests as http_requests
|
|
|
|
from utils.logging_config import get_logger
|
|
from config.settings import config_manager
|
|
|
|
# Import Soulseek data structures for drop-in replacement compatibility
|
|
from core.download_plugins.types import TrackResult, AlbumResult, DownloadStatus
|
|
from core.quality.source_map import quality_from_tidal_tier, quality_tier_for_source
|
|
|
|
logger = get_logger("tidal_download_client")
|
|
|
|
|
|
# Quality tiers definitions (used for display in search results)
|
|
QUALITY_MAP = {
|
|
'low': {
|
|
'label': 'AAC 96kbps',
|
|
'extension': 'm4a',
|
|
'bitrate': 96,
|
|
'codec': 'aac',
|
|
},
|
|
'high': {
|
|
'label': 'AAC 320kbps',
|
|
'extension': 'm4a',
|
|
'bitrate': 320,
|
|
'codec': 'aac',
|
|
},
|
|
'lossless': {
|
|
'label': 'FLAC 16-bit/44.1kHz',
|
|
'extension': 'flac',
|
|
'bitrate': 1411,
|
|
'codec': 'flac',
|
|
},
|
|
'hires': {
|
|
'label': 'FLAC 24-bit/96kHz',
|
|
'extension': 'flac',
|
|
'bitrate': 9216,
|
|
'codec': 'flac',
|
|
},
|
|
}
|
|
|
|
# HLS-specific format mapping for v2 trackManifests endpoint
|
|
HLS_QUALITY_MAP = {
|
|
'hires': {
|
|
'formats': ['FLAC_HIRES'],
|
|
'manifest_type': 'HLS',
|
|
'extension': 'flac',
|
|
},
|
|
'lossless': {
|
|
'formats': ['FLAC'],
|
|
'manifest_type': 'HLS',
|
|
'extension': 'flac',
|
|
},
|
|
'high': {
|
|
'formats': ['AACLC'],
|
|
'manifest_type': 'HLS',
|
|
'extension': 'm4a',
|
|
},
|
|
'low': {
|
|
'formats': ['HEAACV1'],
|
|
'manifest_type': 'HLS',
|
|
'extension': 'm4a',
|
|
},
|
|
}
|
|
|
|
HLS_MAP_TAG_RE = re.compile(r'#EXT-X-MAP:.*URI="([^"]+)"')
|
|
|
|
|
|
from core.download_plugins.base import DownloadSourcePlugin
|
|
|
|
|
|
def _looks_like_json_decode_error(exc: Exception) -> bool:
|
|
name = exc.__class__.__name__.lower()
|
|
message = str(exc).lower()
|
|
return (
|
|
"jsondecodeerror" in name
|
|
or "expecting value" in message
|
|
or "could not decode json" in message
|
|
)
|
|
|
|
|
|
class TidalDownloadClient(DownloadSourcePlugin):
|
|
"""
|
|
Tidal download client using tidalapi.
|
|
Provides search, matching, and download capabilities as a drop-in alternative to YouTube/Soulseek.
|
|
"""
|
|
|
|
def __init__(self, download_path: str = None):
|
|
if tidalapi is None:
|
|
logger.warning("tidalapi not installed — Tidal downloads unavailable")
|
|
|
|
if download_path is None:
|
|
download_path = config_manager.get('soulseek.download_path', './downloads')
|
|
|
|
self.download_path = Path(download_path)
|
|
try:
|
|
self.download_path.mkdir(parents=True, exist_ok=True)
|
|
except OSError as e:
|
|
logger.warning(f"Could not verify download path {self.download_path}: {e}")
|
|
|
|
logger.info(f"Tidal download client using download path: {self.download_path}")
|
|
|
|
self.shutdown_check = None
|
|
|
|
self.session: Optional['tidalapi.Session'] = None
|
|
self._init_session()
|
|
|
|
self._device_auth_future = None
|
|
self._device_auth_link = None
|
|
self._boot_session_tokens: Optional[dict] = None
|
|
|
|
# Engine reference is populated by set_engine() at registration
|
|
# time. Until then dispatch returns None — orchestrator wires
|
|
# this immediately so the only None case is tests that bypass
|
|
# the orchestrator.
|
|
self._engine = None
|
|
|
|
def set_engine(self, engine):
|
|
"""Engine callback — gives the client access to the central
|
|
thread worker + state store. Engine calls this during
|
|
``register_plugin`` if the plugin defines it."""
|
|
self._engine = engine
|
|
|
|
def set_shutdown_check(self, check_callable):
|
|
self.shutdown_check = check_callable
|
|
|
|
def _init_session(self):
|
|
if tidalapi is None:
|
|
return
|
|
|
|
self.session = tidalapi.Session()
|
|
|
|
saved = config_manager.get('tidal_download.session', {})
|
|
token_type = saved.get('token_type', '')
|
|
access_token = saved.get('access_token', '')
|
|
refresh_token = saved.get('refresh_token', '')
|
|
expiry_time = saved.get('expiry_time', 0)
|
|
|
|
if token_type and access_token:
|
|
from core.boot_phase import is_boot_phase
|
|
if is_boot_phase():
|
|
self._boot_session_tokens = saved
|
|
logger.info(
|
|
"Loaded Tidal download session from config (verification deferred until after boot)"
|
|
)
|
|
return
|
|
|
|
try:
|
|
expiry_dt = datetime.fromtimestamp(expiry_time, tz=timezone.utc) if expiry_time else None
|
|
|
|
restored = self.session.load_oauth_session(
|
|
token_type=token_type,
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
expiry_time=expiry_dt,
|
|
)
|
|
if restored and self.session.check_login():
|
|
logger.info("Restored Tidal download session from saved tokens")
|
|
self._save_session()
|
|
return
|
|
else:
|
|
logger.warning("Saved Tidal session tokens are invalid/expired")
|
|
except Exception as e:
|
|
logger.warning(f"Could not restore Tidal session: {e}")
|
|
|
|
def _complete_deferred_session(self) -> bool:
|
|
"""Finish restoring a session that was deferred during boot."""
|
|
pending = getattr(self, '_boot_session_tokens', None)
|
|
if not pending or tidalapi is None:
|
|
self._boot_session_tokens = None
|
|
return False
|
|
|
|
if not self.session:
|
|
self.session = tidalapi.Session()
|
|
|
|
token_type = pending.get('token_type', '')
|
|
access_token = pending.get('access_token', '')
|
|
refresh_token = pending.get('refresh_token', '')
|
|
expiry_time = pending.get('expiry_time', 0)
|
|
self._boot_session_tokens = None
|
|
|
|
try:
|
|
expiry_dt = datetime.fromtimestamp(expiry_time, tz=timezone.utc) if expiry_time else None
|
|
restored = self.session.load_oauth_session(
|
|
token_type=token_type,
|
|
access_token=access_token,
|
|
refresh_token=refresh_token,
|
|
expiry_time=expiry_dt,
|
|
)
|
|
if restored and self.session.check_login():
|
|
logger.info("Restored Tidal download session from saved tokens")
|
|
self._save_session()
|
|
return True
|
|
logger.warning("Saved Tidal session tokens are invalid/expired")
|
|
except Exception as e:
|
|
logger.warning(f"Could not restore Tidal session: {e}")
|
|
return False
|
|
|
|
def _save_session(self):
|
|
if not self.session:
|
|
return
|
|
config_manager.set('tidal_download.session', {
|
|
'token_type': self.session.token_type or '',
|
|
'access_token': self.session.access_token or '',
|
|
'refresh_token': self.session.refresh_token or '',
|
|
'expiry_time': self.session.expiry_time.timestamp() if self.session.expiry_time else 0,
|
|
})
|
|
|
|
def is_authenticated(self) -> bool:
|
|
from core.boot_phase import is_boot_phase
|
|
if is_boot_phase():
|
|
pending = getattr(self, '_boot_session_tokens', None)
|
|
return bool(pending and pending.get('access_token'))
|
|
|
|
if getattr(self, '_boot_session_tokens', None):
|
|
return self._complete_deferred_session()
|
|
|
|
if not self.session:
|
|
return False
|
|
try:
|
|
return self.session.check_login()
|
|
except Exception:
|
|
return False
|
|
|
|
def start_device_auth(self) -> Optional[Dict[str, str]]:
|
|
if tidalapi is None:
|
|
return None
|
|
|
|
try:
|
|
if not self.session:
|
|
self.session = tidalapi.Session()
|
|
|
|
login, future = self.session.login_oauth()
|
|
self._device_auth_future = future
|
|
raw_uri = login.verification_uri_complete or f"link.tidal.com/{login.user_code}"
|
|
if not raw_uri.startswith(('http://', 'https://')):
|
|
raw_uri = f"https://{raw_uri}"
|
|
self._device_auth_link = {
|
|
'verification_uri': raw_uri,
|
|
'user_code': login.user_code,
|
|
}
|
|
logger.info(f"Tidal device auth started — code: {login.user_code}")
|
|
return self._device_auth_link
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to start Tidal device auth: {e}")
|
|
return None
|
|
|
|
def check_device_auth(self) -> Dict[str, Any]:
|
|
if not self._device_auth_future:
|
|
return {'status': 'error', 'message': 'No auth in progress'}
|
|
|
|
try:
|
|
if self._device_auth_future.running():
|
|
return {
|
|
'status': 'pending',
|
|
'verification_uri': self._device_auth_link.get('verification_uri', ''),
|
|
'user_code': self._device_auth_link.get('user_code', ''),
|
|
}
|
|
|
|
result = self._device_auth_future.result(timeout=0)
|
|
if self.session and self.session.check_login():
|
|
self._save_session()
|
|
logger.info("Tidal device auth completed successfully")
|
|
return {'status': 'completed', 'message': 'Authenticated successfully'}
|
|
else:
|
|
return {'status': 'error', 'message': 'Auth completed but session invalid'}
|
|
|
|
except Exception as e:
|
|
if _looks_like_json_decode_error(e):
|
|
logger.error(
|
|
"Tidal device auth check received a non-JSON response from Tidal's token endpoint: %s",
|
|
e,
|
|
)
|
|
return {
|
|
'status': 'error',
|
|
'message': (
|
|
"Tidal returned an invalid auth response while SoulSync was finishing login. "
|
|
"Try again after disabling VPN/proxy/network filtering, then restart SoulSync if it repeats."
|
|
),
|
|
}
|
|
logger.error(f"Tidal device auth check error: {e}")
|
|
return {'status': 'error', 'message': str(e)}
|
|
|
|
def is_available(self) -> bool:
|
|
return tidalapi is not None and self.is_authenticated()
|
|
|
|
def is_configured(self) -> bool:
|
|
return self.is_available()
|
|
|
|
async def check_connection(self) -> bool:
|
|
try:
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(None, self.is_available)
|
|
except Exception as e:
|
|
logger.error(f"Tidal connection check failed: {e}")
|
|
return False
|
|
|
|
_QUALIFIER_KEYWORDS = frozenset({
|
|
'remix', 'mix', 'edit', 'version', 'dub', 'rmx', 'vip', 'cut',
|
|
'rework', 'bootleg', 'flip',
|
|
'live', 'concert', 'unplugged', 'acoustic', 'session',
|
|
'instrumental', 'karaoke', 'demo', 'bonus',
|
|
'extended', 'radio',
|
|
})
|
|
|
|
@classmethod
|
|
def _extract_qualifiers(cls, query: str) -> List[str]:
|
|
if not query:
|
|
return []
|
|
found = []
|
|
q_lower = query.lower()
|
|
for kw in cls._QUALIFIER_KEYWORDS:
|
|
if re.search(r'\b' + re.escape(kw) + r'\b', q_lower):
|
|
found.append(kw)
|
|
return found
|
|
|
|
@staticmethod
|
|
def _track_name_contains_qualifiers(track_name: str, qualifiers: List[str]) -> bool:
|
|
if not qualifiers:
|
|
return True
|
|
if not track_name:
|
|
return False
|
|
name_lower = track_name.lower()
|
|
for kw in qualifiers:
|
|
if not re.search(r'\b' + re.escape(kw) + r'\b', name_lower):
|
|
return False
|
|
return True
|
|
|
|
@classmethod
|
|
def _track_matches_qualifiers(cls, track, qualifiers: List[str]) -> bool:
|
|
"""Issue #589 — qualifier check must inspect track.name, track.version
|
|
AND track.album.name. Tidal stores remix/live/edit qualifiers in a
|
|
dedicated `version` attribute (e.g. name="Emerge", version="Junkie XL
|
|
Remix"), and for MTV Unplugged-style releases the live / unplugged
|
|
signal lives in the album title. A track passes if every required
|
|
qualifier appears as a whole word in the track name, its version, OR
|
|
its album name.
|
|
"""
|
|
if not qualifiers:
|
|
return True
|
|
track_name = (getattr(track, 'name', '') or '').lower()
|
|
version = (getattr(track, 'version', '') or '').lower()
|
|
album = getattr(track, 'album', None)
|
|
album_name = (getattr(album, 'name', '') or '').lower() if album else ''
|
|
haystack = f"{track_name} {version} {album_name}".strip()
|
|
if not haystack:
|
|
return False
|
|
for kw in qualifiers:
|
|
if not re.search(r'\b' + re.escape(kw) + r'\b', haystack):
|
|
return False
|
|
return True
|
|
|
|
@staticmethod
|
|
def _generate_shortened_queries(original: str) -> List[str]:
|
|
variants: List[str] = []
|
|
seen = {original.strip().lower()}
|
|
|
|
def _add(candidate: str) -> None:
|
|
candidate = candidate.strip()
|
|
if candidate and candidate.lower() not in seen:
|
|
variants.append(candidate)
|
|
seen.add(candidate.lower())
|
|
|
|
_add(re.sub(r'\s*[\(\[][^\)\]]*[\)\]]\s*$', '', original))
|
|
_add(re.sub(r'\s*[\(\[][^\)\]]*[\)\]]', ' ', original))
|
|
|
|
tokens = original.split()
|
|
|
|
if len(tokens) >= 3:
|
|
_add(' '.join(tokens[:-1]))
|
|
|
|
if len(tokens) >= 4:
|
|
_add(' '.join(tokens[:-2]))
|
|
|
|
if len(tokens) >= 5:
|
|
_add(' '.join(tokens[:-3]))
|
|
|
|
if len(tokens) >= 7:
|
|
_add(' '.join(tokens[:len(tokens) // 2 + 1]))
|
|
|
|
return variants
|
|
|
|
async def search(self, query: str, timeout: int = None, progress_callback=None) -> Tuple[List[TrackResult], List[AlbumResult]]:
|
|
if not self.is_available():
|
|
logger.warning("Tidal not available for search (not authenticated)")
|
|
return ([], [])
|
|
|
|
if not query or not isinstance(query, str):
|
|
logger.warning(f"Invalid Tidal search query: {query!r}")
|
|
return ([], [])
|
|
|
|
logger.info(f"Searching Tidal for: {query}")
|
|
|
|
try:
|
|
queries_to_try = [query] + self._generate_shortened_queries(query)
|
|
queries_to_try = queries_to_try[:5]
|
|
|
|
required_qualifiers = self._extract_qualifiers(query)
|
|
|
|
tidal_tracks: list = []
|
|
successful_query: Optional[str] = None
|
|
last_error: Optional[Exception] = None
|
|
any_fallback_filtered_out = False
|
|
|
|
loop = asyncio.get_event_loop()
|
|
for attempt_idx, attempt_query in enumerate(queries_to_try):
|
|
try:
|
|
q_copy = attempt_query
|
|
|
|
def _search(q=q_copy):
|
|
results = self.session.search(q, models=[tidalapi.media.Track], limit=50)
|
|
return results.get('tracks', []) if isinstance(results, dict) else []
|
|
|
|
found = await loop.run_in_executor(None, _search)
|
|
|
|
if found:
|
|
# Issue #589 — qualifier filter applies to ALL
|
|
# search attempts, not just fallbacks. If the
|
|
# primary query carries "live" / "unplugged" /
|
|
# etc, the studio cut should never be accepted
|
|
# just because Tidal returned it first. The
|
|
# filter inspects both track.name AND
|
|
# track.album.name (the live signal often lives
|
|
# in the album title for concert releases).
|
|
is_fallback = attempt_idx > 0
|
|
if required_qualifiers:
|
|
filtered = [
|
|
t for t in found
|
|
if self._track_matches_qualifiers(t, required_qualifiers)
|
|
]
|
|
if filtered:
|
|
tidal_tracks = filtered
|
|
successful_query = attempt_query
|
|
logger.info(
|
|
f"Tidal {'fallback' if is_fallback else 'primary'} kept "
|
|
f"{len(filtered)}/{len(found)} tracks after qualifier filter "
|
|
f"{required_qualifiers} for '{attempt_query}'"
|
|
)
|
|
break
|
|
else:
|
|
any_fallback_filtered_out = True
|
|
logger.debug(
|
|
f"Tidal {'fallback' if is_fallback else 'primary'} "
|
|
f"'{attempt_query}' returned {len(found)} tracks but none "
|
|
f"matched required qualifiers {required_qualifiers} — "
|
|
f"trying next variant"
|
|
)
|
|
if attempt_idx < len(queries_to_try) - 1:
|
|
await asyncio.sleep(0.1)
|
|
continue
|
|
else:
|
|
tidal_tracks = found
|
|
successful_query = attempt_query
|
|
break
|
|
|
|
if attempt_idx < len(queries_to_try) - 1:
|
|
logger.debug(f"Tidal returned 0 results for '{attempt_query}' — trying shortened variant")
|
|
await asyncio.sleep(0.1)
|
|
except Exception as e:
|
|
last_error = e
|
|
logger.debug(f"Tidal search attempt {attempt_idx + 1} failed: {e}")
|
|
|
|
if not tidal_tracks:
|
|
if last_error is not None:
|
|
import traceback
|
|
tb_str = ''.join(traceback.format_exception(
|
|
type(last_error), last_error, last_error.__traceback__
|
|
))
|
|
logger.error(
|
|
f"Tidal search failed after {len(queries_to_try)} attempts: {last_error}\n{tb_str}"
|
|
)
|
|
elif any_fallback_filtered_out:
|
|
logger.warning(
|
|
f"No Tidal results for '{query}' — fallbacks found broader matches but "
|
|
f"none preserved required qualifiers {required_qualifiers}"
|
|
)
|
|
else:
|
|
logger.warning(f"No Tidal results for: {query}")
|
|
return ([], [])
|
|
|
|
if successful_query and successful_query != query:
|
|
logger.info(f"Tidal fallback query succeeded: '{successful_query}' (original: '{query}')")
|
|
|
|
quality_key = quality_tier_for_source('tidal', default='lossless')
|
|
quality_info = QUALITY_MAP.get(quality_key, QUALITY_MAP['lossless'])
|
|
# Stamp the configured tier (what will actually be downloaded) so
|
|
# the global ranker sees real sample_rate/bit_depth.
|
|
tier_quality = quality_from_tidal_tier(quality_key)
|
|
|
|
track_results = []
|
|
for track in tidal_tracks:
|
|
try:
|
|
track_result = self._tidal_to_track_result(track, quality_info)
|
|
track_result.set_quality(tier_quality)
|
|
track_results.append(track_result)
|
|
except Exception as e:
|
|
logger.debug(f"Skipping track conversion error: {e}")
|
|
|
|
logger.info(f"Found {len(track_results)} Tidal tracks")
|
|
return (track_results, [])
|
|
|
|
except Exception as e:
|
|
logger.error(f"Tidal search orchestration failed: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
return ([], [])
|
|
|
|
def _tidal_to_track_result(self, track, quality_info: dict) -> TrackResult:
|
|
artist_name = track.artist.name if track.artist else 'Unknown Artist'
|
|
title = track.name or 'Unknown Title'
|
|
# Tidal keeps remix/live/edition qualifiers in a separate `version`
|
|
# attribute (e.g. name="Emerge", version="Junkie XL Remix"). The
|
|
# matcher only scores `title`, so without folding the version in,
|
|
# every remix/live/edit candidate presents as the bare base track
|
|
# and MusicMatchingEngine's strict version check rejects it against
|
|
# a "... (Junkie XL Remix)" request. Append the version (unless it's
|
|
# already in the name) so the candidate title is the full
|
|
# "Title (Version)".
|
|
version = getattr(track, 'version', None)
|
|
if version and version.strip() and version.strip().lower() not in title.lower():
|
|
title = f"{title} ({version.strip()})"
|
|
album_name = track.album.name if track.album else None
|
|
|
|
duration_ms = int(track.duration * 1000) if track.duration else None
|
|
|
|
display_name = f"{artist_name} - {title}"
|
|
filename = f"{track.id}||{display_name}"
|
|
|
|
track_result = TrackResult(
|
|
username='tidal',
|
|
filename=filename,
|
|
size=0,
|
|
bitrate=quality_info.get('bitrate'),
|
|
duration=duration_ms,
|
|
quality=quality_info.get('codec', 'flac'),
|
|
free_upload_slots=999,
|
|
upload_speed=999999,
|
|
queue_length=0,
|
|
artist=artist_name,
|
|
title=title,
|
|
album=album_name,
|
|
track_number=track.track_num,
|
|
_source_metadata={
|
|
'source': 'tidal',
|
|
'track_id': track.id,
|
|
'artist_id': track.artist.id if track.artist else None,
|
|
'isrc': track.isrc or None,
|
|
'bpm': track.bpm if track.bpm and track.bpm > 0 else None,
|
|
'copyright': track.copyright or None,
|
|
},
|
|
)
|
|
|
|
return track_result
|
|
|
|
def _parse_hls_playlist(self, text: str, playlist_url: str):
|
|
init_uri = None
|
|
segment_uris = []
|
|
variant_uri = None
|
|
|
|
lines = [line.strip() for line in text.splitlines() if line.strip()]
|
|
|
|
for index, line in enumerate(lines):
|
|
if line.startswith('#EXTM3U'):
|
|
continue
|
|
|
|
if line.startswith('#EXT-X-STREAM-INF'):
|
|
for next_line in lines[index + 1:]:
|
|
if not next_line.startswith('#'):
|
|
variant_uri = urljoin(playlist_url, next_line)
|
|
break
|
|
break
|
|
|
|
if line.startswith('#EXT-X-MAP'):
|
|
match = HLS_MAP_TAG_RE.search(line)
|
|
if match:
|
|
init_uri = match.group(1)
|
|
continue
|
|
|
|
if line.startswith('#'):
|
|
continue
|
|
|
|
segment_uris.append(urljoin(playlist_url, line))
|
|
|
|
if variant_uri:
|
|
return None, [variant_uri]
|
|
|
|
if not segment_uris:
|
|
raise ValueError('No segment URIs found in the HLS playlist')
|
|
|
|
if init_uri:
|
|
init_uri = urljoin(playlist_url, init_uri)
|
|
|
|
return init_uri, segment_uris
|
|
|
|
# Tidal aggressively rate-limits the trackManifests endpoint. When a
|
|
# batch fans out, a bare request fails 429 instantly, the quality tier is
|
|
# burned, the track is re-queued, and the retry hammers again — a
|
|
# self-amplifying storm (thousands of 429s, downloads stalled, only the
|
|
# lowest tier squeaking through). Honour the server's Retry-After (or back
|
|
# off exponentially) so downloads pace themselves to whatever rate Tidal
|
|
# allows instead of slamming the wall and instant-failing.
|
|
_MANIFEST_MAX_RETRIES = 5
|
|
_MANIFEST_BACKOFF_BASE = 2.0 # seconds → 2, 4, 8, 16, 32 (capped)
|
|
_MANIFEST_BACKOFF_CAP = 30.0
|
|
|
|
@classmethod
|
|
def _retry_after_seconds(cls, response, attempt: int) -> float:
|
|
"""Backoff delay for a rate-limited response: honour a numeric
|
|
Retry-After header when present, else exponential backoff (capped)."""
|
|
retry_after = response.headers.get('Retry-After') if response is not None else None
|
|
if retry_after:
|
|
try:
|
|
return min(max(float(retry_after), 0.0), cls._MANIFEST_BACKOFF_CAP)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return min(cls._MANIFEST_BACKOFF_BASE * (2 ** attempt), cls._MANIFEST_BACKOFF_CAP)
|
|
|
|
def _sleep_with_shutdown(self, delay: float) -> bool:
|
|
"""Sleep up to `delay` seconds in small slices. Returns True if a
|
|
shutdown was requested mid-sleep so the caller can bail early."""
|
|
slept = 0.0
|
|
while slept < delay:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
return True
|
|
chunk = min(0.5, delay - slept)
|
|
time.sleep(chunk)
|
|
slept += chunk
|
|
return False
|
|
|
|
def _get_with_rate_limit_retry(self, url: str, *, params=None, headers=None,
|
|
timeout: int = 20):
|
|
"""HTTP GET that backs off and retries on 429 (and transient 5xx),
|
|
honouring Retry-After. Returns the final Response (the caller still
|
|
calls raise_for_status); a persistent rate-limit after all retries
|
|
returns the last 429 response so existing error handling applies."""
|
|
response = None
|
|
for attempt in range(self._MANIFEST_MAX_RETRIES + 1):
|
|
response = http_requests.get(url, params=params, headers=headers, timeout=timeout)
|
|
if response.status_code != 429 and response.status_code < 500:
|
|
return response
|
|
if attempt >= self._MANIFEST_MAX_RETRIES:
|
|
return response
|
|
delay = self._retry_after_seconds(response, attempt)
|
|
logger.warning(
|
|
f"Tidal returned {response.status_code} on manifest fetch — backing off "
|
|
f"{delay:.1f}s (attempt {attempt + 1}/{self._MANIFEST_MAX_RETRIES})")
|
|
if self._sleep_with_shutdown(delay):
|
|
return response
|
|
return response
|
|
|
|
def _get_hls_manifest(self, track_id: int, quality: str = 'lossless') -> Optional[Dict]:
|
|
q_info = HLS_QUALITY_MAP.get(quality, HLS_QUALITY_MAP['lossless'])
|
|
formats = q_info['formats']
|
|
|
|
access_token = self.session.access_token
|
|
if not access_token:
|
|
logger.error("No Tidal access token available")
|
|
return None
|
|
|
|
url = f"https://openapi.tidal.com/v2/trackManifests/{track_id}"
|
|
params = [
|
|
('adaptive', 'true'),
|
|
('manifestType', 'HLS'),
|
|
('uriScheme', 'HTTPS'),
|
|
('usage', 'DOWNLOAD'),
|
|
]
|
|
for fmt in formats:
|
|
params.append(('formats', fmt))
|
|
|
|
headers = {
|
|
'Authorization': f'Bearer {access_token}',
|
|
'Accept': 'application/vnd.api+json',
|
|
}
|
|
|
|
try:
|
|
response = self._get_with_rate_limit_retry(url, params=params, headers=headers, timeout=20)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
except http_requests.HTTPError as e:
|
|
logger.warning(f"Failed to fetch HLS manifest for track {track_id}: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch HLS manifest for track {track_id}: {e}")
|
|
return None
|
|
|
|
try:
|
|
attrs = data.get('data', {}).get('attributes', {})
|
|
uri = attrs.get('uri')
|
|
except (AttributeError, KeyError) as e:
|
|
logger.warning(f"Failed to extract playlist URI from manifest response for track {track_id}: {e}")
|
|
return None
|
|
|
|
if not uri:
|
|
logger.warning(f"No playlist URI in manifest for track {track_id}")
|
|
return None
|
|
|
|
try:
|
|
playlist_resp = http_requests.get(uri, allow_redirects=True, timeout=30)
|
|
playlist_resp.raise_for_status()
|
|
playlist_text = playlist_resp.text
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch HLS playlist for track {track_id}: {e}")
|
|
return None
|
|
|
|
try:
|
|
init_uri, segment_uris = self._parse_hls_playlist(playlist_text, uri)
|
|
except ValueError as e:
|
|
logger.warning(f"Failed to parse HLS playlist for track {track_id}: {e}")
|
|
return None
|
|
|
|
if '#EXT-X-STREAM-INF' in playlist_text and segment_uris:
|
|
playlist_uri = segment_uris[0]
|
|
try:
|
|
logger.debug(f"Detected master HLS playlist, following variant: {playlist_uri}")
|
|
variant_resp = http_requests.get(playlist_uri, allow_redirects=True, timeout=30)
|
|
variant_resp.raise_for_status()
|
|
variant_text = variant_resp.text
|
|
init_uri, segment_uris = self._parse_hls_playlist(variant_text, playlist_uri)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch variant playlist for track {track_id}: {e}")
|
|
return None
|
|
|
|
if init_uri:
|
|
logger.info(f"Tidal HLS manifest for track {track_id}: "
|
|
f"init segment + {len(segment_uris)} segments ({quality})")
|
|
else:
|
|
logger.info(f"Tidal HLS manifest for track {track_id}: "
|
|
f"{len(segment_uris)} segments ({quality})")
|
|
|
|
return {
|
|
'init_uri': init_uri,
|
|
'segment_uris': segment_uris,
|
|
'extension': QUALITY_MAP.get(quality, {}).get('extension', 'flac'),
|
|
'codec': QUALITY_MAP.get(quality, {}).get('codec', 'flac'),
|
|
'quality': quality,
|
|
}
|
|
|
|
def _demux_flac(self, input_path: Path, output_path: Path) -> None:
|
|
ffmpeg = shutil.which('ffmpeg')
|
|
if not ffmpeg:
|
|
tools_dir = Path(__file__).parent.parent / 'tools'
|
|
ffmpeg_candidate = tools_dir / ('ffmpeg.exe' if os.name == 'nt' else 'ffmpeg')
|
|
if ffmpeg_candidate.exists():
|
|
ffmpeg = str(ffmpeg_candidate)
|
|
else:
|
|
raise RuntimeError('ffmpeg is required to demux FLAC from MP4. Install ffmpeg and retry.')
|
|
|
|
try:
|
|
result = subprocess.run(
|
|
[
|
|
ffmpeg,
|
|
'-y',
|
|
'-hide_banner',
|
|
'-loglevel', 'error',
|
|
'-i', str(input_path),
|
|
'-map', '0:a:0',
|
|
'-c', 'copy',
|
|
str(output_path),
|
|
],
|
|
check=True,
|
|
capture_output=True,
|
|
text=True,
|
|
)
|
|
except subprocess.CalledProcessError as exc:
|
|
raise RuntimeError(
|
|
f'ffmpeg failed while demuxing {input_path} -> {output_path}: '
|
|
f'{exc.returncode}\n{exc.stderr}'
|
|
) from exc
|
|
|
|
async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]:
|
|
if '||' not in filename:
|
|
logger.error(f"Invalid filename format: {filename}")
|
|
return None
|
|
if self._engine is None:
|
|
# Raise rather than return None so the orchestrator's
|
|
# download_with_fallback surfaces a real warning + tries
|
|
# the next source. Returning None silently dropped the
|
|
# download with no user feedback (per JohnBaumb).
|
|
raise RuntimeError("Tidal client has no engine reference — cannot dispatch download")
|
|
|
|
track_id_str, display_name = filename.split('||', 1)
|
|
try:
|
|
track_id = int(track_id_str)
|
|
except ValueError:
|
|
logger.error(f"Invalid Tidal track ID: {track_id_str}")
|
|
return None
|
|
|
|
logger.info(f"Starting Tidal download: {display_name}")
|
|
|
|
return self._engine.worker.dispatch(
|
|
source_name='tidal',
|
|
target_id=track_id,
|
|
display_name=display_name,
|
|
original_filename=filename,
|
|
impl_callable=self._download_sync,
|
|
extra_record_fields={
|
|
'track_id': track_id,
|
|
'display_name': display_name,
|
|
},
|
|
)
|
|
|
|
def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]:
|
|
if not self.session or not self.session.check_login():
|
|
logger.error("Tidal session not authenticated")
|
|
return None
|
|
|
|
quality_key = quality_tier_for_source('tidal', default='lossless')
|
|
chain = ['hires', 'lossless', 'high', 'low']
|
|
start = chain.index(quality_key) if quality_key in chain else 1
|
|
allow_fallback = config_manager.get('tidal_download.allow_fallback', True)
|
|
chain = chain[start:] if allow_fallback else [quality_key]
|
|
|
|
MIN_AUDIO_SIZE = 100 * 1024
|
|
|
|
for q_key in chain:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
logger.info("Shutdown detected, aborting Tidal download")
|
|
return None
|
|
|
|
manifest_info = self._get_hls_manifest(track_id, quality=q_key)
|
|
if not manifest_info or not manifest_info.get('segment_uris'):
|
|
logger.warning(f"No HLS manifest at quality {q_key}, trying next")
|
|
continue
|
|
|
|
extension = manifest_info['extension']
|
|
safe_name = re.sub(r'[<>:"/\\|?*]', '_', display_name)
|
|
out_filename = f"{safe_name}.{extension}"
|
|
out_path = self.download_path / out_filename
|
|
|
|
is_flac = q_key in ('hires', 'lossless')
|
|
intermediate_path = out_path.with_suffix('.m4a') if is_flac else out_path
|
|
|
|
try:
|
|
init_uri = manifest_info.get('init_uri')
|
|
segment_uris = manifest_info['segment_uris']
|
|
total_segments = len(segment_uris) + (1 if init_uri else 0)
|
|
|
|
logger.info(f"Downloading from Tidal ({q_key}): {out_filename} "
|
|
f"({total_segments} segments)")
|
|
|
|
downloaded = 0
|
|
speed_start = time.time()
|
|
segments_completed = 0
|
|
|
|
if self._engine is not None:
|
|
self._engine.update_record('tidal', download_id, {'size': 0})
|
|
|
|
with intermediate_path.open('wb') as output_file:
|
|
if init_uri:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
logger.info("Shutdown detected, aborting Tidal download")
|
|
intermediate_path.unlink(missing_ok=True)
|
|
return None
|
|
|
|
logger.debug(f"Downloading init segment: {init_uri}")
|
|
init_data = self._download_segment_with_retry(init_uri)
|
|
output_file.write(init_data)
|
|
downloaded += len(init_data)
|
|
segments_completed += 1
|
|
|
|
self._update_download_progress(download_id, downloaded,
|
|
segments_completed, total_segments, speed_start)
|
|
|
|
for segment_url in segment_uris:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
logger.info("Shutdown detected, aborting Tidal download")
|
|
intermediate_path.unlink(missing_ok=True)
|
|
return None
|
|
|
|
segment_data = self._download_segment_with_retry(segment_url)
|
|
output_file.write(segment_data)
|
|
downloaded += len(segment_data)
|
|
segments_completed += 1
|
|
|
|
self._update_download_progress(download_id, downloaded,
|
|
segments_completed, total_segments, speed_start)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Download failed at quality {q_key}: {e}")
|
|
intermediate_path.unlink(missing_ok=True)
|
|
continue
|
|
|
|
if downloaded < MIN_AUDIO_SIZE:
|
|
logger.warning(f"File too small at {q_key} ({downloaded} bytes), trying next")
|
|
intermediate_path.unlink(missing_ok=True)
|
|
continue
|
|
|
|
try:
|
|
if is_flac:
|
|
logger.info(f"Demuxing FLAC from MP4 container: {intermediate_path} -> {out_path}")
|
|
self._demux_flac(intermediate_path, out_path)
|
|
intermediate_path.unlink(missing_ok=True)
|
|
final_size = out_path.stat().st_size if out_path.exists() else 0
|
|
else:
|
|
final_size = intermediate_path.stat().st_size if intermediate_path.exists() else 0
|
|
|
|
if final_size < MIN_AUDIO_SIZE:
|
|
logger.warning(f"Final file too small after processing at {q_key} "
|
|
f"({final_size} bytes), trying next")
|
|
out_path.unlink(missing_ok=True)
|
|
continue
|
|
|
|
logger.info(f"Tidal download complete ({q_key}): {out_path} "
|
|
f"({final_size / (1024*1024):.1f} MB)")
|
|
return str(out_path)
|
|
|
|
except Exception as e:
|
|
logger.warning(f"Post-processing failed at quality {q_key}: {e}")
|
|
out_path.unlink(missing_ok=True)
|
|
intermediate_path.unlink(missing_ok=True)
|
|
continue
|
|
|
|
logger.error(f"All quality tiers exhausted for '{display_name}'")
|
|
return None
|
|
|
|
def _download_segment_with_retry(self, url: str) -> bytes:
|
|
"""Download a single HLS segment with 3 retries and 2s fixed backoff."""
|
|
last_error = None
|
|
for attempt in range(4):
|
|
try:
|
|
resp = http_requests.get(url, allow_redirects=True, timeout=30)
|
|
resp.raise_for_status()
|
|
return resp.content
|
|
except http_requests.exceptions.HTTPError as e:
|
|
status = e.response.status_code if e.response is not None else 0
|
|
if 400 <= status < 500:
|
|
raise
|
|
last_error = e
|
|
except (http_requests.exceptions.Timeout,
|
|
http_requests.exceptions.ConnectionError) as e:
|
|
last_error = e
|
|
|
|
if attempt < 3:
|
|
if self.shutdown_check and self.shutdown_check():
|
|
raise RuntimeError("Shutdown requested")
|
|
logger.warning(f"Tidal segment download failed (attempt {attempt + 1}/4), "
|
|
f"retrying in 2s: {url}")
|
|
time.sleep(2)
|
|
|
|
raise last_error
|
|
|
|
def _update_download_progress(self, download_id: str, downloaded: int,
|
|
segments_completed: int, total_segments: int,
|
|
speed_start: float):
|
|
if self._engine is None:
|
|
return
|
|
record = self._engine.get_record('tidal', download_id)
|
|
if record is None:
|
|
return
|
|
|
|
now = time.time()
|
|
elapsed_total = now - speed_start
|
|
speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0
|
|
|
|
progress = record.get('progress', 0.0)
|
|
if total_segments > 0:
|
|
progress = round(min((segments_completed / total_segments) * 100, 99.9), 1)
|
|
|
|
time_remaining = None
|
|
if speed > 0:
|
|
remaining_bytes = downloaded * (total_segments / max(segments_completed, 1)) - downloaded
|
|
if remaining_bytes > 0:
|
|
time_remaining = int(remaining_bytes / speed)
|
|
|
|
self._engine.update_record('tidal', download_id, {
|
|
'transferred': downloaded,
|
|
'speed': speed,
|
|
'progress': progress,
|
|
'time_remaining': time_remaining,
|
|
})
|
|
|
|
def _record_to_status(self, record):
|
|
return DownloadStatus(
|
|
id=record['id'],
|
|
filename=record['filename'],
|
|
username=record['username'],
|
|
state=record['state'],
|
|
progress=record['progress'],
|
|
size=record.get('size', 0),
|
|
transferred=record.get('transferred', 0),
|
|
speed=record.get('speed', 0),
|
|
time_remaining=record.get('time_remaining'),
|
|
file_path=record.get('file_path'),
|
|
)
|
|
|
|
async def get_all_downloads(self) -> List[DownloadStatus]:
|
|
if self._engine is None:
|
|
return []
|
|
return [
|
|
self._record_to_status(record)
|
|
for record in self._engine.iter_records_for_source('tidal')
|
|
]
|
|
|
|
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
|
|
if self._engine is None:
|
|
return None
|
|
record = self._engine.get_record('tidal', download_id)
|
|
return self._record_to_status(record) if record is not None else None
|
|
|
|
async def cancel_download(self, download_id: str, username: str = None, remove: bool = False) -> bool:
|
|
if self._engine is None:
|
|
return False
|
|
if self._engine.get_record('tidal', download_id) is None:
|
|
logger.warning(f"Tidal download {download_id} not found")
|
|
return False
|
|
self._engine.update_record('tidal', download_id, {'state': 'Cancelled'})
|
|
logger.info(f"Marked Tidal download {download_id} as cancelled")
|
|
if remove:
|
|
self._engine.remove_record('tidal', download_id)
|
|
logger.info(f"Removed Tidal download {download_id} from queue")
|
|
return True
|
|
|
|
async def clear_all_completed_downloads(self) -> bool:
|
|
if self._engine is None:
|
|
return True
|
|
try:
|
|
terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
|
|
for record in list(self._engine.iter_records_for_source('tidal')):
|
|
if record.get('state') in terminal:
|
|
self._engine.remove_record('tidal', record['id'])
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error clearing downloads: {e}")
|
|
return False
|