Check HiFi download capability via manifests

Probe public HiFi instances with the same trackManifests endpoint used by real downloads instead of the legacy /track endpoint. This prevents compatible instances from being falsely labeled search-only in Settings.

Centralize HiFi instance capability checks in HiFiClient and reuse manifest URI parsing with the download path.

Tests cover manifest-based capability detection, no legacy /track probe, and limited instances without a manifest URI.

core/hifi_client.py

@@ -244,6 +244,75 @@ class HiFiClient(DownloadSourcePlugin):
             return data.get('version') or data.get('data', {}).get('version')
         return None
 
+    @staticmethod
+    def _extract_manifest_uri(data: Any) -> Optional[str]:
+        try:
+            inner = data.get('data', data) if isinstance(data, dict) else data
+            attrs = inner.get('data', {}).get('attributes', {})
+            return attrs.get('uri')
+        except (AttributeError, KeyError):
+            return None
+
+    def check_instance_capabilities(self, url: str, timeout: int = 5) -> Dict[str, Any]:
+        """Probe one public HiFi instance using the endpoints SoulSync needs."""
+        entry = {
+            'url': url,
+            'status': 'unknown',
+            'version': None,
+            'can_search': False,
+            'can_download': False,
+        }
+        try:
+            root = self.session.get(
+                f'{url}/',
+                timeout=timeout,
+                headers={'Accept': 'application/json'},
+            )
+            if not root.ok:
+                entry['status'] = f'error (HTTP {root.status_code})'
+                return entry
+
+            data = root.json()
+            entry['version'] = data.get('version') or data.get('data', {}).get('version')
+            entry['status'] = 'online'
+
+            search = self.session.get(
+                f'{url}/search/',
+                params={'s': 'test', 'limit': 1},
+                timeout=timeout,
+            )
+            entry['can_search'] = search.ok
+
+            manifest = self.session.get(
+                f'{url}/trackManifests/',
+                params={
+                    'id': '1550546',
+                    'formats': 'FLAC',
+                    'usage': 'DOWNLOAD',
+                    'manifestType': 'HLS',
+                    'adaptive': 'true',
+                    'uriScheme': 'HTTPS',
+                },
+                timeout=timeout,
+            )
+            entry['can_download'] = (
+                manifest.ok
+                and bool(self._extract_manifest_uri(manifest.json()))
+            )
+            if not manifest.ok:
+                entry['download_error'] = f'HTTP {manifest.status_code}'
+            elif not entry['can_download']:
+                entry['download_error'] = 'No HLS manifest URI'
+        except http_requests.exceptions.SSLError:
+            entry['status'] = 'ssl_error'
+        except http_requests.exceptions.ConnectTimeout:
+            entry['status'] = 'timeout'
+        except http_requests.exceptions.ConnectionError:
+            entry['status'] = 'offline'
+        except Exception as e:
+            entry['status'] = f'error ({type(e).__name__})'
+        return entry
+
     def search_tracks(self, title: str = None, artist: str = None,
                       album: str = None, limit: int = 20) -> List[Dict]:
         params = {'limit': limit}
@@ -435,12 +504,9 @@ class HiFiClient(DownloadSourcePlugin):
         if not data:
             return None
 
-        try:
+        uri = self._extract_manifest_uri(data)
-            inner = data.get('data', data) if isinstance(data, dict) else data
+        if uri is None:
-            attrs = inner.get('data', {}).get('attributes', {})
+            logger.warning("Failed to extract playlist URI from manifest response")
-            uri = attrs.get('uri')
-        except (AttributeError, KeyError) as e:
-            logger.warning(f"Failed to extract playlist URI from manifest response: {e}")
             return None
 
         if not uri:

tests/downloads/test_hifi_pinning.py

