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

PR5h: lift _run_quality_scanner to core/discovery/quality_scanner.py

Browse source 
Final lift in the PR5 discovery-workers series. Pulls the 328-line
library quality scanner out of `web_server.py` into its own focused
module under `core/discovery/`. Pure 1:1 lift — wrapper keeps the
original entry-point name.

What the quality scanner does:

1. Reset scanner state (counters, results), load quality profile +
   minimum acceptable tier from QUALITY_TIERS.
2. Load tracks from DB based on scope:
   - 'watchlist' → tracks for watchlisted artists only.
   - 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).
   - 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
     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.

Wishlist service interaction is via the public
`add_spotify_track_to_wishlist` API only — no overlap with kettui's
planned `core/wishlist/` package extraction (the import lives inside
the function, exactly as in the original, and will follow whatever
path that package takes).

Dependencies injected via `QualityScannerDeps` (8 fields) —
quality_scanner_state dict, quality_scanner_lock, QUALITY_TIERS
constant, spotify_client, matching_engine, automation_engine, plus 2
callable helpers (get_quality_tier_from_extension, add_activity_item).

Diff vs original after `deps.X` → global X normalization is **zero
differences** — 328 lines orig = 328 lines lifted, byte-identical body
(including all whitespace, comments, log strings, and the inline
`from core.wishlist_service import get_wishlist_service` /
`from database.music_database import MusicDatabase` imports at the
top of the function).

Tests: 11 new under tests/discovery/test_discovery_quality_scanner.py
covering state init/reset, no-watchlist-artists short-circuit,
unauthenticated Spotify error, high-quality skip, low-quality search
trigger, match → wishlist add (with full source_context payload),
no-match no-add, mid-loop stop request, completion phase + progress,
automation engine event emission, all-library scope load.

Full suite: 1152 passing (was 1141). Ruff clean.

End of the PR5 series — `web_server.py` lost ~328 lines on this commit
alone; total trim across PR5a–PR5h is ~2,400 lines of discovery worker
code moved into focused `core/discovery/*.py` modules. The remaining
discovery-adjacent worker `_process_watchlist_scan_automatically` was
deliberately deferred to avoid overlap with kettui's planned wishlist
extraction.
This commit is contained in:
Broque Thomas 2026-04-28 18:41:29 -07:00
parent 1c86d70386
commit a38bfcba55
3 changed files with 770 additions and 327 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

388
core/discovery/quality_scanner.py Normal file
View file

