danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Merge pull request #393 from elmerohueso/hifi-fixes

Browse source 
fix HIFI downloads to get LOSSLESS and HI_RES again
This commit is contained in:
BoulderBadgeDad 2026-04-28 21:06:41 -07:00 committed by GitHub
commit e504099439
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 1865 additions and 791 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

504
core/hifi_client.py
View file

@ -12,20 +12,22 @@ Supports:
- Track search by title, artist, album
- Album lookup by ID
- Artist lookup by ID
- Direct FLAC download URLs from Tidal CDN
- Quality selection: HI_RES_LOSSLESS, LOSSLESS, HIGH, LOW
- HLS manifest-based downloads via /trackManifests/ endpoint
- Quality selection: HIRES_LOSSLESS, LOSSLESS, HIGH, LOW
- Multiple API instance failover
- FFmpeg demuxing for FLAC extraction from MP4 containers
"""
import os
import re
import json
import base64
import uuid
import time
import shutil
import subprocess
import threading
from typing import List, Optional, Dict, Any, Tuple
from pathlib import Path
from urllib.parse import urljoin
import requests as http_requests
@ -35,38 +37,44 @@ from core.soulseek_client import TrackResult, AlbumResult, DownloadStatus
logger = get_logger("hifi_client")
# Quality tiers matching Tidal's internal quality labels
HIFI_QUALITY_MAP = {
# HLS quality presets mapping to /trackManifests/ format parameters
HLS_QUALITY_MAP = {
'hires': {
'api_value': 'HI_RES_LOSSLESS',
'label': 'FLAC 24-bit/96kHz',
'formats': ['FLAC_HIRES'],
'manifest_type': 'HLS',
'extension': 'flac',
'label': 'FLAC 24-bit/96kHz',
'bitrate': 9216,
'codec': 'flac',
},
'lossless': {
'api_value': 'LOSSLESS',
'label': 'FLAC 16-bit/44.1kHz',
'formats': ['FLAC'],
'manifest_type': 'HLS',
'extension': 'flac',
'label': 'FLAC 16-bit/44.1kHz',
'bitrate': 1411,
'codec': 'flac',
},
'high': {
'api_value': 'HIGH',
'label': 'AAC 320kbps',
'formats': ['AACLC'],
'manifest_type': 'HLS',
'extension': 'm4a',
'label': 'AAC 320kbps',
'bitrate': 320,
'codec': 'aac',
},
'low': {
'api_value': 'LOW',
'label': 'AAC 96kbps',
'formats': ['HEAACV1'],
'manifest_type': 'HLS',
'extension': 'm4a',
'label': 'AAC 96kbps',
'bitrate': 96,
'codec': 'aac',
},
}
HLS_MAP_TAG_RE = re.compile(r'#EXT-X-MAP:.*URI="([^"]+)"')
# Default public hifi-api instances (ordered by preference)
DEFAULT_INSTANCES = [
'https://triton.squid.wtf',
@ -85,56 +93,68 @@ class HiFiClient:
"""
def __init__(self, download_path: str = None, base_url: str = None):
# Download path (use Soulseek path for consistency with post-processing)
if download_path is None:
download_path = config_manager.get('soulseek.download_path', './downloads')
self.download_path = Path(download_path)
self.download_path.mkdir(parents=True, exist_ok=True)
# API instance management
self._instances = list(DEFAULT_INSTANCES)
if base_url:
# User-provided instance gets top priority
self._instances.insert(0, base_url.rstrip('/'))
self._instances = []
self._instance_lock = threading.Lock()
self._load_instances_from_db()
self._current_instance = self._instances[0] if self._instances else None
self._instance_lock = threading.Lock()
# HTTP session with retry-friendly settings
self.session = http_requests.Session()
self.session.headers.update({
'User-Agent': 'SoulSync/1.0',
'Accept': 'application/json',
})
# Download tracking (mirrors TidalDownloadClient pattern)
self.active_downloads: Dict[str, Dict[str, Any]] = {}
self._download_lock = threading.Lock()
# Shutdown check callback
self.shutdown_check = None
# Rate limiting
self._last_api_call = 0
self._api_lock = threading.Lock()
self._min_interval = 0.5 # 500ms between calls
self._min_interval = 0.5
logger.info(f"HiFi client initialized (instance: {self._current_instance}, "
f"download path: {self.download_path})")
def set_shutdown_check(self, check_callable):
"""Set a callback function to check for system shutdown."""
self.shutdown_check = check_callable
# ===================== Instance Management =====================
def _load_instances_from_db(self):
try:
from database.music_database import get_database
db = get_database()
db.seed_hifi_instances(DEFAULT_INSTANCES)
rows = db.get_hifi_instances()
urls = [r['url'] for r in rows if r['enabled']]
if urls:
self._instances = urls
else:
self._instances = list(DEFAULT_INSTANCES)
except Exception as e:
logger.warning(f"Failed to load HiFi instances from DB, using defaults: {e}")
self._instances = list(DEFAULT_INSTANCES)
def reload_instances(self):
with self._instance_lock:
old_current = self._current_instance
self._load_instances_from_db()
self._current_instance = self._instances[0] if self._instances else None
if self._current_instance != old_current:
logger.info(f"HiFi instances reloaded, active: {self._current_instance}")
else:
logger.info("HiFi instances reloaded")
def _get_instance(self) -> Optional[str]:
"""Get the current active API instance URL."""
with self._instance_lock:
return self._current_instance
def _rotate_instance(self, failed_url: str):
"""Move a failed instance to the back of the list and switch to next."""
with self._instance_lock:
if failed_url in self._instances:
self._instances.remove(failed_url)
@ -146,7 +166,6 @@ class HiFiClient:
self._current_instance = None
def _rate_limit(self):
"""Enforce minimum interval between API calls."""
with self._api_lock:
now = time.time()
elapsed = now - self._last_api_call
@ -155,10 +174,6 @@ class HiFiClient:
self._last_api_call = time.time()
def _api_get(self, path: str, params: dict = None, timeout: int = 15) -> Optional[dict]:
"""
Make a GET request to the hifi-api, with instance failover.
Tries each instance up to once before giving up.
"""
tried = set()
while True:
@ -176,7 +191,6 @@ class HiFiClient:
response.raise_for_status()
data = response.json()
# Check for API-level errors
if isinstance(data, dict) and data.get('error'):
logger.warning(f"HiFi API error from {instance}: {data['error']}")
return None
@ -201,10 +215,7 @@ class HiFiClient:
logger.error(f"HiFi API unexpected error: {e}")
return None
# ===================== Availability =====================
def is_available(self) -> bool:
"""Check if the HiFi API is reachable."""
try:
data = self._api_get('/', timeout=5)
return data is not None
@ -212,11 +223,9 @@ class HiFiClient:
return False
def is_configured(self) -> bool:
"""Check if HiFi client is configured and ready (matches Soulseek interface)."""
return self._current_instance is not None
async def check_connection(self) -> bool:
"""Test if HiFi API is accessible (async, Soulseek-compatible)."""
try:
import asyncio
loop = asyncio.get_event_loop()
@ -226,28 +235,13 @@ class HiFiClient:
return False
def get_version(self) -> Optional[str]:
"""Get the API version of the current instance."""
data = self._api_get('/')
if data and isinstance(data, dict):
return data.get('version') or data.get('data', {}).get('version')
return None
# ===================== Search =====================
def search_tracks(self, title: str = None, artist: str = None,
album: str = None, limit: int = 20) -> List[Dict]:
"""
Search for tracks on Tidal via hifi-api.
Args:
title: Track title to search for
artist: Artist name to search for
album: Album name to search for
limit: Max results to return
Returns:
List of track dicts with id, title, artist, album, duration, etc.
"""
params = {'limit': limit}
if title:
params['s'] = title
@ -264,7 +258,6 @@ class HiFiClient:
if not data:
return []
# Handle response format: {data: {items: [...]}} or {data: [...]}
items = []
if isinstance(data, dict):
inner = data.get('data', data)
@ -285,15 +278,9 @@ class HiFiClient:
return results
def search_raw(self, query: str, limit: int = 20) -> List[Dict]:
"""
Generic search (free-text query). Maps to title search.
Returns raw dicts (not TrackResult).
"""
return self.search_tracks(title=query, limit=limit)
def _parse_track(self, item: dict) -> Dict:
"""Parse a track item from hifi-api response into a normalized dict."""
# Artist can be a dict with 'name' or a list of artists
artist_name = 'Unknown Artist'
artists_raw = item.get('artists', item.get('artist'))
if isinstance(artists_raw, list):
@ -309,7 +296,6 @@ class HiFiClient:
elif isinstance(artists_raw, str):
artist_name = artists_raw
# Album
album_raw = item.get('album', {})
album_name = ''
if isinstance(album_raw, dict):
@ -317,7 +303,6 @@ class HiFiClient:
elif isinstance(album_raw, str):
album_name = album_raw
# Duration
duration_s = item.get('duration', 0)
duration_ms = duration_s * 1000 if duration_s and duration_s < 100000 else duration_s
@ -333,10 +318,7 @@ class HiFiClient:
'quality': item.get('audioQuality', item.get('quality', '')),
}
# ===================== Track Info & Stream URL =====================
def get_track_info(self, track_id: int) -> Optional[Dict]:
"""Get detailed metadata for a specific track."""
data = self._api_get('/info/', params={'id': track_id})
if not data:
return None
@ -346,57 +328,7 @@ class HiFiClient:
return self._parse_track(inner)
return None
def get_stream_url(self, track_id: int, quality: str = 'lossless') -> Optional[Dict]:
"""
Get the direct download URL for a track.
Args:
track_id: Tidal track ID
quality: One of 'hires', 'lossless', 'high', 'low'
Returns:
Dict with 'url', 'mime_type', 'codec', 'quality' or None on failure.
"""
q_info = HIFI_QUALITY_MAP.get(quality, HIFI_QUALITY_MAP['lossless'])
api_quality = q_info['api_value']
data = self._api_get('/track/', params={'id': track_id, 'quality': api_quality})
if not data:
return None
# Extract manifest from response
inner = data.get('data', data) if isinstance(data, dict) else data
if not isinstance(inner, dict):
return None
manifest_b64 = inner.get('manifest')
if not manifest_b64:
logger.warning(f"No manifest in track response for {track_id}")
return None
try:
manifest = json.loads(base64.b64decode(manifest_b64))
except Exception as e:
logger.error(f"Failed to decode manifest for track {track_id}: {e}")
return None
urls = manifest.get('urls', [])
if not urls:
logger.warning(f"No URLs in manifest for track {track_id}")
return None
return {
'url': urls[0],
'mime_type': manifest.get('mimeType', ''),
'codec': manifest.get('codecs', ''),
'encryption': manifest.get('encryptionType', 'NONE'),
'quality': quality,
}
# ===================== Album & Artist =====================
def get_album(self, album_id: int, limit: int = 100) -> Optional[Dict]:
"""Get album metadata and track list."""
data = self._api_get('/album/', params={'id': album_id, 'limit': limit})
if not data:
return None
@ -405,7 +337,6 @@ class HiFiClient:
if not isinstance(inner, dict):
return None
# Parse tracks within album
tracks_raw = inner.get('items', inner.get('tracks', []))
tracks = []
for item in tracks_raw:
@ -426,7 +357,6 @@ class HiFiClient:
}
def get_artist(self, artist_id: int) -> Optional[Dict]:
"""Get artist info and top tracks."""
data = self._api_get('/artist/', params={'id': artist_id})
if not data:
return None
@ -434,14 +364,150 @@ class HiFiClient:
inner = data.get('data', data) if isinstance(data, dict) else data
return inner if isinstance(inner, dict) else None
# ===================== Soulseek-Compatible Search =====================
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
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']
params = [
('id', str(track_id)),
('formats', ','.join(formats)),
('usage', 'DOWNLOAD'),
('manifestType', 'HLS'),
('adaptive', 'true'),
('uriScheme', 'HTTPS'),
]
data = self._api_get('/trackManifests/', params=params, timeout=20)
if not data:
return None
try:
inner = data.get('data', data) if isinstance(data, dict) else data
attrs = inner.get('data', {}).get('attributes', {})
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:
logger.warning(f"No playlist URI in manifest for track {track_id}")
return None
try:
playlist_resp = self.session.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 = self.session.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"HiFi HLS manifest for track {track_id}: "
f"init segment + {len(segment_uris)} segments ({quality})")
else:
logger.info(f"HiFi HLS manifest for track {track_id}: "
f"{len(segment_uris)} segments ({quality})")
return {
'init_uri': init_uri,
'segment_uris': segment_uris,
'extension': q_info['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:
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 search(self, query: str, timeout: int = None,
progress_callback=None) -> Tuple[List[TrackResult], List[AlbumResult]]:
"""
Search with Soulseek-compatible return format (TrackResult, AlbumResult).
Matches the interface expected by DownloadOrchestrator.
"""
import asyncio
try:
@ -449,7 +515,7 @@ class HiFiClient:
tracks = await loop.run_in_executor(None, lambda: self.search_raw(query))
quality_key = config_manager.get('hifi_download.quality', 'lossless')
q_info = HIFI_QUALITY_MAP.get(quality_key, HIFI_QUALITY_MAP['lossless'])
q_info = HLS_QUALITY_MAP.get(quality_key, HLS_QUALITY_MAP['lossless'])
results = []
for t in tracks:
@ -466,7 +532,6 @@ class HiFiClient:
return ([], [])
def _to_track_result(self, track: Dict, quality_info: Dict) -> TrackResult:
"""Convert a hifi track dict to a TrackResult."""
display_name = f"{track['artist']} - {track['title']}"
filename = f"{track['id']}||{display_name}"
@ -486,13 +551,7 @@ class HiFiClient:
track_number=track.get('track_number'),
)
# ===================== Download =====================
async def download(self, username: str, filename: str, file_size: int = 0) -> Optional[str]:
"""
Download a track (async, Soulseek-compatible interface).
Filename format: "track_id||display_name"
"""
try:
if '||' not in filename:
logger.error(f"Invalid filename format: {filename}")
@ -537,7 +596,6 @@ class HiFiClient:
return None
def _download_worker(self, download_id: str, track_id: int, display_name: str):
"""Background download thread."""
try:
with self._download_lock:
if download_id in self.active_downloads:
@ -561,111 +619,168 @@ class HiFiClient:
self.active_downloads[download_id]['state'] = 'Errored'
def _download_sync(self, download_id: str, track_id: int, display_name: str) -> Optional[str]:
"""
Synchronous download with quality fallback chain.
Returns file path on success, None on failure.
"""
quality_key = config_manager.get('hifi_download.quality', 'lossless')
chain = ['hires', 'lossless', 'high', 'low']
start = chain.index(quality_key) if quality_key in chain else 1
allow_fallback = config_manager.get('hifi_download.allow_fallback', True)
chain = chain[start:] if allow_fallback else [quality_key]
MIN_AUDIO_SIZE = 100 * 1024 # 100KB
MIN_AUDIO_SIZE = 100 * 1024
for q_key in chain:
if self.shutdown_check and self.shutdown_check():
logger.info("Shutdown detected, aborting HiFi download")
return None
stream_info = self.get_stream_url(track_id, quality=q_key)
if not stream_info or not stream_info.get('url'):
logger.warning(f"No stream URL at quality {q_key}, trying next")
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
download_url = stream_info['url']
codec = stream_info.get('codec', '')
# Determine extension
if 'flac' in codec.lower():
extension = 'flac'
elif 'mp4a' in codec.lower() or 'aac' in codec.lower():
extension = 'm4a'
else:
extension = HIFI_QUALITY_MAP.get(q_key, {}).get('extension', 'flac')
# Build output path
extension = manifest_info['extension']
safe_name = re.sub(r'[<>:"/\\|?*]', '_', display_name)
out_filename = f"{safe_name}.{extension}"
out_path = self.download_path / out_filename
try:
logger.info(f"Downloading from HiFi ({q_key}): {out_filename}")
response = http_requests.get(download_url, stream=True, timeout=120)
response.raise_for_status()
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 HiFi ({q_key}): {out_filename} "
f"({total_segments} segments)")
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
chunk_size = 64 * 1024
speed_start = time.time()
last_speed_update = speed_start
segments_completed = 0
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['size'] = total_size
self.active_downloads[download_id]['size'] = 0
with open(out_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=chunk_size):
if not chunk:
continue
with intermediate_path.open('wb') as output_file:
if init_uri:
if self.shutdown_check and self.shutdown_check():
f.close()
out_path.unlink(missing_ok=True)
logger.info("Shutdown detected, aborting HiFi download")
intermediate_path.unlink(missing_ok=True)
return None
f.write(chunk)
downloaded += len(chunk)
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
if total_size > 0:
progress = (downloaded / total_size) * 100
else:
progress = 0
self._update_download_progress(download_id, downloaded,
segments_completed, total_segments, speed_start)
# Calculate speed every 0.5s
now = time.time()
elapsed_total = now - speed_start
speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0
time_remaining = int((total_size - downloaded) / speed) if speed > 0 and total_size > 0 else None
for segment_url in segment_uris:
if self.shutdown_check and self.shutdown_check():
logger.info("Shutdown detected, aborting HiFi download")
intermediate_path.unlink(missing_ok=True)
return None
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['transferred'] = downloaded
self.active_downloads[download_id]['progress'] = round(progress, 1)
self.active_downloads[download_id]['speed'] = speed
self.active_downloads[download_id]['time_remaining'] = time_remaining
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}")
out_path.unlink(missing_ok=True)
intermediate_path.unlink(missing_ok=True)
continue
# Validate file size
if downloaded < MIN_AUDIO_SIZE:
logger.warning(f"File too small at {q_key} ({downloaded} bytes), trying next")
out_path.unlink(missing_ok=True)
intermediate_path.unlink(missing_ok=True)
continue
logger.info(f"HiFi download complete ({q_key}): {out_path} "
f"({downloaded / (1024*1024):.1f} MB)")
return str(out_path)
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"HiFi 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
# ===================== Status / Cancel / Clear =====================
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 = self.session.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"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):
with self._download_lock:
if download_id not in self.active_downloads:
return
info = self.active_downloads[download_id]
info['transferred'] = downloaded
now = time.time()
elapsed_total = now - speed_start
speed = int(downloaded / elapsed_total) if elapsed_total > 0 else 0
info['speed'] = speed
if total_segments > 0:
progress = (segments_completed / total_segments) * 100
info['progress'] = round(min(progress, 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)
info['time_remaining'] = time_remaining
async def get_all_downloads(self) -> List[DownloadStatus]:
"""Get all active downloads (Soulseek-compatible)."""
statuses = []
with self._download_lock:
for _dl_id, info in self.active_downloads.items():
@ -684,7 +799,6 @@ class HiFiClient:
return statuses
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
"""Get status of a specific download."""
with self._download_lock:
info = self.active_downloads.get(download_id)
if not info:
@ -704,7 +818,6 @@ class HiFiClient:
async def cancel_download(self, download_id: str, username: str = None,
remove: bool = False) -> bool:
"""Cancel an active download."""
with self._download_lock:
if download_id not in self.active_downloads:
return False
@ -714,7 +827,6 @@ class HiFiClient:
return True
async def clear_all_completed_downloads(self) -> bool:
"""Clear all terminal downloads."""
with self._download_lock:
to_remove = [
did for did, info in self.active_downloads.items()

746
core/tidal_download_client.py
View file

File diff suppressed because it is too large Load diff

91
database/music_database.py
View file

@ -723,6 +723,17 @@ class MusicDatabase:
except Exception:
pass
# HiFi API instances table
cursor.execute("""
CREATE TABLE IF NOT EXISTS hifi_instances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT NOT NULL UNIQUE,
priority INTEGER NOT NULL DEFAULT 0,
enabled INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
conn.commit()
logger.info("Database initialized successfully")
@ -11594,6 +11605,86 @@ class MusicDatabase:
logger.error(f"Error getting issue counts: {e}")
return {'open': 0, 'in_progress': 0, 'resolved': 0, 'dismissed': 0, 'total': 0}
# ===================== HiFi Instances =====================
def get_hifi_instances(self) -> List[Dict[str, Any]]:
"""Get all enabled HiFi instances ordered by priority."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("SELECT url, priority, enabled FROM hifi_instances WHERE enabled = 1 ORDER BY priority ASC, id ASC")
return [dict(row) for row in cursor.fetchall()]
def get_all_hifi_instances(self) -> List[Dict[str, Any]]:
"""Get all HiFi instances (including disabled) ordered by priority."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("SELECT url, priority, enabled FROM hifi_instances ORDER BY priority ASC, id ASC")
return [dict(row) for row in cursor.fetchall()]
def add_hifi_instance(self, url: str, priority: int = 0) -> bool:
"""Add a new HiFi instance."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT OR IGNORE INTO hifi_instances (url, priority, enabled) VALUES (?, ?, 1)",
(url, priority)
)
conn.commit()
return cursor.rowcount > 0
def remove_hifi_instance(self, url: str) -> bool:
"""Remove a HiFi instance."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("DELETE FROM hifi_instances WHERE url = ?", (url,))
conn.commit()
return cursor.rowcount > 0
def toggle_hifi_instance(self, url: str, enabled: bool) -> bool:
"""Enable or disable a HiFi instance."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("UPDATE hifi_instances SET enabled = ? WHERE url = ?", (1 if enabled else 0, url))
conn.commit()
return cursor.rowcount > 0
def reorder_hifi_instances(self, urls: List[str]) -> bool:
"""Update priorities based on the given URL order.
Returns False if any URL does not exist in the database.
"""
if not urls:
return True
conn = self._get_connection()
cursor = conn.cursor()
placeholders = ",".join("?" for _ in urls)
cursor.execute(
f"SELECT url FROM hifi_instances WHERE url IN ({placeholders})",
urls
)
existing = {row["url"] for row in cursor.fetchall()}
missing = [u for u in urls if u not in existing]
if missing:
return False
for i, url in enumerate(urls):
cursor.execute("UPDATE hifi_instances SET priority = ? WHERE url = ?", (i, url))
conn.commit()
return True
def seed_hifi_instances(self, default_urls: List[str]) -> None:
"""Insert default instances if the table is empty."""
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) as cnt FROM hifi_instances")
count = cursor.fetchone()['cnt']
if count == 0:
for i, url in enumerate(default_urls):
cursor.execute(
"INSERT OR IGNORE INTO hifi_instances (url, priority, enabled) VALUES (?, ?, 1)",
(url, i)
)
conn.commit()
logger.info(f"Seeded {len(default_urls)} default HiFi instances")
# Thread-safe singleton pattern for database access
_database_instances: Dict[int, MusicDatabase] = {} # Thread ID -> Database instance
_database_lock = threading.Lock()

368
tests/test_hifi_instance_methods.py Normal file
View file

@ -0,0 +1,368 @@
"""Tests for the HiFi instance CRUD helpers on ``MusicDatabase``:
- ``get_hifi_instances()`` — returns enabled instances ordered by priority
- ``get_all_hifi_instances()`` — returns all instances (enabled + disabled)
- ``add_hifi_instance(url, priority)`` — inserts a new instance
- ``remove_hifi_instance(url)`` — deletes an instance by URL
- ``toggle_hifi_instance(url, enabled)`` — enables/disables an instance
- ``reorder_hifi_instances(urls)`` — updates priority ordering
- ``seed_hifi_instances(default_urls)`` — seeds defaults when table is empty
These are isolated DB-method tests so the SQL itself is verified
without spinning up Flask or any HiFi client.
"""
import sqlite3
import sys
import types
import pytest
# ── stubs (same shape used elsewhere in the test suite) ───────────────────
if "spotipy" not in sys.modules:
spotipy = types.ModuleType("spotipy")
spotipy.Spotify = object
oauth2 = types.ModuleType("spotipy.oauth2")
oauth2.SpotifyOAuth = object
oauth2.SpotifyClientCredentials = object
spotipy.oauth2 = oauth2
sys.modules["spotipy"] = spotipy
sys.modules["spotipy.oauth2"] = oauth2
if "config.settings" not in sys.modules:
config_pkg = types.ModuleType("config")
settings_mod = types.ModuleType("config.settings")
class _DummyConfigManager:
def get(self, key, default=None):
return default
def get_active_media_server(self):
return "primary"
settings_mod.config_manager = _DummyConfigManager()
config_pkg.settings = settings_mod
sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod
from database.music_database import MusicDatabase # noqa: E402
# ── helpers ───────────────────────────────────────────────────────────────
class _InMemoryDB(MusicDatabase):
"""MusicDatabase that uses an in-memory sqlite that survives across
`_get_connection()` calls."""
def __init__(self):
self._conn = sqlite3.connect(":memory:")
self._conn.row_factory = sqlite3.Row
self._conn.execute("""
CREATE TABLE hifi_instances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
url TEXT NOT NULL UNIQUE,
priority INTEGER NOT NULL DEFAULT 0,
enabled INTEGER NOT NULL DEFAULT 1,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
""")
self._conn.commit()
def _get_connection(self):
return _NonClosingConn(self._conn)
class _NonClosingConn:
"""Wraps the shared sqlite connection so `with db._get_connection()
as conn:` doesn't close the underlying handle between calls."""
def __init__(self, real):
self._real = real
def cursor(self):
return self._real.cursor()
def commit(self):
return self._real.commit()
def close(self):
pass
def __enter__(self):
return self
def __exit__(self, *args):
pass
@pytest.fixture
def db():
return _InMemoryDB()
def _seed(db, *, instances=()):
"""Seed hifi_instances rows. Each tuple: (url, priority, enabled)."""
cur = db._conn.cursor()
for url, priority, enabled in instances:
cur.execute(
"INSERT INTO hifi_instances (url, priority, enabled) VALUES (?, ?, ?)",
(url, priority, enabled),
)
db._conn.commit()
# ── get_hifi_instances ────────────────────────────────────────────────────
def test_get_hifi_instances_returns_enabled_ordered_by_priority(db):
_seed(db, instances=[
("http://b.com", 10, 1),
("http://a.com", 5, 1),
("http://c.com", 1, 1),
])
rows = db.get_hifi_instances()
assert [r["url"] for r in rows] == ["http://c.com", "http://a.com", "http://b.com"]
assert [r["priority"] for r in rows] == [1, 5, 10]
def test_get_hifi_instances_excludes_disabled(db):
_seed(db, instances=[
("http://a.com", 0, 1),
("http://b.com", 1, 0),
("http://c.com", 2, 1),
])
rows = db.get_hifi_instances()
assert {r["url"] for r in rows} == {"http://a.com", "http://c.com"}
def test_get_hifi_instances_returns_empty_when_no_rows(db):
assert db.get_hifi_instances() == []
def test_get_hifi_instances_tiebreaks_on_id(db):
"""Same priority → ordered by insertion order (autoincrement id)."""
_seed(db, instances=[
("http://first.com", 0, 1),
("http://second.com", 0, 1),
("http://third.com", 0, 1),
])
rows = db.get_hifi_instances()
assert [r["url"] for r in rows] == ["http://first.com", "http://second.com", "http://third.com"]
# ── get_all_hifi_instances ────────────────────────────────────────────────
def test_get_all_hifi_instances_returns_all_including_disabled(db):
_seed(db, instances=[
("http://a.com", 0, 1),
("http://b.com", 1, 0),
])
rows = db.get_all_hifi_instances()
assert {r["url"] for r in rows} == {"http://a.com", "http://b.com"}
def test_get_all_hifi_instances_ordered_by_priority(db):
_seed(db, instances=[
("http://c.com", 20, 0),
("http://a.com", 0, 1),
("http://b.com", 10, 1),
])
rows = db.get_all_hifi_instances()
assert [r["url"] for r in rows] == ["http://a.com", "http://b.com", "http://c.com"]
def test_get_all_hifi_instances_returns_empty_when_no_rows(db):
assert db.get_all_hifi_instances() == []
# ── add_hifi_instance ─────────────────────────────────────────────────────
def test_add_hifi_instance_returns_true_on_insert(db):
assert db.add_hifi_instance("http://new.com", priority=3) is True
rows = db.get_all_hifi_instances()
assert len(rows) == 1
assert rows[0]["url"] == "http://new.com"
assert rows[0]["priority"] == 3
assert rows[0]["enabled"] == 1
def test_add_hifi_instance_returns_false_on_duplicate(db):
_seed(db, instances=[("http://dup.com", 0, 1)])
# INSERT OR IGNORE — should not raise, but return False (rowcount == 0)
assert db.add_hifi_instance("http://dup.com", priority=5) is False
rows = db.get_all_hifi_instances()
assert len(rows) == 1
def test_add_hifi_instance_default_priority(db):
db.add_hifi_instance("http://x.com")
row = db.get_all_hifi_instances()[0]
assert row["priority"] == 0
# ── remove_hifi_instance ──────────────────────────────────────────────────
def test_remove_hifi_instance_returns_true_on_delete(db):
_seed(db, instances=[("http://go.com", 0, 1)])
assert db.remove_hifi_instance("http://go.com") is True
assert db.get_all_hifi_instances() == []
def test_remove_hifi_instance_returns_false_when_not_found(db):
assert db.remove_hifi_instance("http://missing.com") is False
def test_remove_hifi_instance_only_removes_matching_url(db):
_seed(db, instances=[
("http://keep.com", 0, 1),
("http://delete.com", 1, 1),
])
db.remove_hifi_instance("http://delete.com")
rows = db.get_all_hifi_instances()
assert len(rows) == 1
assert rows[0]["url"] == "http://keep.com"
# ── toggle_hifi_instance ──────────────────────────────────────────────────
def test_toggle_hifi_instance_disable(db):
_seed(db, instances=[("http://x.com", 0, 1)])
assert db.toggle_hifi_instance("http://x.com", enabled=False) is True
row = db.get_all_hifi_instances()[0]
assert row["enabled"] == 0
def test_toggle_hifi_instance_enable(db):
_seed(db, instances=[("http://x.com", 0, 0)])
assert db.toggle_hifi_instance("http://x.com", enabled=True) is True
row = db.get_all_hifi_instances()[0]
assert row["enabled"] == 1
def test_toggle_hifi_instance_returns_false_when_not_found(db):
assert db.toggle_hifi_instance("http://missing.com", enabled=True) is False
def test_toggle_hifi_instance_noop_when_already_set(db):
"""Toggling to the same value should still return True (row matched)."""
_seed(db, instances=[("http://x.com", 0, 1)])
# SQLite rowcount for UPDATE is 1 even if value didn't change
assert db.toggle_hifi_instance("http://x.com", enabled=True) is True
# ── reorder_hifi_instances ────────────────────────────────────────────────
def test_reorder_hifi_instances_updates_priorities(db):
_seed(db, instances=[
("http://a.com", 0, 1),
("http://b.com", 1, 1),
("http://c.com", 2, 1),
])
db.reorder_hifi_instances(["http://c.com", "http://a.com", "http://b.com"])
rows = db.get_all_hifi_instances()
by_url = {r["url"]: r["priority"] for r in rows}
assert by_url == {"http://c.com": 0, "http://a.com": 1, "http://b.com": 2}
def test_reorder_hifi_instances_returns_true_on_empty_list(db):
assert db.reorder_hifi_instances([]) is True
def test_reorder_hifi_instances_returns_false_with_unknown_urls(db):
"""Reorder should fail when any URL doesn't exist."""
_seed(db, instances=[("http://a.com", 0, 1)])
assert db.reorder_hifi_instances(["http://a.com", "http://phantom.com"]) is False
# ── seed_hifi_instances ───────────────────────────────────────────────────
def test_seed_hifi_instances_inserts_when_empty(db):
db.seed_hifi_instances(["http://a.com", "http://b.com"])
rows = db.get_all_hifi_instances()
assert len(rows) == 2
by_url = {r["url"]: r["priority"] for r in rows}
assert by_url == {"http://a.com": 0, "http://b.com": 1}
def test_seed_hifi_instances_does_nothing_when_table_has_rows(db):
_seed(db, instances=[("http://existing.com", 0, 1)])
db.seed_hifi_instances(["http://new.com"])
rows = db.get_all_hifi_instances()
assert len(rows) == 1
assert rows[0]["url"] == "http://existing.com"
def test_seed_hifi_instances_does_not_duplicate_on_reseed(db):
db.seed_hifi_instances(["http://a.com"])
db.seed_hifi_instances(["http://a.com"])
rows = db.get_all_hifi_instances()
assert len(rows) == 1
# ── error propagation ────────────────────────────────────────────────────
# These methods now let DB errors bubble up so the route layer turns them
# into a 500 — the user sees a real failure instead of a phantom empty state.
def _db_without_hifi_table():
"""Returns a MusicDatabase with NO hifi_instances table."""
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
class _NoTableDB(MusicDatabase):
def __init__(self):
self._conn = conn
def _get_connection(self):
return _NonClosingConn(self._conn)
return _NoTableDB()
def test_get_hifi_instances_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.get_hifi_instances()
def test_get_all_hifi_instances_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.get_all_hifi_instances()
def test_add_hifi_instance_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.add_hifi_instance("http://x.com")
def test_remove_hifi_instance_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.remove_hifi_instance("http://x.com")
def test_toggle_hifi_instance_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.toggle_hifi_instance("http://x.com", enabled=True)
def test_reorder_hifi_instances_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.reorder_hifi_instances(["http://x.com"])
def test_seed_hifi_instances_propagates_db_errors():
db = _db_without_hifi_table()
with pytest.raises(sqlite3.OperationalError):
db.seed_hifi_instances(["http://x.com"])

384
tests/test_hls_parsing.py Normal file
View file

@ -0,0 +1,384 @@
"""Tests for HLS-related methods on ``HiFiClient``:
- ``_parse_hls_playlist(text, playlist_url)`` — parses m3u8 playlists
- ``_demux_flac(input_path, output_path)`` — ffmpeg demuxing error paths
The parsing logic is identical in ``tidal_download_client.py``; these tests
cover the shared behavior via ``hifi_client.HiFiClient``.
"""
import shutil
import subprocess
import sys
import types
from pathlib import Path
from unittest.mock import patch
import pytest
# ── stubs for module-level imports ────────────────────────────────────────
if "spotipy" not in sys.modules:
spotipy = types.ModuleType("spotipy")
spotipy.Spotify = object
oauth2 = types.ModuleType("spotipy.oauth2")
oauth2.SpotifyOAuth = object
oauth2.SpotifyClientCredentials = object
spotipy.oauth2 = oauth2
sys.modules["spotipy"] = spotipy
sys.modules["spotipy.oauth2"] = oauth2
if "config.settings" not in sys.modules:
config_pkg = types.ModuleType("config")
settings_mod = types.ModuleType("config.settings")
class _DummyConfigManager:
def get(self, key, default=None):
return default
def get_active_media_server(self):
return "primary"
settings_mod.config_manager = _DummyConfigManager()
config_pkg.settings = settings_mod
sys.modules["config"] = config_pkg
sys.modules["config.settings"] = settings_mod
from core.hifi_client import HiFiClient # noqa: E402
# ── fixture ───────────────────────────────────────────────────────────────
@pytest.fixture
def client(tmp_path):
"""HiFiClient with a temp download dir and no DB dependency."""
return HiFiClient(download_path=str(tmp_path / "downloads"))
# ── _parse_hls_playlist: master playlist ─────────────────────────────────
MASTER_PLAYLIST = """\
#EXTM3U
#EXT-X-VERSION:3
#EXT-X-STREAM-INF:BANDWIDTH=256000,CODECS="mp4a.40.2"
stream/low.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=512000,CODECS="mp4a.40.5"
stream/high.m3u8
"""
def test_parse_master_playlist_returns_variant_uri(client):
init, segments = client._parse_hls_playlist(
MASTER_PLAYLIST, "https://cdn.example.com/master.m3u8"
)
assert init is None
assert segments == ["https://cdn.example.com/stream/low.m3u8"]
def test_parse_master_playlist_picks_first_variant(client):
"""Master playlists should return only the first variant URI."""
init, segments = client._parse_hls_playlist(
MASTER_PLAYLIST, "https://cdn.example.com/master.m3u8"
)
assert len(segments) == 1
def test_parse_master_playlist_resolves_relative_uri(client):
init, segments = client._parse_hls_playlist(
MASTER_PLAYLIST, "https://cdn.example.com/playlists/master.m3u8"
)
assert segments[0] == "https://cdn.example.com/playlists/stream/low.m3u8"
# ── _parse_hls_playlist: variant playlist with init segment ──────────────
VARIANT_WITH_INIT = """\
#EXTM3U
#EXT-X-VERSION:6
#EXT-X-TARGETDURATION:4
#EXT-X-MAP:URI="init.mp4"
#EXTINF:3.840000,
seg001.m4s
#EXTINF:3.840000,
seg002.m4s
#EXTINF:2.560000,
seg003.m4s
"""
def test_parse_variant_with_init_returns_init_and_segments(client):
init, segments = client._parse_hls_playlist(
VARIANT_WITH_INIT, "https://cdn.example.com/variant.m3u8"
)
assert init == "https://cdn.example.com/init.mp4"
assert segments == [
"https://cdn.example.com/seg001.m4s",
"https://cdn.example.com/seg002.m4s",
"https://cdn.example.com/seg003.m4s",
]
def test_parse_variant_with_init_resolves_relative_uris(client):
init, segments = client._parse_hls_playlist(
VARIANT_WITH_INIT, "https://cdn.example.com/audio/variant.m3u8"
)
assert init == "https://cdn.example.com/audio/init.mp4"
assert segments[0] == "https://cdn.example.com/audio/seg001.m4s"
def test_parse_variant_with_absolute_init_uri(client):
playlist = """\
#EXTM3U
#EXT-X-MAP:URI="https://other.cdn/init.mp4"
#EXTINF:3.0,
https://cdn.example.com/seg001.m4s
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/variant.m3u8"
)
assert init == "https://other.cdn/init.mp4"
assert segments == ["https://cdn.example.com/seg001.m4s"]
# ── _parse_hls_playlist: variant playlist without init segment ───────────
VARIANT_NO_INIT = """\
#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXTINF:9.984000,
seg001.ts
#EXTINF:9.984000,
seg002.ts
#EXT-X-ENDLIST
"""
def test_parse_variant_without_init_returns_none_init(client):
init, segments = client._parse_hls_playlist(
VARIANT_NO_INIT, "https://cdn.example.com/variant.m3u8"
)
assert init is None
assert segments == [
"https://cdn.example.com/seg001.ts",
"https://cdn.example.com/seg002.ts",
]
# ── _parse_hls_playlist: error cases ─────────────────────────────────────
def test_parse_empty_playlist_raises_value_error(client):
with pytest.raises(ValueError, match="No segment URIs"):
client._parse_hls_playlist("#EXTM3U", "https://cdn.example.com/x.m3u8")
def test_parse_only_tags_raises_value_error(client):
"""Playlist with only header and MAP tag but no segment URIs."""
playlist = """\
#EXTM3U
#EXT-X-VERSION:6
#EXT-X-MAP:URI="init.mp4"
"""
with pytest.raises(ValueError, match="No segment URIs"):
client._parse_hls_playlist(playlist, "https://cdn.example.com/x.m3u8")
def test_parse_only_extm3u_raises_value_error(client):
with pytest.raises(ValueError, match="No segment URIs"):
client._parse_hls_playlist("#EXTM3U\n", "https://cdn.example.com/x.m3u8")
# ── _parse_hls_playlist: edge cases ──────────────────────────────────────
def test_parse_skips_blank_lines(client):
playlist = """\
#EXTM3U
#EXT-X-MAP:URI="init.mp4"
#EXTINF:3.0,
seg001.m4s
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/x.m3u8"
)
assert init == "https://cdn.example.com/init.mp4"
assert segments == ["https://cdn.example.com/seg001.m4s"]
def test_parse_skips_unknown_tags(client):
"""Tags like #EXT-X-VERSION, #EXT-X-TARGETDURATION should be ignored."""
playlist = """\
#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-PLAYLIST-TYPE:VOD
#EXTINF:5.0,
seg001.ts
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/x.m3u8"
)
assert init is None
assert segments == ["https://cdn.example.com/seg001.ts"]
def test_parse_captures_last_map_tag(client):
"""If multiple EXT-X-MAP tags appear, the last one wins (overwrites init_uri)."""
playlist = """\
#EXTM3U
#EXT-X-MAP:URI="init-first.mp4"
#EXTINF:3.0,
seg001.m4s
#EXT-X-MAP:URI="init-second.mp4"
#EXTINF:3.0,
seg002.m4s
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/x.m3u8"
)
assert init == "https://cdn.example.com/init-second.mp4"
assert len(segments) == 2
def test_parse_master_breaks_on_first_variant(client):
"""Parser should stop after finding the first variant URI."""
playlist = """\
#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=256000
variant-low.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=512000
variant-high.m3u8
#EXTINF:3.0,
should-not-appear.ts
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/master.m3u8"
)
assert init is None
assert segments == ["https://cdn.example.com/variant-low.m3u8"]
def test_parse_master_skips_comment_after_stream_inf(client):
"""The line immediately after #EXT-X-STREAM-INF must be a non-comment URI."""
playlist = """\
#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=256000
# some comment
variant.m3u8
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/master.m3u8"
)
assert segments == ["https://cdn.example.com/variant.m3u8"]
def test_parse_handles_mixed_absolute_and_relative_uris(client):
playlist = """\
#EXTM3U
#EXTINF:3.0,
https://cdn-a.example.com/seg001.m4s
#EXTINF:3.0,
seg002.m4s
#EXTINF:3.0,
https://cdn-b.example.com/seg003.m4s
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/variant.m3u8"
)
assert segments == [
"https://cdn-a.example.com/seg001.m4s",
"https://cdn.example.com/seg002.m4s",
"https://cdn-b.example.com/seg003.m4s",
]
def test_parse_single_segment(client):
playlist = """\
#EXTM3U
#EXTINF:3.0,
only-segment.m4s
"""
init, segments = client._parse_hls_playlist(
playlist, "https://cdn.example.com/x.m3u8"
)
assert init is None
assert segments == ["https://cdn.example.com/only-segment.m4s"]
# ── _demux_flac: error paths ─────────────────────────────────────────────
def test_demux_flac_raises_when_ffmpeg_not_found(client, tmp_path, monkeypatch):
"""When ffmpeg is nowhere on PATH and not in tools/, RuntimeError is raised."""
monkeypatch.setattr("shutil.which", lambda _: None)
inp = tmp_path / "input.mp4"
inp.touch()
out = tmp_path / "output.flac"
with patch.object(Path, "exists", return_value=False):
with pytest.raises(RuntimeError, match="ffmpeg is required"):
client._demux_flac(inp, out)
def test_demux_flac_raises_on_ffmpeg_failure(client, tmp_path, monkeypatch):
"""When ffmpeg exits non-zero, RuntimeError includes stderr."""
monkeypatch.setattr("shutil.which", lambda _: "/usr/bin/ffmpeg")
inp = tmp_path / "input.mp4"
inp.touch()
out = tmp_path / "output.flac"
fake_result = subprocess.CalledProcessError(
returncode=1,
cmd=["ffmpeg"],
stderr="Invalid data found when processing input",
)
with patch("subprocess.run", side_effect=fake_result):
with pytest.raises(RuntimeError, match="ffmpeg failed"):
client._demux_flac(inp, out)
def test_demux_flac_uses_tools_dir_fallback(client, tmp_path, monkeypatch):
"""When shutil.which fails but tools/ffmpeg exists, it should be used."""
monkeypatch.setattr("shutil.which", lambda _: None)
inp = tmp_path / "input.mp4"
inp.touch()
out = tmp_path / "output.flac"
tools_dir = Path(__file__).parent.parent / "tools"
original_exists = Path.exists
original_which = shutil.which
def fake_exists(self):
if str(self) == str(tools_dir / "ffmpeg"):
return True
return original_exists(self)
def fake_which(name):
if name == "ffmpeg":
return None
return original_which(name)
monkeypatch.setattr(shutil, "which", fake_which)
monkeypatch.setattr(Path, "exists", fake_exists)
fake_result = subprocess.CalledProcessError(
returncode=1, cmd=["ffmpeg"], stderr="fail"
)
with patch("subprocess.run", side_effect=fake_result) as mock_run:
with pytest.raises(RuntimeError, match="ffmpeg failed"):
client._demux_flac(inp, out)
call_args = mock_run.call_args[0][0]
assert call_args[0] == str(tools_dir / "ffmpeg")

