From 2d271cfacfb02862858b174221ef8a1766152d37 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Mon, 27 Apr 2026 23:26:26 -0700 Subject: [PATCH] PR4e: lift status helpers + 3 routes to core/downloads/status.py MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fifth sub-PR in the download orchestrator series. Strict 1:1 lift — zero behavior change. What moved: - _build_batch_status_data → build_batch_status_data - get_batch_download_status route body → build_single_batch_status - get_batched_download_statuses route body → build_batched_status - get_all_downloads_unified route body → build_unified_downloads_response - Status priority dict → module-level _STATUS_PRIORITY constant Dependencies bundled in `StatusDeps` dataclass: - config_manager, docker_resolve_path, find_completed_file, make_context_key, submit_post_processing (lambda wrapping missing_download_executor.submit + _run_post_processing_worker), get_cached_transfer_data Direct imports from core.runtime_state for download_tasks / download_batches / tasks_lock (already lifted by kettui). Behavior parity: - Same response payload shape across all 3 endpoints - Same safety-valve mutation: stuck downloading task with file recovered → status='post_processing' + submit worker; stuck searching → not_found; stuck downloading no file → failed - Same live transfer state mapping (Cancelled/Canceled, Failed/Errored/ Rejected/TimedOut, Completed/Succeeded with byte-mismatch verification, InProgress, default queued) - Same intermediate post_processing status promotion + single-shot worker submission (only when status != 'post_processing') - Same 'Errored' handling: keeps current status to let monitor retry - Same 17-key item dict in unified response with same field order - Same artist/album/artwork normalization (handles string, dict, list, list-of-dicts, list-of-strings variants) - Same sort: (priority asc, -timestamp desc) - Same batch summary aggregation - Same items[:limit] slicing - Same logger messages text-for-text - Same lock scope (single tasks_lock per call) — no new contention Pre-existing bug preserved (will fix in follow-up PR): - batched_status `debug_info` block iterates `response["batches"]` and guards with `if "error" not in batch_status`. Every successful payload includes `"error": batch.get('error')` (key always present, value usually None) so the guard is always False and debug_info never populates in production. Test documents the buggy behavior so the next PR can flip the check to `batch_status.get('error') is None`. Tests: 32 new under tests/downloads/test_downloads_status.py covering phase routing (analysis vs downloading vs unknown), task formatting + sort + V2 fields, every live transfer state mapping (Cancelled, Succeeded with full + partial bytes, InProgress, Errored, terminal- not-overridden), safety valve (stuck searching → not_found, stuck downloading recovered → post_processing, stuck downloading no file → failed), all 3 route helpers (single, batched, unified), unified artist/album/artwork normalization, batch summary aggregation, limit slicing, plus debug_info bug documentation. Full suite: 982 passing (was 950). Ruff clean. --- core/downloads/status.py | 410 +++++++++++++++++++ tests/downloads/test_downloads_status.py | 477 +++++++++++++++++++++++ web_server.py | 399 ++----------------- 3 files changed, 916 insertions(+), 370 deletions(-) create mode 100644 core/downloads/status.py create mode 100644 tests/downloads/test_downloads_status.py diff --git a/core/downloads/status.py b/core/downloads/status.py new file mode 100644 index 00000000..a3359272 --- /dev/null +++ b/core/downloads/status.py @@ -0,0 +1,410 @@ +"""Batch + unified download status helpers. + +`build_batch_status_data` is the per-batch payload formatter shared by: +- /api/playlists//download_status (single batch) +- /api/download_status/batch (multiple batches in one call) + +It's NOT pure read-only — it has a safety valve that mutates task state +when slskd reports terminal-but-stuck downloads, and it submits the +post-processing worker when slskd reports 'Succeeded' or when a stuck +file is recovered. Those side effects are preserved exactly. + +`build_unified_downloads_response` powers /api/downloads/all — flattens +all tasks across batches into one sorted list with per-row metadata for +the centralized Downloads page. + +Lifted verbatim from web_server.py. Dependencies that touch the live +runtime (config, file finder, post-processing submitter, transfer cache) +are passed via `StatusDeps` so the module is web_server-import-free. +""" + +from __future__ import annotations + +import logging +import time +from dataclasses import dataclass +from typing import Any, Callable, Optional + +from core.runtime_state import ( + download_batches, + download_tasks, + tasks_lock, +) + +logger = logging.getLogger(__name__) + + +@dataclass +class StatusDeps: + """Cross-cutting deps the status helpers need.""" + config_manager: Any + docker_resolve_path: Callable[[str], str] + find_completed_file: Callable + make_context_key: Callable[[str, str], str] + submit_post_processing: Callable[[str, str], None] # (task_id, batch_id) -> None + get_cached_transfer_data: Callable[[], dict] + + +def build_batch_status_data(batch_id: str, batch: dict, live_transfers_lookup: dict, deps: StatusDeps) -> dict: + """Build status payload for a single batch. + + Includes a safety-valve that mutates stuck task state and submits the + post-processing worker when slskd reports 'Succeeded' or when a + stuck-but-recovered file is found on disk. + """ + response_data = { + "phase": batch.get('phase', 'unknown'), + "error": batch.get('error'), + "auto_initiated": batch.get('auto_initiated', False), + "playlist_id": batch.get('playlist_id'), # Include playlist_id for rehydration + "playlist_name": batch.get('playlist_name'), # Include playlist_name for reference + } + + if response_data["phase"] == 'analysis': + response_data['analysis_progress'] = { + 'total': batch.get('analysis_total', 0), + 'processed': batch.get('analysis_processed', 0), + } + response_data['analysis_results'] = batch.get('analysis_results', []) + + elif response_data["phase"] in ['downloading', 'complete', 'error']: + response_data['analysis_results'] = batch.get('analysis_results', []) + batch_tasks = [] + for task_id in batch.get('queue', []): + task = download_tasks.get(task_id) + if not task: + continue + + # SAFETY VALVE: Check for downloads stuck too long + current_time = time.time() + task_start_time = task.get('status_change_time', current_time) + task_age = current_time - task_start_time + + # If task has been running too long, check if file completed + _dl_timeout = deps.config_manager.get('soulseek.download_timeout', 600) or 600 + if task_age > _dl_timeout and task['status'] in ['downloading', 'queued', 'searching']: + stuck_state = task['status'] + task_filename = task.get('filename') or (task.get('track_info') or {}).get('filename') + + # Before failing, check if the file actually downloaded successfully + recovered = False + if task_filename and stuck_state == 'downloading': + try: + 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')) + found_file, file_location = deps.find_completed_file(download_dir, task_filename, transfer_dir) + if found_file: + logger.info(f"[Safety Valve] Task {task_id} stuck but file found in {file_location} — routing to post-processing") + task['status'] = 'post_processing' + task['status_change_time'] = current_time + deps.submit_post_processing(task_id, batch_id) + recovered = True + except Exception as e: + logger.error(f"[Safety Valve] Error checking for completed file: {e}") + + if not recovered: + if stuck_state == 'searching': + logger.info(f"⏰ [Safety Valve] Task {task_id} stuck in searching for {task_age:.1f}s - marking not_found") + task['status'] = 'not_found' + task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out' + else: + logger.error(f"⏰ [Safety Valve] Task {task_id} stuck for {task_age:.1f}s - forcing failure") + task['status'] = 'failed' + task['error_message'] = f'Task stuck in {stuck_state} state for {int(task_age // 60)} minutes — forcibly stopped' + + task_status = { + 'task_id': task_id, + 'track_index': task['track_index'], + 'status': task['status'], + 'track_info': task['track_info'], + 'progress': 0, + # V2 SYSTEM: Add persistent state information + 'cancel_requested': task.get('cancel_requested', False), + 'cancel_timestamp': task.get('cancel_timestamp'), + 'ui_state': task.get('ui_state', 'normal'), # normal|cancelling|cancelled + 'playlist_id': task.get('playlist_id'), # For V2 system identification + 'error_message': task.get('error_message'), # Surface failure reasons to UI + 'has_candidates': bool(task.get('cached_candidates')), # Whether search found results (for clickable review) + } + _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} + task_filename = task.get('filename') or _ti.get('filename') + task_username = task.get('username') or _ti.get('username') + if task_filename and task_username: + lookup_key = deps.make_context_key(task_username, task_filename) + + if lookup_key in live_transfers_lookup: + live_info = live_transfers_lookup[lookup_key] + state_str = live_info.get('state', 'Unknown') + + # Don't override tasks that are already in terminal states or post-processing + if task['status'] not in ['completed', 'failed', 'cancelled', 'not_found', 'post_processing']: + # SYNC.PY PARITY: Prioritized state checking (Errored/Cancelled before Completed) + # This prevents "Completed, Errored" states from being marked as completed + if 'Cancelled' in state_str or 'Canceled' in state_str: + task_status['status'] = 'cancelled' + task['status'] = 'cancelled' + elif 'Failed' in state_str or 'Errored' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str: + # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency + # Monitor will detect errored state and trigger retry within 5 seconds + logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry") + + # Keep task in current status (downloading/queued) so monitor can detect error + # Don't mark as failed here - let the unified retry system handle it + if task['status'] in ['searching', 'downloading', 'queued']: + task_status['status'] = task['status'] # Keep current status for monitor + else: + task_status['status'] = 'downloading' # Default to downloading for error detection + task['status'] = 'downloading' + elif 'Completed' in state_str or 'Succeeded' in state_str: + # Verify bytes actually transferred before trusting state string + expected_size = live_info.get('size', 0) + transferred = live_info.get('bytesTransferred', 0) + if expected_size > 0 and transferred < expected_size: + # State says complete but bytes don't match — keep current status + task_status['status'] = task['status'] + logger.info(f"Task {task_id} state says complete but bytes incomplete ({transferred}/{expected_size})") + # NEW VERIFICATION WORKFLOW: Use intermediate post_processing status + # Only set this status once to prevent multiple worker submissions + elif task['status'] != 'post_processing': + task_status['status'] = 'post_processing' + task['status'] = 'post_processing' + logger.info(f"Task {task_id} API reports 'Succeeded' - starting post-processing verification") + + # Submit post-processing worker to verify file and complete the task + deps.submit_post_processing(task_id, batch_id) + else: + # FIXED: Always require verification workflow - no bypass for stream processed tasks + # Stream processing only handles metadata, not file verification + task_status['status'] = 'post_processing' + logger.info(f"Task {task_id} waiting for verification worker to complete") + elif 'InProgress' in state_str: + task_status['status'] = 'downloading' + else: + task_status['status'] = 'queued' + task_status['progress'] = live_info.get('percentComplete', 0) + # For completed/post-processing tasks, keep appropriate progress + elif task['status'] == 'completed': + task_status['progress'] = 100 + elif task['status'] == 'post_processing': + task_status['progress'] = 95 # Nearly complete, just verifying + else: + # If task is completed but not in live transfers, keep appropriate status + if task['status'] == 'completed': + task_status['progress'] = 100 + elif task['status'] == 'post_processing': + task_status['progress'] = 95 # Nearly complete, just verifying + batch_tasks.append(task_status) + batch_tasks.sort(key=lambda x: x['track_index']) + response_data['tasks'] = batch_tasks + + # CRITICAL: Add batch worker management metadata (was missing!) + # This is essential for client-side worker validation and prevents false desync warnings + response_data['active_count'] = batch.get('active_count', 0) + response_data['max_concurrent'] = batch.get('max_concurrent', 3) + + # Add wishlist summary if batch is complete (matching sync.py behavior) + if response_data["phase"] == 'complete' and 'wishlist_summary' in batch: + response_data['wishlist_summary'] = batch['wishlist_summary'] + + return response_data + + +# --------------------------------------------------------------------------- +# Route-shaped builders +# --------------------------------------------------------------------------- + +def build_single_batch_status(batch_id: str, deps: StatusDeps) -> tuple[Optional[dict], int]: + """For /api/playlists//download_status. Returns (response, status).""" + live_transfers_lookup = deps.get_cached_transfer_data() + + with tasks_lock: + if batch_id not in download_batches: + return {"error": "Batch not found"}, 404 + + batch = download_batches[batch_id] + return build_batch_status_data(batch_id, batch, live_transfers_lookup, deps), 200 + + +def build_batched_status(requested_batch_ids: list, deps: StatusDeps) -> dict: + """For /api/download_status/batch. Returns the full response dict (always 200).""" + live_transfers_lookup = deps.get_cached_transfer_data() + response: dict[str, Any] = {"batches": {}} + + with tasks_lock: + if requested_batch_ids: + target_batches = { + bid: batch for bid, batch in download_batches.items() + if bid in requested_batch_ids + } + else: + target_batches = download_batches.copy() + + for batch_id, batch in target_batches.items(): + try: + response["batches"][batch_id] = build_batch_status_data( + batch_id, batch, live_transfers_lookup, deps, + ) + except Exception as batch_error: + logger.error(f"Error processing batch {batch_id}: {batch_error}") + response["batches"][batch_id] = {"error": str(batch_error)} + + response["metadata"] = { + "total_batches": len(response["batches"]), + "requested_batch_ids": requested_batch_ids, + "timestamp": time.time(), + } + + debug_info = {} + for batch_id, batch_status in response["batches"].items(): + if "error" not in batch_status: + active_count = batch_status.get("active_count", 0) + max_concurrent = batch_status.get("max_concurrent", 3) + task_count = len(batch_status.get("tasks", [])) + active_tasks = len([t for t in batch_status.get("tasks", []) if t.get("status") in ['searching', 'downloading', 'queued']]) + + debug_info[batch_id] = { + "reported_active": active_count, + "actual_active_tasks": active_tasks, + "max_concurrent": max_concurrent, + "total_tasks": task_count, + "worker_discrepancy": active_count != active_tasks, + } + + response["debug_info"] = debug_info + + logger.info(f"[Batched Status] Returning status for {len(response['batches'])} batches") + + discrepancies = [bid for bid, info in debug_info.items() if info.get("worker_discrepancy")] + if discrepancies: + logger.info(f"[Batched Status] Worker count discrepancies in batches: {discrepancies}") + + return response + + +_STATUS_PRIORITY = { + 'downloading': 0, 'searching': 1, 'post_processing': 2, + 'queued': 3, 'pending': 3, + 'completed': 4, 'skipped': 5, 'already_owned': 5, + 'not_found': 6, 'failed': 7, 'cancelled': 8, +} + + +def build_unified_downloads_response(limit: int, deps: StatusDeps) -> dict: + """Flat list of every task across batches, sorted active-first then by recency. + + Powers /api/downloads/all for the centralized Downloads page. + """ + items = [] + with tasks_lock: + for task_id, task in download_tasks.items(): + track_info = task.get('track_info') or {} + batch_id = task.get('batch_id', '') + batch = download_batches.get(batch_id, {}) + + # Extract track metadata — handle all format variations + title = '' + artist = '' + album = '' + artwork = '' + if isinstance(track_info, dict): + title = track_info.get('title') or track_info.get('name') or track_info.get('track_name') or '' + + # Artist can be: string, list of strings, list of dicts with 'name' + raw_artist = track_info.get('artist') or track_info.get('artist_name') or track_info.get('artists') or '' + if isinstance(raw_artist, list): + parts = [] + for a in raw_artist: + if isinstance(a, dict): + parts.append(a.get('name', '')) + else: + parts.append(str(a)) + artist = ', '.join(p for p in parts if p) + elif isinstance(raw_artist, dict): + artist = raw_artist.get('name', '') + else: + artist = str(raw_artist) if raw_artist else '' + + # Album can be: string or dict with 'name' + raw_album = track_info.get('album') or track_info.get('album_name') or '' + if isinstance(raw_album, dict): + album = raw_album.get('name', '') + else: + album = str(raw_album) if raw_album else '' + + artwork = track_info.get('artwork_url') or track_info.get('image_url') or track_info.get('album_art') or '' + # Try album images + if not artwork: + raw_alb = track_info.get('album') + if isinstance(raw_alb, dict): + images = raw_alb.get('images') or [] + if images and isinstance(images, list) and len(images) > 0: + artwork = images[0].get('url', '') if isinstance(images[0], dict) else str(images[0]) + + status = task.get('status', 'queued') + # Determine download progress percentage + progress = 0 + if status == 'completed': + progress = 100 + elif status == 'post_processing': + progress = 95 + elif status in ('downloading', 'searching'): + # Check live transfer data for real progress + task_filename = task.get('filename') or track_info.get('filename') + task_username = task.get('username') or track_info.get('username') + if task_filename and task_username: + lookup_key = deps.make_context_key(task_username, task_filename) + live_info = deps.get_cached_transfer_data().get(lookup_key) + if live_info: + progress = live_info.get('percentComplete', 0) + + items.append({ + 'task_id': task_id, + 'title': title, + 'artist': artist, + 'album': album, + 'artwork': artwork, + 'status': status, + 'progress': progress, + 'error': task.get('error_message'), + 'batch_id': batch_id, + 'batch_name': batch.get('playlist_name') or batch.get('album_name') or '', + 'batch_source': batch.get('source_page') or batch.get('initiated_from') or '', + # playlist_id is needed by per-row cancel (cancel_task_v2 + # takes playlist_id + track_index). Surfacing it here so + # the frontend doesn't need a second lookup. + 'playlist_id': batch.get('playlist_id', ''), + 'track_index': task.get('track_index', 0), + 'batch_total': len(batch.get('queue', [])), + 'timestamp': task.get('status_change_time', 0), + 'priority': _STATUS_PRIORITY.get(status, 9), + }) + + # Sort: active first (by priority), then by timestamp desc within each group + items.sort(key=lambda x: (x['priority'], -x['timestamp'])) + + # Build batch summaries for the batch context panel + batch_summaries = [] + with tasks_lock: + for bid, batch in download_batches.items(): + queue = batch.get('queue', []) + statuses = [download_tasks[tid]['status'] for tid in queue if tid in download_tasks] + batch_summaries.append({ + 'batch_id': bid, + 'playlist_id': batch.get('playlist_id', ''), + 'batch_name': batch.get('playlist_name') or batch.get('album_name') or '', + 'source_page': batch.get('source_page') or batch.get('initiated_from') or '', + 'phase': batch.get('phase', 'unknown'), + 'total': len(queue), + 'completed': sum(1 for s in statuses if s in ('completed', 'skipped', 'already_owned')), + 'failed': sum(1 for s in statuses if s in ('failed', 'not_found', 'cancelled')), + 'active': sum(1 for s in statuses if s in ('downloading', 'searching', 'post_processing')), + 'queued': sum(1 for s in statuses if s in ('queued', 'pending')), + }) + + return { + 'success': True, + 'downloads': items[:limit], + 'total': len(items), + 'batches': batch_summaries, + 'timestamp': time.time(), + } diff --git a/tests/downloads/test_downloads_status.py b/tests/downloads/test_downloads_status.py new file mode 100644 index 00000000..757f4ddc --- /dev/null +++ b/tests/downloads/test_downloads_status.py @@ -0,0 +1,477 @@ +"""Tests for core/downloads/status.py — batch + unified status helpers.""" + +from __future__ import annotations + +import pytest + +from core.downloads import status as st +from core.runtime_state import ( + download_batches, + download_tasks, +) + + +@pytest.fixture(autouse=True) +def reset_state(): + download_tasks.clear() + download_batches.clear() + yield + download_tasks.clear() + download_batches.clear() + + +# --------------------------------------------------------------------------- +# Fakes +# --------------------------------------------------------------------------- + +class _FakeConfig: + def __init__(self, values=None): + self._v = values or {} + + def get(self, key, default=None): + return self._v.get(key, default) + + +def _build_deps( + *, + config=None, + docker_resolve=None, + find_completed=None, + make_key=None, + submit_pp=None, + cached_transfers=None, +): + submitted = [] + + def _default_submit(task_id, batch_id): + submitted.append((task_id, batch_id)) + + deps = st.StatusDeps( + config_manager=config or _FakeConfig({'soulseek.download_timeout': 600}), + docker_resolve_path=docker_resolve or (lambda p: p), + find_completed_file=find_completed or (lambda *a, **kw: (None, None)), + make_context_key=make_key or (lambda u, f: f"{u}::{f}"), + submit_post_processing=submit_pp or _default_submit, + get_cached_transfer_data=cached_transfers or (lambda: {}), + ) + return deps, submitted + + +# --------------------------------------------------------------------------- +# build_batch_status_data — phase routing +# --------------------------------------------------------------------------- + +def test_unknown_phase_includes_basic_fields_only(): + deps, _ = _build_deps() + batch = {'phase': 'unknown', 'playlist_id': 'pl1', 'playlist_name': 'My PL'} + out = st.build_batch_status_data('b1', batch, {}, deps) + assert out['phase'] == 'unknown' + assert out['playlist_id'] == 'pl1' + assert out['playlist_name'] == 'My PL' + assert 'tasks' not in out + assert 'analysis_progress' not in out + + +def test_analysis_phase_includes_analysis_progress_and_results(): + deps, _ = _build_deps() + batch = { + 'phase': 'analysis', + 'analysis_total': 10, + 'analysis_processed': 4, + 'analysis_results': [{'track_index': 0, 'found': True}], + } + out = st.build_batch_status_data('b1', batch, {}, deps) + assert out['analysis_progress'] == {'total': 10, 'processed': 4} + assert out['analysis_results'] == [{'track_index': 0, 'found': True}] + + +def test_complete_phase_includes_wishlist_summary_when_present(): + deps, _ = _build_deps() + batch = { + 'phase': 'complete', + 'queue': [], + 'wishlist_summary': {'added': 3, 'failed': 0}, + } + out = st.build_batch_status_data('b1', batch, {}, deps) + assert out['wishlist_summary'] == {'added': 3, 'failed': 0} + + +def test_complete_phase_omits_wishlist_summary_when_missing(): + deps, _ = _build_deps() + batch = {'phase': 'complete', 'queue': []} + out = st.build_batch_status_data('b1', batch, {}, deps) + assert 'wishlist_summary' not in out + + +# --------------------------------------------------------------------------- +# Task formatting +# --------------------------------------------------------------------------- + +def test_downloading_phase_includes_active_count_and_max_concurrent(): + deps, _ = _build_deps() + batch = {'phase': 'downloading', 'queue': [], 'active_count': 2, 'max_concurrent': 5} + out = st.build_batch_status_data('b1', batch, {}, deps) + assert out['active_count'] == 2 + assert out['max_concurrent'] == 5 + + +def test_tasks_sorted_by_track_index(): + deps, _ = _build_deps() + download_tasks['t1'] = {'track_index': 2, 'status': 'queued', 'track_info': {}} + download_tasks['t2'] = {'track_index': 0, 'status': 'queued', 'track_info': {}} + download_tasks['t3'] = {'track_index': 1, 'status': 'queued', 'track_info': {}} + batch = {'phase': 'downloading', 'queue': ['t1', 't2', 't3']} + out = st.build_batch_status_data('b1', batch, {}, deps) + assert [t['track_index'] for t in out['tasks']] == [0, 1, 2] + + +def test_missing_task_in_queue_is_skipped(): + deps, _ = _build_deps() + download_tasks['t1'] = {'track_index': 0, 'status': 'queued', 'track_info': {}} + batch = {'phase': 'downloading', 'queue': ['t1', 'absent']} + out = st.build_batch_status_data('b1', batch, {}, deps) + assert len(out['tasks']) == 1 + + +def test_task_status_includes_v2_state_fields(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'cancelling', 'track_info': {}, + 'cancel_requested': True, 'cancel_timestamp': 12345, + 'ui_state': 'cancelling', 'playlist_id': 'pl1', + 'error_message': 'oh no', 'cached_candidates': [{'x': 1}], + } + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, {}, deps) + t = out['tasks'][0] + assert t['cancel_requested'] is True + assert t['cancel_timestamp'] == 12345 + assert t['ui_state'] == 'cancelling' + assert t['playlist_id'] == 'pl1' + assert t['error_message'] == 'oh no' + assert t['has_candidates'] is True + + +# --------------------------------------------------------------------------- +# Live transfer state mapping +# --------------------------------------------------------------------------- + +def test_live_state_cancelled_maps_to_cancelled(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': {'state': 'Cancelled', 'percentComplete': 50}} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['status'] == 'cancelled' + assert download_tasks['t1']['status'] == 'cancelled' # mutates source state + + +def test_live_state_succeeded_with_full_bytes_marks_post_processing_and_submits(): + deps, submitted = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': { + 'state': 'Succeeded', 'size': 100, 'bytesTransferred': 100, 'percentComplete': 100, + }} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['status'] == 'post_processing' + assert download_tasks['t1']['status'] == 'post_processing' + assert submitted == [('t1', 'b1')] + + +def test_live_state_succeeded_with_byte_mismatch_keeps_status(): + deps, submitted = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': { + 'state': 'Succeeded', 'size': 100, 'bytesTransferred': 50, + }} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['status'] == 'downloading' # not promoted to post_processing + assert submitted == [] # not submitted + + +def test_live_state_inprogress_maps_to_downloading(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'queued', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': {'state': 'InProgress', 'percentComplete': 42}} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['status'] == 'downloading' + assert out['tasks'][0]['progress'] == 42 + + +def test_live_state_errored_keeps_active_status_for_monitor(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': {'state': 'Errored'}} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + # Keeps current status so monitor handles retry + assert out['tasks'][0]['status'] == 'downloading' + assert download_tasks['t1']['status'] == 'downloading' # not marked failed + + +def test_terminal_status_not_overridden_by_live_state(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'completed', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {'u1::song.flac': {'state': 'InProgress', 'percentComplete': 50}} + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['status'] == 'completed' + assert out['tasks'][0]['progress'] == 100 + + +def test_post_processing_status_progress_is_95(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'post_processing', 'track_info': {}, + 'filename': 'song.flac', 'username': 'u1', + } + live = {} # no live entry + batch = {'phase': 'downloading', 'queue': ['t1']} + out = st.build_batch_status_data('b1', batch, live, deps) + assert out['tasks'][0]['progress'] == 95 + + +# --------------------------------------------------------------------------- +# Safety valve (stuck task handling) +# --------------------------------------------------------------------------- + +def test_safety_valve_stuck_searching_marks_not_found(): + deps, _ = _build_deps(config=_FakeConfig({'soulseek.download_timeout': 1})) + download_tasks['t1'] = { + 'track_index': 0, 'status': 'searching', 'track_info': {}, + 'status_change_time': 0, # ancient + } + batch = {'phase': 'downloading', 'queue': ['t1']} + st.build_batch_status_data('b1', batch, {}, deps) + assert download_tasks['t1']['status'] == 'not_found' + assert 'Search stuck' in download_tasks['t1']['error_message'] + + +def test_safety_valve_stuck_downloading_with_recovered_file_routes_to_post_processing(): + deps, submitted = _build_deps( + config=_FakeConfig({'soulseek.download_timeout': 1, 'soulseek.download_path': '/d', 'soulseek.transfer_path': '/t'}), + find_completed=lambda *a, **kw: ('/found.flac', 'transfer'), + ) + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', + 'status_change_time': 0, + } + batch = {'phase': 'downloading', 'queue': ['t1']} + st.build_batch_status_data('b1', batch, {}, deps) + assert download_tasks['t1']['status'] == 'post_processing' + assert submitted == [('t1', 'b1')] + + +def test_safety_valve_stuck_downloading_no_file_marks_failed(): + deps, _ = _build_deps(config=_FakeConfig({'soulseek.download_timeout': 1})) + download_tasks['t1'] = { + 'track_index': 0, 'status': 'downloading', 'track_info': {}, + 'filename': 'song.flac', + 'status_change_time': 0, + } + batch = {'phase': 'downloading', 'queue': ['t1']} + st.build_batch_status_data('b1', batch, {}, deps) + assert download_tasks['t1']['status'] == 'failed' + assert 'Task stuck' in download_tasks['t1']['error_message'] + + +# --------------------------------------------------------------------------- +# build_single_batch_status (route helper) +# --------------------------------------------------------------------------- + +def test_single_batch_status_404_for_unknown_batch(): + deps, _ = _build_deps() + body, status = st.build_single_batch_status('absent', deps) + assert status == 404 + assert body['error'] == 'Batch not found' + + +def test_single_batch_status_returns_payload_for_known_batch(): + deps, _ = _build_deps() + download_batches['b1'] = {'phase': 'unknown', 'playlist_id': 'pl1'} + body, status = st.build_single_batch_status('b1', deps) + assert status == 200 + assert body['phase'] == 'unknown' + assert body['playlist_id'] == 'pl1' + + +# --------------------------------------------------------------------------- +# build_batched_status (route helper) +# --------------------------------------------------------------------------- + +def test_batched_status_filters_to_requested_ids(): + deps, _ = _build_deps() + download_batches['b1'] = {'phase': 'unknown'} + download_batches['b2'] = {'phase': 'unknown'} + download_batches['b3'] = {'phase': 'unknown'} + out = st.build_batched_status(['b1', 'b3'], deps) + assert set(out['batches'].keys()) == {'b1', 'b3'} + + +def test_batched_status_no_filter_returns_all_batches(): + deps, _ = _build_deps() + download_batches['b1'] = {'phase': 'unknown'} + download_batches['b2'] = {'phase': 'unknown'} + out = st.build_batched_status([], deps) + assert set(out['batches'].keys()) == {'b1', 'b2'} + + +def test_batched_status_metadata_present(): + deps, _ = _build_deps() + download_batches['b1'] = {'phase': 'unknown'} + out = st.build_batched_status([], deps) + assert out['metadata']['total_batches'] == 1 + assert out['metadata']['requested_batch_ids'] == [] + assert isinstance(out['metadata']['timestamp'], (int, float)) + + +def test_batched_status_per_batch_failure_isolated(): + deps, _ = _build_deps() + # b1 valid, b2 raises + download_batches['b1'] = {'phase': 'downloading', 'queue': []} + + class _BadBatch(dict): + def get(self, k, default=None): + if k == 'phase': + raise RuntimeError("batch boom") + return super().get(k, default) + + download_batches['b2'] = _BadBatch() + + out = st.build_batched_status([], deps) + assert 'error' in out['batches']['b2'] + assert out['batches']['b1'].get('phase') == 'downloading' + + +def test_batched_status_debug_info_pre_existing_bug_never_populates(): + """Documents a pre-existing bug: every batch payload includes + `"error": batch.get('error')` (key always present, value usually None). + The debug_info loop checks `if "error" not in batch_status:` which is + therefore always False → debug_info stays empty in production. + + Lift preserves this exactly. A future PR can flip the check to + `if batch_status.get('error') is None` to fix it. + """ + deps, _ = _build_deps() + download_tasks['t1'] = {'track_index': 0, 'status': 'downloading', 'track_info': {}} + download_tasks['t2'] = {'track_index': 1, 'status': 'downloading', 'track_info': {}} + download_batches['b1'] = { + 'phase': 'downloading', 'queue': ['t1', 't2'], + 'active_count': 5, + 'max_concurrent': 5, + } + out = st.build_batched_status([], deps) + # Bug: stays empty because "error" key is always present in payload + assert out['debug_info'] == {} + + +# --------------------------------------------------------------------------- +# build_unified_downloads_response +# --------------------------------------------------------------------------- + +def test_unified_response_sorts_by_priority_then_recency(): + deps, _ = _build_deps() + download_tasks['old_complete'] = { + 'track_index': 0, 'status': 'completed', 'track_info': {'name': 'A'}, + 'status_change_time': 100, + } + download_tasks['new_active'] = { + 'track_index': 1, 'status': 'downloading', 'track_info': {'name': 'B'}, + 'status_change_time': 50, + } + out = st.build_unified_downloads_response(100, deps) + # Active downloads come first regardless of timestamp + assert out['downloads'][0]['title'] == 'B' + + +def test_unified_response_artist_list_of_dicts_normalized(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'queued', + 'track_info': {'name': 'X', 'artists': [{'name': 'Pink Floyd'}, {'name': 'Roger'}]}, + } + out = st.build_unified_downloads_response(100, deps) + assert out['downloads'][0]['artist'] == 'Pink Floyd, Roger' + + +def test_unified_response_album_dict_extracted_to_name(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'queued', + 'track_info': {'name': 'X', 'album': {'name': 'DSOTM', 'images': [{'url': 'http://thumb.jpg'}]}}, + } + out = st.build_unified_downloads_response(100, deps) + assert out['downloads'][0]['album'] == 'DSOTM' + assert out['downloads'][0]['artwork'] == 'http://thumb.jpg' + + +def test_unified_response_completed_task_progress_is_100(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'completed', + 'track_info': {'name': 'X'}, + } + out = st.build_unified_downloads_response(100, deps) + assert out['downloads'][0]['progress'] == 100 + + +def test_unified_response_post_processing_progress_is_95(): + deps, _ = _build_deps() + download_tasks['t1'] = { + 'track_index': 0, 'status': 'post_processing', + 'track_info': {'name': 'X'}, + } + out = st.build_unified_downloads_response(100, deps) + assert out['downloads'][0]['progress'] == 95 + + +def test_unified_response_includes_batch_summaries(): + deps, _ = _build_deps() + download_tasks['t1'] = {'track_index': 0, 'status': 'completed', 'track_info': {}} + download_tasks['t2'] = {'track_index': 1, 'status': 'failed', 'track_info': {}} + download_tasks['t3'] = {'track_index': 2, 'status': 'downloading', 'track_info': {}} + download_tasks['t4'] = {'track_index': 3, 'status': 'queued', 'track_info': {}} + download_batches['b1'] = { + 'phase': 'downloading', 'playlist_name': 'PL', + 'queue': ['t1', 't2', 't3', 't4'], + } + out = st.build_unified_downloads_response(100, deps) + bs = out['batches'][0] + assert bs['total'] == 4 + assert bs['completed'] == 1 + assert bs['failed'] == 1 + assert bs['active'] == 1 + assert bs['queued'] == 1 + + +def test_unified_response_respects_limit(): + deps, _ = _build_deps() + for i in range(20): + download_tasks[f't{i}'] = { + 'track_index': i, 'status': 'completed', 'track_info': {}, + } + out = st.build_unified_downloads_response(5, deps) + assert len(out['downloads']) == 5 + assert out['total'] == 20 # total still reflects all diff --git a/web_server.py b/web_server.py index 1e6df444..1a6008c8 100644 --- a/web_server.py +++ b/web_server.py @@ -23764,402 +23764,61 @@ def get_active_processes(): logger.info(f"Active processes check: {len([p for p in active_processes if p['type'] == 'batch'])} download batches, {len([p for p in active_processes if p['type'] == 'youtube_playlist'])} YouTube playlists") return jsonify({"active_processes": active_processes}) +# Status payload helpers live in core/downloads/status.py. +from core.downloads import status as _downloads_status + + +def _build_status_deps(): + """Build StatusDeps bundle from web_server.py globals on each call.""" + return _downloads_status.StatusDeps( + config_manager=config_manager, + docker_resolve_path=docker_resolve_path, + find_completed_file=_find_completed_file_robust, + make_context_key=_make_context_key, + submit_post_processing=lambda task_id, batch_id: missing_download_executor.submit( + _run_post_processing_worker, task_id, batch_id + ), + get_cached_transfer_data=get_cached_transfer_data, + ) + + def _build_batch_status_data(batch_id, batch, live_transfers_lookup): - """ - Helper function to build status data for a single batch. + """Helper function to build status data for a single batch. Extracted from get_batch_download_status for reuse in batched endpoint. """ - response_data = { - "phase": batch.get('phase', 'unknown'), - "error": batch.get('error'), - "auto_initiated": batch.get('auto_initiated', False), - "playlist_id": batch.get('playlist_id'), # Include playlist_id for rehydration - "playlist_name": batch.get('playlist_name') # Include playlist_name for reference - } + return _downloads_status.build_batch_status_data(batch_id, batch, live_transfers_lookup, _build_status_deps()) - if response_data["phase"] == 'analysis': - response_data['analysis_progress'] = { - 'total': batch.get('analysis_total', 0), - 'processed': batch.get('analysis_processed', 0) - } - response_data['analysis_results'] = batch.get('analysis_results', []) - - elif response_data["phase"] in ['downloading', 'complete', 'error']: - response_data['analysis_results'] = batch.get('analysis_results', []) - batch_tasks = [] - for task_id in batch.get('queue', []): - task = download_tasks.get(task_id) - if not task: continue - - # SAFETY VALVE: Check for downloads stuck too long - import time - current_time = time.time() - task_start_time = task.get('status_change_time', current_time) - task_age = current_time - task_start_time - - # If task has been running too long, check if file completed - _dl_timeout = config_manager.get('soulseek.download_timeout', 600) or 600 - if task_age > _dl_timeout and task['status'] in ['downloading', 'queued', 'searching']: - stuck_state = task['status'] - task_filename = task.get('filename') or (task.get('track_info') or {}).get('filename') - - # Before failing, check if the file actually downloaded successfully - recovered = False - if task_filename and stuck_state == 'downloading': - try: - download_dir = docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')) - transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer')) - found_file, file_location = _find_completed_file_robust(download_dir, task_filename, transfer_dir) - if found_file: - logger.info(f"[Safety Valve] Task {task_id} stuck but file found in {file_location} — routing to post-processing") - task['status'] = 'post_processing' - task['status_change_time'] = current_time - missing_download_executor.submit(_run_post_processing_worker, task_id, batch_id) - recovered = True - except Exception as e: - logger.error(f"[Safety Valve] Error checking for completed file: {e}") - - if not recovered: - if stuck_state == 'searching': - logger.info(f"⏰ [Safety Valve] Task {task_id} stuck in searching for {task_age:.1f}s - marking not_found") - task['status'] = 'not_found' - task['error_message'] = f'Search stuck for {int(task_age // 60)} minutes with no results — timed out' - else: - logger.error(f"⏰ [Safety Valve] Task {task_id} stuck for {task_age:.1f}s - forcing failure") - task['status'] = 'failed' - task['error_message'] = f'Task stuck in {stuck_state} state for {int(task_age // 60)} minutes — forcibly stopped' - - task_status = { - 'task_id': task_id, - 'track_index': task['track_index'], - 'status': task['status'], - 'track_info': task['track_info'], - 'progress': 0, - # V2 SYSTEM: Add persistent state information - 'cancel_requested': task.get('cancel_requested', False), - 'cancel_timestamp': task.get('cancel_timestamp'), - 'ui_state': task.get('ui_state', 'normal'), # normal|cancelling|cancelled - 'playlist_id': task.get('playlist_id'), # For V2 system identification - 'error_message': task.get('error_message'), # Surface failure reasons to UI - 'has_candidates': bool(task.get('cached_candidates')), # Whether search found results (for clickable review) - } - _ti = task.get('track_info') if isinstance(task.get('track_info'), dict) else {} - task_filename = task.get('filename') or _ti.get('filename') - task_username = task.get('username') or _ti.get('username') - if task_filename and task_username: - lookup_key = _make_context_key(task_username, task_filename) - - if lookup_key in live_transfers_lookup: - live_info = live_transfers_lookup[lookup_key] - state_str = live_info.get('state', 'Unknown') - - # Don't override tasks that are already in terminal states or post-processing - if task['status'] not in ['completed', 'failed', 'cancelled', 'not_found', 'post_processing']: - # SYNC.PY PARITY: Prioritized state checking (Errored/Cancelled before Completed) - # This prevents "Completed, Errored" states from being marked as completed - if 'Cancelled' in state_str or 'Canceled' in state_str: - task_status['status'] = 'cancelled' - task['status'] = 'cancelled' - elif 'Failed' in state_str or 'Errored' in state_str or 'Rejected' in state_str or 'TimedOut' in state_str: - # UNIFIED ERROR HANDLING: Let monitor handle errors for consistency - # Monitor will detect errored state and trigger retry within 5 seconds - logger.error(f"Task {task_id} API shows error state: {state_str} - letting monitor handle retry") - - # Keep task in current status (downloading/queued) so monitor can detect error - # Don't mark as failed here - let the unified retry system handle it - if task['status'] in ['searching', 'downloading', 'queued']: - task_status['status'] = task['status'] # Keep current status for monitor - else: - task_status['status'] = 'downloading' # Default to downloading for error detection - task['status'] = 'downloading' - elif 'Completed' in state_str or 'Succeeded' in state_str: - # Verify bytes actually transferred before trusting state string - expected_size = live_info.get('size', 0) - transferred = live_info.get('bytesTransferred', 0) - if expected_size > 0 and transferred < expected_size: - # State says complete but bytes don't match — keep current status - task_status['status'] = task['status'] - logger.info(f"Task {task_id} state says complete but bytes incomplete ({transferred}/{expected_size})") - # NEW VERIFICATION WORKFLOW: Use intermediate post_processing status - # Only set this status once to prevent multiple worker submissions - elif task['status'] != 'post_processing': - task_status['status'] = 'post_processing' - task['status'] = 'post_processing' - logger.info(f"Task {task_id} API reports 'Succeeded' - starting post-processing verification") - - # Submit post-processing worker to verify file and complete the task - missing_download_executor.submit(_run_post_processing_worker, task_id, batch_id) - else: - # FIXED: Always require verification workflow - no bypass for stream processed tasks - # Stream processing only handles metadata, not file verification - task_status['status'] = 'post_processing' - logger.info(f"Task {task_id} waiting for verification worker to complete") - elif 'InProgress' in state_str: - task_status['status'] = 'downloading' - else: - task_status['status'] = 'queued' - task_status['progress'] = live_info.get('percentComplete', 0) - # For completed/post-processing tasks, keep appropriate progress - elif task['status'] == 'completed': - task_status['progress'] = 100 - elif task['status'] == 'post_processing': - task_status['progress'] = 95 # Nearly complete, just verifying - else: - # If task is completed but not in live transfers, keep appropriate status - if task['status'] == 'completed': - task_status['progress'] = 100 - elif task['status'] == 'post_processing': - task_status['progress'] = 95 # Nearly complete, just verifying - batch_tasks.append(task_status) - batch_tasks.sort(key=lambda x: x['track_index']) - response_data['tasks'] = batch_tasks - - # CRITICAL: Add batch worker management metadata (was missing!) - # This is essential for client-side worker validation and prevents false desync warnings - response_data['active_count'] = batch.get('active_count', 0) - response_data['max_concurrent'] = batch.get('max_concurrent', 3) - - # Add wishlist summary if batch is complete (matching sync.py behavior) - if response_data["phase"] == 'complete' and 'wishlist_summary' in batch: - response_data['wishlist_summary'] = batch['wishlist_summary'] - - return response_data @app.route('/api/playlists//download_status', methods=['GET']) def get_batch_download_status(batch_id): - """ - Returns real-time status for a single batch. - Now uses shared helper function for consistency with batched endpoint. - """ + """Returns real-time status for a single batch.""" try: - # Use cached transfer data to reduce API calls with multiple concurrent modals - live_transfers_lookup = get_cached_transfer_data() - - with tasks_lock: - if batch_id not in download_batches: - return jsonify({"error": "Batch not found"}), 404 - - batch = download_batches[batch_id] - response_data = _build_batch_status_data(batch_id, batch, live_transfers_lookup) - return jsonify(response_data) - + body, status = _downloads_status.build_single_batch_status(batch_id, _build_status_deps()) + return jsonify(body), status except Exception as e: import traceback traceback.print_exc() return jsonify({"error": str(e)}), 500 + @app.route('/api/download_status/batch', methods=['GET']) def get_batched_download_statuses(): - """ - NEW: Returns status for multiple download batches in a single request. - Dramatically reduces API calls when multiple download modals are active. - - Query params: - - batch_ids: Optional list of specific batch IDs to include - - If no batch_ids provided, returns all active batches - """ + """Returns status for multiple download batches in one request.""" try: - # Get optional batch ID filtering from query params requested_batch_ids = request.args.getlist('batch_ids') - - # Use shared cached transfer data - single lookup for all batches - live_transfers_lookup = get_cached_transfer_data() - - response = {"batches": {}} - - with tasks_lock: - # Determine which batches to include - if requested_batch_ids: - # Filter to only requested batch IDs that exist - target_batches = { - bid: batch for bid, batch in download_batches.items() - if bid in requested_batch_ids - } - else: - # Return all active batches - target_batches = download_batches.copy() - - # Build status data for each batch using shared helper - for batch_id, batch in target_batches.items(): - try: - response["batches"][batch_id] = _build_batch_status_data( - batch_id, batch, live_transfers_lookup - ) - except Exception as batch_error: - # Don't fail entire request if one batch has issues - logger.error(f"Error processing batch {batch_id}: {batch_error}") - response["batches"][batch_id] = {"error": str(batch_error)} - - # Add metadata for debugging/monitoring - response["metadata"] = { - "total_batches": len(response["batches"]), - "requested_batch_ids": requested_batch_ids, - "timestamp": time.time() - } - - # ENHANCED: Add comprehensive debug info for worker tracking - debug_info = {} - for batch_id, batch_status in response["batches"].items(): - if "error" not in batch_status: - active_count = batch_status.get("active_count", 0) - max_concurrent = batch_status.get("max_concurrent", 3) - task_count = len(batch_status.get("tasks", [])) - active_tasks = len([t for t in batch_status.get("tasks", []) if t.get("status") in ['searching', 'downloading', 'queued']]) - - debug_info[batch_id] = { - "reported_active": active_count, - "actual_active_tasks": active_tasks, - "max_concurrent": max_concurrent, - "total_tasks": task_count, - "worker_discrepancy": active_count != active_tasks - } - - response["debug_info"] = debug_info - - logger.info(f"[Batched Status] Returning status for {len(response['batches'])} batches") - - # Log worker discrepancies for debugging - discrepancies = [bid for bid, info in debug_info.items() if info.get("worker_discrepancy")] - if discrepancies: - logger.info(f"[Batched Status] Worker count discrepancies in batches: {discrepancies}") - - return jsonify(response) - + return jsonify(_downloads_status.build_batched_status(requested_batch_ids, _build_status_deps())) except Exception as e: import traceback traceback.print_exc() return jsonify({"error": str(e)}), 500 + @app.route('/api/downloads/all', methods=['GET']) def get_all_downloads_unified(): - """ - Unified downloads list for the centralized Downloads page. - Returns a flat list of all download tasks across all batches, - sorted: downloading/searching first, then queued, then completed/failed. - """ + """Unified downloads list for the centralized Downloads page.""" try: limit = int(request.args.get('limit', 200)) - status_priority = { - 'downloading': 0, 'searching': 1, 'post_processing': 2, - 'queued': 3, 'pending': 3, - 'completed': 4, 'skipped': 5, 'already_owned': 5, - 'not_found': 6, 'failed': 7, 'cancelled': 8, - } - - items = [] - with tasks_lock: - for task_id, task in download_tasks.items(): - track_info = task.get('track_info') or {} - batch_id = task.get('batch_id', '') - batch = download_batches.get(batch_id, {}) - - # Extract track metadata — handle all format variations - title = '' - artist = '' - album = '' - artwork = '' - if isinstance(track_info, dict): - title = track_info.get('title') or track_info.get('name') or track_info.get('track_name') or '' - - # Artist can be: string, list of strings, list of dicts with 'name' - raw_artist = track_info.get('artist') or track_info.get('artist_name') or track_info.get('artists') or '' - if isinstance(raw_artist, list): - parts = [] - for a in raw_artist: - if isinstance(a, dict): - parts.append(a.get('name', '')) - else: - parts.append(str(a)) - artist = ', '.join(p for p in parts if p) - elif isinstance(raw_artist, dict): - artist = raw_artist.get('name', '') - else: - artist = str(raw_artist) if raw_artist else '' - - # Album can be: string or dict with 'name' - raw_album = track_info.get('album') or track_info.get('album_name') or '' - if isinstance(raw_album, dict): - album = raw_album.get('name', '') - else: - album = str(raw_album) if raw_album else '' - - artwork = track_info.get('artwork_url') or track_info.get('image_url') or track_info.get('album_art') or '' - # Try album images - if not artwork: - raw_alb = track_info.get('album') - if isinstance(raw_alb, dict): - images = raw_alb.get('images') or [] - if images and isinstance(images, list) and len(images) > 0: - artwork = images[0].get('url', '') if isinstance(images[0], dict) else str(images[0]) - - status = task.get('status', 'queued') - # Determine download progress percentage - progress = 0 - if status == 'completed': - progress = 100 - elif status == 'post_processing': - progress = 95 - elif status in ('downloading', 'searching'): - # Check live transfer data for real progress - task_filename = task.get('filename') or track_info.get('filename') - task_username = task.get('username') or track_info.get('username') - if task_filename and task_username: - lookup_key = _make_context_key(task_username, task_filename) - live_info = get_cached_transfer_data().get(lookup_key) - if live_info: - progress = live_info.get('percentComplete', 0) - - items.append({ - 'task_id': task_id, - 'title': title, - 'artist': artist, - 'album': album, - 'artwork': artwork, - 'status': status, - 'progress': progress, - 'error': task.get('error_message'), - 'batch_id': batch_id, - 'batch_name': batch.get('playlist_name') or batch.get('album_name') or '', - 'batch_source': batch.get('source_page') or batch.get('initiated_from') or '', - # playlist_id is needed by per-row cancel (cancel_task_v2 - # takes playlist_id + track_index). Surfacing it here so - # the frontend doesn't need a second lookup. - 'playlist_id': batch.get('playlist_id', ''), - 'track_index': task.get('track_index', 0), - 'batch_total': len(batch.get('queue', [])), - 'timestamp': task.get('status_change_time', 0), - 'priority': status_priority.get(status, 9), - }) - - # Sort: active first (by priority), then by timestamp desc within each group - items.sort(key=lambda x: (x['priority'], -x['timestamp'])) - - # Build batch summaries for the batch context panel - batch_summaries = [] - with tasks_lock: - for bid, batch in download_batches.items(): - queue = batch.get('queue', []) - statuses = [download_tasks[tid]['status'] for tid in queue if tid in download_tasks] - batch_summaries.append({ - 'batch_id': bid, - 'playlist_id': batch.get('playlist_id', ''), - 'batch_name': batch.get('playlist_name') or batch.get('album_name') or '', - 'source_page': batch.get('source_page') or batch.get('initiated_from') or '', - 'phase': batch.get('phase', 'unknown'), - 'total': len(queue), - 'completed': sum(1 for s in statuses if s in ('completed', 'skipped', 'already_owned')), - 'failed': sum(1 for s in statuses if s in ('failed', 'not_found', 'cancelled')), - 'active': sum(1 for s in statuses if s in ('downloading', 'searching', 'post_processing')), - 'queued': sum(1 for s in statuses if s in ('queued', 'pending')), - }) - - return jsonify({ - 'success': True, - 'downloads': items[:limit], - 'total': len(items), - 'batches': batch_summaries, - 'timestamp': time.time(), - }) + return jsonify(_downloads_status.build_unified_downloads_response(limit, _build_status_deps())) except Exception as e: logger.error(f"Error getting unified downloads: {e}") return jsonify({'success': False, 'error': str(e)}), 500