feat(quality): source mappers + populate real quality on streaming results

Add core/quality/source_map.py centralising each source's tier->AudioQuality
mapping (Tidal/HiFi tiers, Qobuz real kHz/bit-depth, Deezer codes, Amazon
codec/tier). Add TrackResult.set_quality() to merge a mapped AudioQuality
onto a result. Wire HiFi, Qobuz, Deezer, Tidal, Amazon search results to
stamp real sample_rate/bit_depth so the global ranker no longer relies on
crude kbps heuristics for streaming sources. Fixes Qobuz/Amazon display
labels ('FLAC 24-bit/192kHz', 'Lossless') breaking format derivation.

Tested: 22 passing (mappers + set_quality merge semantics).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

core/amazon_download_client.py

@@ -32,6 +32,7 @@ from config.settings import config_manager
 from core.amazon_client import AmazonClient, AmazonClientError
 from core.download_plugins.base import DownloadSourcePlugin
 from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult
+from core.quality.model import AudioQuality
 from utils.logging_config import get_logger
 
 logger = get_logger("amazon_download_client")
@@ -133,11 +134,17 @@ class AmazonDownloadClient(DownloadSourcePlugin):
         album_map: Dict[str, AlbumResult] = {}
         album_order: List[str] = []
         preferred = self._client.preferred_codec
+        # Search results only carry the codec (real sample_rate arrives at
+        # stream time). Claim the format honestly — FLAC for the lossless
+        # codec, lossy otherwise — so audio_quality derives a real format
+        # instead of the display label ("Lossless"), and the post-download
+        # probe pins the actual sample_rate/bit_depth.
+        amazon_q = AudioQuality(format='flac' if _codec_key(preferred) == 'flac' else 'aac')
 
         for item in items:
             quality = _quality_label(preferred)
             if item.is_track:
-                track_results.append(TrackResult(
+                tr = TrackResult(
                     username="amazon",
                     filename=f"{item.asin}||{item.artist_name} - {item.title}",
                     size=0,
@@ -155,7 +162,9 @@ class AmazonDownloadClient(DownloadSourcePlugin):
                         "album_asin": item.album_asin,
                         "isrc": item.isrc,
                     },
-                ))
+                )
+                tr.set_quality(amazon_q)
+                track_results.append(tr)
             elif item.is_album:
                 album_asin = item.album_asin or item.asin
                 if album_asin not in album_map:
@@ -173,6 +182,7 @@ class AmazonDownloadClient(DownloadSourcePlugin):
                         title=item.title,
                         album=item.album_name,
                     )
+                    placeholder.set_quality(amazon_q)
                     album_map[album_asin] = AlbumResult(
                         username="amazon",
                         album_path=album_asin,

core/deezer_download_client.py

@@ -21,6 +21,7 @@ from typing import Any, Dict, List, Optional, Tuple
 import requests
 
 from core.download_plugins.types import AlbumResult, DownloadStatus, TrackResult
+from core.quality.source_map import quality_from_deezer
 from utils.logging_config import get_logger
 
 logger = get_logger("deezer_download")
@@ -581,7 +582,7 @@ class DeezerDownloadClient(DownloadSourcePlugin):
                     bitrate = 128
                     quality = 'mp3'
 
-                results.append(TrackResult(
+                tr = TrackResult(
                     username='deezer_dl',
                     filename=f"{track_id}||{artist} - {title}",
                     size=est_size,
@@ -595,7 +596,10 @@ class DeezerDownloadClient(DownloadSourcePlugin):
                     title=title,
                     album=album,
                     track_number=item.get('track_position'),
-                ))
+                )
+                # Stamp CD-quality FLAC (16/44.1) so lossless ranks correctly.
+                tr.set_quality(quality_from_deezer(self._quality))
+                results.append(tr)
 
             logger.info(f"Deezer search for '{query}' returned {len(results)} results")
             return results, []

core/download_plugins/types.py

@@ -48,6 +48,22 @@ class SearchResult:
             bit_depth=self.bit_depth,
         )
 
