feat(quality): unify quality scanner onto the real ranked-target core (strict)

The library quality scanner judged quality by FILE EXTENSION only
(get_quality_tier_from_extension) and read the legacy v2 `qualities` dict —
so every FLAC was "lossless tier 1" regardless of bit depth / sample rate. It
could never flag a 16-bit FLAC as upgradeable under a 24-bit profile, and it
ignored the v3 ranked_targets entirely. Completely inconsistent with the
download guard.

Now both share one core:
- selection.targets_from_profile(profile) — single profile→targets conversion
  (v2→v3 migration), reused by load_profile_targets.
- selection.quality_meets_profile(aq, targets) — strict: meets iff the real
  measured quality satisfies a ranked target (fallback ignored — it's a
  download concession, not a definition of "good enough").
- guards.check_quality_target refactored to use both.
- quality_scanner probes real quality (probe_audio_quality) and checks against
  the v3 targets via quality_meets_profile. Extension tier kept only as a
  fallback label when a file can't be probed.

Result: the scan flags exactly what the download gate would reject — 16-bit
when you want 24-bit, wrong sample rate, MP3 when you want FLAC.

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

core/discovery/quality_scanner.py

@@ -11,8 +11,11 @@ and add provider matches to the wishlist:
    - other → all library tracks.
 3. For each track:
    - Stop-request gate (state['status'] != 'running').
-   - Quality-tier check via _get_quality_tier_from_extension(file_path).
-   - Skip tracks meeting standards (tier_num <= min_acceptable_tier).
+   - Probe REAL audio quality (bit depth / sample rate / bitrate) and check it
+     against the user's v3 ranked targets via quality_meets_profile — the SAME
+     strict core the download import guard uses (fallback ignored). Extension
+     tier is only a fallback label when the file can't be probed.
+   - Skip tracks that satisfy a target (quality_met).
    - For low-quality tracks: matching_engine search query gen, score
      candidates against the configured metadata source priority
      (artist + title similarity, album-type bonus), pick best match >=
@@ -66,6 +69,10 @@ class QualityScannerDeps:
     automation_engine: Any
     get_quality_tier_from_extension: Callable
     add_activity_item: Callable
+    # Probe real audio quality (mutagen-read bit depth / sample rate / bitrate)
+    # so the scan checks against the SAME ranked-target core as the download
+    # quality guard — not just the file extension. Injected for testability.
+    probe_audio_quality: Callable = None
 
 
 def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any:
@@ -325,27 +332,20 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
         # Get database instance
         db = MusicDatabase()
 
-        # Get quality profile to determine preferred quality
+        # Load the user's v3 ranked targets — the SAME quality definition the
+        # download import guard uses. A track is "low quality" when its REAL
+        # measured quality (bit depth + sample rate + bitrate) satisfies none of
+        # the targets. Strict: fallback is ignored (it's a download-time
+        # concession, not a definition of what counts as good enough), so the
+        # scanner surfaces every genuinely-upgradeable track.
+        from core.quality.selection import targets_from_profile, quality_meets_profile
         quality_profile = db.get_quality_profile()
-        preferred_qualities = quality_profile.get('qualities', {})
+        profile_targets, _fallback_enabled = targets_from_profile(quality_profile)
 
-        # Determine minimum acceptable tier based on enabled qualities
-        min_acceptable_tier = 999
-        for quality_name, quality_config in preferred_qualities.items():
-            if quality_config.get('enabled', False):
-                # Map quality profile names to tier names
-                tier_map = {
-                    'flac': 'lossless',
-                    'mp3_320': 'low_lossy',
-                    'mp3_256': 'low_lossy',
-                    'mp3_192': 'low_lossy'
-                }
-                tier_name = tier_map.get(quality_name)
-                if tier_name:
-                    tier_num = deps.QUALITY_TIERS[tier_name]['tier']
-                    min_acceptable_tier = min(min_acceptable_tier, tier_num)
-
-        logger.info(f"[Quality Scanner] Minimum acceptable tier: {min_acceptable_tier}")
+        logger.info(
+            "[Quality Scanner] Profile targets (strict): %s",
+            [t.label for t in profile_targets] or "(none — no constraint)",
+        )
 
         # Get tracks to scan based on scope
         with deps.quality_scanner_lock:
