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

@@ -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()

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/'