+    def set_quality(self, aq: AudioQuality) -> None:
+        """Merge a mapped :class:`AudioQuality` onto this result's fields.
+
+        Used by streaming sources to stamp their claimed tier (Tidal/HiFi
+        tier strings, Qobuz API values, …) so ``audio_quality`` ranks
+        correctly. Mapper-provided fields win; a ``None`` from the mapper
+        leaves any already-reported value (e.g. a probed bitrate) intact.
+        """
+        self.quality = aq.format
+        if aq.bitrate is not None:
+            self.bitrate = aq.bitrate
+        if aq.sample_rate is not None:
+            self.sample_rate = aq.sample_rate
+        if aq.bit_depth is not None:
+            self.bit_depth = aq.bit_depth
+
     @property
     def quality_score(self) -> float:
         quality_weights = {

core/hifi_client.py

@@ -36,6 +36,7 @@ import requests as http_requests
 from utils.logging_config import get_logger
 from config.settings import config_manager
 from core.download_plugins.types import TrackResult, AlbumResult, DownloadStatus
+from core.quality.source_map import quality_from_tidal_tier
 
 logger = get_logger("hifi_client")
 
@@ -754,10 +755,15 @@ class HiFiClient(DownloadSourcePlugin):
             quality_key = config_manager.get('hifi_download.quality', 'lossless')
             q_info = HLS_QUALITY_MAP.get(quality_key, HLS_QUALITY_MAP['lossless'])
 
+            # HiFi is Tidal-backed; stamp the configured tier so the global
+            # ranker sees real sample_rate/bit_depth, not just 'flac'.
+            tier_quality = quality_from_tidal_tier(quality_key)
+
             results = []
             for t in tracks:
                 try:
                     tr = self._to_track_result(t, q_info)
+                    tr.set_quality(tier_quality)
                     results.append(tr)
                 except Exception as e:
                     logger.debug(f"Skipping track result conversion: {e}")

core/qobuz_client.py

@@ -29,6 +29,7 @@ from config.settings import config_manager
 
 # Import Soulseek data structures for drop-in replacement compatibility
 from core.download_plugins.types import TrackResult, AlbumResult, DownloadStatus
+from core.quality.source_map import quality_from_qobuz
 
 logger = get_logger("qobuz_client")
 
@@ -1072,6 +1073,14 @@ class QobuzClient(DownloadSourcePlugin):
             track_number=track.get('track_number'),
         )
 
+        # Stamp real API quality so the global ranker sees actual
+        # sample_rate/bit_depth. Hi-res only when the track itself is
+        # hires-streamable; otherwise it streams as CD-quality FLAC.
+        if hires_streamable and max_bit_depth >= 24:
+            track_result.set_quality(quality_from_qobuz(max_sample_rate, max_bit_depth))
+        else:
+            track_result.set_quality(quality_from_qobuz(44.1, 16))
+
         return track_result
 
     # ===================== Download =====================

core/quality/source_map.py

@@ -0,0 +1,104 @@
+"""Per-source quality mappers — turn each download source's tier string or
+API values into a unified :class:`~core.quality.model.AudioQuality`.
+
+Every streaming source describes quality differently (Tidal/HiFi use tier
+strings, Qobuz reports real kHz + bit depth, Deezer uses config codes,
+Amazon mixes real values with HD/UHD tiers). Centralising the knowledge
+here keeps the per-client code to a single call and keeps the tier tables
+in one auditable place.
+
+Each value is a *claim*: the download client populates its ``TrackResult``
+from it so the global ranker can choose a source, and the post-download
+quality guard later verifies the real file. Over-claiming is the danger —
+an unknown tier maps to ``format='unknown'`` rather than pretending to be
+lossless.
+"""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from core.quality.model import AudioQuality
+
+
+# ── Tidal / HiFi (Monochrome is Tidal-backed) ──────────────────────────────
+#
+# Tidal exposes UPPER_SNAKE tier strings (``HI_RES_LOSSLESS``); HiFi's config
+# uses lowercase keys (``hires``/``lossless``). We normalise both into the
+# same lookup so one mapper serves both sources.
+
+_TIDAL_HIRES = AudioQuality(format='flac', sample_rate=96000, bit_depth=24)
+_TIDAL_LOSSLESS = AudioQuality(format='flac', sample_rate=44100, bit_depth=16)
+_TIDAL_HIGH = AudioQuality(format='aac', bitrate=320)
+_TIDAL_LOW = AudioQuality(format='aac', bitrate=96)
+
+TIDAL_TIER_MAP = {
+    'HI_RES_LOSSLESS': _TIDAL_HIRES,
+    'HI_RES': _TIDAL_HIRES,
+    'HIRES': _TIDAL_HIRES,
+    'LOSSLESS': _TIDAL_LOSSLESS,
+    'HIGH': _TIDAL_HIGH,
+    'LOW': _TIDAL_LOW,
+}
+
+
+def quality_from_tidal_tier(tier: str) -> AudioQuality:
+    """Map a Tidal/HiFi quality tier string to an AudioQuality.
+
+    Case-insensitive; accepts both ``HI_RES`` and ``hires`` spellings.
+    Unrecognised tiers map to ``format='unknown'`` so they never
+    over-claim lossless quality.
+    """
+    key = (tier or '').strip().upper()
+    return TIDAL_TIER_MAP.get(key, AudioQuality(format='unknown'))
+
+
+# ── Qobuz (real API values) ────────────────────────────────────────────────
+
+def quality_from_qobuz(sampling_rate_khz: float, bit_depth: int) -> AudioQuality:
+    """Qobuz reports ``maximum_sampling_rate`` in kHz (e.g. 44.1, 96, 192)
+    and ``maximum_bit_depth``. These are real values from the API.
+    """
+    sample_rate = int(round(sampling_rate_khz * 1000)) if sampling_rate_khz else None
+    return AudioQuality(format='flac', sample_rate=sample_rate, bit_depth=bit_depth)
+
+
+# ── Deezer (config code) ───────────────────────────────────────────────────
+
+DEEZER_CODE_MAP = {
+    'flac': AudioQuality(format='flac', sample_rate=44100, bit_depth=16),
+    'mp3_320': AudioQuality(format='mp3', bitrate=320),
+    'mp3_128': AudioQuality(format='mp3', bitrate=128),
+}
+
+
+def quality_from_deezer(code: str) -> AudioQuality:
+    """Map a Deezer download quality code to AudioQuality.
+
+    Deezer FLAC is always CD-quality (16-bit/44.1 kHz).
+    """
+    return DEEZER_CODE_MAP.get((code or '').lower(), AudioQuality(format='unknown'))
+
+
+# ── Amazon Music (real sampleRate preferred, HD/UHD tier fallback) ─────────
+
+_AMAZON_TIER_MAP = {
+    'UHD': AudioQuality(format='flac', sample_rate=96000, bit_depth=24),
+    'HD': AudioQuality(format='flac', sample_rate=44100, bit_depth=16),
+}
+
+
+def quality_from_amazon(
+    tier: str,
+    sample_rate: Optional[int] = None,
+    bit_depth: Optional[int] = None,
+) -> AudioQuality:
+    """Amazon Music is FLAC; prefer the real ``sampleRate``/``bitDepth`` from
+    the stream info when present, otherwise fall back to the HD/UHD tier.
+    """
+    base = _AMAZON_TIER_MAP.get((tier or '').strip().upper(), AudioQuality(format='flac'))
+    return AudioQuality(
+        format='flac',
+        sample_rate=sample_rate if sample_rate is not None else base.sample_rate,
+        bit_depth=bit_depth if bit_depth is not None else base.bit_depth,
+    )

