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

Refactor quality scanner to respect primary metadata provider

Browse source 
- search metadata providers in source-priority order for each generated query instead of caching one client for the whole scan
- keep the quality-scanner worker provider-neutral and preserve the no-provider error path
- update the quality-scanner tests and remove the obsolete web_server spotify_client injection
This commit is contained in:
Antti Kettunen 2026-04-30 21:24:51 +03:00
parent a7ac4e48a4
commit c97a072f54
No known key found for this signature in database
GPG key ID: C6B2A3D250359BD7
3 changed files with 384 additions and 133 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

371
core/discovery/quality_scanner.py
View file

@ -3,7 +3,7 @@
`run_quality_scanner(scope, profile_id, deps)` is the function the
quality-scanner endpoint kicks off in a thread to scan the library
for low-quality tracks (below the user's configured quality profile)
and add their Spotify matches to the wishlist:
and add provider matches to the wishlist:
1. Reset scanner state, load quality profile + minimum acceptable tier.
2. Load tracks from DB based on scope:
@ -14,24 +14,16 @@ and add their Spotify matches to the wishlist:
- Quality-tier check via _get_quality_tier_from_extension(file_path).
- Skip tracks meeting standards (tier_num <= min_acceptable_tier).
- For low-quality tracks: matching_engine search query gen, score
candidates against Spotify (artist + title similarity, album-type
bonus), pick best match >= 0.7 confidence.
- On match: add full Spotify track to wishlist via
`wishlist_service.add_spotify_track_to_wishlist` with
candidates against the configured metadata source priority
(artist + title similarity, album-type bonus), pick best match >=
0.7 confidence.
- On match: add normalized track data to wishlist via
`wishlist_service.add_track_to_wishlist` with
source_type='quality_scanner' and a source_context that captures
original file_path, format tier, bitrate, and match confidence.
4. After all tracks: status='finished', progress=100, activity feed
entry, emit `quality_scan_completed` event for automation engine.
5. On critical exception: status='error', error message captured.
Note: This worker uses `wishlist_service` via its public
`add_spotify_track_to_wishlist` API only — it does not modify wishlist
internals. Safe to lift even before kettui's planned `core/wishlist/`
package extraction lands.
Lifted verbatim from web_server.py. Wide dependency surface (Spotify
client, matching engine, automation engine, quality state and lock,
quality-tier helper) all injected via `QualityScannerDeps`.
"""
from __future__ import annotations
@ -42,6 +34,9 @@ from dataclasses import dataclass
from datetime import datetime
from typing import Any, Callable
from core.metadata.registry import get_client_for_source, get_primary_source, get_source_priority
from core.wishlist.payloads import ensure_wishlist_track_format
logger = logging.getLogger(__name__)
@ -51,13 +46,145 @@ class QualityScannerDeps:
quality_scanner_state: dict
quality_scanner_lock: Any # threading.Lock
QUALITY_TIERS: dict
spotify_client: Any
matching_engine: Any
automation_engine: Any
get_quality_tier_from_extension: Callable
add_activity_item: Callable
def _extract_lookup_value(value: Any, *names: str, default: Any = None) -> Any:
if value is None:
return default
if isinstance(value, (str, bytes)):
return value
for name in names:
if isinstance(value, dict):
if name in value and value[name] is not None:
return value[name]
else:
candidate = getattr(value, name, None)
if candidate is not None:
return candidate
return default
def _normalize_track_artists(track_item: Any) -> list[dict]:
artists = _extract_lookup_value(track_item, 'artists', default=[]) or []
if isinstance(artists, (str, bytes)):
artists = [artists]
elif isinstance(artists, dict):
artists = [artists]
else:
try:
artists = list(artists)
except TypeError:
artists = [artists]
normalized = []
for artist in artists:
artist_name = _extract_lookup_value(artist, 'name', 'artist_name', 'title')
if not artist_name and isinstance(artist, (str, bytes)):
artist_name = artist
if artist_name:
normalized.append({'name': str(artist_name)})
if not normalized:
normalized.append({'name': 'Unknown Artist'})
return normalized
def _normalize_track_album(track_item: Any) -> dict:
album = _extract_lookup_value(track_item, 'album', default={})
if isinstance(album, dict):
album_data = dict(album)
else:
album_data = {
'name': _extract_lookup_value(album, 'name', 'title', default=str(album) if album else '') or '',
'images': _extract_lookup_value(album, 'images', default=[]) or [],
'album_type': _extract_lookup_value(album, 'album_type', default='album') or 'album',
'total_tracks': _extract_lookup_value(album, 'total_tracks', 'track_count', default=0) or 0,
'release_date': _extract_lookup_value(album, 'release_date', default='') or '',
}
album_data.setdefault('name', _extract_lookup_value(track_item, 'album_name', default='Unknown Album') or 'Unknown Album')
album_data.setdefault('album_type', _extract_lookup_value(track_item, 'album_type', default='album') or 'album')
album_data.setdefault('total_tracks', _extract_lookup_value(track_item, 'total_tracks', 'track_count', default=0) or 0)
album_data.setdefault('release_date', _extract_lookup_value(track_item, 'release_date', default='') or '')
if isinstance(album, dict):
album_data.setdefault('images', album.get('images', []) or [])
else:
album_data.setdefault('images', [])
album_data.setdefault('artists', _normalize_track_artists(track_item))
return album_data
def _normalize_track_match(track_item: Any, provider: str) -> dict:
track_data = {
'id': _extract_lookup_value(track_item, 'id', 'track_id', default='') or '',
'name': _extract_lookup_value(track_item, 'name', 'title', default='Unknown Track') or 'Unknown Track',
'artists': _normalize_track_artists(track_item),
'album': _normalize_track_album(track_item),
'duration_ms': _extract_lookup_value(track_item, 'duration_ms', default=0) or 0,
'track_number': _extract_lookup_value(track_item, 'track_number', default=1) or 1,
'disc_number': _extract_lookup_value(track_item, 'disc_number', default=1) or 1,
'preview_url': _extract_lookup_value(track_item, 'preview_url', default=None),
'external_urls': _extract_lookup_value(track_item, 'external_urls', default={}) or {},
'popularity': _extract_lookup_value(track_item, 'popularity', default=0) or 0,
'provider': provider,
'source': provider,
}
return ensure_wishlist_track_format(track_data)
def _track_name(track_item: Any) -> str:
return str(_extract_lookup_value(track_item, 'name', 'title', default='Unknown Track') or 'Unknown Track')
def _track_artist_names(track_item: Any) -> list[str]:
artists = _extract_lookup_value(track_item, 'artists', default=[]) or []
if isinstance(artists, (str, bytes)):
artists = [artists]
elif isinstance(artists, dict):
artists = [artists]
else:
try:
artists = list(artists)
except TypeError:
artists = [artists]
normalized = []
for artist in artists:
artist_name = _extract_lookup_value(artist, 'name', 'artist_name', 'title')
if not artist_name and isinstance(artist, (str, bytes)):
artist_name = artist
if artist_name:
normalized.append(str(artist_name))
return normalized
def _search_tracks_for_source(source: str, query: str, limit: int = 5, client: Any = None):
if client is None:
client = get_client_for_source(source)
if not client or not hasattr(client, 'search_tracks'):
return []
try:
if source == 'spotify':
return client.search_tracks(query, limit=limit, allow_fallback=False) or []
return client.search_tracks(query, limit=limit) or []
except TypeError:
try:
return client.search_tracks(query, limit=limit) or []
except Exception as exc:
logger.debug("Could not search %s for %s: %s", source, query, exc)
return []
except Exception as exc:
logger.debug("Could not search %s for %s: %s", source, query, exc)
return []
def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDeps = None):
"""Main quality scanner worker function"""
from core.wishlist_service import get_wishlist_service
@ -156,16 +283,23 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
deps.quality_scanner_state["total"] = total_tracks
deps.quality_scanner_state["phase"] = f"Scanning {total_tracks} tracks..."
# Use the module-level spotify_client (already authenticated with cached token)
if not deps.spotify_client or not deps.spotify_client.is_spotify_authenticated():
source_priority = get_source_priority(get_primary_source())
if not source_priority:
with deps.quality_scanner_lock:
deps.quality_scanner_state["status"] = "error"
deps.quality_scanner_state["phase"] = "Spotify not authenticated"
deps.quality_scanner_state["error_message"] = "Please authenticate with Spotify first"
logger.info("[Quality Scanner] Spotify not authenticated")
deps.quality_scanner_state["phase"] = "No metadata provider available"
deps.quality_scanner_state["error_message"] = "No metadata provider is available for quality scanning"
logger.info("[Quality Scanner] No metadata provider available")
return
logger.info("[Quality Scanner] Using metadata source priority: %s", source_priority)
wishlist_service = get_wishlist_service()
add_to_wishlist = getattr(wishlist_service, 'add_track_to_wishlist', None)
if add_to_wishlist is None:
add_to_wishlist = getattr(wishlist_service, 'add_spotify_track_to_wishlist', None)
if add_to_wishlist is None:
raise AttributeError("Wishlist service does not expose an add-to-wishlist method")
# Scan each track
for idx, track_row in enumerate(tracks_to_scan, 1):
@ -199,9 +333,11 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
logger.info(f"[Quality Scanner] Low quality: {artist_name} - {title} ({tier_name}, {file_path})")
# Attempt to match to Spotify using matching_engine
# Attempt to match using the active metadata provider
matched = False
matched_track_data = None
best_source = None
attempted_any_provider = False
try:
# Generate search queries using matching engine
@ -221,83 +357,99 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
for _query_idx, search_query in enumerate(search_queries):
try:
spotify_matches = deps.spotify_client.search_tracks(search_query, limit=5)
time.sleep(0.5) # Rate limit Spotify API calls
if not spotify_matches:
continue
# Score each result using matching engine
for spotify_track in spotify_matches:
try:
# Calculate artist confidence
artist_confidence = 0.0
if spotify_track.artists:
for result_artist in spotify_track.artists:
artist_sim = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(artist_name),
deps.matching_engine.normalize_string(result_artist)
)
artist_confidence = max(artist_confidence, artist_sim)
# Calculate title confidence
title_confidence = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(title),
deps.matching_engine.normalize_string(spotify_track.name)
)
# Combined confidence (50% artist + 50% title)
combined_confidence = (artist_confidence * 0.5 + title_confidence * 0.5)
# Small bonus for album tracks over singles
_at = getattr(spotify_track, 'album_type', None) or ''
if _at == 'album':
combined_confidence += 0.02
elif _at == 'ep':
combined_confidence += 0.01
logger.info(f"[Quality Scanner] Candidate: '{spotify_track.artists[0]}' - '{spotify_track.name}' (confidence: {combined_confidence:.3f})")
# Update best match if this is better
if combined_confidence > best_confidence and combined_confidence >= min_confidence:
best_confidence = combined_confidence
best_match = spotify_track
logger.info(f"[Quality Scanner] New best match: {spotify_track.artists[0]} - {spotify_track.name} (confidence: {combined_confidence:.3f})")
except Exception as e:
logger.error(f"[Quality Scanner] Error scoring result: {e}")
for source in source_priority:
client = get_client_for_source(source)
if not client or not hasattr(client, 'search_tracks'):
continue
# If we found a very high confidence match, stop searching
if best_confidence >= 0.9:
logger.info(f"[Quality Scanner] High confidence match found ({best_confidence:.3f}), stopping search")
break
attempted_any_provider = True
provider_matches = _search_tracks_for_source(source, search_query, limit=5, client=client)
time.sleep(0.5) # Rate limit metadata API calls
if not provider_matches:
continue
# Score each result using matching engine
for provider_track in provider_matches:
try:
# Calculate artist confidence
artist_confidence = 0.0
provider_artists = _track_artist_names(provider_track)
if provider_artists:
for result_artist in provider_artists:
artist_sim = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(artist_name),
deps.matching_engine.normalize_string(result_artist)
)
artist_confidence = max(artist_confidence, artist_sim)
# Calculate title confidence
title_confidence = deps.matching_engine.similarity_score(
deps.matching_engine.normalize_string(title),
deps.matching_engine.normalize_string(_track_name(provider_track))
)
# Combined confidence (50% artist + 50% title)
combined_confidence = (artist_confidence * 0.5 + title_confidence * 0.5)
# Small bonus for album tracks over singles
_at = _extract_lookup_value(provider_track, 'album_type', default='') or ''
if _at == 'album':
combined_confidence += 0.02
elif _at == 'ep':
combined_confidence += 0.01
candidate_artist = provider_artists[0] if provider_artists else 'Unknown Artist'
candidate_name = _track_name(provider_track)
logger.info(
f"[Quality Scanner] Candidate ({source}): '{candidate_artist}' - "
f"'{candidate_name}' (confidence: {combined_confidence:.3f})"
)
# Update best match if this is better
if combined_confidence > best_confidence and combined_confidence >= min_confidence:
best_confidence = combined_confidence
best_match = provider_track
best_source = source
logger.info(
f"[Quality Scanner] New best match ({source}): {candidate_artist} - "
f"{candidate_name} (confidence: {combined_confidence:.3f})"
)
except Exception as e:
logger.error(f"[Quality Scanner] Error scoring result: {e}")
continue
# If we found a very high confidence match, stop searching this query
if best_confidence >= 0.9:
logger.info(f"[Quality Scanner] High confidence match found ({best_confidence:.3f}), stopping search")
break
except Exception as e:
logger.debug(f"[Quality Scanner] Error searching with query '{search_query}': {e}")
continue
if not attempted_any_provider:
with deps.quality_scanner_lock:
deps.quality_scanner_state["status"] = "error"
deps.quality_scanner_state["phase"] = "No metadata provider available"
deps.quality_scanner_state["error_message"] = "No metadata provider is available for quality scanning"
logger.info("[Quality Scanner] No metadata provider available")
return
# Process best match
if best_match:
matched = True
logger.info(f"[Quality Scanner] Final match: {best_match.artists[0]} - {best_match.name} (confidence: {best_confidence:.3f})")
final_artist = _track_artist_names(best_match)[0] if _track_artist_names(best_match) else 'Unknown Artist'
final_name = _track_name(best_match)
final_source = best_source or 'metadata'
logger.info(
f"[Quality Scanner] Final match ({final_source}): {final_artist} - "
f"{final_name} (confidence: {best_confidence:.3f})"
)
# Build full Spotify track data for wishlist
matched_track_data = {
'id': best_match.id,
'name': best_match.name,
'artists': [{'name': artist} for artist in best_match.artists],
'album': {
'name': best_match.album,
'artists': [{'name': artist} for artist in best_match.artists],
'album_type': 'album', # Default to 'album' for quality scanner matches
'release_date': getattr(best_match, 'release_date', '') or ''
},
'duration_ms': best_match.duration_ms,
'popularity': best_match.popularity,
'preview_url': best_match.preview_url,
'external_urls': best_match.external_urls or {}
}
# Build normalized track data for wishlist
matched_track_data = _normalize_track_match(best_match, final_source)
# Add to wishlist
source_context = {
@ -309,8 +461,8 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
'scan_date': datetime.now().isoformat()
}
success = wishlist_service.add_spotify_track_to_wishlist(
spotify_track_data=matched_track_data,
success = add_to_wishlist(
track_data=matched_track_data,
failure_reason=f"Low quality - {tier_name.replace('_', ' ').title()} format",
source_type='quality_scanner',
source_context=source_context,
@ -324,29 +476,34 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
else:
logger.error(f"[Quality Scanner] Failed to add to wishlist: {artist_name} - {title}")
else:
logger.warning(f"[Quality Scanner] No suitable match found (best confidence: {best_confidence:.3f}, required: {min_confidence:.3f})")
logger.warning(
f"[Quality Scanner] No suitable metadata match found "
f"(best confidence: {best_confidence:.3f}, required: {min_confidence:.3f})"
)
except Exception as matching_error:
logger.error(f"[Quality Scanner] Matching error for {artist_name} - {title}: {matching_error}")
# Store result
result_entry = {
'track_id': track_id,
'title': title,
'artist': artist_name,
'album': album_title,
'file_path': file_path,
'current_format': tier_name,
'bitrate': bitrate,
'matched': matched,
'spotify_id': matched_track_data['id'] if matched_track_data else None
}
# Store result
result_entry = {
'track_id': track_id,
'title': title,
'artist': artist_name,
'album': album_title,
'file_path': file_path,
'current_format': tier_name,
'bitrate': bitrate,
'matched': matched,
'match_id': matched_track_data['id'] if matched_track_data else None,
'provider': best_source if matched else None,
'spotify_id': matched_track_data['id'] if matched_track_data else None,
}
with deps.quality_scanner_lock:
deps.quality_scanner_state["results"].append(result_entry)
with deps.quality_scanner_lock:
deps.quality_scanner_state["results"].append(result_entry)
if not matched:
logger.warning(f"[Quality Scanner] No Spotify match found for: {artist_name} - {title}")
if not matched:
logger.warning(f"[Quality Scanner] No metadata match found for: {artist_name} - {title}")
except Exception as track_error:
logger.error(f"[Quality Scanner] Error processing track: {track_error}")
@ -361,7 +518,7 @@ def run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDep
deps.quality_scanner_state["phase"] = "Scan complete"
logger.info(f"[Quality Scanner] Scan {'stopped' if was_stopped else 'complete'}: {deps.quality_scanner_state['processed']} processed, "
f"{deps.quality_scanner_state['low_quality']} low quality, {deps.quality_scanner_state['matched']} matched to Spotify")
f"{deps.quality_scanner_state['low_quality']} low quality, {deps.quality_scanner_state['matched']} matched to metadata providers")
# Add activity
deps.add_activity_item("", "Quality Scan Complete",

145
tests/discovery/test_discovery_quality_scanner.py
View file

@ -4,7 +4,6 @@ from __future__ import annotations
import threading
from dataclasses import dataclass
from unittest.mock import MagicMock
import pytest
@ -35,20 +34,30 @@ class _FakeSpotifyTrack:
self.external_urls = {}
class _FakeSpotifyClient:
def __init__(self, results=None, authenticated=True):
class _FakeMetadataClient:
def __init__(self, results=None):
self._results = results if results is not None else []
self._authenticated = authenticated
self.search_calls = []
def is_spotify_authenticated(self):
return self._authenticated
def search_tracks(self, query, limit=5):
self.search_calls.append((query, limit))
def search_tracks(self, query, limit=5, allow_fallback=True):
self.search_calls.append((query, limit, allow_fallback))
return self._results
_TEST_PRIMARY_SOURCE = 'spotify'
_TEST_SOURCE_CLIENTS = {}
@pytest.fixture(autouse=True)
def _patch_source_resolution(monkeypatch):
monkeypatch.setattr(qs, 'get_primary_source', lambda: _TEST_PRIMARY_SOURCE)
monkeypatch.setattr(qs, 'get_client_for_source', lambda source, **_kwargs: _TEST_SOURCE_CLIENTS.get(source))
monkeypatch.setattr(qs.time, 'sleep', lambda *_args, **_kwargs: None)
yield
_TEST_SOURCE_CLIENTS.clear()
globals()['_TEST_PRIMARY_SOURCE'] = 'spotify'
class _FakeMatchingEngine:
def generate_download_queries(self, t):
return [f"{t.artists[0]} {t.name}"]
@ -64,6 +73,14 @@ class _FakeMatchingEngine:
return 0.95 if a in b or b in a else 0.0
class _MultiQueryMatchingEngine(_FakeMatchingEngine):
def generate_download_queries(self, t):
return [
f"{t.artists[0]} {t.name} first",
f"{t.artists[0]} {t.name} second",
]
class _FakeAutomationEngine:
def __init__(self):
self.events = []
@ -125,16 +142,22 @@ class _WatchlistArtist:
def _build_deps(
*,
state=None,
spotify_results=None,
spotify_auth=True,
source_clients=None,
primary_source='spotify',
quality_tier_result=('lossless', 1),
automation=None,
):
globals()['_TEST_PRIMARY_SOURCE'] = primary_source
_TEST_SOURCE_CLIENTS.clear()
if source_clients is not None:
_TEST_SOURCE_CLIENTS.update(source_clients)
elif primary_source:
_TEST_SOURCE_CLIENTS[primary_source] = _FakeMetadataClient(results=[])
deps = qs.QualityScannerDeps(
quality_scanner_state=state if state is not None else {},
quality_scanner_lock=threading.Lock(),
QUALITY_TIERS={'lossless': {'tier': 1}, 'low_lossy': {'tier': 4}, 'lossy': {'tier': 3}},
spotify_client=_FakeSpotifyClient(results=spotify_results or [], authenticated=spotify_auth),
matching_engine=_FakeMatchingEngine(),
automation_engine=automation or _FakeAutomationEngine(),
get_quality_tier_from_extension=lambda fp: quality_tier_result,
@ -190,21 +213,21 @@ def test_no_watchlist_artists_short_circuit(mock_db_and_wishlist):
# ---------------------------------------------------------------------------
# Spotify auth gate
# Provider availability gate
# ---------------------------------------------------------------------------
def test_unauthenticated_spotify_marks_error(mock_db_and_wishlist):
"""Spotify not authenticated → state['status']='error'."""
def test_no_available_provider_marks_error(mock_db_and_wishlist):
"""No available metadata providers → state['status']='error'."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('A')]
db._tracks = [_track_row()]
state = {}
deps = _build_deps(state=state, spotify_auth=False)
deps = _build_deps(state=state, source_clients={}, quality_tier_result=('low_lossy', 4))
qs.run_quality_scanner('watchlist', 1, deps)
assert state['status'] == 'error'
assert 'authenticate' in state['error_message'].lower()
assert 'metadata provider' in state['error_message'].lower()
# ---------------------------------------------------------------------------
@ -228,19 +251,83 @@ def test_high_quality_tracks_skipped(mock_db_and_wishlist):
def test_low_quality_tracks_attempted(mock_db_and_wishlist):
"""Low-quality tracks (tier_num > min) trigger Spotify search."""
"""Low-quality tracks (tier_num > min) trigger a metadata search."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('Artist')]
db._tracks = [_track_row(file_path='/x.mp3', artist_name='Artist', title='Track')]
state = {}
match = _FakeSpotifyTrack(name='Track', artists=['Artist'])
deps = _build_deps(state=state, quality_tier_result=('low_lossy', 4),
spotify_results=[match])
spotify_client = _FakeMetadataClient(results=[match])
deps = _build_deps(
state=state,
quality_tier_result=('low_lossy', 4),
source_clients={'spotify': spotify_client},
primary_source='spotify',
)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['low_quality'] == 1
assert deps.spotify_client.search_calls # spotify queried
assert spotify_client.search_calls
assert spotify_client.search_calls[0][2] is False
def test_client_lookup_happens_per_query(mock_db_and_wishlist, monkeypatch):
"""Each generated query re-resolves the source client."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('Artist')]
db._tracks = [_track_row(file_path='/x.mp3', artist_name='Artist', title='Track')]
state = {}
spotify_client = _FakeMetadataClient(results=[])
lookups = []
monkeypatch.setattr(
qs,
'get_client_for_source',
lambda source, **_kwargs: (lookups.append(source), _TEST_SOURCE_CLIENTS.get(source))[1],
)
deps = _build_deps(
state=state,
quality_tier_result=('low_lossy', 4),
source_clients={'spotify': spotify_client},
primary_source='spotify',
)
deps.matching_engine = _MultiQueryMatchingEngine()
qs.run_quality_scanner('watchlist', 1, deps)
assert len(lookups) % 2 == 0
midpoint = len(lookups) // 2
assert lookups[:midpoint] == lookups[midpoint:]
assert len(spotify_client.search_calls) == 2
def test_low_quality_tracks_follow_source_priority(mock_db_and_wishlist):
"""Primary source is searched before Spotify."""
db, ws = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('Artist')]
db._tracks = [_track_row(file_path='/x.mp3', artist_name='Artist', title='Track')]
state = {}
match = _FakeSpotifyTrack(name='Track', artists=['Artist'])
deezer_client = _FakeMetadataClient(results=[match])
spotify_client = _FakeMetadataClient(results=[])
deps = _build_deps(
state=state,
source_clients={'deezer': deezer_client, 'spotify': spotify_client},
primary_source='deezer',
quality_tier_result=('low_lossy', 4),
)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['matched'] == 1
assert deezer_client.search_calls
assert spotify_client.search_calls == []
assert len(ws.added) == 1
add_args = ws.added[0]
assert add_args['track_data']['provider'] == 'deezer'
assert add_args['track_data']['source'] == 'deezer'
# ---------------------------------------------------------------------------
@ -254,8 +341,12 @@ def test_match_adds_to_wishlist(mock_db_and_wishlist):
db._tracks = [_track_row(artist_name='Artist', title='Track', file_path='/x.mp3', bitrate=128)]
state = {}
match = _FakeSpotifyTrack(name='Track', artists=['Artist'])
deps = _build_deps(state=state, quality_tier_result=('low_lossy', 4),
spotify_results=[match])
deps = _build_deps(
state=state,
quality_tier_result=('low_lossy', 4),
source_clients={'spotify': _FakeMetadataClient(results=[match])},
primary_source='spotify',
)
qs.run_quality_scanner('watchlist', 1, deps)
@ -273,8 +364,12 @@ def test_no_match_no_wishlist_add(mock_db_and_wishlist):
db._tracks = [_track_row(artist_name='A', title='Z', file_path='/x.mp3')]
state = {}
# No spotify results → no match
deps = _build_deps(state=state, quality_tier_result=('low_lossy', 4),
spotify_results=[])
deps = _build_deps(
state=state,
quality_tier_result=('low_lossy', 4),
source_clients={'spotify': _FakeMetadataClient(results=[])},
primary_source='spotify',
)
qs.run_quality_scanner('watchlist', 1, deps)

1
web_server.py
View file

@ -16563,7 +16563,6 @@ def _build_quality_scanner_deps():
quality_scanner_state=quality_scanner_state,
quality_scanner_lock=quality_scanner_lock,
QUALITY_TIERS=QUALITY_TIERS,
spotify_client=spotify_client,
matching_engine=matching_engine,
automation_engine=automation_engine,
get_quality_tier_from_extension=_get_quality_tier_from_extension,