162
tests/test_tidal_stream_tier_verification.py
View file

@ -1,162 +0,0 @@
"""Tests for `_verify_stream_tier` — the guard that rejects silent Tidal
quality downgrades so the fallback chain (or "HiRes only" with fallback
disabled) behaves the way users configure it to.
Without this check, a user with "HiRes only, no quality fallback" who
asks Tidal for a track that's only available in AAC 320kbps would
receive the 320kbps stream silently — Tidal never raises, it just
serves the highest tier available — and the downloader would accept
the m4a file and report success. Reported by Netti93.
Tiers ranked worst-to-best:
LOW < HIGH < LOSSLESS < HI_RES < HI_RES_LOSSLESS
Accepting matches and upgrades, rejecting downgrades, rejecting
unrecognized values.
Note on the fake Quality values: tidalapi's real Quality enum has
VALUES that differ from the member names (e.g., `low_320k.value ==
'HIGH'`, `high_lossless.value == 'LOSSLESS'`). The stub mirrors real
values so the tests catch case-sensitivity regressions.
"""
import sys
import types
if 'tidalapi' not in sys.modules:
_fake = types.ModuleType('tidalapi')
class _FakeQuality:
low_96k = 'LOW'
low_320k = 'HIGH'
high_lossless = 'LOSSLESS'
hi_res = 'HI_RES'
hi_res_lossless = 'HI_RES_LOSSLESS'
_fake.Quality = _FakeQuality
_fake.media = types.SimpleNamespace(Track=object)
sys.modules['tidalapi'] = _fake
from core.tidal_download_client import QUALITY_MAP, _verify_stream_tier # noqa: E402
class _FakeStream:
"""Minimal stand-in for tidalapi.media.Stream."""
def __init__(self, audio_quality=None):
if audio_quality is not None:
self.audio_quality = audio_quality
# ---------------------------------------------------------------------------
# Match — served quality is exactly what was requested
# ---------------------------------------------------------------------------
def test_served_quality_matches_request():
stream = _FakeStream(audio_quality='HI_RES_LOSSLESS')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['hires'], 'hires')
assert ok is True
assert reason is None
# ---------------------------------------------------------------------------
# Upgrades — Tidal serving a higher tier than requested is accepted
# ---------------------------------------------------------------------------
def test_lossless_request_upgraded_to_hires_is_accepted():
"""If Tidal serves HI_RES_LOSSLESS on a LOSSLESS-tier request (rare
but possible on tracks flagged as such in Tidal's catalog), we take
the upgrade — rejecting a better-than-asked tier would be user-
hostile."""
stream = _FakeStream(audio_quality='HI_RES_LOSSLESS')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['lossless'], 'lossless')
assert ok is True
assert reason is None
def test_lossless_request_upgraded_to_mqa_hires_is_accepted():
stream = _FakeStream(audio_quality='HI_RES')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['lossless'], 'lossless')
assert ok is True
assert reason is None
def test_low_request_upgraded_to_any_higher_tier_is_accepted():
stream = _FakeStream(audio_quality='LOSSLESS')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['low'], 'low')
assert ok is True
assert reason is None
# ---------------------------------------------------------------------------
# Downgrades — the reported bug
# ---------------------------------------------------------------------------
def test_hires_downgraded_to_aac_is_rejected():
"""The exact case Netti93 reported: asked HiRes, Tidal served
AAC 320kbps (`'HIGH'` in Tidal's API vocabulary)."""
stream = _FakeStream(audio_quality='HIGH')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['hires'], 'hires')
assert ok is False
assert 'HIGH' in reason
assert 'HI_RES_LOSSLESS' in reason
def test_hires_lossless_downgraded_to_mqa_hires_is_rejected():
"""User explicitly asked for HI_RES_LOSSLESS (true lossless HiRes).
Getting MQA-encoded HI_RES is a downgrade even though both are
"HiRes tier" marketing-wise — MQA is lossy."""
stream = _FakeStream(audio_quality='HI_RES')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['hires'], 'hires')
assert ok is False
assert 'HI_RES_LOSSLESS' in reason
def test_lossless_downgraded_to_aac_is_rejected():
stream = _FakeStream(audio_quality='HIGH')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['lossless'], 'lossless')
assert ok is False
assert 'LOSSLESS' in reason
# ---------------------------------------------------------------------------
# Unknown quality strings — reject conservatively
# ---------------------------------------------------------------------------
def test_unknown_served_quality_is_rejected():
"""If Tidal introduces a new tier we haven't mapped yet, we can't
prove it's acceptable — reject rather than silently pass through,
so the next fallback tier gets a chance and the final diagnostic
log names the unknown value."""
stream = _FakeStream(audio_quality='SPATIAL_360_DREAM_TIER')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['hires'], 'hires')
assert ok is False
assert 'SPATIAL_360_DREAM_TIER' in reason
assert 'unrecognized' in reason.lower() or 'can\'t verify' in reason.lower()
# ---------------------------------------------------------------------------
# Defensive — missing attributes must not spuriously fail downloads
# ---------------------------------------------------------------------------
def test_stream_without_audio_quality_attr_is_accepted():
"""Older tidalapi versions may not expose audio_quality — treat as
"can't verify" and let pre-existing codec / file-size guards decide.
Better to miss a downgrade than break every Tidal download after a
library upgrade."""
stream = _FakeStream()
assert not hasattr(stream, 'audio_quality')
ok, reason = _verify_stream_tier(stream, QUALITY_MAP['hires'], 'hires')
assert ok is True
assert reason is None
def test_quality_info_without_tidal_quality_is_accepted():
"""If QUALITY_MAP somehow lacks 'tidal_quality' (tidalapi failed to
import at module load), don't spuriously reject streams."""
stream = _FakeStream(audio_quality='HI_RES_LOSSLESS')
ok, reason = _verify_stream_tier(stream, {'label': 'x'}, 'hires')
assert ok is True
assert reason is None