core/tidal_download_client.py

@@ -33,6 +33,7 @@ from config.settings import config_manager
 
 # Import Soulseek data structures for drop-in replacement compatibility
 from core.download_plugins.types import TrackResult, AlbumResult, DownloadStatus
+from core.quality.source_map import quality_from_tidal_tier
 
 logger = get_logger("tidal_download_client")
 
@@ -457,11 +458,15 @@ class TidalDownloadClient(DownloadSourcePlugin):
 
             quality_key = config_manager.get('tidal_download.quality', 'lossless')
             quality_info = QUALITY_MAP.get(quality_key, QUALITY_MAP['lossless'])
+            # Stamp the configured tier (what will actually be downloaded) so
+            # the global ranker sees real sample_rate/bit_depth.
+            tier_quality = quality_from_tidal_tier(quality_key)
 
             track_results = []
             for track in tidal_tracks:
                 try:
                     track_result = self._tidal_to_track_result(track, quality_info)
+                    track_result.set_quality(tier_quality)
                     track_results.append(track_result)
                 except Exception as e:
                     logger.debug(f"Skipping track conversion error: {e}")

tests/quality/test_source_map.py

@@ -0,0 +1,122 @@
+"""Tests for core.quality.source_map — each download source's tier/string
+mapped into a unified AudioQuality.
+
+These mappers express each source's *claimed* capability tier. The values
+are later verified post-download by the quality guard, so over-claiming is
+the failure mode to avoid: an unknown tier must NOT pretend to be lossless.
+"""
+
+import pytest
+
+from core.quality.source_map import (
+    quality_from_tidal_tier,
+    quality_from_qobuz,
+    quality_from_deezer,
+    quality_from_amazon,
+)
+
+
+# ── Tidal / HiFi (Tidal-backed) ────────────────────────────────────────────
+
+@pytest.mark.parametrize("tier", ["HI_RES_LOSSLESS", "HI_RES", "hi_res", "hires", "HIRES"])
+def test_tidal_hires_is_flac_24_96(tier):
+    aq = quality_from_tidal_tier(tier)
+    assert aq.format == "flac"
+    assert aq.bit_depth == 24
+    assert aq.sample_rate == 96000
+
+
+@pytest.mark.parametrize("tier", ["LOSSLESS", "lossless"])
+def test_tidal_lossless_is_flac_16_44(tier):
+    aq = quality_from_tidal_tier(tier)
+    assert aq.format == "flac"
+    assert aq.bit_depth == 16
+    assert aq.sample_rate == 44100
+
+
+def test_tidal_high_is_lossy_aac_320():
+    aq = quality_from_tidal_tier("HIGH")
+    assert aq.format == "aac"
+    assert aq.bitrate == 320
+    assert aq.bit_depth is None  # lossy: no bit depth
+
+
+def test_tidal_low_is_lossy_aac_96():
+    aq = quality_from_tidal_tier("LOW")
+    assert aq.format == "aac"
+    assert aq.bitrate == 96
+
+
+def test_tidal_unknown_tier_does_not_overclaim():
+    # An unrecognised tier must not masquerade as lossless.
+    aq = quality_from_tidal_tier("SOMETHING_NEW")
+    assert aq.format == "unknown"
+    assert aq.bit_depth is None
+    assert aq.sample_rate is None
+
+
+# ── Qobuz (real API values) ────────────────────────────────────────────────
+
+def test_qobuz_hires_khz_to_hz():
+    aq = quality_from_qobuz(96.0, 24)
+    assert aq.format == "flac"
+    assert aq.sample_rate == 96000
+    assert aq.bit_depth == 24
+
+
+def test_qobuz_cd_quality_fractional_khz():
+    aq = quality_from_qobuz(44.1, 16)
+    assert aq.format == "flac"
+    assert aq.sample_rate == 44100
+    assert aq.bit_depth == 16
+
+
+def test_qobuz_192k():
+    aq = quality_from_qobuz(192.0, 24)
+    assert aq.sample_rate == 192000
+    assert aq.bit_depth == 24
+
+
+# ── Deezer (config code) ───────────────────────────────────────────────────
+
+def test_deezer_flac_is_16_44():
+    aq = quality_from_deezer("flac")
+    assert aq.format == "flac"
+    assert aq.bit_depth == 16
+    assert aq.sample_rate == 44100
+
+
+def test_deezer_mp3_320():
+    aq = quality_from_deezer("mp3_320")
+    assert aq.format == "mp3"
+    assert aq.bitrate == 320
+    assert aq.bit_depth is None
+
+
+def test_deezer_mp3_128():
+    aq = quality_from_deezer("mp3_128")
+    assert aq.format == "mp3"
+    assert aq.bitrate == 128
+
+
+# ── Amazon (real sampleRate preferred, tier fallback) ──────────────────────
+
+def test_amazon_prefers_real_sample_rate():
+    aq = quality_from_amazon("HD", sample_rate=88200, bit_depth=24)
+    assert aq.format == "flac"
+    assert aq.sample_rate == 88200
+    assert aq.bit_depth == 24
+
+
+def test_amazon_hd_tier_fallback():
+    aq = quality_from_amazon("HD")
+    assert aq.format == "flac"
+    assert aq.sample_rate == 44100
+    assert aq.bit_depth == 16
+
+
+def test_amazon_uhd_tier_fallback():
+    aq = quality_from_amazon("UHD")
+    assert aq.format == "flac"
+    assert aq.bit_depth == 24
+    assert aq.sample_rate == 96000

