From a133448a6e9c1aeade182ff7ac3cb6845ac1c8cf Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Mon, 27 Apr 2026 22:50:07 -0700 Subject: [PATCH] PR4d: lift _run_post_processing_worker to core/downloads/post_processing.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fourth sub-PR in the download orchestrator series. Strict 1:1 lift — zero behavior change. ~407 lines moved out of web_server.py. What moved: - _run_post_processing_worker → run_post_processing_worker The lifted function is intentionally kept as one ~400-line block to preserve byte-for-byte parity with the original. Refactoring it into smaller helpers (context lookup, file search loop, transfer-folder handler, downloads-folder handler) gets its own follow-up PR. Dependencies: 9 callbacks bundled in `PostProcessDeps` dataclass. - config_manager, soulseek_client, run_async (live refs) - docker_resolve_path, extract_filename, make_context_key (small utilities still in web_server.py — will lift in a future PR alongside other shared utilities) - find_completed_file (file search helper, still in web_server.py) - enhance_file_metadata, wipe_source_tags (web_server wrappers around core.metadata.enrichment) - post_process_with_verification (web_server wrapper around core.imports.pipeline) - mark_task_completed (wraps runtime_state.mark_task_completed + session counter) - on_download_completed (deferred to PR4g batch lifecycle) Direct imports for already-lifted helpers (no injection needed): - core.imports.album_naming.resolve_album_group - core.imports.context.{get_import_clean_title, get_import_clean_album, get_import_original_search, get_import_context_artist, get_import_context_album, normalize_import_context} - core.imports.filename.extract_track_number_from_filename - core.metadata.enrichment (re-exported as metadata_enrichment) - core.runtime_state.{download_tasks, tasks_lock, matched_downloads_context, matched_context_lock} Behavior parity: - Same control flow: missing-task short-circuit → cancelled/completed short-circuit → missing-filename failure → docker path resolution → context lookup with fuzzy fallback → expected filename generation → YouTube special-case path resolution → 5-attempt search loop with Strategy 1 (original filename in download+transfer) and Strategy 2 (expected final filename in transfer) → file-not-found failure → transfer-folder handler with metadata enhancement → downloads-folder handler with full post-process verification - Same retry count (5), sleep duration (5s), per-attempt logging - Same album_info dict construction with is_album=True for explicit album downloads - Same album grouping skip when context.is_album_download is True - Same wipe_source_tags fallback when enhancement context missing - Same matched_downloads_context cleanup on success - Same exception swallowing at processing-error and critical-error layers, both setting status='failed' + error_message + calling on_download_completed(b, t, success=False) - Every logger message text preserved verbatim (so log filters keep working) Tests: 16 new under tests/downloads/test_downloads_post_processing.py covering missing task, cancelled, already-completed, stream_processed, missing filename + username, file-not-found-after-retries with sleep mocked, stream-processor-completes-mid-search, transfer-folder with metadata enhanced + with no context (wipes tags), downloads-folder with + without context, processing exception, critical outer exception, YouTube special path, fuzzy context matching. Full suite: 950 passing (was 934). Ruff clean. --- core/downloads/post_processing.py | 481 ++++++++++++++++++ .../test_downloads_post_processing.py | 344 +++++++++++++ web_server.py | 427 +--------------- 3 files changed, 849 insertions(+), 403 deletions(-) create mode 100644 core/downloads/post_processing.py create mode 100644 tests/downloads/test_downloads_post_processing.py diff --git a/core/downloads/post_processing.py b/core/downloads/post_processing.py new file mode 100644 index 00000000..4a3df9d2 --- /dev/null +++ b/core/downloads/post_processing.py @@ -0,0 +1,481 @@ +"""Post-processing worker for completed downloads. + +The verification workflow that runs AFTER a slskd transfer reports as +'Succeeded' but BEFORE the task is marked completed in the UI. Locates +the file on disk (with retries + multiple search strategies), routes +it through metadata enhancement and the import pipeline, and finally +calls the batch lifecycle completion callback. + +Lifted verbatim from web_server.py's `_run_post_processing_worker`. +The single function is intentionally kept as one ~400-line block to +preserve byte-for-byte parity with the original — refactoring into +smaller helpers gets its own follow-up PR. + +Dependencies are passed in via `PostProcessDeps` since the function +needs ~9 callbacks/refs and direct injection beats hidden imports. +""" + +from __future__ import annotations + +import logging +import os +import time +import traceback +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Callable, Optional + +from core.imports.album_naming import resolve_album_group as _resolve_album_group +from core.imports.context import ( + get_import_clean_album, + get_import_clean_title, + get_import_context_album, + get_import_context_artist, + get_import_original_search, + normalize_import_context, +) +from core.imports.filename import extract_track_number_from_filename +from core.metadata import enrichment as metadata_enrichment +from core.runtime_state import ( + download_tasks, + matched_context_lock, + matched_downloads_context, + tasks_lock, +) + +logger = logging.getLogger(__name__) + + +@dataclass +class PostProcessDeps: + """Bundle of dependencies the post-processing worker needs. + + Constructed per-call by the route layer so client / config refs are + always live (no caching of pre-init Spotify clients etc). + """ + config_manager: Any + soulseek_client: Any + run_async: Callable + docker_resolve_path: Callable[[str], str] + extract_filename: Callable[[str], str] + make_context_key: Callable[[str, str], str] + find_completed_file: Callable + enhance_file_metadata: Callable + wipe_source_tags: Callable[[str], bool] + post_process_with_verification: Callable + mark_task_completed: Callable[[str, Optional[dict]], None] + on_download_completed: Callable[[str, str, bool], None] + + +def run_post_processing_worker(task_id: str, batch_id: str, deps: PostProcessDeps) -> None: + """NEW VERIFICATION WORKFLOW: Post-processing worker that only sets 'completed' + status after successful file verification and processing. This matches sync.py's + reliability. + """ + try: + logger.info(f"[Post-Processing] Starting verification for task {task_id}") + + # Retrieve task details from global state + with tasks_lock: + if task_id not in download_tasks: + logger.warning(f"[Post-Processing] Task {task_id} not found in download_tasks") + return + task = download_tasks[task_id].copy() + + # Check if task was cancelled or already completed during post-processing + if task['status'] == 'cancelled': + logger.warning(f"[Post-Processing] Task {task_id} was cancelled, skipping verification") + return + if task['status'] == 'completed' or task.get('stream_processed'): + logger.info(f"[Post-Processing] Task {task_id} already completed by stream processor, skipping verification") + return + + # Extract file information for verification + track_info = task.get('track_info', {}) + task_filename = task.get('filename') or track_info.get('filename') + task_username = task.get('username') or track_info.get('username') + + if not task_filename or not task_username: + logger.warning(f"[Post-Processing] Missing filename or username for task {task_id}") + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = 'Post-processing failed: missing file or source information from Soulseek transfer' + deps.on_download_completed(batch_id, task_id, False) + return + + download_dir = deps.docker_resolve_path(deps.config_manager.get('soulseek.download_path', './downloads')) + transfer_dir = deps.docker_resolve_path(deps.config_manager.get('soulseek.transfer_path', './Transfer')) + + # Try to get context for generating the correct final filename + task_basename = deps.extract_filename(task_filename) + context_key = deps.make_context_key(task_username, task_filename) + expected_final_filename = None + + logger.info(f"[Post-Processing] Looking up context with key: {context_key}") + + with matched_context_lock: + context = matched_downloads_context.get(context_key) + # Debug: Show all available context keys + available_keys = list(matched_downloads_context.keys()) + logger.info(f"[Post-Processing] Available context keys: {available_keys[:10]}...") # Show first 10 keys + + if context: + logger.info(f"[Post-Processing] Found context for key: {context_key}") + try: + original_search = context.get("original_search_result", {}) + logger.info(f"[Post-Processing] original_search keys: {list(original_search.keys())}") + + clean_title = get_import_clean_title(context, default=original_search.get('title', '')) + track_number = original_search.get('track_number') + + logger.info(f"[Post-Processing] clean_title: '{clean_title}', track_number: {track_number}") + + if clean_title and track_number: + # Generate expected final filename that stream processor would create + # Pattern: f"{track_number:02d} - {clean_title}.flac" + sanitized_title = clean_title.replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_') + expected_final_filename = f"{track_number:02d} - {sanitized_title}.flac" + logger.info(f"[Post-Processing] Generated expected final filename: {expected_final_filename}") + else: + logger.warning(f"[Post-Processing] Missing required data - clean_title: {bool(clean_title)}, track_number: {bool(track_number)}") + except Exception as e: + logger.error(f"[Post-Processing] Error generating expected filename: {e}") + traceback.print_exc() + else: + logger.warning(f"[Post-Processing] No context found for key: {context_key}") + # Try fuzzy matching with similar keys containing the filename + # SAFETY: Constrain to same Soulseek username to prevent cross-album + # metadata contamination during mass downloads (e.g., two albums both + # having "01 - Intro.flac" would match the wrong context without this) + with matched_context_lock: + similar_keys = [k for k in matched_downloads_context.keys() + if k.startswith(f"{task_username}::") and task_basename in k] + if similar_keys: + # Use the first similar key found + fuzzy_key = similar_keys[0] + context = matched_downloads_context.get(fuzzy_key) + logger.info(f"[Post-Processing] Found context using fuzzy key matching: {fuzzy_key}") + + # Generate expected final filename using the found context + try: + original_search = context.get("original_search_result", {}) + logger.info(f"[Post-Processing] fuzzy context original_search keys: {list(original_search.keys())}") + + clean_title = get_import_clean_title(context, default=original_search.get('title', '')) + track_number = original_search.get('track_number') + + logger.info(f"[Post-Processing] fuzzy context clean_title: '{clean_title}', track_number: {track_number}") + + if clean_title and track_number: + # Generate expected final filename that stream processor would create + # Pattern: f"{track_number:02d} - {clean_title}.flac" + sanitized_title = clean_title.replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_') + expected_final_filename = f"{track_number:02d} - {sanitized_title}.flac" + logger.info(f"[Post-Processing] Generated expected final filename from fuzzy match: {expected_final_filename}") + else: + logger.warning(f"[Post-Processing] Missing required data from fuzzy match - clean_title: {bool(clean_title)}, track_number: {bool(track_number)}") + except Exception as e: + logger.error(f"[Post-Processing] Error generating expected filename from fuzzy match: {e}") + traceback.print_exc() + else: + logger.warning(f"[Post-Processing] No similar keys found containing '{task_basename}'") + # Show a sample of what keys actually exist for debugging + sample_keys = list(matched_downloads_context.keys())[:5] + logger.info(f"[Post-Processing] Sample of existing keys: {sample_keys}") + + # RESILIENT FILE-FINDING LOOP: Try up to 3 times with delays + found_file = None + file_location = None + + # CRITICAL FIX: For YouTube downloads, the filename in task is 'id||title' (metadata), + # but the actual file on disk is 'Title.mp3'. We must ask the client for the real path. + if (task.get('username') == 'youtube' or '||' in str(task_filename)) and not found_file: + logger.info(f"[Post-Processing] Detected YouTube download task: {task_id}") + try: + # Query the download orchestrator for the status which contains the real file path + # CRITICAL FIX: Use the actual download_id designated by the client, not the internal task_id + actual_download_id = task.get('download_id') or task_id + status = deps.run_async(deps.soulseek_client.get_download_status(actual_download_id)) + if status and status.file_path: + real_path = status.file_path + if os.path.exists(real_path): + # Determine if it's in download or transfer directory + real_path_obj = Path(real_path) + download_dir_obj = Path(download_dir) + transfer_dir_obj = Path(transfer_dir) + + # Use absolute path comparison + try: + if download_dir_obj.resolve() in real_path_obj.resolve().parents: + file_location = 'download' + elif transfer_dir_obj.resolve() in real_path_obj.resolve().parents: + file_location = 'transfer' + else: + file_location = 'absolute' + except: # noqa: E722 -- byte-faithful to original (catches even KeyboardInterrupt) + # Fallback if resolve fails (e.g. permission or path issues) + file_location = 'absolute' + + if file_location: + # We found the file! Use the absolute path if it confuses the joining logic, + # but usually we want just the filename if location is 'download'/'transfer' + # CRITICAL FIX: Always use the absolute real_path. + # Stripping to basename causes FileNotFoundError because post-processing + # runs with CWD as project root, not download dir. + found_file = real_path + + logger.info(f"[Post-Processing] Resolved actual YouTube filename: {found_file} (Location: {file_location})") + else: + logger.warning(f"[Post-Processing] YouTube status reported path but file missing: {real_path}") + else: + logger.warning(f"[Post-Processing] YouTube status returned no file_path for task {task_id}") + except Exception as e: + logger.error(f"[Post-Processing] Failed to retrieve YouTube task status: {e}") + + _file_search_max_retries = 5 + for retry_count in range(_file_search_max_retries): + # If we already resolved the file (e.g. via YouTube status), skip searching + if found_file: + logger.info(f"[Post-Processing] Skipping search loop, file already resolved: {found_file}") + break + + # Check if stream processor already completed this task while we were waiting + with tasks_lock: + if task_id in download_tasks: + if download_tasks[task_id].get('stream_processed') or download_tasks[task_id]['status'] == 'completed': + logger.info(f"[Post-Processing] Task {task_id} was completed by stream processor during file search - done") + return + + logger.warning(f"[Post-Processing] Attempt {retry_count + 1}/{_file_search_max_retries} to find file") + logger.info(f"[Post-Processing] Original filename: {task_basename}") + if expected_final_filename: + logger.info(f"[Post-Processing] Expected final filename: {expected_final_filename}") + else: + logger.warning("[Post-Processing] No expected final filename available") + + # Strategy 1: Try with original filename in both downloads and transfer + logger.info("[Post-Processing] Strategy 1: Searching with original filename...") + found_file, file_location = deps.find_completed_file(download_dir, task_filename, transfer_dir) + + if found_file: + logger.info(f"[Post-Processing] Strategy 1 SUCCESS: Found file with original filename in {file_location}: {found_file}") + else: + logger.error("[Post-Processing] Strategy 1 FAILED: Original filename not found in either location") + + # Strategy 2: If not found and we have an expected final filename, try that in transfer folder + if not found_file and expected_final_filename: + logger.info("[Post-Processing] Strategy 2: Searching transfer folder with expected final filename...") + found_result = deps.find_completed_file(transfer_dir, expected_final_filename) + if found_result and found_result[0]: + found_file, file_location = found_result[0], 'transfer' + logger.info(f"[Post-Processing] Strategy 2 SUCCESS: Found file with expected final filename: {found_file}") + else: + logger.error("[Post-Processing] Strategy 2 FAILED: Expected final filename not found in transfer folder") + elif not expected_final_filename: + logger.warning("[Post-Processing] Strategy 2 SKIPPED: No expected final filename available") + + if found_file: + logger.warning(f"[Post-Processing] FILE FOUND after {retry_count + 1} attempts in {file_location}: {found_file}") + break + else: + logger.error(f"[Post-Processing] All search strategies failed on attempt {retry_count + 1}/{_file_search_max_retries}") + if retry_count < _file_search_max_retries - 1: # Don't sleep on final attempt + logger.info("[Post-Processing] Waiting 5 seconds before next attempt...") + time.sleep(5) + + if not found_file: + # CRITICAL: Before marking as failed, check if stream processor already handled this + # The /api/downloads/status polling endpoint processes files independently and may have + # already moved/renamed/tagged the file successfully while we were searching + with tasks_lock: + if task_id in download_tasks: + if download_tasks[task_id].get('stream_processed') or download_tasks[task_id]['status'] == 'completed': + logger.error(f"[Post-Processing] Task {task_id} was completed by stream processor - not marking as failed") + return + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = f'File not found on disk after {_file_search_max_retries} search attempts. Expected: {os.path.basename(task_filename)}' + deps.on_download_completed(batch_id, task_id, False) + return + + # Handle file found in transfer folder - already completed by stream processor + if file_location == 'transfer': + logger.info(f"[Post-Processing] File found in transfer folder - already completed by stream processor: {found_file}") + + # Check if metadata enhancement was completed + metadata_enhanced = False + with tasks_lock: + if task_id in download_tasks: + metadata_enhanced = download_tasks[task_id].get('metadata_enhanced', False) + + if not metadata_enhanced: + logger.warning("[Post-Processing] File in transfer folder missing metadata enhancement - completing now") + # Attempt to complete metadata enhancement using context + if context and expected_final_filename: + try: + context = normalize_import_context(context) + # Extract required data from context + original_search = get_import_original_search(context) + artist_context = get_import_context_artist(context) + album_context = get_import_context_album(context) + + if artist_context and album_context: + # CRITICAL FIX: Create album_info dict with proper structure for metadata enhancement + # This must match the format used in main stream processor to ensure consistency + + # Extract track number from context (should be available from fuzzy match) + track_number = original_search.get('track_number', 1) + + # If no track number in context, extract from filename + if track_number == 1 and found_file: + track_number = extract_track_number_from_filename(found_file) + logger.warning( + "[Verification] missing track_number; extracted from filename=%r -> %s", + os.path.basename(found_file), + track_number, + ) + + # Ensure track_number is valid + if not isinstance(track_number, int) or track_number < 1: + logger.error(f"[Verification] Invalid track number ({track_number}), defaulting to 1") + track_number = 1 + + # Get clean track name + clean_track_name = get_import_clean_title(context, default=original_search.get('title', 'Unknown Track')) + album_name = get_import_clean_album(context, default=album_context.get('name', 'Unknown Album')) + album_image_url = album_context.get('image_url') + if not album_image_url and album_context.get('images'): + album_images = album_context.get('images', []) + if album_images and isinstance(album_images[0], dict): + album_image_url = album_images[0].get('url') + + album_info = { + 'is_album': True, # CRITICAL: Mark as album track + 'album_name': album_name, + 'track_number': track_number, # CORRECTED TRACK NUMBER + 'disc_number': original_search.get('disc_number', 1), + 'clean_track_name': clean_track_name, + 'album_image_url': album_image_url, + 'confidence': 0.9, + 'source': 'verification_worker_corrected', + } + + # Apply album grouping for consistency with stream processor path. + # Only for singles/auto-detected — explicit album downloads already + # have the correct Spotify name and re-grouping would mangle it. + if not context.get("is_album_download", False): + try: + raw_album_ctx = original_search.get('album') + if isinstance(raw_album_ctx, str): + original_album_ctx = raw_album_ctx + elif isinstance(raw_album_ctx, dict) and 'name' in raw_album_ctx: + original_album_ctx = raw_album_ctx['name'] + else: + original_album_ctx = None + consistent_album_name = _resolve_album_group(artist_context, album_info, original_album_ctx) + album_info['album_name'] = consistent_album_name + except Exception as group_err: + logger.error(f"[Verification] Album grouping failed, using raw name: {group_err}") + else: + logger.info(f"[Verification] Explicit album download - preserving album name: '{album_info['album_name']}'") + + logger.info(f"[Verification] Created proper album_info - track_number: {track_number}, album: {album_info['album_name']}") + + logger.info(f"[Post-Processing] Attempting metadata enhancement for: {found_file}") + logger.warning(f"[Metadata Input] Verification worker - artist: '{artist_context.get('name', 'MISSING')}' (id: {artist_context.get('id', 'MISSING')})") + logger.warning(f"[Metadata Input] Verification worker - album: '{album_info.get('album_name', 'MISSING')}', track#: {album_info.get('track_number', 'MISSING')}, source: {album_info.get('source', 'unknown')}") + enhancement_success = deps.enhance_file_metadata(found_file, context, artist_context, album_info) + + if enhancement_success: + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['metadata_enhanced'] = True + logger.info(f"[Post-Processing] Successfully completed metadata enhancement for: {os.path.basename(found_file)}") + else: + logger.info(f"[Post-Processing] Metadata enhancement returned False for: {os.path.basename(found_file)}") + else: + logger.warning("[Post-Processing] Missing artist or album in context") + logger.info(f"[Post-Processing] artist_context: {artist_context is not None}, album_context: {album_context is not None}") + # Wipe source tags even without full enhancement — prevents + # Soulseek uploader's MusicBrainz IDs from causing album splits + if found_file and os.path.exists(found_file): + deps.wipe_source_tags(found_file) + except Exception as enhancement_error: + logger.error(f"[Post-Processing] Error during metadata enhancement: {enhancement_error}\n{traceback.format_exc()}") + if found_file and os.path.exists(found_file): + deps.wipe_source_tags(found_file) + else: + logger.warning("[Post-Processing] Cannot complete metadata enhancement - missing context or expected filename") + if found_file and os.path.exists(found_file): + deps.wipe_source_tags(found_file) + else: + logger.info("[Post-Processing] File already has metadata enhancement completed") + + with tasks_lock: + if task_id in download_tasks: + track_info = download_tasks[task_id].get('track_info') + deps.mark_task_completed(task_id, track_info) + + # Clean up context now that both stream processor and verification worker are done + with matched_context_lock: + if context_key in matched_downloads_context: + del matched_downloads_context[context_key] + logger.info(f"[Verification] Cleaned up context after successful verification: {context_key}") + + deps.on_download_completed(batch_id, task_id, True) + return + + # File found in downloads folder - attempt post-processing + try: + # Rebuild the context key using the same function that stored it + context_key = deps.make_context_key(task_username, task_filename) + + # Check if this download has matched context for post-processing + with matched_context_lock: + context = matched_downloads_context.get(context_key) + + if context: + logger.info(f"[Post-Processing] Found matched context, running full post-processing for: {context_key}") + # Run the existing post-processing logic with verification + deps.post_process_with_verification(context_key, context, found_file, task_id, batch_id) + else: + # No matched context - just mark as completed since file exists + logger.warning(f"[Post-Processing] No matched context, marking as completed: {os.path.basename(found_file)}") + with tasks_lock: + if task_id in download_tasks: + track_info = download_tasks[task_id].get('track_info') + deps.mark_task_completed(task_id, track_info) + + # Clean up context if it exists (might be leftover from stream processor) + with matched_context_lock: + if context_key in matched_downloads_context: + del matched_downloads_context[context_key] + logger.info(f"[Verification] Cleaned up leftover context: {context_key}") + + # Call completion callback since there's no other post-processing to handle it + deps.on_download_completed(batch_id, task_id, True) + + except Exception as processing_error: + logger.error(f"[Post-Processing] Processing failed for task {task_id}: {processing_error}") + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = f"Post-processing failed: {str(processing_error)}" + deps.on_download_completed(batch_id, task_id, False) + + except Exception as e: + logger.error(f"[Post-Processing] Critical error in post-processing worker for task {task_id}: {e}") + with tasks_lock: + if task_id in download_tasks: + download_tasks[task_id]['status'] = 'failed' + download_tasks[task_id]['error_message'] = f"Critical post-processing error: {str(e)}" + deps.on_download_completed(batch_id, task_id, False) + + +# Re-export the metadata helper so callers can wrap it in a callback without +# importing from core.metadata directly. +__all__ = [ + 'PostProcessDeps', + 'run_post_processing_worker', + 'metadata_enrichment', +] diff --git a/tests/downloads/test_downloads_post_processing.py b/tests/downloads/test_downloads_post_processing.py new file mode 100644 index 00000000..e3b03406 --- /dev/null +++ b/tests/downloads/test_downloads_post_processing.py @@ -0,0 +1,344 @@ +"""Tests for core/downloads/post_processing.py — verification worker for completed downloads. + +The worker is large + side-effecty. Tests cover the major control-flow +branches: missing task, cancelled, already-completed, missing +filename/username, file-found-in-transfer with + without metadata, file- +found-in-downloads with + without context, file-not-found-after-retries, +youtube special path, and top-level exception swallow. +""" + +from __future__ import annotations + +import os +import pytest + +from core.downloads import post_processing as pp +from core.runtime_state import ( + download_tasks, + matched_context_lock, + matched_downloads_context, + tasks_lock, +) + + +@pytest.fixture(autouse=True) +def reset_state(): + download_tasks.clear() + matched_downloads_context.clear() + yield + download_tasks.clear() + matched_downloads_context.clear() + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + +class _Recorder: + """Captures every call into a list of (name, args, kwargs).""" + def __init__(self): + self.calls = [] + + def __call__(self, name): + def _inner(*args, **kwargs): + self.calls.append((name, args, kwargs)) + return None + return _inner + + +def _build_deps( + *, + config=None, + soulseek_client=None, + run_async=None, + docker_resolve_path=None, + extract_filename=None, + make_context_key=None, + find_completed_file=None, + enhance_file_metadata=None, + wipe_source_tags=None, + post_process_with_verification=None, + mark_task_completed=None, + on_download_completed=None, +): + rec = _Recorder() + return pp.PostProcessDeps( + config_manager=config or _FakeConfig(), + soulseek_client=soulseek_client, + run_async=run_async or (lambda c: None), + docker_resolve_path=docker_resolve_path or (lambda p: p), + extract_filename=extract_filename or (lambda f: os.path.basename(f) if f else ''), + make_context_key=make_context_key or (lambda u, f: f"{u}::{f}"), + find_completed_file=find_completed_file or (lambda *a, **kw: (None, None)), + enhance_file_metadata=enhance_file_metadata or rec('enhance'), + wipe_source_tags=wipe_source_tags or rec('wipe'), + post_process_with_verification=post_process_with_verification or rec('post_process'), + mark_task_completed=mark_task_completed or rec('mark_completed'), + on_download_completed=on_download_completed or rec('on_complete'), + ), rec + + +class _FakeConfig: + def __init__(self, values=None): + self._v = values or {} + + def get(self, key, default=None): + return self._v.get(key, default) + + +# --------------------------------------------------------------------------- +# Branch coverage tests +# --------------------------------------------------------------------------- + +def test_missing_task_returns_early_no_callbacks(): + deps, rec = _build_deps() + pp.run_post_processing_worker('absent', 'b1', deps) + assert rec.calls == [] + + +def test_cancelled_task_returns_early_no_callbacks(): + download_tasks['t1'] = {'status': 'cancelled'} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert rec.calls == [] + + +def test_already_completed_task_returns_early(): + download_tasks['t1'] = {'status': 'completed'} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert rec.calls == [] + + +def test_stream_processed_task_returns_early(): + download_tasks['t1'] = {'status': 'post_processing', 'stream_processed': True} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert rec.calls == [] + + +def test_missing_filename_marks_failed_and_calls_on_complete(): + download_tasks['t1'] = {'status': 'post_processing', 'username': 'u1', 'track_info': {}} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' + assert 'Post-processing failed' in download_tasks['t1']['error_message'] + assert ('on_complete', ('b1', 't1', False), {}) in rec.calls + + +def test_missing_username_marks_failed_and_calls_on_complete(): + download_tasks['t1'] = {'status': 'post_processing', 'filename': 'song.flac', 'track_info': {}} + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' + + +def test_file_not_found_after_retries_marks_failed(monkeypatch): + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {}, + } + # Skip sleeps to keep test fast + monkeypatch.setattr(pp.time, 'sleep', lambda s: None) + deps, rec = _build_deps() + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' + assert 'File not found on disk' in download_tasks['t1']['error_message'] + assert ('on_complete', ('b1', 't1', False), {}) in rec.calls + + +def test_stream_processor_completes_during_search_loop_returns_no_failure(monkeypatch): + """If task gets marked completed by stream processor mid-retry, abort without failing.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {}, + } + monkeypatch.setattr(pp.time, 'sleep', lambda s: None) + call_count = [0] + + def _stream_completes_after_first_search(*a, **kw): + call_count[0] += 1 + if call_count[0] >= 1: + download_tasks['t1']['stream_processed'] = True + return (None, None) + + deps, rec = _build_deps(find_completed_file=_stream_completes_after_first_search) + pp.run_post_processing_worker('t1', 'b1', deps) + # Worker should detect stream_processed, return early, not mark failed + assert download_tasks['t1']['status'] == 'post_processing' # original status preserved + assert ('on_complete', ('b1', 't1', False), {}) not in rec.calls + + +def test_file_found_in_transfer_with_metadata_enhanced_skips_enhancement_and_completes(): + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {'name': 'Money'}, + 'metadata_enhanced': True, + } + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: ('/transfer/song.flac', 'transfer'), + ) + pp.run_post_processing_worker('t1', 'b1', deps) + # No enhance call because metadata_enhanced=True + assert not any(c[0] == 'enhance' for c in rec.calls) + # Mark + on-complete called + assert any(c[0] == 'mark_completed' for c in rec.calls) + assert ('on_complete', ('b1', 't1', True), {}) in rec.calls + + +def test_file_found_in_transfer_no_context_no_filename_wipes_tags(monkeypatch): + """Transfer file but missing context AND expected filename -> wipe tags only.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {}, + 'metadata_enhanced': False, + } + monkeypatch.setattr(pp.os.path, 'exists', lambda p: True) + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: ('/transfer/song.flac', 'transfer'), + ) + pp.run_post_processing_worker('t1', 'b1', deps) + # wipe_source_tags called (no full enhancement possible) + assert any(c[0] == 'wipe' for c in rec.calls) + # Still completed + assert ('on_complete', ('b1', 't1', True), {}) in rec.calls + + +def test_file_found_in_downloads_with_context_runs_post_process_with_verification(): + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {'name': 'Money'}, + } + matched_downloads_context['u1::song.flac'] = { + 'original_search_result': {'title': 'Money', 'track_number': 1}, + 'context_artist': {'name': 'Pink Floyd', 'id': 'art1'}, + 'context_album': {'name': 'DSOTM'}, + } + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: ('/downloads/song.flac', 'download'), + ) + pp.run_post_processing_worker('t1', 'b1', deps) + # post_process_with_verification called with the context + file + assert any(c[0] == 'post_process' for c in rec.calls) + + +def test_file_found_in_downloads_no_context_marks_completed_directly(): + """No matched context for the file → just mark completed since file exists.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {'name': 'Money'}, + } + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: ('/downloads/song.flac', 'download'), + ) + pp.run_post_processing_worker('t1', 'b1', deps) + # No post_process call (no context) + assert not any(c[0] == 'post_process' for c in rec.calls) + # Mark + on-complete called + assert any(c[0] == 'mark_completed' for c in rec.calls) + assert ('on_complete', ('b1', 't1', True), {}) in rec.calls + + +def test_processing_exception_marks_failed_and_calls_on_complete(): + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {'name': 'Money'}, + } + matched_downloads_context['u1::song.flac'] = {'original_search_result': {}} + + def _exploding_post_process(*a, **kw): + raise RuntimeError("post-process boom") + + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: ('/downloads/song.flac', 'download'), + post_process_with_verification=_exploding_post_process, + ) + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' + assert 'Post-processing failed' in download_tasks['t1']['error_message'] + assert ('on_complete', ('b1', 't1', False), {}) in rec.calls + + +def test_critical_outer_exception_marks_failed(): + """Top-level exception (e.g. broken deps) still marks task failed.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {}, + } + + def _broken_resolve(p): + raise RuntimeError("config dead") + + deps, rec = _build_deps(docker_resolve_path=_broken_resolve) + # Must NOT raise + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' + assert 'Critical post-processing error' in download_tasks['t1']['error_message'] + assert ('on_complete', ('b1', 't1', False), {}) in rec.calls + + +def test_youtube_task_uses_get_download_status_to_resolve_path(monkeypatch): + """YouTube downloads use a different filename scheme — worker queries soulseek client for real path.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'vid_id||Money', + 'username': 'youtube', + 'download_id': 'dl-yt-1', + 'track_info': {}, + } + + class _FakeStatus: + file_path = '/downloads/Money.mp3' + + class _FakeYTClient: + def get_download_status(self, dl_id): + assert dl_id == 'dl-yt-1' + return _FakeStatus() + + # File exists on disk (mock) + monkeypatch.setattr(pp.os.path, 'exists', lambda p: p == '/downloads/Money.mp3') + + deps, rec = _build_deps( + soulseek_client=_FakeYTClient(), + run_async=lambda coro: coro, # not async — direct call + ) + pp.run_post_processing_worker('t1', 'b1', deps) + # mark_completed should fire (file resolved from YouTube status) + assert any(c[0] == 'mark_completed' for c in rec.calls) + + +def test_fuzzy_context_matching_when_exact_key_missing(): + """When exact key isn't in matched_downloads_context, worker tries fuzzy match + constrained to same Soulseek username.""" + download_tasks['t1'] = { + 'status': 'post_processing', + 'filename': 'song.flac', + 'username': 'u1', + 'track_info': {}, + } + # Different exact key but same user + filename substring + matched_downloads_context['u1::folder/song.flac'] = { + 'original_search_result': {'title': 'Money', 'track_number': 1}, + } + deps, rec = _build_deps( + find_completed_file=lambda *a, **kw: (None, None), # file not found + ) + # Won't find file → marks failed. But the fuzzy match log path executes. + pp.run_post_processing_worker('t1', 'b1', deps) + assert download_tasks['t1']['status'] == 'failed' diff --git a/web_server.py b/web_server.py index b37a2d9a..1e6df444 100644 --- a/web_server.py +++ b/web_server.py @@ -22509,412 +22509,33 @@ def _run_full_missing_tracks_process(batch_id, playlist_id, tracks_json): 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 + + +def _build_post_processing_deps(): + """Build the PostProcessDeps bundle from web_server.py globals on each call.""" + return _downloads_post_processing.PostProcessDeps( + config_manager=config_manager, + soulseek_client=soulseek_client, + run_async=run_async, + docker_resolve_path=docker_resolve_path, + extract_filename=extract_filename, + make_context_key=_make_context_key, + find_completed_file=_find_completed_file_robust, + enhance_file_metadata=_enhance_file_metadata, + wipe_source_tags=_wipe_source_tags, + post_process_with_verification=_post_process_matched_download_with_verification, + mark_task_completed=_mark_task_completed, + on_download_completed=_on_download_completed, + ) + + def _run_post_processing_worker(task_id, batch_id): - """ - NEW VERIFICATION WORKFLOW: Post-processing worker that only sets 'completed' status - after successful file verification and processing. This matches sync.py's reliability. - """ - try: - logger.info(f"[Post-Processing] Starting verification for task {task_id}") - - # Retrieve task details from global state - with tasks_lock: - if task_id not in download_tasks: - logger.warning(f"[Post-Processing] Task {task_id} not found in download_tasks") - return - task = download_tasks[task_id].copy() - - # Check if task was cancelled or already completed during post-processing - if task['status'] == 'cancelled': - logger.warning(f"[Post-Processing] Task {task_id} was cancelled, skipping verification") - return - if task['status'] == 'completed' or task.get('stream_processed'): - logger.info(f"[Post-Processing] Task {task_id} already completed by stream processor, skipping verification") - return - - # Extract file information for verification - track_info = task.get('track_info', {}) - task_filename = task.get('filename') or track_info.get('filename') - task_username = task.get('username') or track_info.get('username') - - if not task_filename or not task_username: - logger.warning(f"[Post-Processing] Missing filename or username for task {task_id}") - with tasks_lock: - if task_id in download_tasks: - download_tasks[task_id]['status'] = 'failed' - download_tasks[task_id]['error_message'] = 'Post-processing failed: missing file or source information from Soulseek transfer' - _on_download_completed(batch_id, task_id, success=False) - return - - download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer')) - - # Try to get context for generating the correct final filename - task_basename = extract_filename(task_filename) - context_key = _make_context_key(task_username, task_filename) - expected_final_filename = None - - logger.info(f"[Post-Processing] Looking up context with key: {context_key}") - - with matched_context_lock: - context = matched_downloads_context.get(context_key) - # Debug: Show all available context keys - available_keys = list(matched_downloads_context.keys()) - logger.info(f"[Post-Processing] Available context keys: {available_keys[:10]}...") # Show first 10 keys - - if context: - logger.info(f"[Post-Processing] Found context for key: {context_key}") - try: - original_search = context.get("original_search_result", {}) - logger.info(f"[Post-Processing] original_search keys: {list(original_search.keys())}") - - clean_title = get_import_clean_title(context, default=original_search.get('title', '')) - track_number = original_search.get('track_number') - - logger.info(f"[Post-Processing] clean_title: '{clean_title}', track_number: {track_number}") - - if clean_title and track_number: - # Generate expected final filename that stream processor would create - # Pattern: f"{track_number:02d} - {clean_title}.flac" - sanitized_title = clean_title.replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_') - expected_final_filename = f"{track_number:02d} - {sanitized_title}.flac" - logger.info(f"[Post-Processing] Generated expected final filename: {expected_final_filename}") - else: - logger.warning(f"[Post-Processing] Missing required data - clean_title: {bool(clean_title)}, track_number: {bool(track_number)}") - except Exception as e: - logger.error(f"[Post-Processing] Error generating expected filename: {e}") - import traceback - traceback.print_exc() - else: - logger.warning(f"[Post-Processing] No context found for key: {context_key}") - # Try fuzzy matching with similar keys containing the filename - # SAFETY: Constrain to same Soulseek username to prevent cross-album - # metadata contamination during mass downloads (e.g., two albums both - # having "01 - Intro.flac" would match the wrong context without this) - with matched_context_lock: - similar_keys = [k for k in matched_downloads_context.keys() - if k.startswith(f"{task_username}::") and task_basename in k] - if similar_keys: - # Use the first similar key found - fuzzy_key = similar_keys[0] - context = matched_downloads_context.get(fuzzy_key) - logger.info(f"[Post-Processing] Found context using fuzzy key matching: {fuzzy_key}") - - # Generate expected final filename using the found context - try: - original_search = context.get("original_search_result", {}) - logger.info(f"[Post-Processing] fuzzy context original_search keys: {list(original_search.keys())}") - - clean_title = get_import_clean_title(context, default=original_search.get('title', '')) - track_number = original_search.get('track_number') - - logger.info(f"[Post-Processing] fuzzy context clean_title: '{clean_title}', track_number: {track_number}") - - if clean_title and track_number: - # Generate expected final filename that stream processor would create - # Pattern: f"{track_number:02d} - {clean_title}.flac" - sanitized_title = clean_title.replace('/', '_').replace('\\', '_').replace(':', '_').replace('*', '_').replace('?', '_').replace('"', '_').replace('<', '_').replace('>', '_').replace('|', '_') - expected_final_filename = f"{track_number:02d} - {sanitized_title}.flac" - logger.info(f"[Post-Processing] Generated expected final filename from fuzzy match: {expected_final_filename}") - else: - logger.warning(f"[Post-Processing] Missing required data from fuzzy match - clean_title: {bool(clean_title)}, track_number: {bool(track_number)}") - except Exception as e: - logger.error(f"[Post-Processing] Error generating expected filename from fuzzy match: {e}") - import traceback - traceback.print_exc() - else: - logger.warning(f"[Post-Processing] No similar keys found containing '{task_basename}'") - # Show a sample of what keys actually exist for debugging - sample_keys = list(matched_downloads_context.keys())[:5] - logger.info(f"[Post-Processing] Sample of existing keys: {sample_keys}") - - # RESILIENT FILE-FINDING LOOP: Try up to 3 times with delays - found_file = None - file_location = None + """Post-processing verification worker — see core/downloads/post_processing.py.""" + _downloads_post_processing.run_post_processing_worker(task_id, batch_id, _build_post_processing_deps()) - # CRITICAL FIX: For YouTube downloads, the filename in task is 'id||title' (metadata), - # but the actual file on disk is 'Title.mp3'. We must ask the client for the real path. - if (task.get('username') == 'youtube' or '||' in str(task_filename)) and not found_file: - logger.info(f"[Post-Processing] Detected YouTube download task: {task_id}") - try: - # Query the download orchestrator for the status which contains the real file path - # CRITICAL FIX: Use the actual download_id designated by the client, not the internal task_id - actual_download_id = task.get('download_id') or task_id - status = run_async(soulseek_client.get_download_status(actual_download_id)) - if status and status.file_path: - real_path = status.file_path - if os.path.exists(real_path): - # Determine if it's in download or transfer directory - real_path_obj = Path(real_path) - download_dir_obj = Path(download_dir) - transfer_dir_obj = Path(transfer_dir) - - # Use absolute path comparison - try: - if download_dir_obj.resolve() in real_path_obj.resolve().parents: - file_location = 'download' - elif transfer_dir_obj.resolve() in real_path_obj.resolve().parents: - file_location = 'transfer' - else: - file_location = 'absolute' - except: - # Fallback if resolve fails (e.g. permission or path issues) - file_location = 'absolute' - if file_location: - # We found the file! Use the absolute path if it confuses the joining logic, - # but usually we want just the filename if location is 'download'/'transfer' - # CRITICAL FIX: Always use the absolute real_path. - # Stripping to basename causes FileNotFoundError because post-processing - # runs with CWD as project root, not download dir. - found_file = real_path - - logger.info(f"[Post-Processing] Resolved actual YouTube filename: {found_file} (Location: {file_location})") - else: - logger.warning(f"[Post-Processing] YouTube status reported path but file missing: {real_path}") - else: - logger.warning(f"[Post-Processing] YouTube status returned no file_path for task {task_id}") - except Exception as e: - logger.error(f"[Post-Processing] Failed to retrieve YouTube task status: {e}") - - _file_search_max_retries = 5 - for retry_count in range(_file_search_max_retries): - # If we already resolved the file (e.g. via YouTube status), skip searching - if found_file: - logger.info(f"[Post-Processing] Skipping search loop, file already resolved: {found_file}") - break - - # Check if stream processor already completed this task while we were waiting - with tasks_lock: - if task_id in download_tasks: - if download_tasks[task_id].get('stream_processed') or download_tasks[task_id]['status'] == 'completed': - logger.info(f"[Post-Processing] Task {task_id} was completed by stream processor during file search - done") - return - - logger.warning(f"[Post-Processing] Attempt {retry_count + 1}/{_file_search_max_retries} to find file") - logger.info(f"[Post-Processing] Original filename: {task_basename}") - if expected_final_filename: - logger.info(f"[Post-Processing] Expected final filename: {expected_final_filename}") - else: - logger.warning("[Post-Processing] No expected final filename available") - - # Strategy 1: Try with original filename in both downloads and transfer - logger.info("[Post-Processing] Strategy 1: Searching with original filename...") - found_file, file_location = _find_completed_file_robust(download_dir, task_filename, transfer_dir) - - if found_file: - logger.info(f"[Post-Processing] Strategy 1 SUCCESS: Found file with original filename in {file_location}: {found_file}") - else: - logger.error("[Post-Processing] Strategy 1 FAILED: Original filename not found in either location") - - # Strategy 2: If not found and we have an expected final filename, try that in transfer folder - if not found_file and expected_final_filename: - logger.info("[Post-Processing] Strategy 2: Searching transfer folder with expected final filename...") - found_result = _find_completed_file_robust(transfer_dir, expected_final_filename) - if found_result and found_result[0]: - found_file, file_location = found_result[0], 'transfer' - logger.info(f"[Post-Processing] Strategy 2 SUCCESS: Found file with expected final filename: {found_file}") - else: - logger.error("[Post-Processing] Strategy 2 FAILED: Expected final filename not found in transfer folder") - elif not expected_final_filename: - logger.warning("[Post-Processing] Strategy 2 SKIPPED: No expected final filename available") - - if found_file: - logger.warning(f"[Post-Processing] FILE FOUND after {retry_count + 1} attempts in {file_location}: {found_file}") - break - else: - logger.error(f"[Post-Processing] All search strategies failed on attempt {retry_count + 1}/{_file_search_max_retries}") - if retry_count < _file_search_max_retries - 1: # Don't sleep on final attempt - logger.info("[Post-Processing] Waiting 5 seconds before next attempt...") - time.sleep(5) - - if not found_file: - # CRITICAL: Before marking as failed, check if stream processor already handled this - # The /api/downloads/status polling endpoint processes files independently and may have - # already moved/renamed/tagged the file successfully while we were searching - with tasks_lock: - if task_id in download_tasks: - if download_tasks[task_id].get('stream_processed') or download_tasks[task_id]['status'] == 'completed': - logger.error(f"[Post-Processing] Task {task_id} was completed by stream processor - not marking as failed") - return - download_tasks[task_id]['status'] = 'failed' - download_tasks[task_id]['error_message'] = f'File not found on disk after {_file_search_max_retries} search attempts. Expected: {os.path.basename(task_filename)}' - _on_download_completed(batch_id, task_id, success=False) - return - - # Handle file found in transfer folder - already completed by stream processor - if file_location == 'transfer': - logger.info(f"[Post-Processing] File found in transfer folder - already completed by stream processor: {found_file}") - - # Check if metadata enhancement was completed - metadata_enhanced = False - with tasks_lock: - if task_id in download_tasks: - metadata_enhanced = download_tasks[task_id].get('metadata_enhanced', False) - - if not metadata_enhanced: - logger.warning("[Post-Processing] File in transfer folder missing metadata enhancement - completing now") - # Attempt to complete metadata enhancement using context - if context and expected_final_filename: - try: - context = normalize_import_context(context) - # Extract required data from context - original_search = get_import_original_search(context) - artist_context = get_import_context_artist(context) - album_context = get_import_context_album(context) - - if artist_context and album_context: - # CRITICAL FIX: Create album_info dict with proper structure for metadata enhancement - # This must match the format used in main stream processor to ensure consistency - - # Extract track number from context (should be available from fuzzy match) - track_number = original_search.get('track_number', 1) - - # If no track number in context, extract from filename - if track_number == 1 and found_file: - track_number = extract_track_number_from_filename(found_file) - logger.warning( - "[Verification] missing track_number; extracted from filename=%r -> %s", - os.path.basename(found_file), - track_number, - ) - - # Ensure track_number is valid - if not isinstance(track_number, int) or track_number < 1: - logger.error(f"[Verification] Invalid track number ({track_number}), defaulting to 1") - track_number = 1 - - # Get clean track name - clean_track_name = get_import_clean_title(context, default=original_search.get('title', 'Unknown Track')) - album_name = get_import_clean_album(context, default=album_context.get('name', 'Unknown Album')) - album_image_url = album_context.get('image_url') - if not album_image_url and album_context.get('images'): - album_images = album_context.get('images', []) - if album_images and isinstance(album_images[0], dict): - album_image_url = album_images[0].get('url') - - album_info = { - 'is_album': True, # CRITICAL: Mark as album track - 'album_name': album_name, - 'track_number': track_number, # CORRECTED TRACK NUMBER - 'disc_number': original_search.get('disc_number', 1), - 'clean_track_name': clean_track_name, - 'album_image_url': album_image_url, - 'confidence': 0.9, - 'source': 'verification_worker_corrected' - } - - # Apply album grouping for consistency with stream processor path. - # Only for singles/auto-detected — explicit album downloads already - # have the correct Spotify name and re-grouping would mangle it. - if not context.get("is_album_download", False): - try: - raw_album_ctx = original_search.get('album') - if isinstance(raw_album_ctx, str): - original_album_ctx = raw_album_ctx - elif isinstance(raw_album_ctx, dict) and 'name' in raw_album_ctx: - original_album_ctx = raw_album_ctx['name'] - else: - original_album_ctx = None - consistent_album_name = _resolve_album_group(artist_context, album_info, original_album_ctx) - album_info['album_name'] = consistent_album_name - except Exception as group_err: - logger.error(f"[Verification] Album grouping failed, using raw name: {group_err}") - else: - logger.info(f"[Verification] Explicit album download - preserving album name: '{album_info['album_name']}'") - - logger.info(f"[Verification] Created proper album_info - track_number: {track_number}, album: {album_info['album_name']}") - - logger.info(f"[Post-Processing] Attempting metadata enhancement for: {found_file}") - logger.warning(f"[Metadata Input] Verification worker - artist: '{artist_context.get('name', 'MISSING')}' (id: {artist_context.get('id', 'MISSING')})") - logger.warning(f"[Metadata Input] Verification worker - album: '{album_info.get('album_name', 'MISSING')}', track#: {album_info.get('track_number', 'MISSING')}, source: {album_info.get('source', 'unknown')}") - enhancement_success = _enhance_file_metadata(found_file, context, artist_context, album_info) - - if enhancement_success: - with tasks_lock: - if task_id in download_tasks: - download_tasks[task_id]['metadata_enhanced'] = True - logger.info(f"[Post-Processing] Successfully completed metadata enhancement for: {os.path.basename(found_file)}") - else: - logger.info(f"[Post-Processing] Metadata enhancement returned False for: {os.path.basename(found_file)}") - else: - logger.warning("[Post-Processing] Missing artist or album in context") - logger.info(f"[Post-Processing] artist_context: {artist_context is not None}, album_context: {album_context is not None}") - # Wipe source tags even without full enhancement — prevents - # Soulseek uploader's MusicBrainz IDs from causing album splits - if found_file and os.path.exists(found_file): - _wipe_source_tags(found_file) - except Exception as enhancement_error: - import traceback - logger.error(f"[Post-Processing] Error during metadata enhancement: {enhancement_error}\n{traceback.format_exc()}") - if found_file and os.path.exists(found_file): - _wipe_source_tags(found_file) - else: - logger.warning("[Post-Processing] Cannot complete metadata enhancement - missing context or expected filename") - if found_file and os.path.exists(found_file): - _wipe_source_tags(found_file) - else: - logger.info("[Post-Processing] File already has metadata enhancement completed") - - with tasks_lock: - if task_id in download_tasks: - track_info = download_tasks[task_id].get('track_info') - _mark_task_completed(task_id, track_info) - - # Clean up context now that both stream processor and verification worker are done - with matched_context_lock: - if context_key in matched_downloads_context: - del matched_downloads_context[context_key] - logger.info(f"[Verification] Cleaned up context after successful verification: {context_key}") - - _on_download_completed(batch_id, task_id, success=True) - return - - # File found in downloads folder - attempt post-processing - try: - # Rebuild the context key using the same function that stored it - context_key = _make_context_key(task_username, task_filename) - - # Check if this download has matched context for post-processing - with matched_context_lock: - context = matched_downloads_context.get(context_key) - - if context: - logger.info(f"[Post-Processing] Found matched context, running full post-processing for: {context_key}") - # Run the existing post-processing logic with verification - _post_process_matched_download_with_verification(context_key, context, found_file, task_id, batch_id) - else: - # No matched context - just mark as completed since file exists - logger.warning(f"[Post-Processing] No matched context, marking as completed: {os.path.basename(found_file)}") - with tasks_lock: - if task_id in download_tasks: - track_info = download_tasks[task_id].get('track_info') - _mark_task_completed(task_id, track_info) - - # Clean up context if it exists (might be leftover from stream processor) - with matched_context_lock: - if context_key in matched_downloads_context: - del matched_downloads_context[context_key] - logger.info(f"[Verification] Cleaned up leftover context: {context_key}") - - # Call completion callback since there's no other post-processing to handle it - _on_download_completed(batch_id, task_id, success=True) - - except Exception as processing_error: - logger.error(f"[Post-Processing] Processing failed for task {task_id}: {processing_error}") - with tasks_lock: - if task_id in download_tasks: - download_tasks[task_id]['status'] = 'failed' - download_tasks[task_id]['error_message'] = f"Post-processing failed: {str(processing_error)}" - _on_download_completed(batch_id, task_id, success=False) - - except Exception as e: - logger.error(f"[Post-Processing] Critical error in post-processing worker for task {task_id}: {e}") - with tasks_lock: - if task_id in download_tasks: - download_tasks[task_id]['status'] = 'failed' - download_tasks[task_id]['error_message'] = f"Critical post-processing error: {str(e)}" - _on_download_completed(batch_id, task_id, success=False) def _download_track_worker(task_id, batch_id=None): """