From 763888e6713b5a2044b57c98bedb31e3bf3e80eb Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Thu, 21 May 2026 18:11:14 -0700 Subject: [PATCH] Support legacy HiFi track manifests Add fallback support for public hifi-api instances that expose playback through /track/ instead of /trackManifests/. The capability checker now accepts either manifest shape, and downloads can use direct URLs decoded from the legacy base64 manifest. Tests cover legacy instance capability detection and download-manifest fallback while preserving the newer trackManifests path. --- core/hifi_client.py | 99 ++++++++++++++++++++++++++-- tests/downloads/test_hifi_pinning.py | 77 +++++++++++++++++++++- 2 files changed, 168 insertions(+), 8 deletions(-) diff --git a/core/hifi_client.py b/core/hifi_client.py index 82163f19..b6c854ee 100644 --- a/core/hifi_client.py +++ b/core/hifi_client.py @@ -23,6 +23,8 @@ import re import uuid import time import shutil +import json +import base64 import subprocess import threading from typing import List, Optional, Dict, Any, Tuple @@ -75,6 +77,13 @@ HLS_QUALITY_MAP = { HLS_MAP_TAG_RE = re.compile(r'#EXT-X-MAP:.*URI="([^"]+)"') +TRACK_ENDPOINT_QUALITY_MAP = { + 'hires': 'HI_RES_LOSSLESS', + 'lossless': 'LOSSLESS', + 'high': 'HIGH', + 'low': 'LOW', +} + # Default public hifi-api instances (ordered by preference) DEFAULT_INSTANCES = [ 'https://triton.squid.wtf', @@ -253,6 +262,37 @@ class HiFiClient(DownloadSourcePlugin): except (AttributeError, KeyError): return None + @staticmethod + def _extract_track_manifest_urls(data: Any) -> List[str]: + try: + inner = data.get('data', data) if isinstance(data, dict) else data + manifest_b64 = inner.get('manifest') + if not manifest_b64: + return [] + manifest = json.loads(base64.b64decode(manifest_b64)) + if manifest.get('encryptionType') not in (None, 'NONE'): + return [] + urls = manifest.get('urls') or [] + return [url for url in urls if isinstance(url, str) and url] + except Exception as e: + logger.debug("Failed to extract legacy HiFi track manifest URLs: %s", e) + return [] + + @staticmethod + def _extension_from_track_manifest(data: Any, fallback: str) -> str: + try: + inner = data.get('data', data) if isinstance(data, dict) else data + manifest = json.loads(base64.b64decode(inner.get('manifest') or '')) + mime = (manifest.get('mimeType') or '').lower() + codecs = (manifest.get('codecs') or '').lower() + if 'flac' in mime or 'flac' in codecs: + return 'flac' + if 'mp4' in mime or 'aac' in codecs: + return 'm4a' + except Exception: + pass + return fallback + def check_instance_capabilities(self, url: str, timeout: int = 5) -> Dict[str, Any]: """Probe one public HiFi instance using the endpoints SoulSync needs.""" entry = { @@ -302,7 +342,23 @@ class HiFiClient(DownloadSourcePlugin): if not manifest.ok: entry['download_error'] = f'HTTP {manifest.status_code}' elif not entry['can_download']: - entry['download_error'] = 'No HLS manifest URI' + legacy = self.session.get( + f'{url}/track/', + params={'id': '1550546', 'quality': 'LOSSLESS'}, + timeout=timeout, + ) + entry['can_download'] = ( + legacy.ok + and bool(self._extract_track_manifest_urls(legacy.json())) + ) + if not legacy.ok: + entry['download_error'] = f'HTTP {legacy.status_code}' + elif not entry['can_download']: + entry['download_error'] = 'No playable manifest URL' + else: + entry['download_probe'] = 'track' + else: + entry['download_probe'] = 'trackManifests' except http_requests.exceptions.SSLError: entry['status'] = 'ssl_error' except http_requests.exceptions.ConnectTimeout: @@ -502,12 +558,12 @@ class HiFiClient(DownloadSourcePlugin): data = self._api_get('/trackManifests/', params=params, timeout=20) if not data: - return None + return self._get_legacy_track_manifest(track_id, quality) uri = self._extract_manifest_uri(data) if uri is None: logger.warning("Failed to extract playlist URI from manifest response") - return None + return self._get_legacy_track_manifest(track_id, quality) if not uri: logger.warning(f"No playlist URI in manifest for track {track_id}") @@ -554,6 +610,28 @@ class HiFiClient(DownloadSourcePlugin): 'quality': quality, } + def _get_legacy_track_manifest(self, track_id: int, quality: str = 'lossless') -> Optional[Dict]: + q_info = HLS_QUALITY_MAP.get(quality, HLS_QUALITY_MAP['lossless']) + api_quality = TRACK_ENDPOINT_QUALITY_MAP.get(quality, 'LOSSLESS') + data = self._api_get('/track/', params={'id': track_id, 'quality': api_quality}, timeout=20) + if not data: + return None + + direct_urls = self._extract_track_manifest_urls(data) + if not direct_urls: + logger.warning(f"No playable URL in legacy HiFi manifest for track {track_id}") + return None + + extension = self._extension_from_track_manifest(data, q_info['extension']) + logger.info(f"HiFi legacy track manifest for track {track_id}: " + f"{len(direct_urls)} direct URL(s) ({quality})") + return { + 'direct_urls': direct_urls, + 'extension': extension, + 'codec': q_info['codec'], + 'quality': quality, + } + def _demux_flac(self, input_path: Path, output_path: Path) -> None: ffmpeg = shutil.which('ffmpeg') if not ffmpeg: @@ -685,7 +763,13 @@ class HiFiClient(DownloadSourcePlugin): return None manifest_info = self._get_hls_manifest(track_id, quality=q_key) - if not manifest_info or not manifest_info.get('segment_uris'): + if ( + not manifest_info + or ( + not manifest_info.get('segment_uris') + and not manifest_info.get('direct_urls') + ) + ): logger.warning(f"No HLS manifest at quality {q_key}, trying next") continue @@ -694,16 +778,17 @@ class HiFiClient(DownloadSourcePlugin): out_filename = f"{safe_name}.{extension}" out_path = self.download_path / out_filename - is_flac = q_key in ('hires', 'lossless') + is_direct = bool(manifest_info.get('direct_urls')) + is_flac = q_key in ('hires', 'lossless') and not is_direct 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'] + segment_uris = manifest_info.get('segment_uris') or manifest_info.get('direct_urls') or [] total_segments = len(segment_uris) + (1 if init_uri else 0) logger.info(f"Downloading from HiFi ({q_key}): {out_filename} " - f"({total_segments} segments)") + f"({total_segments} {'URL(s)' if is_direct else 'segments'})") downloaded = 0 speed_start = time.time() diff --git a/tests/downloads/test_hifi_pinning.py b/tests/downloads/test_hifi_pinning.py index de04b62d..d5fc9a8b 100644 --- a/tests/downloads/test_hifi_pinning.py +++ b/tests/downloads/test_hifi_pinning.py @@ -176,4 +176,79 @@ def test_instance_capability_probe_reports_manifest_without_uri_as_limited(): assert result['can_search'] is True assert result['can_download'] is False - assert result['download_error'] == 'No HLS manifest URI' + assert result['download_error'] == 'No playable manifest URL' + + +def test_instance_capability_probe_accepts_legacy_track_manifest(): + import base64 + import json + + class _Response: + def __init__(self, payload, *, ok=True, status_code=200): + self._payload = payload + self.ok = ok + self.status_code = status_code + + def json(self): + return self._payload + + class _Session: + def __init__(self): + self.calls = [] + + def get(self, url, **kwargs): + self.calls.append((url, kwargs)) + if url.endswith('/search/'): + return _Response({'items': []}) + if url.endswith('/trackManifests/'): + return _Response({'data': {'data': {'attributes': {}}}}) + if url.endswith('/track/'): + manifest = base64.b64encode(json.dumps({ + 'mimeType': 'audio/flac', + 'codecs': 'flac', + 'encryptionType': 'NONE', + 'urls': ['https://cdn.example/track.flac'], + }).encode()).decode() + return _Response({'data': {'manifest': manifest}}) + if url.endswith('/'): + return _Response({'version': 'test'}) + return _Response({}, ok=False, status_code=404) + + client = HiFiClient.__new__(HiFiClient) + client.session = _Session() + + result = client.check_instance_capabilities('https://hifi.example') + + assert result['can_search'] is True + assert result['can_download'] is True + assert result['download_probe'] == 'track' + assert any(url.endswith('/track/') for url, _ in client.session.calls) + + +def test_get_hls_manifest_falls_back_to_legacy_track_endpoint(): + import base64 + import json + + client = HiFiClient.__new__(HiFiClient) + calls = [] + + def _fake_api_get(path, params=None, timeout=15): + calls.append((path, params)) + if path == '/trackManifests/': + return None + manifest = base64.b64encode(json.dumps({ + 'mimeType': 'audio/flac', + 'codecs': 'flac', + 'encryptionType': 'NONE', + 'urls': ['https://cdn.example/track.flac'], + }).encode()).decode() + return {'data': {'manifest': manifest}} + + client._api_get = _fake_api_get + + result = client._get_hls_manifest(123, quality='lossless') + + assert result['direct_urls'] == ['https://cdn.example/track.flac'] + assert result['extension'] == 'flac' + assert calls[0][0] == '/trackManifests/' + assert calls[1][0] == '/track/'