tests/quality/test_track_result_quality.py

@@ -0,0 +1,39 @@
+"""TrackResult.set_quality() — merge a mapped AudioQuality onto a result so
+its derived ``audio_quality`` reflects the source's real/claimed tier.
+"""
+
+from core.download_plugins.types import TrackResult
+from core.quality.model import AudioQuality
+
+
+def _tr(**kw):
+    base = dict(
+        username='x', filename='f', size=0, bitrate=None, duration=None,
+        quality='', free_upload_slots=0, upload_speed=0, queue_length=0,
+    )
+    base.update(kw)
+    return TrackResult(**base)
+
+
+def test_set_quality_populates_lossless_fields():
+    tr = _tr(quality='flac', bitrate=1411)
+    tr.set_quality(AudioQuality('flac', sample_rate=96000, bit_depth=24))
+    assert tr.quality == 'flac'
+    assert tr.sample_rate == 96000
+    assert tr.bit_depth == 24
+    # derived descriptor must reflect the merge
+    assert tr.audio_quality.sample_rate == 96000
+    assert tr.audio_quality.bit_depth == 24
+
+
+def test_set_quality_keeps_existing_bitrate_when_mapper_has_none():
+    tr = _tr(quality='flac', bitrate=950)
+    tr.set_quality(AudioQuality('flac', sample_rate=44100, bit_depth=16))
+    assert tr.bitrate == 950  # mapper bitrate is None → preserve probed/reported
+
+
+def test_set_quality_overwrites_bitrate_for_lossy():
+    tr = _tr(quality='mp3', bitrate=128)
+    tr.set_quality(AudioQuality('mp3', bitrate=320))
+    assert tr.bitrate == 320
+    assert tr.bit_depth is None