@@ -428,8 +428,15 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
             try:
                 track_id, title, artist_id, album_id, file_path, bitrate, artist_name, album_title = track_row
 
-                # Check quality tier
-                tier_name, tier_num = deps.get_quality_tier_from_extension(file_path)
+                # Probe the REAL audio quality (bit depth / sample rate / bitrate)
+                # and check it against the ranked targets — same core as the
+                # download guard. Extension tier is only a fallback label when the
+                # file can't be probed.
+                aq = deps.probe_audio_quality(file_path) if deps.probe_audio_quality else None
+                if aq is not None:
+                    tier_name = aq.label()
+                else:
+                    tier_name = deps.get_quality_tier_from_extension(file_path)[0]
 
                 # Update progress
                 with deps.quality_scanner_lock:
@@ -437,8 +444,8 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
                     deps.quality_scanner_state["progress"] = (idx / total_tracks) * 100
                     deps.quality_scanner_state["phase"] = f"Scanning: {artist_name} - {title}"
 
-                # Check if meets quality standards
-                if tier_num <= min_acceptable_tier:
+                # Check if it meets the profile (strict — satisfies a real target).
+                if quality_meets_profile(aq, profile_targets):
                     # Quality met
                     with deps.quality_scanner_lock:
                         deps.quality_scanner_state["quality_met"] += 1

core/imports/guards.py

@@ -125,7 +125,7 @@ def check_quality_target(file_path: str, context: dict) -> Optional[str]:
     logic here.
     """
     from core.imports.file_ops import probe_audio_quality
-    from core.quality.model import QualityTarget, rank_candidate, v2_qualities_to_ranked_targets
+    from core.quality.selection import targets_from_profile, quality_meets_profile
 
     aq = probe_audio_quality(file_path)
     if aq is None:
@@ -133,26 +133,21 @@ def check_quality_target(file_path: str, context: dict) -> Optional[str]:
         return None
 
     profile = MusicDatabase().get_quality_profile()
-    raw_targets = profile.get("ranked_targets")
-    if not raw_targets and "qualities" in profile:
-        raw_targets = v2_qualities_to_ranked_targets(profile["qualities"])
+    targets, fallback_enabled = targets_from_profile(profile)
 
-    if not raw_targets:
+    if not targets:
         return None
 
-    targets = [QualityTarget.from_dict(t) for t in raw_targets]
-    fallback_enabled = profile.get("fallback_enabled", True)
     downsample_enabled = _get_config_manager().get("lossy_copy.downsample_hires", False)
 
-    target_idx, _ = rank_candidate(aq, targets)
-    matched = target_idx < len(targets)
+    matched = quality_meets_profile(aq, targets)
 
     track_info = context.get("track_info", {})
     track_name = track_info.get("name", os.path.basename(file_path))
     actual_label = aq.label()
 
     if matched:
-        logger.info("[QualityGuard] %s matched target '%s': %s", track_name, targets[target_idx].label, actual_label)
+        logger.info("[QualityGuard] %s meets profile: %s", track_name, actual_label)
         return None
 
     # No target matched

core/quality/selection.py

@@ -51,6 +51,19 @@ def rank_with_targets(
     return [], False
 
 
+def targets_from_profile(profile: dict) -> Tuple[List[QualityTarget], bool]:
+    """Convert a quality-profile dict into ``(targets, fallback_enabled)`` with
+    v2->v3 migration applied. The single conversion path shared by the import
+    guard, the download ranker and the library quality scanner."""
+    raw_targets = profile.get('ranked_targets')
+    if not raw_targets and 'qualities' in profile:
+        raw_targets = v2_qualities_to_ranked_targets(profile['qualities'])
+
+    targets = [QualityTarget.from_dict(t) for t in (raw_targets or [])]
+    fallback_enabled = profile.get('fallback_enabled', True)
+    return targets, fallback_enabled
+
+
 def load_profile_targets() -> Tuple[List[QualityTarget], bool]:
     """Load the user's quality profile from the DB and return
     ``(targets, fallback_enabled)`` with v2->v3 migration applied.
