|
|
"""BackgroundDownloadWorker — engine-owned thread spawning + state
|
|
|
lifecycle for downloads.
|
|
|
|
|
|
Today every streaming download client (YouTube, Tidal, Qobuz, HiFi,
|
|
|
Deezer, SoundCloud) hand-rolls the same thread-spawn pattern:
|
|
|
|
|
|
```python
|
|
|
async def download(self, ...):
|
|
|
download_id = str(uuid.uuid4())
|
|
|
with self._download_lock:
|
|
|
self.active_downloads[download_id] = {...initial state...}
|
|
|
threading.Thread(
|
|
|
target=self._download_thread_worker,
|
|
|
args=(download_id, target_id, display_name, ...),
|
|
|
daemon=True,
|
|
|
).start()
|
|
|
return download_id
|
|
|
|
|
|
def _download_thread_worker(self, download_id, target_id, display_name, ...):
|
|
|
with self._download_semaphore:
|
|
|
# rate-limit sleep
|
|
|
# update state to 'InProgress, Downloading'
|
|
|
file_path = self._download_sync(...) # the source-specific atomic op
|
|
|
# update state to 'Completed, Succeeded' / 'Errored'
|
|
|
```
|
|
|
|
|
|
That pattern is duplicated 6+ times across the codebase (~70 LOC
|
|
|
each, ~490 total). The worker class lifts it into the engine — each
|
|
|
plugin only has to provide the atomic op (``impl_callable``) and
|
|
|
declare its rate-limit policy. Adding a new download source becomes
|
|
|
a much smaller patch.
|
|
|
|
|
|
Phase C1 scope: introduce the worker. No client migrated yet — the
|
|
|
worker just exists for C2–C7 to migrate sources one at a time, each
|
|
|
under a passing pinning test.
|
|
|
"""
|
|
|
|
|
|
from __future__ import annotations
|
|
|
|
|
|
import threading
|
|
|
import time
|
|
|
import uuid
|
|
|
from typing import Any, Callable, Dict, Optional
|
|
|
|
|
|
from utils.logging_config import get_logger
|
|
|
|
|
|
logger = get_logger("download_engine.worker")
|
|
|
|
|
|
|
|
|
# Type aliases for clarity. ``ImplCallable`` is the per-plugin
|
|
|
# atomic download operation — synchronous, returns a file path on
|
|
|
# success or raises (or returns None) on failure.
|
|
|
ImplCallable = Callable[[str, Any, str], Optional[str]]
|
|
|
|
|
|
|
|
|
class BackgroundDownloadWorker:
|
|
|
"""Engine-owned thread spawner for per-source downloads.
|
|
|
|
|
|
State-machine semantics (preserved verbatim from the legacy
|
|
|
per-client workers so consumers reading these fields keep
|
|
|
working):
|
|
|
|
|
|
- ``Initializing`` — set on dispatch, before the thread starts.
|
|
|
- ``InProgress, Downloading`` — set when the worker thread
|
|
|
acquires the semaphore and is about to call the impl.
|
|
|
- ``Completed, Succeeded`` — set when impl returns a non-None
|
|
|
file path. ``progress=100.0`` and ``file_path=<the path>``
|
|
|
also written.
|
|
|
- ``Errored`` — set when impl returns None OR raises. The
|
|
|
record is left in place so downstream consumers can inspect
|
|
|
what failed.
|
|
|
|
|
|
Per-source serialization: each source gets a ``threading.Semaphore``
|
|
|
(default size 1, configurable per-source via ``set_concurrency``).
|
|
|
Same shape the existing clients use today (each source defines
|
|
|
its own semaphore). Engine owning them centrally lets a future
|
|
|
Phase E rate-limiter swap the semaphore for a smarter pool.
|
|
|
|
|
|
Per-source delay-between-downloads: default 0 seconds (most
|
|
|
sources don't need it). YouTube currently uses 3s, Qobuz uses
|
|
|
1s — the legacy values get configured in via ``set_delay``
|
|
|
when the source registers.
|
|
|
"""
|
|
|
|
|
|
def __init__(self, engine: Any) -> None:
|
|
|
self._engine = engine
|
|
|
# Per-source semaphores + delay state. The first dispatch
|
|
|
# for a source auto-creates a semaphore with concurrency=1
|
|
|
# if the source hasn't been configured explicitly.
|
|
|
self._semaphores: Dict[str, threading.Semaphore] = {}
|
|
|
self._delays: Dict[str, float] = {}
|
|
|
self._last_download_at: Dict[str, float] = {}
|
|
|
self._config_lock = threading.Lock()
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
|
# Per-source rate-limit configuration
|
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def set_concurrency(self, source_name: str, max_concurrent: int) -> None:
|
|
|
"""Set the max number of concurrent downloads for a source.
|
|
|
Default is 1 (serial). Most sources will keep the default —
|
|
|
the streaming APIs all rate-limit at the API gateway level
|
|
|
anyway, parallel downloads just trade rate-limit errors for
|
|
|
thread overhead."""
|
|
|
with self._config_lock:
|
|
|
self._semaphores[source_name] = threading.Semaphore(max_concurrent)
|
|
|
|
|
|
def set_delay(self, source_name: str, seconds: float) -> None:
|
|
|
"""Set a minimum delay between successive downloads from the
|
|
|
same source. YouTube uses 3s today (avoid yt-dlp 429s),
|
|
|
Qobuz uses 1s. Other sources use 0 (no delay)."""
|
|
|
with self._config_lock:
|
|
|
self._delays[source_name] = float(seconds)
|
|
|
|
|
|
def _get_semaphore(self, source_name: str) -> threading.Semaphore:
|
|
|
with self._config_lock:
|
|
|
sem = self._semaphores.get(source_name)
|
|
|
if sem is None:
|
|
|
sem = threading.Semaphore(1)
|
|
|
self._semaphores[source_name] = sem
|
|
|
return sem
|
|
|
|
|
|
def _get_delay(self, source_name: str) -> float:
|
|
|
with self._config_lock:
|
|
|
return self._delays.get(source_name, 0.0)
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
|
# Dispatch — public API
|
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def dispatch(
|
|
|
self,
|
|
|
source_name: str,
|
|
|
target_id: Any,
|
|
|
display_name: str,
|
|
|
original_filename: str,
|
|
|
impl_callable: ImplCallable,
|
|
|
extra_record_fields: Optional[Dict[str, Any]] = None,
|
|
|
username_override: Optional[str] = None,
|
|
|
thread_name: Optional[str] = None,
|
|
|
) -> str:
|
|
|
"""Kick off a background download.
|
|
|
|
|
|
Args:
|
|
|
source_name: Canonical source name (e.g. 'youtube',
|
|
|
'tidal'). Used as the engine state key + the
|
|
|
username slot in the record (unless overridden).
|
|
|
target_id: Source-specific identifier (track_id, video_id,
|
|
|
permalink_url, album_foreign_id, etc.). Passed
|
|
|
verbatim to ``impl_callable``.
|
|
|
display_name: Human-readable label for logs / UI.
|
|
|
original_filename: The encoded filename the orchestrator
|
|
|
received (e.g. ``'12345||Song Title'``). Stored in
|
|
|
the record's ``filename`` slot for context-key lookups.
|
|
|
impl_callable: Synchronous function that performs the
|
|
|
actual download. Signature:
|
|
|
``impl_callable(download_id, target_id, display_name) -> Optional[str]``.
|
|
|
Returns the final file path on success or None /
|
|
|
raises on failure.
|
|
|
extra_record_fields: Per-source extras to merge into the
|
|
|
initial record (e.g. ``{'video_id': '...', 'url':
|
|
|
'...', 'title': '...'}`` for YouTube). Used to
|
|
|
preserve source-specific slots that downstream
|
|
|
consumers + status APIs read.
|
|
|
username_override: Use this instead of ``source_name``
|
|
|
in the record's ``username`` slot. Required for
|
|
|
Deezer (legacy ``'deezer_dl'``) — every other source
|
|
|
uses the canonical name.
|
|
|
thread_name: Optional thread name for diagnostics. Deezer
|
|
|
uses ``'deezer-dl-<track_id>'`` — Phase A pinning
|
|
|
tests catch any drift in this convention.
|
|
|
|
|
|
Returns:
|
|
|
download_id (UUID4 string). The orchestrator polls via
|
|
|
``engine.get_download_status(download_id)`` for progress.
|
|
|
"""
|
|
|
download_id = str(uuid.uuid4())
|
|
|
|
|
|
record: Dict[str, Any] = {
|
|
|
'id': download_id,
|
|
|
'filename': original_filename,
|
|
|
'username': username_override or source_name,
|
|
|
'state': 'Initializing',
|
|
|
'progress': 0.0,
|
|
|
'size': 0,
|
|
|
'transferred': 0,
|
|
|
'speed': 0,
|
|
|
'time_remaining': None,
|
|
|
'file_path': None,
|
|
|
}
|
|
|
if extra_record_fields:
|
|
|
record.update(extra_record_fields)
|
|
|
|
|
|
self._engine.add_record(source_name, download_id, record)
|
|
|
|
|
|
thread = threading.Thread(
|
|
|
target=self._worker_loop,
|
|
|
args=(source_name, download_id, target_id, display_name, impl_callable),
|
|
|
daemon=True,
|
|
|
name=thread_name,
|
|
|
)
|
|
|
thread.start()
|
|
|
|
|
|
return download_id
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
|
# Worker thread — the lifted boilerplate
|
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _worker_loop(
|
|
|
self,
|
|
|
source_name: str,
|
|
|
download_id: str,
|
|
|
target_id: Any,
|
|
|
display_name: str,
|
|
|
impl_callable: ImplCallable,
|
|
|
) -> None:
|
|
|
"""Runs on the spawned daemon thread. Handles semaphore
|
|
|
acquisition, rate-limit sleep, state lifecycle, exception
|
|
|
capture. The plugin-specific work happens entirely inside
|
|
|
``impl_callable``."""
|
|
|
try:
|
|
|
with self._get_semaphore(source_name):
|
|
|
# Rate-limit delay against the LAST download from
|
|
|
# this source (not just this worker — semaphore
|
|
|
# ensures serial access while delay is configured).
|
|
|
delay = self._get_delay(source_name)
|
|
|
if delay > 0:
|
|
|
last_at = self._last_download_at.get(source_name, 0.0)
|
|
|
elapsed = time.time() - last_at
|
|
|
if last_at > 0 and elapsed < delay:
|
|
|
wait_time = delay - elapsed
|
|
|
logger.info(
|
|
|
"Rate-limit delay for %s: waiting %.1fs before next download",
|
|
|
source_name, wait_time,
|
|
|
)
|
|
|
time.sleep(wait_time)
|
|
|
|
|
|
self._engine.update_record(source_name, download_id, {
|
|
|
'state': 'InProgress, Downloading',
|
|
|
})
|
|
|
|
|
|
try:
|
|
|
file_path = impl_callable(download_id, target_id, display_name)
|
|
|
except Exception as exc:
|
|
|
logger.error(
|
|
|
"%s download %s failed (impl raised): %s",
|
|
|
source_name, download_id, exc,
|
|
|
)
|
|
|
self._mark_terminal(
|
|
|
source_name, download_id,
|
|
|
success=False, error=str(exc),
|
|
|
)
|
|
|
return
|
|
|
|
|
|
self._last_download_at[source_name] = time.time()
|
|
|
|
|
|
if file_path:
|
|
|
# Atomic write — preserve Cancelled if user cancelled
|
|
|
# between impl returning and this write. Same guard
|
|
|
# _mark_terminal uses; Cin flagged both split sites.
|
|
|
self._engine.update_record_unless_state(
|
|
|
source_name, download_id,
|
|
|
{
|
|
|
'state': 'Completed, Succeeded',
|
|
|
'progress': 100.0,
|
|
|
'file_path': file_path,
|
|
|
},
|
|
|
skip_if_state_in=('Cancelled',),
|
|
|
)
|
|
|
logger.info(
|
|
|
"%s download %s completed: %s",
|
|
|
source_name, download_id, file_path,
|
|
|
)
|
|
|
else:
|
|
|
self._mark_terminal(source_name, download_id, success=False)
|
|
|
logger.error(
|
|
|
"%s download %s failed (impl returned None)",
|
|
|
source_name, download_id,
|
|
|
)
|
|
|
|
|
|
except Exception as exc:
|
|
|
# Defensive — semaphore / sleep shouldn't blow up the
|
|
|
# thread, but if they do the record needs SOME terminal
|
|
|
# state or it sits at 'Initializing' forever.
|
|
|
logger.exception(
|
|
|
"%s worker_loop crashed for download %s: %s",
|
|
|
source_name, download_id, exc,
|
|
|
)
|
|
|
self._mark_terminal(
|
|
|
source_name, download_id,
|
|
|
success=False, error=f'worker crash: {exc}',
|
|
|
)
|
|
|
|
|
|
def _mark_terminal(self, source_name: str, download_id: str,
|
|
|
success: bool, error: Optional[str] = None) -> None:
|
|
|
"""Write a terminal state, but DON'T clobber an explicit
|
|
|
'Cancelled' state set by the user via cancel_download.
|
|
|
Mirrors the legacy per-client guard
|
|
|
(``if state != 'Cancelled': state = 'Errored'``) every
|
|
|
client used to hand-roll inside its thread worker.
|
|
|
|
|
|
Uses ``update_record_unless_state`` so the check + write are
|
|
|
atomic under the engine's per-source lock. Cin caught a race
|
|
|
where a cancel landing between the read-snapshot + write
|
|
|
could overwrite Cancelled back to Errored / Completed.
|
|
|
"""
|
|
|
patch: Dict[str, Any] = {
|
|
|
'state': 'Completed, Succeeded' if success else 'Errored',
|
|
|
}
|
|
|
if error is not None:
|
|
|
patch['error'] = error
|
|
|
self._engine.update_record_unless_state(
|
|
|
source_name, download_id, patch, skip_if_state_in=('Cancelled',),
|
|
|
)
|