111
web_server.py
View file

@ -5071,7 +5071,7 @@ def get_status():
(download_mode == 'hybrid' and 'soulseek' in hybrid_order))
# Serverless sources (YouTube, HiFi, Qobuz) are always available
serverless_sources = ('youtube', 'hifi', 'qobuz')
serverless_sources = ('youtube', 'hifi', 'qobuz', 'tidal', 'deezer_dl')
is_serverless = (download_mode in serverless_sources or
(download_mode == 'hybrid' and
hybrid_order and any(s in serverless_sources for s in hybrid_order)))
@ -23425,6 +23425,115 @@ def hifi_instances():
return jsonify({'error': str(e)}), 500
@app.route('/api/hifi/instances', methods=['POST'])
@admin_only
def hifi_add_instance():
"""Add a new HiFi API instance."""
try:
data = request.get_json() or {}
url = data.get('url', '').strip().rstrip('/')
if not url:
return jsonify({'success': False, 'error': 'URL is required'}), 400
if not url.startswith(('http://', 'https://')):
return jsonify({'success': False, 'error': 'URL must start with http:// or https://'}), 400
from database.music_database import get_database
db = get_database()
# Get current count to assign next priority
existing = db.get_all_hifi_instances()
priority = len(existing)
added = db.add_hifi_instance(url, priority)
if not added:
return jsonify({'success': False, 'error': 'Instance already exists'}), 400
# Reload the client
if soulseek_client and hasattr(soulseek_client, 'hifi') and soulseek_client.hifi:
soulseek_client.hifi.reload_instances()
return jsonify({'success': True, 'url': url})
except Exception as e:
logger.error(f"Error adding HiFi instance: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/hifi/instances', methods=['DELETE'])
@admin_only
def hifi_remove_instance():
"""Remove a HiFi API instance."""
try:
url = (request.args.get('url') or '').strip().rstrip('/')
if not url:
return jsonify({'success': False, 'error': 'URL is required'}), 400
from database.music_database import get_database
db = get_database()
removed = db.remove_hifi_instance(url)
if not removed:
return jsonify({'success': False, 'error': 'Instance not found'}), 404
# Reload the client
if soulseek_client and hasattr(soulseek_client, 'hifi') and soulseek_client.hifi:
soulseek_client.hifi.reload_instances()
return jsonify({'success': True, 'url': url})
except Exception as e:
logger.error(f"Error removing HiFi instance: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/hifi/instances/toggle', methods=['POST'])
@admin_only
def hifi_toggle_instance():
"""Toggle enabled state of a HiFi API instance."""
try:
data = request.get_json() or {}
url = data.get('url', '').strip().rstrip('/')
enabled = data.get('enabled', True)
if not url:
return jsonify({'success': False, 'error': 'URL is required'}), 400
from database.music_database import get_database
db = get_database()
db.toggle_hifi_instance(url, enabled)
if soulseek_client and hasattr(soulseek_client, 'hifi') and soulseek_client.hifi:
soulseek_client.hifi.reload_instances()
return jsonify({'success': True})
except Exception as e:
logger.error(f"Error toggling HiFi instance: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/hifi/instances/reorder', methods=['POST'])
@admin_only
def hifi_reorder_instances():
"""Reorder HiFi API instances."""
try:
data = request.get_json() or {}
urls = data.get('urls', [])
if not urls:
return jsonify({'success': False, 'error': 'URL list is required'}), 400
from database.music_database import get_database
db = get_database()
if not db.reorder_hifi_instances(urls):
return jsonify({'success': False, 'error': 'One or more URLs not found'}), 400
# Reload the client
if soulseek_client and hasattr(soulseek_client, 'hifi') and soulseek_client.hifi:
soulseek_client.hifi.reload_instances()
return jsonify({'success': True})
except Exception as e:
logger.error(f"Error reordering HiFi instances: {e}")
return jsonify({'success': False, 'error': str(e)}), 500
@app.route('/api/hifi/instances/list', methods=['GET'])
@admin_only
def hifi_list_instances():
"""Get editable list of HiFi API instances."""
try:
from database.music_database import get_database
from core.hifi_client import DEFAULT_INSTANCES
db = get_database()
db.seed_hifi_instances(DEFAULT_INSTANCES)
instances = db.get_all_hifi_instances()
return jsonify({'instances': instances})
except Exception as e:
logger.error(f"Error listing HiFi instances: {e}")
return jsonify({'error': str(e)}), 500
# ===================================================================
# DEEZER DOWNLOAD ENDPOINTS
# ===================================================================

7
webui/index.html
View file

@ -4622,8 +4622,13 @@
</div>
<div class="form-group">
<label>API Instances:</label>
<div id="hifi-instances-list" style="margin-bottom: 8px;"></div>
<div style="display:flex;gap:8px;margin-bottom:8px;">
<input type="url" id="hifi-new-instance" class="form-input" placeholder="https://example.com" style="flex:1;">
<button class="test-button" onclick="addHiFiInstance()">Add</button>
</div>
<button class="test-button" onclick="checkHiFiInstances()" id="hifi-instances-check-btn" style="margin-bottom: 8px;">Check All Instances</button>
<div id="hifi-instances-panel" style="display: none;"></div>
<div id="hifi-instances-status-panel" style="display: none;"></div>
</div>
</div>

206
webui/static/settings.js
View file

@ -797,6 +797,7 @@ async function loadSettingsData() {
document.getElementById('qobuz-allow-fallback').checked = settings.qobuz?.allow_fallback !== false;
document.getElementById('hifi-download-quality').value = settings.hifi_download?.quality || 'lossless';
document.getElementById('hifi-allow-fallback').checked = settings.hifi_download?.allow_fallback !== false;
loadHiFiInstances();
document.getElementById('deezer-download-quality').value = settings.deezer_download?.quality || 'flac';
document.getElementById('deezer-allow-fallback').checked = settings.deezer_download?.allow_fallback !== false;
document.getElementById('deezer-download-arl').value = settings.deezer_download?.arl || '';
@ -1833,9 +1834,22 @@ function _genreWhitelistRender(genres) {
const searchVal = (document.getElementById('genre-whitelist-search')?.value || '').toLowerCase();
const filtered = searchVal ? _genreWhitelistCache.filter(g => g.toLowerCase().includes(searchVal)) : _genreWhitelistCache;
container.innerHTML = filtered.map(g =>
`<span class="genre-chip">${escapeHtml(g)}<button class="genre-chip-x" onclick="_genreWhitelistRemove('${escapeHtml(g.replace(/'/g, "\\'"))}')">&times;</button></span>`
`<span class="genre-chip">${escapeHtml(g)}<button class="genre-chip-x" data-genre="${escapeHtml(g)}">&times;</button></span>`
).join('');
if (countEl) countEl.textContent = `${_genreWhitelistCache.length} genres`;
_initGenreChipClickHandler();
}
function _initGenreChipClickHandler() {
const container = document.getElementById('genre-whitelist-chips');
if (!container) return;
container.onclick = (e) => {
const btn = e.target.closest('.genre-chip-x');
if (btn) {
e.preventDefault();
_genreWhitelistRemove(btn.dataset.genre);
}
};
}
function _genreWhitelistRemove(genre) {
@ -2786,19 +2800,32 @@ function renderApiKeys(keys) {
container.innerHTML = keys.map(k => `
<div style="display: flex; align-items: center; justify-content: space-between; padding: 8px 10px; margin-bottom: 4px; background: rgba(255,255,255,0.03); border: 1px solid rgba(255,255,255,0.06); border-radius: 6px;">
<div style="flex: 1; min-width: 0;">
<div style="font-size: 13px; color: #e0e0e0; font-weight: 500;">${k.label || 'Unnamed'}</div>
<div style="font-size: 13px; color: #e0e0e0; font-weight: 500;">${escapeHtml(k.label || 'Unnamed')}</div>
<div style="font-size: 11px; color: #666; margin-top: 2px;">
<code>${k.key_prefix || 'sk_...'}...</code>
<code>${escapeHtml(k.key_prefix || 'sk_...')}...</code>
&middot; Created ${k.created_at ? new Date(k.created_at).toLocaleDateString() : 'unknown'}
${k.last_used_at ? '&middot; Last used ' + new Date(k.last_used_at).toLocaleDateString() : ''}
</div>
</div>
<button onclick="revokeApiKey('${k.id}', '${(k.label || 'this key').replace(/'/g, "\\'")}')"
<button class="revoke-api-key-btn" data-key-id="${escapeHtml(k.id)}" data-key-label="${escapeHtml(k.label || 'this key')}"
style="padding: 4px 10px; background: rgba(255,82,82,0.1); border: 1px solid rgba(255,82,82,0.2); color: #ff5252; border-radius: 4px; cursor: pointer; font-size: 11px; white-space: nowrap;">
Revoke
</button>
</div>
`).join('');
_initApiKeyClickHandler();
}
function _initApiKeyClickHandler() {
const container = document.getElementById('api-keys-list');
if (!container) return;
container.onclick = (e) => {
const btn = e.target.closest('.revoke-api-key-btn');
if (btn) {
e.preventDefault();
revokeApiKey(btn.dataset.keyId, btn.dataset.keyLabel);
}
};
}
async function generateApiKey() {
@ -3277,8 +3304,177 @@ async function testLidarrConnection() {
}
}
async function loadHiFiInstances() {
const listEl = document.getElementById('hifi-instances-list');
if (!listEl) return;
try {
const resp = await fetch('/api/hifi/instances/list');
const data = await resp.json();
if (!data.instances || data.instances.length === 0) {
listEl.innerHTML = '<div style="color: rgba(255,255,255,0.4); font-size: 0.85em;">No instances configured.</div>';
return;
}
listEl.innerHTML = data.instances.map((inst, i) => {
const enabledClass = inst.enabled ? '' : 'hifi-instance-disabled';
const checkHtml = inst.enabled
? `<span class="hifi-instance-toggle on" data-url="${escapeHtml(inst.url)}" title="Click to disable">&#x2714;</span>`
: `<span class="hifi-instance-toggle off" data-url="${escapeHtml(inst.url)}" title="Click to enable">&#x2718;</span>`;
return `<div class="hifi-instance-item${inst.enabled ? '' : ' disabled'}" draggable="true" data-url="${escapeHtml(inst.url)}">
<span class="hifi-instance-grip">&#x2630;</span>
<span class="hifi-instance-url">${escapeHtml(inst.url)}</span>
${checkHtml}
<span class="hifi-instance-remove" data-url="${escapeHtml(inst.url)}" title="Remove instance">&#x2716;</span>
</div>`;
}).join('');
_initHiFiDragDrop();
_initHiFiClickHandlers();
} catch (e) {
listEl.innerHTML = `<div style="color:#f44336;font-size:0.85em;">Error loading instances: ${escapeHtml(e.message)}</div>`;
}
}
function _initHiFiDragDrop() {
const listEl = document.getElementById('hifi-instances-list');
if (!listEl) return;
let dragIdx = null;
listEl.querySelectorAll('.hifi-instance-item').forEach((item, idx) => {
item.addEventListener('dragstart', (e) => {
dragIdx = idx;
item.classList.add('dragging');
e.dataTransfer.effectAllowed = 'move';
});
item.addEventListener('dragend', () => {
item.classList.remove('dragging');
dragIdx = null;
listEl.querySelectorAll('.hifi-instance-item').forEach(i => i.classList.remove('drag-over'));
});
item.addEventListener('dragover', (e) => {
e.preventDefault();
e.dataTransfer.dropEffect = 'move';
item.classList.add('drag-over');
});
item.addEventListener('dragleave', () => {
item.classList.remove('drag-over');
});
item.addEventListener('drop', async (e) => {
e.preventDefault();
item.classList.remove('drag-over');
if (dragIdx === null) return;
const items = [...listEl.querySelectorAll('.hifi-instance-item')];
const dragged = items[dragIdx];
if (dragIdx !== idx) {
if (dragIdx < idx) {
item.after(dragged);
} else {
item.before(dragged);
}
const urls = [...listEl.querySelectorAll('.hifi-instance-item')].map(el => el.dataset.url);
await _saveHiFiInstanceOrder(urls);
}
});
});
}
function _initHiFiClickHandlers() {
const listEl = document.getElementById('hifi-instances-list');
if (!listEl) return;
listEl.onclick = (e) => {
const toggle = e.target.closest('.hifi-instance-toggle');
if (toggle) {
e.preventDefault();
toggleHiFiInstance(toggle.dataset.url);
return;
}
const remove = e.target.closest('.hifi-instance-remove');
if (remove) {
e.preventDefault();
removeHiFiInstance(remove.dataset.url);
}
};
}
async function _saveHiFiInstanceOrder(urls) {
try {
await fetch('/api/hifi/instances/reorder', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ urls })
});
} catch (e) {
console.error('Failed to save HiFi instance order:', e);
}
}
async function toggleHiFiInstance(url) {
const listEl = document.getElementById('hifi-instances-list');
if (!listEl) return;
const item = listEl.querySelector(`.hifi-instance-item[data-url="${url}"]`);
const toggle = item?.querySelector('.hifi-instance-toggle');
const currentlyEnabled = toggle?.classList.contains('on');
const newEnabled = !currentlyEnabled;
try {
const resp = await fetch('/api/hifi/instances/toggle', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ url, enabled: newEnabled })
});
const data = await resp.json();
if (data.success) {
loadHiFiInstances();
} else {
alert(data.error || 'Failed to toggle instance');
}
} catch (e) {
alert(`Error: ${e.message}`);
}
}
async function addHiFiInstance() {
const input = document.getElementById('hifi-new-instance');
if (!input) return;
const url = input.value.trim();
if (!url) return;
if (!url.startsWith('http://') && !url.startsWith('https://')) {
alert('URL must start with http:// or https://');
return;
}
try {
const resp = await fetch('/api/hifi/instances', {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ url })
});
const data = await resp.json();
if (data.success) {
input.value = '';
loadHiFiInstances();
} else {
alert(data.error || 'Failed to add instance');
}
} catch (e) {
alert(`Error: ${e.message}`);
}
}
async function removeHiFiInstance(url) {
try {
const resp = await fetch(`/api/hifi/instances?url=${encodeURIComponent(url)}`, {
method: 'DELETE'
});
const data = await resp.json();
if (data.success) {
loadHiFiInstances();
} else {
alert(data.error || 'Failed to remove instance');
}
} catch (e) {
alert(`Error: ${e.message}`);
}
}
async function checkHiFiInstances() {
const panel = document.getElementById('hifi-instances-panel');
const panel = document.getElementById('hifi-instances-status-panel');
const btn = document.getElementById('hifi-instances-check-btn');
if (!panel) return;
panel.style.display = 'block';

77
webui/static/style.css
View file

@ -60262,3 +60262,80 @@ body[data-artist-source="source"] #artist-detail-page #library-artist-enhance-bt
#artist-detail-page .release-card.album-card .mb-card-icon:hover {
opacity: 1;
}
/* ── HiFi API instances list (drag and drop) ── */
#hifi-instances-list {
display: flex;
flex-direction: column;
gap: 4px;
}
.hifi-instance-item {
display: flex;
align-items: center;
gap: 8px;
padding: 6px 10px;
background: rgba(255, 255, 255, 0.03);
border: 1px solid rgba(255, 255, 255, 0.06);
border-radius: 8px;
font-size: 0.82em;
user-select: none;
transition: all 0.2s;
cursor: grab;
}
.hifi-instance-item:hover {
background: rgba(255, 255, 255, 0.05);
border-color: rgba(255, 255, 255, 0.1);
}
.hifi-instance-item.dragging {
opacity: 0.4;
border-color: rgb(var(--accent-rgb));
}
.hifi-instance-item.drag-over {
border-color: rgb(var(--accent-rgb));
background: rgba(var(--accent-rgb), 0.08);
}
.hifi-instance-item.disabled {
opacity: 0.4;
}
.hifi-instance-grip {
color: rgba(255, 255, 255, 0.3);
cursor: grab;
flex-shrink: 0;
font-size: 1em;
}
.hifi-instance-url {
flex: 1;
color: rgba(255, 255, 255, 0.7);
font-family: monospace;
overflow: hidden;
text-overflow: ellipsis;
white-space: nowrap;
}
.hifi-instance-toggle {
flex-shrink: 0;
cursor: pointer;
font-size: 0.9em;
transition: color 0.15s;
}
.hifi-instance-toggle.on {
color: #4caf50;
}
.hifi-instance-toggle.on:hover {
color: #66bb6a;
}
.hifi-instance-toggle.off {
color: #666;
}
.hifi-instance-toggle.off:hover {
color: #999;
}
.hifi-instance-remove {
flex-shrink: 0;
color: #f44336;
cursor: pointer;
font-size: 0.9em;
transition: color 0.15s;
}
.hifi-instance-remove:hover {
color: #ef5350;
}