@@ -61,14 +74,27 @@ def load_profile_targets() -> Tuple[List[QualityTarget], bool]:
     """
     from database.music_database import MusicDatabase
 
-    profile = MusicDatabase().get_quality_profile()
-    raw_targets = profile.get('ranked_targets')
-    if not raw_targets and 'qualities' in profile:
-        raw_targets = v2_qualities_to_ranked_targets(profile['qualities'])
+    return targets_from_profile(MusicDatabase().get_quality_profile())
 
-    targets = [QualityTarget.from_dict(t) for t in (raw_targets or [])]
-    fallback_enabled = profile.get('fallback_enabled', True)
-    return targets, fallback_enabled
+
+def quality_meets_profile(aq, targets: List[QualityTarget]) -> bool:
+    """Strict: True iff *aq* satisfies at least one ranked *target*.
+
+    The shared definition of "good enough" for both the import guard and the
+    library scanner — bit depth + sample rate are minimums (see
+    :meth:`AudioQuality.matches_target`). Fallback is NOT consulted here; it's a
+    download-time last-resort concession, not part of what counts as meeting the
+    profile. ``targets`` empty → no constraint (True). ``aq`` None (probe
+    failed) → True, so an unreadable file is never falsely flagged.
+    """
+    if not targets:
+        return True
+    if aq is None:
+        return True
+    from core.quality.model import rank_candidate
+
+    idx, _ = rank_candidate(aq, targets)
+    return idx < len(targets)
 
 
 _VALID_SEARCH_MODES = ("priority", "best_quality")

tests/discovery/test_discovery_quality_scanner.py

@@ -146,6 +146,7 @@ def _build_deps(
     primary_source='spotify',
     quality_tier_result=('lossless', 1),
     automation=None,
+    probe_fn=None,
 ):
     globals()['_TEST_PRIMARY_SOURCE'] = primary_source
     _TEST_SOURCE_CLIENTS.clear()
@@ -154,6 +155,17 @@ def _build_deps(
     elif primary_source:
         _TEST_SOURCE_CLIENTS[primary_source] = _FakeMetadataClient(results=[])
 
+    # Quality is now decided by probing the real file against the v3 targets.
+    # Derive a stand-in AudioQuality from the legacy tier param so existing
+    # tests keep their intent: 'lossless' → a FLAC that meets the default
+    # flac-enabled profile; anything else → an MP3 that doesn't.
+    from core.quality.model import AudioQuality
+
+    def _default_probe(_fp):
+        if quality_tier_result[0] == 'lossless':
+            return AudioQuality('flac', bit_depth=16, sample_rate=44100)
+        return AudioQuality('mp3', bitrate=128)
+
     deps = qs.QualityScannerDeps(
         quality_scanner_state=state if state is not None else {},
         quality_scanner_lock=threading.Lock(),
@@ -162,6 +174,7 @@ def _build_deps(
         automation_engine=automation or _FakeAutomationEngine(),
         get_quality_tier_from_extension=lambda fp: quality_tier_result,
         add_activity_item=lambda *a, **kw: None,
+        probe_audio_quality=probe_fn or _default_probe,
     )
     return deps
 
@@ -423,20 +436,21 @@ def test_stop_request_halts_loop(mock_db_and_wishlist):
     db._watchlist_artists = [_WatchlistArtist('A')]
     db._tracks = [_track_row(track_id=1), _track_row(track_id=2)]
     state = {}
-    deps = _build_deps(state=state, quality_tier_result=('lossless', 1))
+    from core.quality.model import AudioQuality
 
-    # Override get_quality_tier_from_extension to set stop after first track
+    # Probe is called once per track; use it to stop after the first.
     call_count = [0]
 
     def stop_after_first(fp):
         call_count[0] += 1
         if call_count[0] == 1:
             # Set status to non-running BEFORE second track iter checks
-            with deps.quality_scanner_lock:
+            with state_lock:
                 state['status'] = 'stopping'
-        return ('lossless', 1)
+        return AudioQuality('flac', bit_depth=16, sample_rate=44100)  # meets profile
 
-    deps.get_quality_tier_from_extension = stop_after_first
+    state_lock = threading.Lock()
+    deps = _build_deps(state=state, quality_tier_result=('lossless', 1), probe_fn=stop_after_first)
 
     qs.run_quality_scanner('watchlist', 1, deps)
 

tests/quality/test_quality_meets_profile.py

@@ -0,0 +1,60 @@
+"""quality_meets_profile / targets_from_profile — the shared strict
+quality-decision core used by BOTH the import quality guard and the library
+quality scanner. A file "meets" the profile iff its real measured quality
+satisfies at least one ranked target (bit depth + sample rate are minimums;
+fallback is NOT consulted — that's a download-time concession, not a definition
+of quality).
+"""
+
+from core.quality.model import AudioQuality, QualityTarget
+from core.quality.selection import quality_meets_profile, targets_from_profile
+
+
+FLAC_HI = AudioQuality('flac', sample_rate=96000, bit_depth=24)
+FLAC_CD = AudioQuality('flac', sample_rate=44100, bit_depth=16)
+MP3 = AudioQuality('mp3', bitrate=320)
+
+WANT_24BIT = [
+    QualityTarget(label='FLAC 24/96', format='flac', bit_depth=24, min_sample_rate=96000),
+    QualityTarget(label='FLAC 24/44.1', format='flac', bit_depth=24, min_sample_rate=44100),
+]
+
+
+def test_24bit_meets_24bit_target():
+    assert quality_meets_profile(FLAC_HI, WANT_24BIT) is True
+
+
+def test_16bit_does_not_meet_24bit_target():
+    assert quality_meets_profile(FLAC_CD, WANT_24BIT) is False
+
+
+def test_mp3_does_not_meet_flac_target():
+    assert quality_meets_profile(MP3, WANT_24BIT) is False
+
+
+def test_no_targets_means_no_constraint():
+    assert quality_meets_profile(FLAC_CD, []) is True
+
+
+def test_unprobeable_file_is_not_flagged():
+    # aq=None (probe failed) → don't act (avoid false re-downloads).
+    assert quality_meets_profile(None, WANT_24BIT) is True
+
+
+def test_targets_from_profile_reads_v3_ranked_targets():
+    profile = {
+        'version': 3,
+        'fallback_enabled': False,
+        'ranked_targets': [
+            {'label': 'FLAC 24/96', 'format': 'flac', 'bit_depth': 24, 'min_sample_rate': 96000},
+        ],
+    }
+    targets, fallback = targets_from_profile(profile)
+    assert [t.label for t in targets] == ['FLAC 24/96']
+    assert fallback is False
+
+
+def test_targets_from_profile_migrates_v2_qualities():
+    profile = {'qualities': {'flac': {'enabled': True, 'priority': 1}}}
+    targets, _ = targets_from_profile(profile)
+    assert any(t.format == 'flac' for t in targets)

web_server.py

@@ -15012,6 +15012,14 @@ def _get_file_path_from_template_raw(template: str, context: dict) -> tuple:
         return '', _sanitize_filename(full_path)
 
 
+def _probe_audio_quality(file_path):
+    """Probe real measured audio quality (bit depth / sample rate / bitrate)
+    as an AudioQuality, for the library quality scanner. Delegates to the same
+    core the download import guard uses. Returns None on any error."""
+    from core.imports.file_ops import probe_audio_quality
+    return probe_audio_quality(file_path)
+
+
 def _get_audio_quality_string(file_path):
     """
     Read audio file and return a quality descriptor string.
@@ -17653,6 +17661,7 @@ def _build_quality_scanner_deps():
         automation_engine=automation_engine,
         get_quality_tier_from_extension=_get_quality_tier_from_extension,
         add_activity_item=add_activity_item,
+        probe_audio_quality=_probe_audio_quality,
     )