@ -0,0 +1,388 @@
"""Background worker for the library quality scanner.
`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:
1. Reset scanner state, load quality profile + minimum acceptable tier.
2. Load tracks from DB based on scope:
- 'watchlist' → tracks for watchlisted artists only.
- 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).
- 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
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
import logging
import time
from dataclasses import dataclass
from datetime import datetime
from typing import Any, Callable
logger = logging.getLogger(__name__)
@dataclass
class QualityScannerDeps:
"""Bundle of cross-cutting deps the quality scanner needs."""
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 run_quality_scanner(scope='watchlist', profile_id=1, deps: QualityScannerDeps = None):
"""Main quality scanner worker function"""
from core.wishlist_service import get_wishlist_service
from database.music_database import MusicDatabase
try:
with deps.quality_scanner_lock:
deps.quality_scanner_state["status"] = "running"
deps.quality_scanner_state["phase"] = "Initializing scan..."
deps.quality_scanner_state["progress"] = 0
deps.quality_scanner_state["processed"] = 0
deps.quality_scanner_state["total"] = 0
deps.quality_scanner_state["quality_met"] = 0
deps.quality_scanner_state["low_quality"] = 0
deps.quality_scanner_state["matched"] = 0
deps.quality_scanner_state["results"] = []
deps.quality_scanner_state["error_message"] = ""
logger.info(f"[Quality Scanner] Starting scan with scope: {scope}")
# Get database instance
db = MusicDatabase()
# Get quality profile to determine preferred quality
quality_profile = db.get_quality_profile()
preferred_qualities = quality_profile.get('qualities', {})
# 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}")
# Get tracks to scan based on scope
with deps.quality_scanner_lock:
deps.quality_scanner_state["phase"] = "Loading tracks from database..."
if scope == 'watchlist':
# Get watchlist artists
watchlist_artists = db.get_watchlist_artists(profile_id=profile_id)
if not watchlist_artists:
with deps.quality_scanner_lock:
deps.quality_scanner_state["status"] = "finished"
deps.quality_scanner_state["phase"] = "No watchlist artists found"
deps.quality_scanner_state["error_message"] = "Please add artists to watchlist first"
logger.warning("[Quality Scanner] No watchlist artists found")
return
# Get artist names from watchlist
artist_names = [artist.artist_name for artist in watchlist_artists]
logger.info(f"[Quality Scanner] Scanning {len(artist_names)} watchlist artists")
# Get all tracks for these artists by name
conn = db._get_connection()
placeholders = ','.join(['?' for _ in artist_names])
tracks_to_scan = conn.execute(
f"SELECT t.id, t.title, t.artist_id, t.album_id, t.file_path, t.bitrate, a.name as artist_name, al.title as album_title "
f"FROM tracks t "
f"JOIN artists a ON t.artist_id = a.id "
f"JOIN albums al ON t.album_id = al.id "
f"WHERE a.name IN ({placeholders}) AND t.file_path IS NOT NULL",
artist_names
).fetchall()
conn.close()
else:
# Scan all library tracks
with deps.quality_scanner_lock:
deps.quality_scanner_state["phase"] = "Loading all library tracks..."
conn = db._get_connection()
tracks_to_scan = conn.execute(
"SELECT t.id, t.title, t.artist_id, t.album_id, t.file_path, t.bitrate, a.name as artist_name, al.title as album_title "
"FROM tracks t "
"JOIN artists a ON t.artist_id = a.id "
"JOIN albums al ON t.album_id = al.id "
"WHERE t.file_path IS NOT NULL"
).fetchall()
conn.close()
total_tracks = len(tracks_to_scan)
logger.info(f"[Quality Scanner] Found {total_tracks} tracks to scan")
with deps.quality_scanner_lock:
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():
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")
return
wishlist_service = get_wishlist_service()
# Scan each track
for idx, track_row in enumerate(tracks_to_scan, 1):
# Check for stop request
if deps.quality_scanner_state.get('status') != 'running':
logger.info(f"[Quality Scanner] Stop requested, halting at track {idx}/{total_tracks}")
break
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)
# Update progress
with deps.quality_scanner_lock:
deps.quality_scanner_state["processed"] = idx
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:
# Quality met
with deps.quality_scanner_lock:
deps.quality_scanner_state["quality_met"] += 1
continue
# Low quality track found
with deps.quality_scanner_lock:
deps.quality_scanner_state["low_quality"] += 1
logger.info(f"[Quality Scanner] Low quality: {artist_name} - {title} ({tier_name}, {file_path})")
# Attempt to match to Spotify using matching_engine
matched = False
matched_track_data = None
try:
# Generate search queries using matching engine
temp_track = type('TempTrack', (), {
'name': title,
'artists': [artist_name],
'album': album_title
})()
search_queries = deps.matching_engine.generate_download_queries(temp_track)
logger.info(f"[Quality Scanner] Generated {len(search_queries)} search queries for {artist_name} - {title}")
# Find best match using confidence scoring
best_match = None
best_confidence = 0.0
min_confidence = 0.7 # Match existing standard
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}")
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
except Exception as e:
logger.debug(f"[Quality Scanner] Error searching with query '{search_query}': {e}")
continue
# 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})")
# 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 {}
}
# Add to wishlist
source_context = {
'quality_scanner': True,
'original_file_path': file_path,
'original_format': tier_name,
'original_bitrate': bitrate,
'match_confidence': best_confidence,
'scan_date': datetime.now().isoformat()
}
success = wishlist_service.add_spotify_track_to_wishlist(
spotify_track_data=matched_track_data,
failure_reason=f"Low quality - {tier_name.replace('_', ' ').title()} format",
source_type='quality_scanner',
source_context=source_context,
profile_id=profile_id
)
if success:
with deps.quality_scanner_lock:
deps.quality_scanner_state["matched"] += 1
logger.info(f"[Quality Scanner] Matched and added to wishlist: {artist_name} - {title}")
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})")
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
}
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}")
except Exception as track_error:
logger.error(f"[Quality Scanner] Error processing track: {track_error}")
continue
# Scan complete (don't overwrite if already stopped by user)
with deps.quality_scanner_lock:
was_stopped = deps.quality_scanner_state["status"] != "running"
deps.quality_scanner_state["status"] = "finished"
deps.quality_scanner_state["progress"] = 100
if not was_stopped:
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")
# Add activity
deps.add_activity_item("", "Quality Scan Complete",
f"{deps.quality_scanner_state['matched']} tracks added to wishlist", "Now")
try:
if deps.automation_engine:
deps.automation_engine.emit('quality_scan_completed', {
'quality_met': str(deps.quality_scanner_state.get('quality_met', 0)),
'low_quality': str(deps.quality_scanner_state.get('low_quality', 0)),
'total_scanned': str(deps.quality_scanner_state.get('processed', 0)),
})
except Exception:
pass
except Exception as e:
logger.error(f"[Quality Scanner] Critical error: {e}")
import traceback
traceback.print_exc()
with deps.quality_scanner_lock:
deps.quality_scanner_state["status"] = "error"
deps.quality_scanner_state["error_message"] = str(e)
deps.quality_scanner_state["phase"] = f"Error: {str(e)}"

361
tests/discovery/test_discovery_quality_scanner.py Normal file
View file

@ -0,0 +1,361 @@
"""Tests for core/discovery/quality_scanner.py — library quality scanner."""
from __future__ import annotations
import threading
from dataclasses import dataclass
from unittest.mock import MagicMock
import pytest
from core.discovery import quality_scanner as qs
# ---------------------------------------------------------------------------
# Fakes
# ---------------------------------------------------------------------------
@dataclass
class _FakeSpotifyTrack:
id: str = 'spt-1'
name: str = 'Found'
artists: list = None
album: str = 'Found Album'
duration_ms: int = 200000
popularity: int = 50
preview_url: str = ''
external_urls: dict = None
album_type: str = 'album'
release_date: str = '2024-01-01'
def __post_init__(self):
if self.artists is None:
self.artists = ['Found Artist']
if self.external_urls is None:
self.external_urls = {}
class _FakeSpotifyClient:
def __init__(self, results=None, authenticated=True):
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))
return self._results
class _FakeMatchingEngine:
def generate_download_queries(self, t):
return [f"{t.artists[0]} {t.name}"]
def normalize_string(self, s):
return (s or '').lower().strip()
def similarity_score(self, a, b):
if a == b:
return 1.0
if not a or not b:
return 0.0
return 0.95 if a in b or b in a else 0.0
class _FakeAutomationEngine:
def __init__(self):
self.events = []
def emit(self, event_type, data):
self.events.append((event_type, data))
class _FakeWishlistService:
def __init__(self):
self.added = []
def add_spotify_track_to_wishlist(self, **kwargs):
self.added.append(kwargs)
return True
class _FakeMusicDB:
def __init__(self, watchlist_artists=None, tracks=None, profile=None):
self._watchlist_artists = watchlist_artists if watchlist_artists is not None else []
self._tracks = tracks if tracks is not None else []
self._profile = profile or {'qualities': {'flac': {'enabled': True}}}
def get_quality_profile(self):
return self._profile
def get_watchlist_artists(self, profile_id=1):
return self._watchlist_artists
def _get_connection(self):
rows = self._tracks
return _FakeConn(rows)
class _FakeConn:
def __init__(self, rows):
self._rows = rows
def execute(self, query, params=None):
return _FakeCursor(self._rows)
def close(self):
pass
class _FakeCursor:
def __init__(self, rows):
self._rows = rows
def fetchall(self):
return self._rows
@dataclass
class _WatchlistArtist:
artist_name: str
def _build_deps(
*,
state=None,
spotify_results=None,
spotify_auth=True,
quality_tier_result=('lossless', 1),
automation=None,
):
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,
add_activity_item=lambda *a, **kw: None,
)
return deps
def _track_row(track_id=1, title='Track', artist_id=1, album_id=1,
file_path='/x.mp3', bitrate=128, artist_name='Artist',
album_title='Album'):
return (track_id, title, artist_id, album_id, file_path, bitrate, artist_name, album_title)
@pytest.fixture
def mock_db_and_wishlist(monkeypatch):
"""Patches MusicDatabase and get_wishlist_service used inside the worker."""
db = _FakeMusicDB()
ws = _FakeWishlistService()
monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db)
monkeypatch.setattr('core.wishlist_service.get_wishlist_service', lambda: ws)
return db, ws
# ---------------------------------------------------------------------------
# State init + DB load
# ---------------------------------------------------------------------------
def test_state_initialized_on_run(mock_db_and_wishlist):
"""Scanner resets state to running with cleared counters."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [] # no artists → exits early but after init
state = {}
deps = _build_deps(state=state)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['status'] == 'finished' # exited early since no artists
assert state['error_message'] == 'Please add artists to watchlist first'
def test_no_watchlist_artists_short_circuit(mock_db_and_wishlist):
"""Scope=watchlist with no artists → status=finished, error message."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = []
state = {}
deps = _build_deps(state=state)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['status'] == 'finished'
assert 'add artists' in state['error_message']
# ---------------------------------------------------------------------------
# Spotify auth gate
# ---------------------------------------------------------------------------
def test_unauthenticated_spotify_marks_error(mock_db_and_wishlist):
"""Spotify not authenticated → 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)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['status'] == 'error'
assert 'authenticate' in state['error_message'].lower()
# ---------------------------------------------------------------------------
# Quality tier check + skip
# ---------------------------------------------------------------------------
def test_high_quality_tracks_skipped(mock_db_and_wishlist):
"""Tracks meeting quality (tier_num <= min_acceptable) → quality_met += 1."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('A')]
db._tracks = [_track_row(file_path='/x.flac')]
state = {}
# Default min_acceptable is from {flac: enabled} → tier 1 (lossless)
# quality_tier_result=('lossless', 1) → 1 <= 1 → skip
deps = _build_deps(state=state, quality_tier_result=('lossless', 1))
qs.run_quality_scanner('watchlist', 1, deps)
assert state['quality_met'] == 1
assert state['low_quality'] == 0
def test_low_quality_tracks_attempted(mock_db_and_wishlist):
"""Low-quality tracks (tier_num > min) trigger Spotify 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])
qs.run_quality_scanner('watchlist', 1, deps)
assert state['low_quality'] == 1
assert deps.spotify_client.search_calls # spotify queried
# ---------------------------------------------------------------------------
# Match → wishlist add
# ---------------------------------------------------------------------------
def test_match_adds_to_wishlist(mock_db_and_wishlist):
"""High-confidence match → wishlist_service.add_spotify_track_to_wishlist called."""
db, ws = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('Artist')]
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])
qs.run_quality_scanner('watchlist', 1, deps)
assert state['matched'] == 1
assert len(ws.added) == 1
add_args = ws.added[0]
assert add_args['source_type'] == 'quality_scanner'
assert add_args['source_context']['original_file_path'] == '/x.mp3'
def test_no_match_no_wishlist_add(mock_db_and_wishlist):
"""No match found → no wishlist add, matched stays 0."""
db, ws = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('A')]
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=[])
qs.run_quality_scanner('watchlist', 1, deps)
assert state['matched'] == 0
assert ws.added == []
# ---------------------------------------------------------------------------
# Stop request gate
# ---------------------------------------------------------------------------
def test_stop_request_halts_loop(mock_db_and_wishlist):
"""Setting state['status'] != 'running' mid-loop halts processing."""
db, _ = 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))
# Override get_quality_tier_from_extension to set stop after first track
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:
state['status'] = 'stopping'
return ('lossless', 1)
deps.get_quality_tier_from_extension = stop_after_first
qs.run_quality_scanner('watchlist', 1, deps)
# Only first track processed
assert state['quality_met'] == 1
# ---------------------------------------------------------------------------
# Completion
# ---------------------------------------------------------------------------
def test_completion_marks_finished(mock_db_and_wishlist):
"""All tracks processed → status='finished', progress=100."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('A')]
db._tracks = [_track_row()]
state = {}
deps = _build_deps(state=state)
qs.run_quality_scanner('watchlist', 1, deps)
assert state['status'] == 'finished'
assert state['progress'] == 100
def test_automation_event_emitted(mock_db_and_wishlist):
"""Successful completion emits 'quality_scan_completed' on automation engine."""
db, _ = mock_db_and_wishlist
db._watchlist_artists = [_WatchlistArtist('A')]
db._tracks = [_track_row()]
automation = _FakeAutomationEngine()
state = {}
deps = _build_deps(state=state, automation=automation)
qs.run_quality_scanner('watchlist', 1, deps)
assert any(name == 'quality_scan_completed' for name, _ in automation.events)
# ---------------------------------------------------------------------------
# All-library scope
# ---------------------------------------------------------------------------
def test_scope_all_loads_all_tracks(mock_db_and_wishlist):
"""scope != 'watchlist' loads all tracks (no watchlist filter)."""
db, _ = mock_db_and_wishlist
db._tracks = [_track_row(track_id=1), _track_row(track_id=2)]
state = {}
deps = _build_deps(state=state)
qs.run_quality_scanner('all', 1, deps)
assert state['total'] == 2

348
web_server.py
View file

@ -20146,334 +20146,28 @@ def _get_quality_tier_from_extension(file_path):
return ('unknown', 999)
# Quality scanner worker logic lives in core/discovery/quality_scanner.py.
from core.discovery import quality_scanner as _discovery_quality_scanner
def _build_quality_scanner_deps():
"""Build the QualityScannerDeps bundle from web_server.py globals on each call."""
return _discovery_quality_scanner.QualityScannerDeps(
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,
add_activity_item=add_activity_item,
)
def _run_quality_scanner(scope='watchlist', profile_id=1):
"""Main quality scanner worker function"""
from core.wishlist_service import get_wishlist_service
from database.music_database import MusicDatabase
try:
with quality_scanner_lock:
quality_scanner_state["status"] = "running"
quality_scanner_state["phase"] = "Initializing scan..."
quality_scanner_state["progress"] = 0
quality_scanner_state["processed"] = 0
quality_scanner_state["total"] = 0
quality_scanner_state["quality_met"] = 0
quality_scanner_state["low_quality"] = 0
quality_scanner_state["matched"] = 0
quality_scanner_state["results"] = []
quality_scanner_state["error_message"] = ""
logger.info(f"[Quality Scanner] Starting scan with scope: {scope}")
# Get database instance
db = MusicDatabase()
# Get quality profile to determine preferred quality
quality_profile = db.get_quality_profile()
preferred_qualities = quality_profile.get('qualities', {})
# 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 = 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}")
# Get tracks to scan based on scope
with quality_scanner_lock:
quality_scanner_state["phase"] = "Loading tracks from database..."
if scope == 'watchlist':
# Get watchlist artists
watchlist_artists = db.get_watchlist_artists(profile_id=profile_id)
if not watchlist_artists:
with quality_scanner_lock:
quality_scanner_state["status"] = "finished"
quality_scanner_state["phase"] = "No watchlist artists found"
quality_scanner_state["error_message"] = "Please add artists to watchlist first"
logger.warning("[Quality Scanner] No watchlist artists found")
return
# Get artist names from watchlist
artist_names = [artist.artist_name for artist in watchlist_artists]
logger.info(f"[Quality Scanner] Scanning {len(artist_names)} watchlist artists")
# Get all tracks for these artists by name
conn = db._get_connection()
placeholders = ','.join(['?' for _ in artist_names])
tracks_to_scan = conn.execute(
f"SELECT t.id, t.title, t.artist_id, t.album_id, t.file_path, t.bitrate, a.name as artist_name, al.title as album_title "
f"FROM tracks t "
f"JOIN artists a ON t.artist_id = a.id "
f"JOIN albums al ON t.album_id = al.id "
f"WHERE a.name IN ({placeholders}) AND t.file_path IS NOT NULL",
artist_names
).fetchall()
conn.close()
else:
# Scan all library tracks
with quality_scanner_lock:
quality_scanner_state["phase"] = "Loading all library tracks..."
conn = db._get_connection()
tracks_to_scan = conn.execute(
"SELECT t.id, t.title, t.artist_id, t.album_id, t.file_path, t.bitrate, a.name as artist_name, al.title as album_title "
"FROM tracks t "
"JOIN artists a ON t.artist_id = a.id "
"JOIN albums al ON t.album_id = al.id "
"WHERE t.file_path IS NOT NULL"
).fetchall()
conn.close()
total_tracks = len(tracks_to_scan)
logger.info(f"[Quality Scanner] Found {total_tracks} tracks to scan")
with quality_scanner_lock:
quality_scanner_state["total"] = total_tracks
quality_scanner_state["phase"] = f"Scanning {total_tracks} tracks..."
# Use the module-level spotify_client (already authenticated with cached token)
if not spotify_client or not spotify_client.is_spotify_authenticated():
with quality_scanner_lock:
quality_scanner_state["status"] = "error"
quality_scanner_state["phase"] = "Spotify not authenticated"
quality_scanner_state["error_message"] = "Please authenticate with Spotify first"
logger.info("[Quality Scanner] Spotify not authenticated")
return
wishlist_service = get_wishlist_service()
# Scan each track
for idx, track_row in enumerate(tracks_to_scan, 1):
# Check for stop request
if quality_scanner_state.get('status') != 'running':
logger.info(f"[Quality Scanner] Stop requested, halting at track {idx}/{total_tracks}")
break
try:
track_id, title, artist_id, album_id, file_path, bitrate, artist_name, album_title = track_row
# Check quality tier
tier_name, tier_num = _get_quality_tier_from_extension(file_path)
# Update progress
with quality_scanner_lock:
quality_scanner_state["processed"] = idx
quality_scanner_state["progress"] = (idx / total_tracks) * 100
quality_scanner_state["phase"] = f"Scanning: {artist_name} - {title}"
# Check if meets quality standards
if tier_num <= min_acceptable_tier:
# Quality met
with quality_scanner_lock:
quality_scanner_state["quality_met"] += 1
continue
# Low quality track found
with quality_scanner_lock:
quality_scanner_state["low_quality"] += 1
logger.info(f"[Quality Scanner] Low quality: {artist_name} - {title} ({tier_name}, {file_path})")
# Attempt to match to Spotify using matching_engine
matched = False
matched_track_data = None
try:
# Generate search queries using matching engine
temp_track = type('TempTrack', (), {
'name': title,
'artists': [artist_name],
'album': album_title
})()
search_queries = matching_engine.generate_download_queries(temp_track)
logger.info(f"[Quality Scanner] Generated {len(search_queries)} search queries for {artist_name} - {title}")
# Find best match using confidence scoring
best_match = None
best_confidence = 0.0
min_confidence = 0.7 # Match existing standard
for _query_idx, search_query in enumerate(search_queries):
try:
spotify_matches = 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 = matching_engine.similarity_score(
matching_engine.normalize_string(artist_name),
matching_engine.normalize_string(result_artist)
)
artist_confidence = max(artist_confidence, artist_sim)
# Calculate title confidence
title_confidence = matching_engine.similarity_score(
matching_engine.normalize_string(title),
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}")
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
except Exception as e:
logger.debug(f"[Quality Scanner] Error searching with query '{search_query}': {e}")
continue
# 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})")
# 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 {}
}
# Add to wishlist
source_context = {
'quality_scanner': True,
'original_file_path': file_path,
'original_format': tier_name,
'original_bitrate': bitrate,
'match_confidence': best_confidence,
'scan_date': datetime.now().isoformat()
}
success = wishlist_service.add_spotify_track_to_wishlist(
spotify_track_data=matched_track_data,
failure_reason=f"Low quality - {tier_name.replace('_', ' ').title()} format",
source_type='quality_scanner',
source_context=source_context,
profile_id=profile_id
)
if success:
with quality_scanner_lock:
quality_scanner_state["matched"] += 1
logger.info(f"[Quality Scanner] Matched and added to wishlist: {artist_name} - {title}")
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})")
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
}
with quality_scanner_lock:
quality_scanner_state["results"].append(result_entry)
if not matched:
logger.warning(f"[Quality Scanner] No Spotify match found for: {artist_name} - {title}")
except Exception as track_error:
logger.error(f"[Quality Scanner] Error processing track: {track_error}")
continue
# Scan complete (don't overwrite if already stopped by user)
with quality_scanner_lock:
was_stopped = quality_scanner_state["status"] != "running"
quality_scanner_state["status"] = "finished"
quality_scanner_state["progress"] = 100
if not was_stopped:
quality_scanner_state["phase"] = "Scan complete"
logger.info(f"[Quality Scanner] Scan {'stopped' if was_stopped else 'complete'}: {quality_scanner_state['processed']} processed, "
f"{quality_scanner_state['low_quality']} low quality, {quality_scanner_state['matched']} matched to Spotify")
# Add activity
add_activity_item("", "Quality Scan Complete",
f"{quality_scanner_state['matched']} tracks added to wishlist", "Now")
try:
if automation_engine:
automation_engine.emit('quality_scan_completed', {
'quality_met': str(quality_scanner_state.get('quality_met', 0)),
'low_quality': str(quality_scanner_state.get('low_quality', 0)),
'total_scanned': str(quality_scanner_state.get('processed', 0)),
})
except Exception:
pass
except Exception as e:
logger.error(f"[Quality Scanner] Critical error: {e}")
import traceback
traceback.print_exc()
with quality_scanner_lock:
quality_scanner_state["status"] = "error"
quality_scanner_state["error_message"] = str(e)
quality_scanner_state["phase"] = f"Error: {str(e)}"
return _discovery_quality_scanner.run_quality_scanner(
scope, profile_id, _build_quality_scanner_deps()
)
def _run_duplicate_cleaner():