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

PR4e: lift status helpers + 3 routes to core/downloads/status.py

Browse source 
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.
This commit is contained in:
Broque Thomas 2026-04-27 23:26:26 -07:00
parent 2e61000ecf
commit 2d271cfacf
3 changed files with 916 additions and 370 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

410
core/downloads/status.py Normal file
View file

@ -0,0 +1,410 @@
"""Batch + unified download status helpers.
`build_batch_status_data` is the per-batch payload formatter shared by:
- /api/playlists/<batch_id>/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/<batch_id>/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(),
}

477
tests/downloads/test_downloads_status.py Normal file
View file

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

399
web_server.py
View file

@ -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/<batch_id>/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