@@ -102,3 +102,78 @@ def test_cancel_download_marks_cancelled(hifi_client_with_engine):
     ok = _run_async(client.cancel_download('dl-1', None, remove=False))
     assert ok is True
     assert engine.get_record('hifi', 'dl-1')['state'] == 'Cancelled'
+
+
+def test_instance_capability_probe_uses_track_manifests_not_legacy_track():
+    class _Response:
+        def __init__(self, *, ok=True, status_code=200, payload=None):
+            self.ok = ok
+            self.status_code = status_code
+            self._payload = payload or {}
+
+        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(payload={'items': []})
+            if url.endswith('/trackManifests/'):
+                return _Response(payload={
+                    'data': {
+                        'data': {
+                            'attributes': {
+                                'uri': 'https://cdn.example/playlist.m3u8',
+                            },
+                        },
+                    },
+                })
+            if url.endswith('/'):
+                return _Response(payload={'version': 'test'})
+            return _Response(ok=False, status_code=404)
+
+    client = HiFiClient.__new__(HiFiClient)
+    client.session = _Session()
+
+    result = client.check_instance_capabilities('https://hifi.example')
+
+    called_urls = [url for url, _ in client.session.calls]
+    assert result['can_search'] is True
+    assert result['can_download'] is True
+    assert any(url.endswith('/trackManifests/') for url in called_urls)
+    assert not any(url.endswith('/track') for url in called_urls)
+
+
+def test_instance_capability_probe_reports_manifest_without_uri_as_limited():
+    class _Response:
+        ok = True
+        status_code = 200
+
+        def __init__(self, payload):
+            self._payload = payload
+
+        def json(self):
+            return self._payload
+
+    class _Session:
+        def get(self, url, **kwargs):
+            if url.endswith('/search/'):
+                return _Response({'items': []})
+            if url.endswith('/trackManifests/'):
+                return _Response({'data': {'data': {'attributes': {}}}})
+            if url.endswith('/'):
+                return _Response({'version': 'test'})
+            return _Response({'data': {'data': {'attributes': {}}}})
+
+    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 False
+    assert result['download_error'] == 'No HLS manifest URI'

web_server.py

@@ -19779,39 +19779,12 @@ def soundcloud_status():
 @app.route('/api/hifi/instances', methods=['GET'])
 def hifi_instances():
     """Check availability of all HiFi API instances."""
-    import requests as req
     try:
         hifi = download_orchestrator.client("hifi")
         instances = list(hifi._instances)
         results = []
         for url in instances:
-            entry = {'url': url, 'status': 'unknown', 'version': None, 'can_search': False, 'can_download': False}
+            results.append(hifi.check_instance_capabilities(url))
-            try:
-                # Check root for version
-                r = req.get(f'{url}/', timeout=5, headers={'Accept': 'application/json'})
-                if r.ok:
-                    data = r.json()
-                    entry['version'] = data.get('version')
-                    entry['status'] = 'online'
-                    # Check search
-                    sr = req.get(f'{url}/search', params={'s': 'test', 'limit': 1}, timeout=5)
-                    entry['can_search'] = sr.ok
-                    # Check track (download capability)
-                    tr = req.get(f'{url}/track', params={'id': '1550546', 'quality': 'LOSSLESS'}, timeout=5)
-                    entry['can_download'] = tr.ok
-                    if not tr.ok:
-                        entry['download_error'] = f'HTTP {tr.status_code}'
-                else:
-                    entry['status'] = f'error (HTTP {r.status_code})'
-            except req.exceptions.SSLError:
-                entry['status'] = 'ssl_error'
-            except req.exceptions.ConnectTimeout:
-                entry['status'] = 'timeout'
-            except req.exceptions.ConnectionError:
-                entry['status'] = 'offline'
-            except Exception as e:
-                entry['status'] = f'error ({type(e).__name__})'
-            results.append(entry)
         return jsonify({'instances': results, 'active': hifi._get_instance()})
     except Exception as e:
         return jsonify({'error': str(e)}), 500