From fa29ee2195cc60d778db53410cc78d7b72be9fd0 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Tue, 28 Apr 2026 10:34:13 -0700 Subject: [PATCH] PR4h: lift _run_full_missing_tracks_process to core/downloads/master.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final extraction in the download orchestrator series. Lifts the 586-line master worker that drives the entire missing-tracks pipeline from `web_server.py` into `core/downloads/master.py`. Pure 1:1 lift — wrappers keep the original entry-point name so the three callers (`missing_download_executor.submit(_run_full_missing_tracks_process, ...)`) continue to work without changes. What the master worker does: 1. PHASE 1 ANALYSIS — per-track DB ownership check with album fast path (lookup album by name+artist, match tracks within it) plus a MusicBrainz release-cache preflight so per-track post-processing all uses the same release MBID (prevents Navidrome album splits). 2. Wishlist removal for tracks already in the library. 3. Explicit-content filter. 4. PHASE 2 transition — if nothing missing, mark batch complete, update per-source playlist phases, kick auto-wishlist completion handler. 5. Soulseek album pre-flight — search for a complete album folder before falling back to track-by-track search, cache the source for reuse. 6. Wishlist album grouping — derive per-album disc counts and resolve ONE artist context per album so collab albums don't fold-split. 7. Task creation with explicit album/artist context injection + playlist-folder-mode flag propagation. 8. Hand off to download_monitor + start_next_batch_of_downloads. 9. Error handler — phase=error, reset YouTube playlist phase to 'discovered', reset auto-wishlist globals on auto-initiated batches. Dependencies injected via `MasterDeps` (21 fields) — wide surface covering config, MB caches/locks, soulseek client, source-page state dicts, multiple callbacks (wishlist removal, explicit filter, executor + auto-completion fn, monitor, start_next_batch). The only behaviour difference from a pure paste is `import traceback` hoisted to module scope (was inline in the except block) — same behaviour. Trailing whitespace on two blank lines also got normalized away by the editor; neither has any runtime effect. `reset_wishlist_auto_processing` callback wraps the `global wishlist_auto_processing, wishlist_auto_processing_timestamp` write + `wishlist_timer_lock` since `global` can't reach back into web_server.py from a separate module. Tests: 21 new under tests/downloads/test_downloads_master.py covering analysis-phase state, force_download_all, found-track wishlist removal, explicit filter, no-missing complete + per-source state updates, auto wishlist completion submit, album fast path (direct + fallthrough), MB preflight (caches both keys, no-mb-worker no-op), task creation (queue + tasks dict, explicit context for albums, wishlist album grouping consistency, playlist folder mode), monitor + next-batch handoff, multi-disc total_discs computation, error handler (phase set, youtube reset, auto wishlist reset), and batch-removed-mid-flight defensive path. Full suite: 1050 passing (was 1029). Ruff clean. End of the PR4 series — `web_server.py` lost ~590 lines on this commit alone; total trim across PR4a–PR4h is ~2900 lines of orchestrator code moved into focused `core/downloads/*.py` modules. --- core/downloads/master.py | 648 +++++++++++++++++++++ tests/downloads/test_downloads_master.py | 684 +++++++++++++++++++++++ web_server.py | 623 ++------------------- 3 files changed, 1372 insertions(+), 583 deletions(-) create mode 100644 core/downloads/master.py create mode 100644 tests/downloads/test_downloads_master.py diff --git a/core/downloads/master.py b/core/downloads/master.py new file mode 100644 index 00000000..9a449b9c --- /dev/null +++ b/core/downloads/master.py @@ -0,0 +1,648 @@ +"""Master worker for the missing-tracks download workflow. + +`run_full_missing_tracks_process(batch_id, playlist_id, tracks_json, deps)` is +the single 580-line worker that orchestrates the entire pipeline: + + 1. PHASE 1 — Analysis: per-track DB ownership check, with album fast path + (lookup album by name+artist, match tracks within it) plus a + MusicBrainz release-cache preflight so per-track post-processing all + uses the same release MBID (prevents Navidrome album splits). + 2. Wishlist removal for tracks already in the library. + 3. Explicit-content filter. + 4. PHASE 2 transition — if nothing missing, mark batch complete, update + per-source playlist phases, kick auto-wishlist completion handler. + 5. Soulseek album pre-flight — search for a complete album folder before + falling back to track-by-track search, cache the source for reuse. + 6. Wishlist album grouping — derive per-album disc counts and resolve + ONE artist context per album so collab albums don't fold-split. + 7. Task creation with explicit album/artist context injection. + 8. Hand off to download monitor + start_next_batch_of_downloads. + +Lifted verbatim from web_server.py. Wide dependency surface (config, MB +caches, Soulseek client, source-page state dicts, multiple helper funcs) +all injected via `MasterDeps`. +""" + +from __future__ import annotations + +import json +import logging +import re +import time +import traceback +import uuid +from dataclasses import dataclass +from typing import Any, Callable + +from core.runtime_state import download_batches, download_tasks, tasks_lock + +logger = logging.getLogger(__name__) + + +@dataclass +class MasterDeps: + """Bundle of cross-cutting deps the master worker needs.""" + config_manager: Any + soulseek_client: Any + run_async: Callable[..., Any] + mb_worker: Any + mb_release_cache: dict + mb_release_cache_lock: Any + mb_release_detail_cache: dict + mb_release_detail_cache_lock: Any + normalize_album_cache_key: Callable[[str], str] + check_and_remove_track_from_wishlist_by_metadata: Callable + is_explicit_blocked: Callable + youtube_playlist_states: dict + tidal_discovery_states: dict + deezer_discovery_states: dict + spotify_public_discovery_states: dict + missing_download_executor: Any + process_failed_tracks_to_wishlist_exact_with_auto_completion: Callable + source_reuse_logger: Any + download_monitor: Any + start_next_batch_of_downloads: Callable[[str], None] + reset_wishlist_auto_processing: Callable[[], None] + + +def run_full_missing_tracks_process(batch_id, playlist_id, tracks_json, deps: MasterDeps): + """ + A master worker that handles the entire missing tracks process: + 1. Runs the analysis. + 2. If missing tracks are found, it automatically queues them for download. + """ + try: + # PHASE 1: ANALYSIS + with tasks_lock: + if batch_id in download_batches: + download_batches[batch_id]['phase'] = 'analysis' + download_batches[batch_id]['analysis_total'] = len(tracks_json) + download_batches[batch_id]['analysis_processed'] = 0 + + from database.music_database import MusicDatabase + db = MusicDatabase() + active_server = deps.config_manager.get_active_media_server() + analysis_results = [] + + # Get force download flag and album context from batch + force_download_all = False + batch_album_context = None + batch_artist_context = None + batch_is_album = False + with tasks_lock: + if batch_id in download_batches: + force_download_all = download_batches[batch_id].get('force_download_all', False) + batch_is_album = download_batches[batch_id].get('is_album_download', False) + batch_album_context = download_batches[batch_id].get('album_context') + batch_artist_context = download_batches[batch_id].get('artist_context') + + if force_download_all: + logger.warning(f"[Force Download] Force download mode enabled for batch {batch_id} - treating all tracks as missing") + + # Allow duplicate tracks across albums — when enabled, only skip tracks already + # owned in THIS album, not tracks owned in other albums + allow_duplicates = deps.config_manager.get('wishlist.allow_duplicate_tracks', True) + if allow_duplicates and batch_is_album: + logger.info("[Duplicates] Allow duplicate tracks enabled — only checking ownership within target album") + + # PREFLIGHT: Pre-populate MusicBrainz release cache for album downloads. + # This ensures ALL tracks in the album use the same release MBID during + # per-track post-processing, preventing Navidrome album splits. + if batch_is_album and batch_album_context and batch_artist_context: + try: + album_name_pf = batch_album_context.get('name', '') + artist_name_pf = batch_artist_context.get('name', '') + if album_name_pf and artist_name_pf: + mb_svc = deps.mb_worker.mb_service if deps.mb_worker else None + if mb_svc: + from core.album_consistency import _find_best_release + release = _find_best_release(album_name_pf, artist_name_pf, len(tracks_json), mb_svc) + if release and release.get('id'): + release_mbid = release['id'] + _artist_key = artist_name_pf.lower().strip() + _rc_key_norm = (deps.normalize_album_cache_key(album_name_pf), _artist_key) + _rc_key_exact = (album_name_pf.lower().strip(), _artist_key) + with deps.mb_release_cache_lock: + deps.mb_release_cache[_rc_key_norm] = release_mbid + deps.mb_release_cache[_rc_key_exact] = release_mbid + # Also cache the full release detail for tag extraction + with deps.mb_release_detail_cache_lock: + deps.mb_release_detail_cache[release_mbid] = release + logger.info(f"[Preflight] Pre-cached MB release for '{album_name_pf}': " + f"'{release.get('title', '')}' ({release_mbid[:8]}...)") + else: + logger.warning(f"[Preflight] No MB release found for '{album_name_pf}' — per-track lookup will be used") + except Exception as pf_err: + logger.error(f"[Preflight] MB release preflight failed: {pf_err}") + + # ALBUM FAST PATH: If this is an album download, try to find the album in the DB first + # and match tracks within it — faster and more accurate than N global searches + album_tracks_map = {} # Maps normalized title -> DatabaseTrack for album-scoped matching + if batch_is_album and batch_album_context and batch_artist_context and not force_download_all: + album_name = batch_album_context.get('name', '') + artist_name = batch_artist_context.get('name', '') + total_tracks = batch_album_context.get('total_tracks', 0) + if album_name and artist_name: + try: + db_album, album_confidence = db.check_album_exists_with_editions( + title=album_name, artist=artist_name, + confidence_threshold=0.7, + expected_track_count=total_tracks if total_tracks > 0 else None, + server_source=active_server + ) + if db_album and album_confidence >= 0.7: + db_album_tracks = db.get_tracks_by_album(db_album.id) + for t in db_album_tracks: + album_tracks_map[t.title.lower().strip()] = t + logger.info(f"[Album Analysis] Found album '{db_album.title}' in DB with {len(db_album_tracks)} tracks (confidence: {album_confidence:.2f})") + else: + logger.warning(f"[Album Analysis] Album '{album_name}' not found in DB — falling back to per-track search") + except Exception as album_err: + logger.error(f"[Album Analysis] Album lookup error: {album_err} — falling back to per-track search") + + for i, track_data in enumerate(tracks_json): + # Use original table index if provided (for partial track selection), + # otherwise fall back to enumeration index + track_index = track_data.get('_original_index', i) + track_name = track_data.get('name', '') + artists = track_data.get('artists', []) + found, confidence = False, 0.0 + + # Skip database check if force download is enabled + if force_download_all: + logger.warning(f"[Force Download] Skipping database check for '{track_name}' - treating as missing") + found, confidence = False, 0.0 + elif album_tracks_map: + # Album-scoped matching: check against known album tracks first + track_name_lower = track_name.lower().strip() + # Direct title match + if track_name_lower in album_tracks_map: + found, confidence = True, 1.0 + else: + # Fuzzy match against album tracks using string similarity + best_sim = 0.0 + for db_title_lower, _db_track in album_tracks_map.items(): + sim = db._string_similarity(track_name_lower, db_title_lower) + if sim > best_sim: + best_sim = sim + if best_sim >= 0.7: + found, confidence = True, best_sim + else: + # Fall back to global per-track search for this track + # When allow_duplicates is on for album downloads, skip global + # search — the track isn't in THIS album so treat as missing + if allow_duplicates and batch_is_album: + found, confidence = False, 0.0 + else: + _fallback_album = batch_album_context.get('name') if batch_album_context else None + for artist in artists: + if isinstance(artist, str): + artist_name = artist + elif isinstance(artist, dict) and 'name' in artist: + artist_name = artist['name'] + else: + artist_name = str(artist) + db_track, track_confidence = db.check_track_exists( + track_name, artist_name, confidence_threshold=0.7, server_source=active_server, album=_fallback_album + ) + if db_track and track_confidence >= 0.7: + found, confidence = True, track_confidence + break + elif allow_duplicates and batch_is_album: + # Allow duplicates + album download + album not in DB yet → treat all as missing + found, confidence = False, 0.0 + else: + # Non-album download (playlist/single track) — always check global + for artist in artists: + # Handle both string format and Spotify API format {'name': 'Artist Name'} + if isinstance(artist, str): + artist_name = artist + elif isinstance(artist, dict) and 'name' in artist: + artist_name = artist['name'] + else: + artist_name = str(artist) + db_track, track_confidence = db.check_track_exists( + track_name, artist_name, confidence_threshold=0.7, server_source=active_server + ) + if db_track and track_confidence >= 0.7: + found, confidence = True, track_confidence + break + + analysis_results.append({ + 'track_index': track_index, 'track': track_data, 'found': found, 'confidence': confidence + }) + + # WISHLIST REMOVAL: If track is found in database, check if it should be removed from wishlist + if found and confidence >= 0.7: + try: + deps.check_and_remove_track_from_wishlist_by_metadata(track_data) + except Exception as wishlist_error: + logger.error(f"[Analysis] Error checking wishlist removal for found track: {wishlist_error}") + + with tasks_lock: + if batch_id in download_batches: + download_batches[batch_id]['analysis_processed'] = i + 1 + # Store incremental results for live updates + download_batches[batch_id]['analysis_results'] = analysis_results.copy() + + missing_tracks = [res for res in analysis_results if not res['found']] + + # Filter explicit tracks if content filter is enabled + if not deps.config_manager.get('content_filter.allow_explicit', True): + before_count = len(missing_tracks) + missing_tracks = [res for res in missing_tracks if not deps.is_explicit_blocked(res.get('track', {}))] + skipped = before_count - len(missing_tracks) + if skipped > 0: + logger.warning(f"[Content Filter] Filtered out {skipped} explicit track(s) from download queue") + + with tasks_lock: + if batch_id in download_batches: + download_batches[batch_id]['analysis_results'] = analysis_results + + # PHASE 2: TRANSITION TO DOWNLOAD (if necessary) + if not missing_tracks: + logger.warning(f"Analysis for batch {batch_id} complete. No missing tracks.") + + # Record sync history — all tracks found, nothing to download + tracks_found = sum(1 for r in analysis_results if r.get('found')) + try: + db_sh = MusicDatabase() + db_sh.update_sync_history_completion(batch_id, tracks_found=tracks_found, tracks_downloaded=0, tracks_failed=0) + # Save per-track results (all found, no downloads) + track_results = [] + for res in analysis_results: + td = res.get('track', {}) + artists = td.get('artists', []) + first_artist = (artists[0].get('name', artists[0]) if isinstance(artists[0], dict) else str(artists[0])) if artists else '' + alb = td.get('album', '') + # Extract image + _img = '' + _alb_obj = td.get('album', {}) + if isinstance(_alb_obj, dict): + _alb_imgs = _alb_obj.get('images', []) + if _alb_imgs and isinstance(_alb_imgs, list) and len(_alb_imgs) > 0: + _img = _alb_imgs[0].get('url', '') if isinstance(_alb_imgs[0], dict) else '' + track_results.append({ + 'index': res.get('track_index', 0), + 'name': td.get('name', ''), + 'artist': first_artist, + 'album': alb.get('name', '') if isinstance(alb, dict) else str(alb or ''), + 'image_url': _img, + 'duration_ms': td.get('duration_ms', 0), + 'source_track_id': td.get('id', ''), + 'status': 'found' if res.get('found') else 'not_found', + 'confidence': round(res.get('confidence', 0.0), 3), + 'matched_track': None, + 'download_status': None, + }) + if track_results: + db_sh.update_sync_history_track_results(batch_id, json.dumps(track_results)) + except Exception: + pass + + is_auto_batch = False + with tasks_lock: + if batch_id in download_batches: + is_auto_batch = download_batches[batch_id].get('auto_initiated', False) + download_batches[batch_id]['phase'] = 'complete' + download_batches[batch_id]['completion_time'] = time.time() # Track for auto-cleanup + + # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist + if playlist_id.startswith('youtube_'): + url_hash = playlist_id.replace('youtube_', '') + if url_hash in deps.youtube_playlist_states: + deps.youtube_playlist_states[url_hash]['phase'] = 'download_complete' + logger.warning(f"Updated YouTube playlist {url_hash} to download_complete phase (no missing tracks)") + + # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist + if playlist_id.startswith('tidal_'): + tidal_playlist_id = playlist_id.replace('tidal_', '') + if tidal_playlist_id in deps.tidal_discovery_states: + deps.tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete' + logger.warning(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase (no missing tracks)") + + # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist + if playlist_id.startswith('deezer_'): + deezer_playlist_id = playlist_id.replace('deezer_', '') + if deezer_playlist_id in deps.deezer_discovery_states: + deps.deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete' + logger.warning(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase (no missing tracks)") + + # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist + if playlist_id.startswith('spotify_public_'): + spotify_public_url_hash = playlist_id.replace('spotify_public_', '') + if spotify_public_url_hash in deps.spotify_public_discovery_states: + deps.spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete' + logger.warning(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase (no missing tracks)") + + # Handle auto-initiated wishlist completion even when no missing tracks + if is_auto_batch and playlist_id == 'wishlist': + logger.warning("[Auto-Wishlist] No missing tracks found - calling auto-completion handler to toggle cycle and reschedule") + deps.missing_download_executor.submit(deps.process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id) + + return + + logger.warning(f" transitioning batch {batch_id} to download phase with {len(missing_tracks)} tracks.") + + # Read batch context (quick lock) before doing any network I/O + with tasks_lock: + if batch_id not in download_batches: return + batch = download_batches[batch_id] + batch_album_context = batch.get('album_context') + batch_artist_context = batch.get('artist_context') + batch_is_album = batch.get('is_album_download', False) + batch_playlist_folder_mode = batch.get('playlist_folder_mode', False) + batch_playlist_name = batch.get('playlist_name', 'Unknown Playlist') + + # === ALBUM PRE-FLIGHT: Search for complete album folder before track-by-track === + # Only run pre-flight when Soulseek is the download source (or hybrid with soulseek) + preflight_source = None + preflight_tracks = None + dl_source_mode = deps.config_manager.get('download_source.mode', 'hybrid') + _dl_hybrid_order = deps.config_manager.get('download_source.hybrid_order', ['hifi', 'youtube', 'soulseek']) + _dl_hybrid_first = _dl_hybrid_order[0] if _dl_hybrid_order else deps.config_manager.get('download_source.hybrid_primary', 'hifi') + soulseek_is_source = dl_source_mode == 'soulseek' or ( + dl_source_mode == 'hybrid' and _dl_hybrid_first == 'soulseek' + ) + if batch_is_album and batch_album_context and batch_artist_context and soulseek_is_source: + artist_name = batch_artist_context.get('name', '') + album_name = batch_album_context.get('name', '') + if artist_name and album_name: + try: + _sr = deps.source_reuse_logger + _sr.info(f"[Album Pre-flight] Searching for '{artist_name} {album_name}'") + logger.info(f"[Album Pre-flight] Searching Soulseek for complete album: '{artist_name} - {album_name}'") + + slsk = deps.soulseek_client.soulseek if hasattr(deps.soulseek_client, 'soulseek') else deps.soulseek_client + + # Try multiple query variations (banned keywords in artist/album name can return 0 results) + album_queries = [f"{artist_name} {album_name}"] + # Clean artist name (remove feat., parentheticals) + clean_artist = re.sub(r'\s*\(.*?\)', '', artist_name).strip() + clean_artist = re.sub(r'\s*(feat\.?|ft\.?|featuring)\s+.*$', '', clean_artist, flags=re.IGNORECASE).strip() + if clean_artist != artist_name: + album_queries.append(f"{clean_artist} {album_name}") + # Album name only (some users file by album) + album_queries.append(album_name) + + album_results = [] + track_results = [] + for aq in album_queries: + _sr.info(f"[Album Pre-flight] Trying query: '{aq}'") + track_results, album_results = deps.run_async(slsk.search(aq, timeout=30)) + if album_results: + _sr.info(f"[Album Pre-flight] Found {len(album_results)} album results with query: '{aq}'") + break + _sr.info(f"[Album Pre-flight] No album results for query: '{aq}'") + + if album_results: + # Filter by quality preference + quality_filtered = [] + for ar in album_results: + filtered_tracks = slsk.filter_results_by_quality_preference(ar.tracks) + if filtered_tracks: + quality_filtered.append((ar, len(filtered_tracks))) + + if quality_filtered: + # Sort by track count (most complete album first), then quality score + quality_filtered.sort(key=lambda x: (x[1], x[0].quality_score), reverse=True) + best_album = quality_filtered[0][0] + + _sr.info(f"[Album Pre-flight] Best album result: {best_album.username}:{best_album.album_path} " + f"({best_album.track_count} tracks, quality={best_album.dominant_quality})") + logger.info(f"[Album Pre-flight] Found album folder: {best_album.username} — " + f"{best_album.track_count} tracks ({best_album.dominant_quality})") + + # Browse the user's folder to get all tracks (may have more than search returned) + browse_files = deps.run_async(slsk.browse_user_directory(best_album.username, best_album.album_path)) + if browse_files: + folder_tracks = slsk.parse_browse_results_to_tracks( + best_album.username, browse_files, directory=best_album.album_path + ) + if folder_tracks: + preflight_source = { + 'username': best_album.username, + 'folder_path': best_album.album_path + } + preflight_tracks = folder_tracks + _sr.info(f"[Album Pre-flight] Browsed folder: {len(folder_tracks)} audio tracks available") + logger.info(f"[Album Pre-flight] Cached {len(folder_tracks)} tracks from {best_album.username} for source reuse") + else: + _sr.info("[Album Pre-flight] Browse returned files but no audio tracks") + else: + # Browse failed — fall back to using the search result tracks directly + _sr.info("[Album Pre-flight] Browse failed, using search result tracks directly") + preflight_source = { + 'username': best_album.username, + 'folder_path': best_album.album_path + } + preflight_tracks = best_album.tracks + logger.info(f"[Album Pre-flight] Using {len(best_album.tracks)} tracks from search results (browse unavailable)") + else: + _sr.info("[Album Pre-flight] No album results passed quality filter") + logger.warning("[Album Pre-flight] No album results matched quality preferences") + else: + _sr.info(f"[Album Pre-flight] Search returned no album results (got {len(track_results)} individual tracks)") + logger.warning("[Album Pre-flight] No complete album folders found, falling back to track-by-track search") + + except Exception as preflight_err: + logger.error(f"[Album Pre-flight] Search failed (non-fatal, falling back to track-by-track): {preflight_err}") + deps.source_reuse_logger.info(f"[Album Pre-flight] Exception: {preflight_err}") + + with tasks_lock: + if batch_id not in download_batches: return + + download_batches[batch_id]['phase'] = 'downloading' + + # Store album pre-flight results on batch for source reuse + if preflight_source and preflight_tracks: + download_batches[batch_id]['last_good_source'] = preflight_source + download_batches[batch_id]['source_folder_tracks'] = preflight_tracks + download_batches[batch_id]['failed_sources'] = set() + logger.info(f"[Album Pre-flight] Pre-loaded source reuse data on batch {batch_id}") + + # Compute total_discs for multi-disc album subfolder support + # Use ALL tracks (tracks_json), not just missing ones, to correctly detect multi-disc + # even when only one disc has missing tracks + if batch_is_album and batch_album_context: + total_discs = max((t.get('disc_number', 1) for t in tracks_json), default=1) + batch_album_context['total_discs'] = total_discs + if total_discs > 1: + logger.info(f"[Multi-Disc] Detected {total_discs} discs for album '{batch_album_context.get('name')}'") + + # Pre-compute per-album data for wishlist tracks (grouped by album ID) + # Wishlist tracks aren't batch_is_album but each track has disc_number in spotify_data + wishlist_album_disc_counts = {} + wishlist_album_artist_map = {} # album_id -> resolved artist context (consistent per album) + if playlist_id == 'wishlist': + import json as _json + # First pass: collect disc_number and resolve ONE artist per album + for t in tracks_json: + sp_data = t.get('spotify_data', {}) + if isinstance(sp_data, str): + try: + sp_data = _json.loads(sp_data) + except: + sp_data = {} + album_val = sp_data.get('album') + album_id = album_val.get('id') if isinstance(album_val, dict) else album_val if isinstance(album_val, str) else None + # Fallback album key: use album name when ID is missing (e.g. mirrored playlist tracks) + if not album_id and isinstance(album_val, dict) and album_val.get('name'): + album_id = f"_name_{album_val['name'].lower().strip()}" + disc_num = sp_data.get('disc_number', t.get('disc_number', 1)) + if album_id: + wishlist_album_disc_counts[album_id] = max( + wishlist_album_disc_counts.get(album_id, 1), disc_num + ) + # Resolve album-level artist once per album (first track wins) + if album_id not in wishlist_album_artist_map: + _wl_source = t.get('source_info') or {} + if isinstance(_wl_source, str): + try: + _wl_source = _json.loads(_wl_source) + except: + _wl_source = {} + _wl_album = album_val if isinstance(album_val, dict) else {} + _wl_album_artists = _wl_album.get('artists', []) + # Priority: watchlist artist > album artists > track artists + if _wl_source.get('watchlist_artist_name'): + wishlist_album_artist_map[album_id] = { + 'name': _wl_source['watchlist_artist_name'], + 'id': _wl_source.get('watchlist_artist_id', '') + } + elif _wl_source.get('artist_name'): + wishlist_album_artist_map[album_id] = {'name': _wl_source['artist_name']} + elif _wl_album_artists: + _fa = _wl_album_artists[0] + wishlist_album_artist_map[album_id] = _fa if isinstance(_fa, dict) else {'name': str(_fa)} + else: + _wl_track_artists = sp_data.get('artists', []) + if _wl_track_artists: + _fa = _wl_track_artists[0] + wishlist_album_artist_map[album_id] = _fa if isinstance(_fa, dict) else {'name': str(_fa)} + else: + # Try top-level 'artists' (wishlist format uses plural) + _tl_artists = t.get('artists', []) + if _tl_artists: + _tla = _tl_artists[0] + _fallback_name = _tla.get('name', str(_tla)) if isinstance(_tla, dict) else str(_tla) + else: + _fallback_name = t.get('artist', '') + wishlist_album_artist_map[album_id] = {'name': _fallback_name or 'Unknown Artist'} + logger.info(f"[Wishlist Album Grouping] Album '{_wl_album.get('name', album_id)}' → artist: '{wishlist_album_artist_map[album_id].get('name', '?')}'") + + + + for res in missing_tracks: + task_id = str(uuid.uuid4()) + track_info = res['track'].copy() + + # Add explicit album context to track_info for artist album downloads + if batch_is_album and batch_album_context and batch_artist_context: + track_info['_explicit_album_context'] = batch_album_context + track_info['_explicit_artist_context'] = batch_artist_context + track_info['_is_explicit_album_download'] = True + logger.info(f"[Task Creation] Added explicit album context for: {track_info.get('name')}") + + # SPECIAL WISHLIST HANDLING: Inject album context if available to force grouping + elif playlist_id == 'wishlist': + # Extract spotify_data again since it might be buried + spotify_data = track_info.get('spotify_data') + if isinstance(spotify_data, str): + try: + spotify_data = json.loads(spotify_data) + except: + spotify_data = {} + + if not spotify_data: + spotify_data = {} + + s_album = spotify_data.get('album') or {} + if isinstance(s_album, str): + s_album = {'name': s_album} # Normalize string album to dict + s_artists = spotify_data.get('artists', []) + + # We need at least an album name and artist + if s_album and isinstance(s_album, dict) and s_album.get('name'): + # Use pre-computed album-level artist for folder consistency. + # All tracks from the same album get the same artist context, + # preventing folder splits on collab albums (KPOP Demon Hunters, etc.) + album_id_for_lookup = s_album.get('id') + # Fallback album key: match first-pass logic for missing IDs + if not album_id_for_lookup and s_album.get('name'): + album_id_for_lookup = f"_name_{s_album['name'].lower().strip()}" + if not album_id_for_lookup: + album_id_for_lookup = 'wishlist_album' + artist_ctx = wishlist_album_artist_map.get(album_id_for_lookup, {}) + if not artist_ctx or not artist_ctx.get('name'): + # Fallback: per-track resolution from artists array + _fb_artists = track_info.get('artists', []) + if _fb_artists: + _fb_a = _fb_artists[0] + _fb_name = _fb_a.get('name', str(_fb_a)) if isinstance(_fb_a, dict) else str(_fb_a) + else: + _fb_name = track_info.get('artist', '') + artist_ctx = {'name': _fb_name or 'Unknown Artist'} + + # Construct minimal album context + # Ensure images are preserved (important for artwork) + album_id = s_album.get('id', 'wishlist_album') + album_ctx = { + 'id': album_id, + 'name': s_album.get('name'), + 'release_date': s_album.get('release_date', ''), + 'total_tracks': s_album.get('total_tracks', 1), + 'total_discs': wishlist_album_disc_counts.get(album_id, 1), + 'album_type': s_album.get('album_type', 'album'), + 'images': s_album.get('images', []) # Pass images array directly + } + + track_info['_explicit_album_context'] = album_ctx + track_info['_explicit_artist_context'] = artist_ctx + track_info['_is_explicit_album_download'] = True + logger.info(f"[Wishlist] Added album context for: '{track_info.get('name')}' -> '{album_ctx['name']}'") + + + # Add playlist folder mode flag for sync page playlists + if batch_playlist_folder_mode: + track_info['_playlist_folder_mode'] = True + track_info['_playlist_name'] = batch_playlist_name + logger.info(f"[Task Creation] Added playlist folder mode for: {track_info.get('name')} → {batch_playlist_name}") + else: + logger.debug(f"[Debug] Task Creation - playlist folder mode NOT enabled for: {track_info.get('name')}") + + download_tasks[task_id] = { + 'status': 'pending', 'track_info': track_info, + 'playlist_id': playlist_id, 'batch_id': batch_id, + 'track_index': res['track_index'], 'retry_count': 0, + 'cached_candidates': [], 'used_sources': set(), + 'status_change_time': time.time(), + 'metadata_enhanced': False + } + download_batches[batch_id]['queue'].append(task_id) + + deps.download_monitor.start_monitoring(batch_id) + deps.start_next_batch_of_downloads(batch_id) + + except Exception as e: + logger.error(f"Master worker for batch {batch_id} failed: {e}") + traceback.print_exc() + + is_auto_batch = False + with tasks_lock: + if batch_id in download_batches: + is_auto_batch = download_batches[batch_id].get('auto_initiated', False) + download_batches[batch_id]['phase'] = 'error' + download_batches[batch_id]['error'] = str(e) + + # Reset YouTube playlist phase to 'discovered' if this is a YouTube playlist on error + if playlist_id.startswith('youtube_'): + url_hash = playlist_id.replace('youtube_', '') + if url_hash in deps.youtube_playlist_states: + deps.youtube_playlist_states[url_hash]['phase'] = 'discovered' + logger.error(f"Reset YouTube playlist {url_hash} to discovered phase (error)") + + # Handle auto-initiated wishlist errors - reset flag + if is_auto_batch and playlist_id == 'wishlist': + logger.error("[Auto-Wishlist] Master worker error - resetting auto-processing flag") + deps.reset_wishlist_auto_processing() diff --git a/tests/downloads/test_downloads_master.py b/tests/downloads/test_downloads_master.py new file mode 100644 index 00000000..5307dd2f --- /dev/null +++ b/tests/downloads/test_downloads_master.py @@ -0,0 +1,684 @@ +"""Tests for core/downloads/master.py — full missing-tracks master worker.""" + +from __future__ import annotations + +import threading + +import pytest + +from core.downloads import master as mw +from core.runtime_state import download_batches, download_tasks, tasks_lock + + +# --------------------------------------------------------------------------- +# Fixtures + fakes +# --------------------------------------------------------------------------- + +@pytest.fixture(autouse=True) +def reset_state(): + download_tasks.clear() + download_batches.clear() + yield + download_tasks.clear() + download_batches.clear() + + +class _FakeConfig: + def __init__(self, values=None): + self._v = values or {} + + def get(self, key, default=None): + return self._v.get(key, default) + + def get_active_media_server(self): + return self._v.get('_active_server', 'plex') + + +class _FakeDB: + def __init__(self, found_tracks=None, album=None, album_tracks=None, album_confidence=0.95): + self.found_tracks = found_tracks or {} # (title_lower, artist_lower) -> confidence + self.album = album + self.album_tracks = album_tracks or [] + self.album_confidence = album_confidence + self.sync_history_calls = [] + self.track_results_calls = [] + + def check_track_exists(self, title, artist, confidence_threshold=0.7, server_source=None, album=None): + key = (title.lower().strip(), artist.lower().strip()) + if key in self.found_tracks: + conf = self.found_tracks[key] + return (object(), conf) # (DatabaseTrack-ish, confidence) + return (None, 0.0) + + def check_album_exists_with_editions(self, title, artist, confidence_threshold=0.7, + expected_track_count=None, server_source=None): + return (self.album, self.album_confidence) + + def get_tracks_by_album(self, album_id): + return self.album_tracks + + def _string_similarity(self, a, b): + if a == b: + return 1.0 + if a in b or b in a: + return 0.85 + return 0.0 + + def update_sync_history_completion(self, batch_id, **kwargs): + self.sync_history_calls.append((batch_id, kwargs)) + + def update_sync_history_track_results(self, batch_id, results_json): + self.track_results_calls.append((batch_id, results_json)) + + +class _DBTrack: + def __init__(self, title): + self.title = title + + +class _DBAlbum: + def __init__(self, id_, title): + self.id = id_ + self.title = title + + +class _FakeSoulseek: + def __init__(self, album_results=None, track_results=None, browse_files=None, parsed_tracks=None): + self._album_results = album_results or [] + self._track_results = track_results or [] + self._browse_files = browse_files + self._parsed_tracks = parsed_tracks or [] + self.search_calls = [] + + async def search(self, query, timeout=30): + self.search_calls.append(query) + return (self._track_results, self._album_results) + + def filter_results_by_quality_preference(self, tracks): + return tracks # no-op, accept all + + async def browse_user_directory(self, username, path): + return self._browse_files + + def parse_browse_results_to_tracks(self, username, browse_files, directory): + return self._parsed_tracks + + +class _FakeSoulseekWrapper: + """Wraps a soulseek client at .soulseek attribute (matches web_server pattern).""" + def __init__(self, inner): + self.soulseek = inner + + +class _FakeMonitor: + def __init__(self): + self.started = [] + + def start_monitoring(self, batch_id): + self.started.append(batch_id) + + +class _FakeExecutor: + def __init__(self): + self.submitted = [] + + def submit(self, fn, *args): + self.submitted.append((fn, args)) + + +class _FakeMBSvc: + pass + + +class _FakeMBWorker: + def __init__(self, svc=None): + self.mb_service = svc + + +def _run_async_sync(coro): + """Synchronously run a coroutine for tests.""" + import asyncio + return asyncio.get_event_loop().run_until_complete(coro) if not asyncio.iscoroutine(coro) else asyncio.new_event_loop().run_until_complete(coro) + + +def _make_run_async(): + import asyncio + def _runner(coro): + loop = asyncio.new_event_loop() + try: + return loop.run_until_complete(coro) + finally: + loop.close() + return _runner + + +def _build_deps( + *, + config=None, + soulseek=None, + run_async=None, + mb_worker=None, + mb_release_cache=None, + mb_release_cache_lock=None, + mb_release_detail_cache=None, + mb_release_detail_cache_lock=None, + normalize_album_cache_key=None, + wishlist_remove=None, + is_explicit_blocked=None, + yt_states=None, + tidal_states=None, + deezer_states=None, + spotify_states=None, + executor=None, + process_failed_auto=None, + source_reuse_logger=None, + monitor=None, + start_next_batch=None, + reset_wishlist_auto=None, +): + return mw.MasterDeps( + config_manager=config or _FakeConfig(), + soulseek_client=soulseek or _FakeSoulseekWrapper(_FakeSoulseek()), + run_async=run_async or _make_run_async(), + mb_worker=mb_worker, + mb_release_cache=mb_release_cache if mb_release_cache is not None else {}, + mb_release_cache_lock=mb_release_cache_lock or threading.Lock(), + mb_release_detail_cache=mb_release_detail_cache if mb_release_detail_cache is not None else {}, + mb_release_detail_cache_lock=mb_release_detail_cache_lock or threading.Lock(), + normalize_album_cache_key=normalize_album_cache_key or (lambda s: s.lower().strip()), + check_and_remove_track_from_wishlist_by_metadata=wishlist_remove or (lambda td: None), + is_explicit_blocked=is_explicit_blocked or (lambda td: False), + youtube_playlist_states=yt_states if yt_states is not None else {}, + tidal_discovery_states=tidal_states if tidal_states is not None else {}, + deezer_discovery_states=deezer_states if deezer_states is not None else {}, + spotify_public_discovery_states=spotify_states if spotify_states is not None else {}, + missing_download_executor=executor or _FakeExecutor(), + process_failed_tracks_to_wishlist_exact_with_auto_completion=process_failed_auto or (lambda bid: None), + source_reuse_logger=source_reuse_logger or _StubLogger(), + download_monitor=monitor or _FakeMonitor(), + start_next_batch_of_downloads=start_next_batch or (lambda bid: None), + reset_wishlist_auto_processing=reset_wishlist_auto or (lambda: None), + ) + + +class _StubLogger: + def info(self, *a, **kw): pass + def warning(self, *a, **kw): pass + def error(self, *a, **kw): pass + def debug(self, *a, **kw): pass + + +def _seed_batch(batch_id, **overrides): + base = { + 'phase': 'queued', + 'queue': [], + 'analysis_total': 0, + 'analysis_processed': 0, + 'analysis_results': [], + } + base.update(overrides) + download_batches[batch_id] = base + + +# --------------------------------------------------------------------------- +# PHASE 1: analysis +# --------------------------------------------------------------------------- + +def test_analysis_phase_sets_state(monkeypatch): + """Analysis phase marks batch counters; phase moves to 'downloading' when there are missing tracks.""" + db = _FakeDB() # found_tracks empty → every track marked missing + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + _seed_batch('B1') + deps = _build_deps() + tracks = [{'name': 'T1', 'artists': ['A']}] + + mw.run_full_missing_tracks_process('B1', 'P1', tracks, deps) + + # Track was missing → progressed to 'downloading' phase + assert download_batches['B1']['phase'] == 'downloading' + assert download_batches['B1']['analysis_processed'] == 1 + assert len(download_batches['B1']['analysis_results']) == 1 + + +def test_force_download_treats_all_as_missing(monkeypatch): + """force_download_all skips DB check — every track marked missing.""" + db = _FakeDB(found_tracks={('t1', 'a'): 1.0, ('t2', 'a'): 1.0}) # would otherwise be found + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + _seed_batch('B2', force_download_all=True) + deps = _build_deps() + tracks = [ + {'name': 'T1', 'artists': ['A']}, + {'name': 'T2', 'artists': ['A']}, + ] + + mw.run_full_missing_tracks_process('B2', 'playlist1', tracks, deps) + + # All 2 tracks should produce queue tasks (treated as missing) + assert len(download_batches['B2']['queue']) == 2 + assert download_batches['B2']['phase'] == 'downloading' + + +def test_found_tracks_trigger_wishlist_removal(monkeypatch): + """When DB lookup succeeds, master worker invokes wishlist removal callback.""" + db = _FakeDB(found_tracks={('t1', 'a'): 0.9}) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + removed = [] + deps = _build_deps(wishlist_remove=lambda td: removed.append(td.get('name'))) + + _seed_batch('B3') + tracks = [{'name': 'T1', 'artists': ['A']}] + + mw.run_full_missing_tracks_process('B3', 'P1', tracks, deps) + + assert removed == ['T1'] + + +def test_explicit_filter_removes_blocked_tracks(monkeypatch): + """When content_filter.allow_explicit=False, blocked tracks dropped from missing set.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + config = _FakeConfig({'content_filter.allow_explicit': False}) + deps = _build_deps( + config=config, + is_explicit_blocked=lambda td: td.get('name') == 'BLOCKED', + ) + + _seed_batch('B4') + tracks = [ + {'name': 'CLEAN', 'artists': ['A']}, + {'name': 'BLOCKED', 'artists': ['A']}, + ] + + mw.run_full_missing_tracks_process('B4', 'P1', tracks, deps) + + # only CLEAN survives the filter + assert len(download_batches['B4']['queue']) == 1 + + +# --------------------------------------------------------------------------- +# PHASE 2: no missing -> complete + state updates +# --------------------------------------------------------------------------- + +def test_no_missing_marks_batch_complete(monkeypatch): + """If every track found in DB, batch transitions directly to complete.""" + db = _FakeDB(found_tracks={('t1', 'a'): 0.9, ('t2', 'a'): 0.9}) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B5') + tracks = [ + {'name': 'T1', 'artists': ['A']}, + {'name': 'T2', 'artists': ['A']}, + ] + + mw.run_full_missing_tracks_process('B5', 'P1', tracks, deps) + + assert download_batches['B5']['phase'] == 'complete' + assert 'completion_time' in download_batches['B5'] + assert db.sync_history_calls # sync history written + + +def test_no_missing_updates_youtube_playlist_state(monkeypatch): + """YouTube playlist phase set to 'download_complete' on no-missing.""" + db = _FakeDB(found_tracks={('t1', 'a'): 0.9}) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + yt_states = {'abc123': {'phase': 'discovered'}} + deps = _build_deps(yt_states=yt_states) + + _seed_batch('B6') + mw.run_full_missing_tracks_process('B6', 'youtube_abc123', [{'name': 'T1', 'artists': ['A']}], deps) + + assert yt_states['abc123']['phase'] == 'download_complete' + + +def test_no_missing_with_auto_wishlist_submits_completion(monkeypatch): + """auto_initiated wishlist batch with no missing tracks submits auto-completion handler.""" + db = _FakeDB(found_tracks={('t1', 'a'): 0.9}) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + executor = _FakeExecutor() + auto_called = [] + deps = _build_deps(executor=executor, process_failed_auto=lambda bid: auto_called.append(bid)) + + _seed_batch('B7', auto_initiated=True) + mw.run_full_missing_tracks_process('B7', 'wishlist', [{'name': 'T1', 'artists': ['A']}], deps) + + assert len(executor.submitted) == 1 + fn, args = executor.submitted[0] + assert args == ('B7',) + + +# --------------------------------------------------------------------------- +# Album fast path +# --------------------------------------------------------------------------- + +def test_album_fast_path_direct_match(monkeypatch): + """Album lookup + direct title match → track marked found, no queue entry.""" + album = _DBAlbum(id_=42, title='Test Album') + album_tracks = [_DBTrack('T1'), _DBTrack('T2')] + db = _FakeDB(album=album, album_tracks=album_tracks) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B8', + is_album_download=True, + album_context={'name': 'Test Album', 'total_tracks': 2}, + artist_context={'name': 'Artist'}) + + tracks = [{'name': 'T1', 'artists': ['Artist']}, {'name': 'T2', 'artists': ['Artist']}] + mw.run_full_missing_tracks_process('B8', 'album:1', tracks, deps) + + assert download_batches['B8']['phase'] == 'complete' # all matched + + +def test_album_fast_path_misses_fall_through_to_global(monkeypatch): + """Album lookup with track not in album → fuzzy fallback or per-track global search.""" + album = _DBAlbum(id_=42, title='Test Album') + album_tracks = [_DBTrack('Existing')] + db = _FakeDB( + album=album, + album_tracks=album_tracks, + found_tracks={}, # global search finds nothing for Other + ) + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B9', + is_album_download=True, + album_context={'name': 'Test Album', 'total_tracks': 2}, + artist_context={'name': 'Artist'}) + + # 'Other' is not in album, allow_duplicates default True → marked missing without global search + tracks = [{'name': 'Other', 'artists': ['Artist']}] + mw.run_full_missing_tracks_process('B9', 'album:1', tracks, deps) + + assert len(download_batches['B9']['queue']) == 1 + + +# --------------------------------------------------------------------------- +# MB release preflight +# --------------------------------------------------------------------------- + +def test_mb_release_preflight_caches_mbid(monkeypatch): + """MB preflight caches release MBID under both normalized and exact keys.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + fake_release = {'id': 'mbid-xyz', 'title': 'Test Album'} + + def fake_find_best_release(album, artist, count, svc): + return fake_release + + import core.album_consistency as ac + monkeypatch.setattr(ac, '_find_best_release', fake_find_best_release) + + cache = {} + detail_cache = {} + deps = _build_deps( + mb_worker=_FakeMBWorker(svc=_FakeMBSvc()), + mb_release_cache=cache, + mb_release_detail_cache=detail_cache, + ) + _seed_batch('B10', + is_album_download=True, + album_context={'name': 'Test Album', 'total_tracks': 1}, + artist_context={'name': 'Artist'}) + + mw.run_full_missing_tracks_process('B10', 'album:1', [{'name': 'T1', 'artists': ['Artist']}], deps) + + # Should have cached under both normalized and exact-lower keys + assert ('test album', 'artist') in cache + assert cache[('test album', 'artist')] == 'mbid-xyz' + assert detail_cache['mbid-xyz'] == fake_release + + +def test_mb_release_preflight_skipped_when_no_mb_worker(monkeypatch): + """Without mb_worker, preflight quietly skips.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + cache = {} + deps = _build_deps(mb_worker=None, mb_release_cache=cache) + _seed_batch('B11', + is_album_download=True, + album_context={'name': 'Album', 'total_tracks': 1}, + artist_context={'name': 'Artist'}) + + mw.run_full_missing_tracks_process('B11', 'album:1', [{'name': 'T1', 'artists': ['Artist']}], deps) + + assert cache == {} # nothing cached + + +# --------------------------------------------------------------------------- +# Task creation +# --------------------------------------------------------------------------- + +def test_missing_tracks_create_queue_tasks(monkeypatch): + """Missing tracks produce download_tasks + are appended to batch queue.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B12') + + tracks = [{'name': 'T1', 'artists': ['A']}, {'name': 'T2', 'artists': ['A']}] + mw.run_full_missing_tracks_process('B12', 'P1', tracks, deps) + + assert len(download_batches['B12']['queue']) == 2 + for task_id in download_batches['B12']['queue']: + assert task_id in download_tasks + assert download_tasks[task_id]['status'] == 'pending' + assert download_tasks[task_id]['batch_id'] == 'B12' + + +def test_album_download_injects_explicit_context(monkeypatch): + """Album downloads embed _explicit_album_context + _explicit_artist_context per task.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + album_ctx = {'name': 'Album', 'total_tracks': 1} + artist_ctx = {'name': 'Artist'} + _seed_batch('B13', + is_album_download=True, + album_context=album_ctx, + artist_context=artist_ctx) + + mw.run_full_missing_tracks_process('B13', 'album:1', [{'name': 'T1', 'artists': ['Artist']}], deps) + + assert len(download_batches['B13']['queue']) == 1 + task_id = download_batches['B13']['queue'][0] + info = download_tasks[task_id]['track_info'] + assert info['_explicit_album_context'] == album_ctx + assert info['_explicit_artist_context'] == artist_ctx + assert info['_is_explicit_album_download'] is True + + +def test_wishlist_album_grouping_resolves_artist(monkeypatch): + """Wishlist tracks sharing an album_id all get the same artist context.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B14') + + # Two tracks on same album with different track-level artists — wishlist grouping + # should resolve ONE artist for the album (first track wins). + tracks = [ + { + 'name': 'T1', 'artists': [{'name': 'Track Artist 1'}], + 'spotify_data': { + 'album': {'id': 'A1', 'name': 'Test Album', 'artists': [{'name': 'Album Artist'}]}, + 'artists': [{'name': 'Track Artist 1'}], + }, + }, + { + 'name': 'T2', 'artists': [{'name': 'Track Artist 2'}], + 'spotify_data': { + 'album': {'id': 'A1', 'name': 'Test Album', 'artists': [{'name': 'Album Artist'}]}, + 'artists': [{'name': 'Track Artist 2'}], + }, + }, + ] + mw.run_full_missing_tracks_process('B14', 'wishlist', tracks, deps) + + assert len(download_batches['B14']['queue']) == 2 + artist_names = set() + for tid in download_batches['B14']['queue']: + info = download_tasks[tid]['track_info'] + artist_names.add(info['_explicit_artist_context']['name']) + + # Both tracks should resolve to the same album-level artist + assert len(artist_names) == 1 + assert 'Album Artist' in artist_names + + +def test_playlist_folder_mode_propagates(monkeypatch): + """Playlist folder mode flag carried through to track_info.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + _seed_batch('B15', + playlist_folder_mode=True, + playlist_name='My Mix') + + mw.run_full_missing_tracks_process('B15', 'P1', [{'name': 'T1', 'artists': ['A']}], deps) + + task_id = download_batches['B15']['queue'][0] + info = download_tasks[task_id]['track_info'] + assert info['_playlist_folder_mode'] is True + assert info['_playlist_name'] == 'My Mix' + + +# --------------------------------------------------------------------------- +# Hand-off to monitor + start_next_batch +# --------------------------------------------------------------------------- + +def test_handoff_starts_monitor_and_next_batch(monkeypatch): + """After task creation, master worker starts monitor + next batch.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + monitor = _FakeMonitor() + started_next = [] + deps = _build_deps(monitor=monitor, start_next_batch=lambda bid: started_next.append(bid)) + + _seed_batch('B16') + mw.run_full_missing_tracks_process('B16', 'P1', [{'name': 'T1', 'artists': ['A']}], deps) + + assert monitor.started == ['B16'] + assert started_next == ['B16'] + + +# --------------------------------------------------------------------------- +# Multi-disc album_context +# --------------------------------------------------------------------------- + +def test_multi_disc_total_discs_computed(monkeypatch): + """For album downloads, total_discs computed from max(disc_number) across all tracks.""" + db = _FakeDB() + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + deps = _build_deps() + album_ctx = {'name': 'Album', 'total_tracks': 3} + _seed_batch('B17', + is_album_download=True, + album_context=album_ctx, + artist_context={'name': 'Artist'}) + + tracks = [ + {'name': 'T1', 'artists': ['Artist'], 'disc_number': 1}, + {'name': 'T2', 'artists': ['Artist'], 'disc_number': 2}, + {'name': 'T3', 'artists': ['Artist'], 'disc_number': 2}, + ] + mw.run_full_missing_tracks_process('B17', 'album:1', tracks, deps) + + assert album_ctx['total_discs'] == 2 + + +# --------------------------------------------------------------------------- +# Error handler +# --------------------------------------------------------------------------- + +def test_error_handler_marks_batch_error(monkeypatch): + """Exception during analysis → batch.phase=error, batch.error=str(exception).""" + def boom(): + raise RuntimeError("DB exploded") + monkeypatch.setattr('database.music_database.MusicDatabase', boom) + + deps = _build_deps() + _seed_batch('B18') + + mw.run_full_missing_tracks_process('B18', 'P1', [{'name': 'T1', 'artists': ['A']}], deps) + + assert download_batches['B18']['phase'] == 'error' + assert 'DB exploded' in download_batches['B18']['error'] + + +def test_error_handler_resets_youtube_phase(monkeypatch): + """Error on a youtube_ playlist resets that playlist's phase to 'discovered'.""" + def boom(): + raise RuntimeError("kaboom") + monkeypatch.setattr('database.music_database.MusicDatabase', boom) + + yt_states = {'abc': {'phase': 'downloading'}} + deps = _build_deps(yt_states=yt_states) + _seed_batch('B19') + + mw.run_full_missing_tracks_process('B19', 'youtube_abc', [{'name': 'T1', 'artists': ['A']}], deps) + + assert yt_states['abc']['phase'] == 'discovered' + + +def test_error_handler_resets_auto_wishlist(monkeypatch): + """Auto-initiated wishlist error invokes reset_wishlist_auto_processing callback.""" + def boom(): + raise RuntimeError("oops") + monkeypatch.setattr('database.music_database.MusicDatabase', boom) + + reset_called = [] + deps = _build_deps(reset_wishlist_auto=lambda: reset_called.append(True)) + _seed_batch('B20', auto_initiated=True) + + mw.run_full_missing_tracks_process('B20', 'wishlist', [{'name': 'T1', 'artists': ['A']}], deps) + + assert reset_called == [True] + + +# --------------------------------------------------------------------------- +# Batch removed mid-flight +# --------------------------------------------------------------------------- + +def test_batch_removed_before_phase_two_returns_cleanly(monkeypatch): + """If batch is deleted between analysis and download phase, function returns without crashing.""" + db = _FakeDB(found_tracks={('t1', 'a'): 0.9}) # marks T1 found → wishlist_remove fires + monkeypatch.setattr('database.music_database.MusicDatabase', lambda: db) + + monitor = _FakeMonitor() + next_batch_calls = [] + + # Wishlist removal callback deletes the batch mid-analysis to simulate cancel. + # T1 will be analyzed as 'found' → callback fires → batch deleted. + def kill_batch(td): + download_batches.pop('B21', None) + + deps = _build_deps( + wishlist_remove=kill_batch, + monitor=monitor, + start_next_batch=lambda bid: next_batch_calls.append(bid), + ) + _seed_batch('B21') + + # Should not raise even though batch vanishes during analysis loop + mw.run_full_missing_tracks_process('B21', 'P1', [{'name': 'T1', 'artists': ['A']}], deps) + + # All tracks were 'found' → no missing → no monitor/next_batch calls + # (batch was deleted, so phase=complete update silently no-ops) + assert monitor.started == [] + assert next_batch_calls == [] diff --git a/web_server.py b/web_server.py index f81dfac8..586a35fc 100644 --- a/web_server.py +++ b/web_server.py @@ -21582,591 +21582,48 @@ def _on_download_completed(batch_id, task_id, success=True): +# Master worker for the missing tracks pipeline lives in core/downloads/master.py. +from core.downloads import master as _downloads_master + + +def _build_master_deps(): + """Build the MasterDeps bundle from web_server.py globals on each call.""" + def _reset_wishlist_auto_processing(): + global wishlist_auto_processing, wishlist_auto_processing_timestamp + with wishlist_timer_lock: + wishlist_auto_processing = False + wishlist_auto_processing_timestamp = 0 + + return _downloads_master.MasterDeps( + config_manager=config_manager, + soulseek_client=soulseek_client, + run_async=run_async, + mb_worker=mb_worker, + mb_release_cache=mb_release_cache, + mb_release_cache_lock=mb_release_cache_lock, + mb_release_detail_cache=mb_release_detail_cache, + mb_release_detail_cache_lock=mb_release_detail_cache_lock, + normalize_album_cache_key=normalize_album_cache_key, + check_and_remove_track_from_wishlist_by_metadata=_check_and_remove_track_from_wishlist_by_metadata, + is_explicit_blocked=_is_explicit_blocked, + youtube_playlist_states=youtube_playlist_states, + tidal_discovery_states=tidal_discovery_states, + deezer_discovery_states=deezer_discovery_states, + spotify_public_discovery_states=spotify_public_discovery_states, + missing_download_executor=missing_download_executor, + process_failed_tracks_to_wishlist_exact_with_auto_completion=_process_failed_tracks_to_wishlist_exact_with_auto_completion, + source_reuse_logger=source_reuse_logger, + download_monitor=download_monitor, + start_next_batch_of_downloads=_start_next_batch_of_downloads, + reset_wishlist_auto_processing=_reset_wishlist_auto_processing, + ) + + def _run_full_missing_tracks_process(batch_id, playlist_id, tracks_json): - """ - A master worker that handles the entire missing tracks process: - 1. Runs the analysis. - 2. If missing tracks are found, it automatically queues them for download. - """ - try: - # PHASE 1: ANALYSIS - with tasks_lock: - if batch_id in download_batches: - download_batches[batch_id]['phase'] = 'analysis' - download_batches[batch_id]['analysis_total'] = len(tracks_json) - download_batches[batch_id]['analysis_processed'] = 0 + return _downloads_master.run_full_missing_tracks_process( + batch_id, playlist_id, tracks_json, _build_master_deps() + ) - from database.music_database import MusicDatabase - db = MusicDatabase() - active_server = config_manager.get_active_media_server() - analysis_results = [] - - # Get force download flag and album context from batch - force_download_all = False - batch_album_context = None - batch_artist_context = None - batch_is_album = False - with tasks_lock: - if batch_id in download_batches: - force_download_all = download_batches[batch_id].get('force_download_all', False) - batch_is_album = download_batches[batch_id].get('is_album_download', False) - batch_album_context = download_batches[batch_id].get('album_context') - batch_artist_context = download_batches[batch_id].get('artist_context') - - if force_download_all: - logger.warning(f"[Force Download] Force download mode enabled for batch {batch_id} - treating all tracks as missing") - - # Allow duplicate tracks across albums — when enabled, only skip tracks already - # owned in THIS album, not tracks owned in other albums - allow_duplicates = config_manager.get('wishlist.allow_duplicate_tracks', True) - if allow_duplicates and batch_is_album: - logger.info("[Duplicates] Allow duplicate tracks enabled — only checking ownership within target album") - - # PREFLIGHT: Pre-populate MusicBrainz release cache for album downloads. - # This ensures ALL tracks in the album use the same release MBID during - # per-track post-processing, preventing Navidrome album splits. - if batch_is_album and batch_album_context and batch_artist_context: - try: - album_name_pf = batch_album_context.get('name', '') - artist_name_pf = batch_artist_context.get('name', '') - if album_name_pf and artist_name_pf: - mb_svc = mb_worker.mb_service if mb_worker else None - if mb_svc: - from core.album_consistency import _find_best_release - release = _find_best_release(album_name_pf, artist_name_pf, len(tracks_json), mb_svc) - if release and release.get('id'): - release_mbid = release['id'] - _artist_key = artist_name_pf.lower().strip() - _rc_key_norm = (normalize_album_cache_key(album_name_pf), _artist_key) - _rc_key_exact = (album_name_pf.lower().strip(), _artist_key) - with mb_release_cache_lock: - mb_release_cache[_rc_key_norm] = release_mbid - mb_release_cache[_rc_key_exact] = release_mbid - # Also cache the full release detail for tag extraction - with mb_release_detail_cache_lock: - mb_release_detail_cache[release_mbid] = release - logger.info(f"[Preflight] Pre-cached MB release for '{album_name_pf}': " - f"'{release.get('title', '')}' ({release_mbid[:8]}...)") - else: - logger.warning(f"[Preflight] No MB release found for '{album_name_pf}' — per-track lookup will be used") - except Exception as pf_err: - logger.error(f"[Preflight] MB release preflight failed: {pf_err}") - - # ALBUM FAST PATH: If this is an album download, try to find the album in the DB first - # and match tracks within it — faster and more accurate than N global searches - album_tracks_map = {} # Maps normalized title -> DatabaseTrack for album-scoped matching - if batch_is_album and batch_album_context and batch_artist_context and not force_download_all: - album_name = batch_album_context.get('name', '') - artist_name = batch_artist_context.get('name', '') - total_tracks = batch_album_context.get('total_tracks', 0) - if album_name and artist_name: - try: - db_album, album_confidence = db.check_album_exists_with_editions( - title=album_name, artist=artist_name, - confidence_threshold=0.7, - expected_track_count=total_tracks if total_tracks > 0 else None, - server_source=active_server - ) - if db_album and album_confidence >= 0.7: - db_album_tracks = db.get_tracks_by_album(db_album.id) - for t in db_album_tracks: - album_tracks_map[t.title.lower().strip()] = t - logger.info(f"[Album Analysis] Found album '{db_album.title}' in DB with {len(db_album_tracks)} tracks (confidence: {album_confidence:.2f})") - else: - logger.warning(f"[Album Analysis] Album '{album_name}' not found in DB — falling back to per-track search") - except Exception as album_err: - logger.error(f"[Album Analysis] Album lookup error: {album_err} — falling back to per-track search") - - for i, track_data in enumerate(tracks_json): - # Use original table index if provided (for partial track selection), - # otherwise fall back to enumeration index - track_index = track_data.get('_original_index', i) - track_name = track_data.get('name', '') - artists = track_data.get('artists', []) - found, confidence = False, 0.0 - - # Skip database check if force download is enabled - if force_download_all: - logger.warning(f"[Force Download] Skipping database check for '{track_name}' - treating as missing") - found, confidence = False, 0.0 - elif album_tracks_map: - # Album-scoped matching: check against known album tracks first - track_name_lower = track_name.lower().strip() - # Direct title match - if track_name_lower in album_tracks_map: - found, confidence = True, 1.0 - else: - # Fuzzy match against album tracks using string similarity - best_sim = 0.0 - for db_title_lower, _db_track in album_tracks_map.items(): - sim = db._string_similarity(track_name_lower, db_title_lower) - if sim > best_sim: - best_sim = sim - if best_sim >= 0.7: - found, confidence = True, best_sim - else: - # Fall back to global per-track search for this track - # When allow_duplicates is on for album downloads, skip global - # search — the track isn't in THIS album so treat as missing - if allow_duplicates and batch_is_album: - found, confidence = False, 0.0 - else: - _fallback_album = batch_album_context.get('name') if batch_album_context else None - for artist in artists: - if isinstance(artist, str): - artist_name = artist - elif isinstance(artist, dict) and 'name' in artist: - artist_name = artist['name'] - else: - artist_name = str(artist) - db_track, track_confidence = db.check_track_exists( - track_name, artist_name, confidence_threshold=0.7, server_source=active_server, album=_fallback_album - ) - if db_track and track_confidence >= 0.7: - found, confidence = True, track_confidence - break - elif allow_duplicates and batch_is_album: - # Allow duplicates + album download + album not in DB yet → treat all as missing - found, confidence = False, 0.0 - else: - # Non-album download (playlist/single track) — always check global - for artist in artists: - # Handle both string format and Spotify API format {'name': 'Artist Name'} - if isinstance(artist, str): - artist_name = artist - elif isinstance(artist, dict) and 'name' in artist: - artist_name = artist['name'] - else: - artist_name = str(artist) - db_track, track_confidence = db.check_track_exists( - track_name, artist_name, confidence_threshold=0.7, server_source=active_server - ) - if db_track and track_confidence >= 0.7: - found, confidence = True, track_confidence - break - - analysis_results.append({ - 'track_index': track_index, 'track': track_data, 'found': found, 'confidence': confidence - }) - - # WISHLIST REMOVAL: If track is found in database, check if it should be removed from wishlist - if found and confidence >= 0.7: - try: - _check_and_remove_track_from_wishlist_by_metadata(track_data) - except Exception as wishlist_error: - logger.error(f"[Analysis] Error checking wishlist removal for found track: {wishlist_error}") - - with tasks_lock: - if batch_id in download_batches: - download_batches[batch_id]['analysis_processed'] = i + 1 - # Store incremental results for live updates - download_batches[batch_id]['analysis_results'] = analysis_results.copy() - - missing_tracks = [res for res in analysis_results if not res['found']] - - # Filter explicit tracks if content filter is enabled - if not config_manager.get('content_filter.allow_explicit', True): - before_count = len(missing_tracks) - missing_tracks = [res for res in missing_tracks if not _is_explicit_blocked(res.get('track', {}))] - skipped = before_count - len(missing_tracks) - if skipped > 0: - logger.warning(f"[Content Filter] Filtered out {skipped} explicit track(s) from download queue") - - with tasks_lock: - if batch_id in download_batches: - download_batches[batch_id]['analysis_results'] = analysis_results - - # PHASE 2: TRANSITION TO DOWNLOAD (if necessary) - if not missing_tracks: - logger.warning(f"Analysis for batch {batch_id} complete. No missing tracks.") - - # Record sync history — all tracks found, nothing to download - tracks_found = sum(1 for r in analysis_results if r.get('found')) - try: - db_sh = MusicDatabase() - db_sh.update_sync_history_completion(batch_id, tracks_found=tracks_found, tracks_downloaded=0, tracks_failed=0) - # Save per-track results (all found, no downloads) - track_results = [] - for res in analysis_results: - td = res.get('track', {}) - artists = td.get('artists', []) - first_artist = (artists[0].get('name', artists[0]) if isinstance(artists[0], dict) else str(artists[0])) if artists else '' - alb = td.get('album', '') - # Extract image - _img = '' - _alb_obj = td.get('album', {}) - if isinstance(_alb_obj, dict): - _alb_imgs = _alb_obj.get('images', []) - if _alb_imgs and isinstance(_alb_imgs, list) and len(_alb_imgs) > 0: - _img = _alb_imgs[0].get('url', '') if isinstance(_alb_imgs[0], dict) else '' - track_results.append({ - 'index': res.get('track_index', 0), - 'name': td.get('name', ''), - 'artist': first_artist, - 'album': alb.get('name', '') if isinstance(alb, dict) else str(alb or ''), - 'image_url': _img, - 'duration_ms': td.get('duration_ms', 0), - 'source_track_id': td.get('id', ''), - 'status': 'found' if res.get('found') else 'not_found', - 'confidence': round(res.get('confidence', 0.0), 3), - 'matched_track': None, - 'download_status': None, - }) - if track_results: - db_sh.update_sync_history_track_results(batch_id, json.dumps(track_results)) - except Exception: - pass - - is_auto_batch = False - with tasks_lock: - if batch_id in download_batches: - is_auto_batch = download_batches[batch_id].get('auto_initiated', False) - download_batches[batch_id]['phase'] = 'complete' - download_batches[batch_id]['completion_time'] = time.time() # Track for auto-cleanup - - # Update YouTube playlist phase to 'download_complete' if this is a YouTube playlist - if playlist_id.startswith('youtube_'): - url_hash = playlist_id.replace('youtube_', '') - if url_hash in youtube_playlist_states: - youtube_playlist_states[url_hash]['phase'] = 'download_complete' - logger.warning(f"Updated YouTube playlist {url_hash} to download_complete phase (no missing tracks)") - - # Update Tidal playlist phase to 'download_complete' if this is a Tidal playlist - if playlist_id.startswith('tidal_'): - tidal_playlist_id = playlist_id.replace('tidal_', '') - if tidal_playlist_id in tidal_discovery_states: - tidal_discovery_states[tidal_playlist_id]['phase'] = 'download_complete' - logger.warning(f"Updated Tidal playlist {tidal_playlist_id} to download_complete phase (no missing tracks)") - - # Update Deezer playlist phase to 'download_complete' if this is a Deezer playlist - if playlist_id.startswith('deezer_'): - deezer_playlist_id = playlist_id.replace('deezer_', '') - if deezer_playlist_id in deezer_discovery_states: - deezer_discovery_states[deezer_playlist_id]['phase'] = 'download_complete' - logger.warning(f"Updated Deezer playlist {deezer_playlist_id} to download_complete phase (no missing tracks)") - - # Update Spotify Public playlist phase to 'download_complete' if this is a Spotify Public playlist - if playlist_id.startswith('spotify_public_'): - spotify_public_url_hash = playlist_id.replace('spotify_public_', '') - if spotify_public_url_hash in spotify_public_discovery_states: - spotify_public_discovery_states[spotify_public_url_hash]['phase'] = 'download_complete' - logger.warning(f"Updated Spotify Public playlist {spotify_public_url_hash} to download_complete phase (no missing tracks)") - - # Handle auto-initiated wishlist completion even when no missing tracks - if is_auto_batch and playlist_id == 'wishlist': - logger.warning("[Auto-Wishlist] No missing tracks found - calling auto-completion handler to toggle cycle and reschedule") - missing_download_executor.submit(_process_failed_tracks_to_wishlist_exact_with_auto_completion, batch_id) - - return - - logger.warning(f" transitioning batch {batch_id} to download phase with {len(missing_tracks)} tracks.") - - # Read batch context (quick lock) before doing any network I/O - with tasks_lock: - if batch_id not in download_batches: return - batch = download_batches[batch_id] - batch_album_context = batch.get('album_context') - batch_artist_context = batch.get('artist_context') - batch_is_album = batch.get('is_album_download', False) - batch_playlist_folder_mode = batch.get('playlist_folder_mode', False) - batch_playlist_name = batch.get('playlist_name', 'Unknown Playlist') - - # === ALBUM PRE-FLIGHT: Search for complete album folder before track-by-track === - # Only run pre-flight when Soulseek is the download source (or hybrid with soulseek) - preflight_source = None - preflight_tracks = None - dl_source_mode = config_manager.get('download_source.mode', 'hybrid') - _dl_hybrid_order = config_manager.get('download_source.hybrid_order', ['hifi', 'youtube', 'soulseek']) - _dl_hybrid_first = _dl_hybrid_order[0] if _dl_hybrid_order else config_manager.get('download_source.hybrid_primary', 'hifi') - soulseek_is_source = dl_source_mode == 'soulseek' or ( - dl_source_mode == 'hybrid' and _dl_hybrid_first == 'soulseek' - ) - if batch_is_album and batch_album_context and batch_artist_context and soulseek_is_source: - artist_name = batch_artist_context.get('name', '') - album_name = batch_album_context.get('name', '') - if artist_name and album_name: - try: - _sr = source_reuse_logger - _sr.info(f"[Album Pre-flight] Searching for '{artist_name} {album_name}'") - logger.info(f"[Album Pre-flight] Searching Soulseek for complete album: '{artist_name} - {album_name}'") - - slsk = soulseek_client.soulseek if hasattr(soulseek_client, 'soulseek') else soulseek_client - - # Try multiple query variations (banned keywords in artist/album name can return 0 results) - album_queries = [f"{artist_name} {album_name}"] - # Clean artist name (remove feat., parentheticals) - clean_artist = re.sub(r'\s*\(.*?\)', '', artist_name).strip() - clean_artist = re.sub(r'\s*(feat\.?|ft\.?|featuring)\s+.*$', '', clean_artist, flags=re.IGNORECASE).strip() - if clean_artist != artist_name: - album_queries.append(f"{clean_artist} {album_name}") - # Album name only (some users file by album) - album_queries.append(album_name) - - album_results = [] - track_results = [] - for aq in album_queries: - _sr.info(f"[Album Pre-flight] Trying query: '{aq}'") - track_results, album_results = run_async(slsk.search(aq, timeout=30)) - if album_results: - _sr.info(f"[Album Pre-flight] Found {len(album_results)} album results with query: '{aq}'") - break - _sr.info(f"[Album Pre-flight] No album results for query: '{aq}'") - - if album_results: - # Filter by quality preference - quality_filtered = [] - for ar in album_results: - filtered_tracks = slsk.filter_results_by_quality_preference(ar.tracks) - if filtered_tracks: - quality_filtered.append((ar, len(filtered_tracks))) - - if quality_filtered: - # Sort by track count (most complete album first), then quality score - quality_filtered.sort(key=lambda x: (x[1], x[0].quality_score), reverse=True) - best_album = quality_filtered[0][0] - - _sr.info(f"[Album Pre-flight] Best album result: {best_album.username}:{best_album.album_path} " - f"({best_album.track_count} tracks, quality={best_album.dominant_quality})") - logger.info(f"[Album Pre-flight] Found album folder: {best_album.username} — " - f"{best_album.track_count} tracks ({best_album.dominant_quality})") - - # Browse the user's folder to get all tracks (may have more than search returned) - browse_files = run_async(slsk.browse_user_directory(best_album.username, best_album.album_path)) - if browse_files: - folder_tracks = slsk.parse_browse_results_to_tracks( - best_album.username, browse_files, directory=best_album.album_path - ) - if folder_tracks: - preflight_source = { - 'username': best_album.username, - 'folder_path': best_album.album_path - } - preflight_tracks = folder_tracks - _sr.info(f"[Album Pre-flight] Browsed folder: {len(folder_tracks)} audio tracks available") - logger.info(f"[Album Pre-flight] Cached {len(folder_tracks)} tracks from {best_album.username} for source reuse") - else: - _sr.info("[Album Pre-flight] Browse returned files but no audio tracks") - else: - # Browse failed — fall back to using the search result tracks directly - _sr.info("[Album Pre-flight] Browse failed, using search result tracks directly") - preflight_source = { - 'username': best_album.username, - 'folder_path': best_album.album_path - } - preflight_tracks = best_album.tracks - logger.info(f"[Album Pre-flight] Using {len(best_album.tracks)} tracks from search results (browse unavailable)") - else: - _sr.info("[Album Pre-flight] No album results passed quality filter") - logger.warning("[Album Pre-flight] No album results matched quality preferences") - else: - _sr.info(f"[Album Pre-flight] Search returned no album results (got {len(track_results)} individual tracks)") - logger.warning("[Album Pre-flight] No complete album folders found, falling back to track-by-track search") - - except Exception as preflight_err: - logger.error(f"[Album Pre-flight] Search failed (non-fatal, falling back to track-by-track): {preflight_err}") - source_reuse_logger.info(f"[Album Pre-flight] Exception: {preflight_err}") - - with tasks_lock: - if batch_id not in download_batches: return - - download_batches[batch_id]['phase'] = 'downloading' - - # Store album pre-flight results on batch for source reuse - if preflight_source and preflight_tracks: - download_batches[batch_id]['last_good_source'] = preflight_source - download_batches[batch_id]['source_folder_tracks'] = preflight_tracks - download_batches[batch_id]['failed_sources'] = set() - logger.info(f"[Album Pre-flight] Pre-loaded source reuse data on batch {batch_id}") - - # Compute total_discs for multi-disc album subfolder support - # Use ALL tracks (tracks_json), not just missing ones, to correctly detect multi-disc - # even when only one disc has missing tracks - if batch_is_album and batch_album_context: - total_discs = max((t.get('disc_number', 1) for t in tracks_json), default=1) - batch_album_context['total_discs'] = total_discs - if total_discs > 1: - logger.info(f"[Multi-Disc] Detected {total_discs} discs for album '{batch_album_context.get('name')}'") - - # Pre-compute per-album data for wishlist tracks (grouped by album ID) - # Wishlist tracks aren't batch_is_album but each track has disc_number in spotify_data - wishlist_album_disc_counts = {} - wishlist_album_artist_map = {} # album_id -> resolved artist context (consistent per album) - if playlist_id == 'wishlist': - import json as _json - # First pass: collect disc_number and resolve ONE artist per album - for t in tracks_json: - sp_data = t.get('spotify_data', {}) - if isinstance(sp_data, str): - try: - sp_data = _json.loads(sp_data) - except: - sp_data = {} - album_val = sp_data.get('album') - album_id = album_val.get('id') if isinstance(album_val, dict) else album_val if isinstance(album_val, str) else None - # Fallback album key: use album name when ID is missing (e.g. mirrored playlist tracks) - if not album_id and isinstance(album_val, dict) and album_val.get('name'): - album_id = f"_name_{album_val['name'].lower().strip()}" - disc_num = sp_data.get('disc_number', t.get('disc_number', 1)) - if album_id: - wishlist_album_disc_counts[album_id] = max( - wishlist_album_disc_counts.get(album_id, 1), disc_num - ) - # Resolve album-level artist once per album (first track wins) - if album_id not in wishlist_album_artist_map: - _wl_source = t.get('source_info') or {} - if isinstance(_wl_source, str): - try: - _wl_source = _json.loads(_wl_source) - except: - _wl_source = {} - _wl_album = album_val if isinstance(album_val, dict) else {} - _wl_album_artists = _wl_album.get('artists', []) - # Priority: watchlist artist > album artists > track artists - if _wl_source.get('watchlist_artist_name'): - wishlist_album_artist_map[album_id] = { - 'name': _wl_source['watchlist_artist_name'], - 'id': _wl_source.get('watchlist_artist_id', '') - } - elif _wl_source.get('artist_name'): - wishlist_album_artist_map[album_id] = {'name': _wl_source['artist_name']} - elif _wl_album_artists: - _fa = _wl_album_artists[0] - wishlist_album_artist_map[album_id] = _fa if isinstance(_fa, dict) else {'name': str(_fa)} - else: - _wl_track_artists = sp_data.get('artists', []) - if _wl_track_artists: - _fa = _wl_track_artists[0] - wishlist_album_artist_map[album_id] = _fa if isinstance(_fa, dict) else {'name': str(_fa)} - else: - # Try top-level 'artists' (wishlist format uses plural) - _tl_artists = t.get('artists', []) - if _tl_artists: - _tla = _tl_artists[0] - _fallback_name = _tla.get('name', str(_tla)) if isinstance(_tla, dict) else str(_tla) - else: - _fallback_name = t.get('artist', '') - wishlist_album_artist_map[album_id] = {'name': _fallback_name or 'Unknown Artist'} - logger.info(f"[Wishlist Album Grouping] Album '{_wl_album.get('name', album_id)}' → artist: '{wishlist_album_artist_map[album_id].get('name', '?')}'") - - - - for res in missing_tracks: - task_id = str(uuid.uuid4()) - track_info = res['track'].copy() - - # Add explicit album context to track_info for artist album downloads - if batch_is_album and batch_album_context and batch_artist_context: - track_info['_explicit_album_context'] = batch_album_context - track_info['_explicit_artist_context'] = batch_artist_context - track_info['_is_explicit_album_download'] = True - logger.info(f"[Task Creation] Added explicit album context for: {track_info.get('name')}") - - # SPECIAL WISHLIST HANDLING: Inject album context if available to force grouping - elif playlist_id == 'wishlist': - # Extract spotify_data again since it might be buried - spotify_data = track_info.get('spotify_data') - if isinstance(spotify_data, str): - try: - spotify_data = json.loads(spotify_data) - except: - spotify_data = {} - - if not spotify_data: - spotify_data = {} - - s_album = spotify_data.get('album') or {} - if isinstance(s_album, str): - s_album = {'name': s_album} # Normalize string album to dict - s_artists = spotify_data.get('artists', []) - - # We need at least an album name and artist - if s_album and isinstance(s_album, dict) and s_album.get('name'): - # Use pre-computed album-level artist for folder consistency. - # All tracks from the same album get the same artist context, - # preventing folder splits on collab albums (KPOP Demon Hunters, etc.) - album_id_for_lookup = s_album.get('id') - # Fallback album key: match first-pass logic for missing IDs - if not album_id_for_lookup and s_album.get('name'): - album_id_for_lookup = f"_name_{s_album['name'].lower().strip()}" - if not album_id_for_lookup: - album_id_for_lookup = 'wishlist_album' - artist_ctx = wishlist_album_artist_map.get(album_id_for_lookup, {}) - if not artist_ctx or not artist_ctx.get('name'): - # Fallback: per-track resolution from artists array - _fb_artists = track_info.get('artists', []) - if _fb_artists: - _fb_a = _fb_artists[0] - _fb_name = _fb_a.get('name', str(_fb_a)) if isinstance(_fb_a, dict) else str(_fb_a) - else: - _fb_name = track_info.get('artist', '') - artist_ctx = {'name': _fb_name or 'Unknown Artist'} - - # Construct minimal album context - # Ensure images are preserved (important for artwork) - album_id = s_album.get('id', 'wishlist_album') - album_ctx = { - 'id': album_id, - 'name': s_album.get('name'), - 'release_date': s_album.get('release_date', ''), - 'total_tracks': s_album.get('total_tracks', 1), - 'total_discs': wishlist_album_disc_counts.get(album_id, 1), - 'album_type': s_album.get('album_type', 'album'), - 'images': s_album.get('images', []) # Pass images array directly - } - - track_info['_explicit_album_context'] = album_ctx - track_info['_explicit_artist_context'] = artist_ctx - track_info['_is_explicit_album_download'] = True - logger.info(f"[Wishlist] Added album context for: '{track_info.get('name')}' -> '{album_ctx['name']}'") - - - # Add playlist folder mode flag for sync page playlists - if batch_playlist_folder_mode: - track_info['_playlist_folder_mode'] = True - track_info['_playlist_name'] = batch_playlist_name - logger.info(f"[Task Creation] Added playlist folder mode for: {track_info.get('name')} → {batch_playlist_name}") - else: - logger.debug(f"[Debug] Task Creation - playlist folder mode NOT enabled for: {track_info.get('name')}") - - download_tasks[task_id] = { - 'status': 'pending', 'track_info': track_info, - 'playlist_id': playlist_id, 'batch_id': batch_id, - 'track_index': res['track_index'], 'retry_count': 0, - 'cached_candidates': [], 'used_sources': set(), - 'status_change_time': time.time(), - 'metadata_enhanced': False - } - download_batches[batch_id]['queue'].append(task_id) - - download_monitor.start_monitoring(batch_id) - _start_next_batch_of_downloads(batch_id) - - except Exception as e: - logger.error(f"Master worker for batch {batch_id} failed: {e}") - import traceback - traceback.print_exc() - - is_auto_batch = False - with tasks_lock: - if batch_id in download_batches: - is_auto_batch = download_batches[batch_id].get('auto_initiated', False) - download_batches[batch_id]['phase'] = 'error' - download_batches[batch_id]['error'] = str(e) - - # Reset YouTube playlist phase to 'discovered' if this is a YouTube playlist on error - if playlist_id.startswith('youtube_'): - url_hash = playlist_id.replace('youtube_', '') - if url_hash in youtube_playlist_states: - youtube_playlist_states[url_hash]['phase'] = 'discovered' - logger.error(f"Reset YouTube playlist {url_hash} to discovered phase (error)") - - # Handle auto-initiated wishlist errors - reset flag - if is_auto_batch and playlist_id == 'wishlist': - logger.error("[Auto-Wishlist] Master worker error - resetting auto-processing flag") - global wishlist_auto_processing, wishlist_auto_processing_timestamp - with wishlist_timer_lock: - wishlist_auto_processing = False - wishlist_auto_processing_timestamp = 0 # Post-processing verification worker logic lives in core/downloads/post_processing.py. from core.downloads import post_processing as _downloads_post_processing