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

Merge pull request #397 from Nezreka/refactor/lift-downloads-post-process

Browse source 
PR4d: lift _run_post_processing_worker to core/downloads/post_process…
This commit is contained in:
BoulderBadgeDad 2026-04-27 23:05:12 -07:00 committed by GitHub
commit 2e61000ecf
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 849 additions and 403 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

481
core/downloads/post_processing.py Normal file
View file

@ -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',
]

344
tests/downloads/test_downloads_post_processing.py Normal file
View file

@ -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'

427
web_server.py
View file

@ -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):
"""