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

C6: Migrate Deezer to engine.worker

Browse source 
Same migration pattern as C2-C5. Deezer-specific quirks
preserved through worker overrides:
- username_override='deezer_dl' (legacy slot frontend reads)
- thread_name='deezer-dl-<track_id>' (diagnostic naming)
- track_id stays as STRING (Deezer GW API uses string IDs)
- Extra 'error' slot in record for ARL re-auth failure messages

Mid-download chunk loop's many state mutations (cancellation
checks, progress updates, error capture across multiple failure
modes) all flow through engine.update_record / get_record now.
Added _set_error and _is_cancelled helpers to keep call sites
readable.

Pinning tests updated. Suite still green (319 download tests).
This commit is contained in:
Broque Thomas 2026-05-04 14:02:03 -07:00
parent 27a97f8af6
commit cf438ba2d6
2 changed files with 194 additions and 215 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

224
core/deezer_download_client.py
View file

@ -91,9 +91,9 @@ class DeezerDownloadClient:
self.download_path = Path(download_path)
self.download_path.mkdir(parents=True, exist_ok=True)
# Download tracking (same pattern as Tidal/Qobuz/HiFi)
self.active_downloads: Dict[str, Dict[str, Any]] = {}
self._download_lock = threading.Lock()
# Engine reference is populated by set_engine() at registration
# time. None until orchestrator wires the registry.
self._engine = None
# Shutdown check callback (set by web_server)
self.shutdown_check = None
@ -125,6 +125,10 @@ class DeezerDownloadClient:
logger.info(f"Deezer download client initialized (download path: {self.download_path})")
def set_engine(self, engine):
"""Engine callback — wires the central thread worker + state store."""
self._engine = engine
# ─── Authentication ──────────────────────────────────────────
def _authenticate(self, arl: str) -> bool:
@ -605,87 +609,64 @@ class DeezerDownloadClient:
if not self._authenticated:
logger.error("Deezer not authenticated — cannot download")
return None
if self._engine is None:
logger.error("Deezer client has no engine reference — cannot dispatch download")
return None
# Parse filename: "track_id||display_name"
parts = filename.split('||', 1)
track_id = parts[0]
display_name = parts[1] if len(parts) > 1 else f"Track {track_id}"
download_id = str(uuid.uuid4())
with self._download_lock:
self.active_downloads[download_id] = {
'id': download_id,
return self._engine.worker.dispatch(
source_name='deezer',
target_id=track_id,
display_name=display_name,
original_filename=filename,
impl_callable=self._download_sync,
extra_record_fields={
'track_id': track_id,
'display_name': display_name,
'filename': filename,
'username': 'deezer_dl',
'state': 'Initializing',
'progress': 0.0,
'size': file_size,
'transferred': 0,
'speed': 0,
'file_path': None,
'error': None,
}
thread = threading.Thread(
target=self._download_thread_worker,
args=(download_id, track_id, display_name),
daemon=True,
name=f'deezer-dl-{track_id}'
},
# Legacy username slot — frontend status indicators key off
# ``deezer_dl``, not the canonical ``deezer``.
username_override='deezer_dl',
# Diagnostic thread name for multi-thread debugging.
thread_name=f'deezer-dl-{track_id}',
)
thread.start()
logger.info(f"Started Deezer download {download_id}: {display_name}")
return download_id
def _set_error(self, download_id: str, message: str) -> None:
"""Helper: set the engine record's `error` slot. No-op if
engine isn't wired or record was already removed."""
if self._engine is None:
return
self._engine.update_record('deezer', download_id, {'error': message})
def _download_thread_worker(self, download_id: str, track_id: str, display_name: str):
"""Background worker for a single download."""
try:
result_path = self._download_sync(download_id, track_id, display_name)
with self._download_lock:
if download_id in self.active_downloads:
dl = self.active_downloads[download_id]
if dl['state'] == 'Cancelled':
return
if result_path:
dl['state'] = 'Completed, Succeeded'
dl['progress'] = 100.0
dl['file_path'] = result_path
logger.info(f"Deezer download {download_id} completed: {result_path}")
else:
dl['state'] = 'Errored'
logger.error(f"Deezer download {download_id} failed: {dl.get('error', 'unknown')}")
except Exception as e:
logger.error(f"Deezer download thread error: {e}")
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['state'] = 'Errored'
self.active_downloads[download_id]['error'] = str(e)
def _is_cancelled(self, download_id: str) -> bool:
if self._engine is None:
return False
record = self._engine.get_record('deezer', download_id)
return record is not None and record.get('state') == 'Cancelled'
def _download_sync(self, download_id: str, track_id: str, display_name: str) -> Optional[str]:
"""Synchronous download: get URL, download, decrypt, save."""
# Check for shutdown
if self.shutdown_check and self.shutdown_check():
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['state'] = 'Aborted'
if self._engine is not None:
self._engine.update_record('deezer', download_id, {'state': 'Aborted'})
return None
# Get track data from private API
track_data = self._get_track_data(track_id)
if not track_data:
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['error'] = 'Failed to get track data'
self._set_error(download_id, 'Failed to get track data')
return None
track_token = track_data.get('TRACK_TOKEN', '')
if not track_token:
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['error'] = 'No track token available'
self._set_error(download_id, 'No track token available')
return None
# Determine quality and get media URL with fallback
@ -695,7 +676,6 @@ class DeezerDownloadClient:
if allow_fallback:
quality_order = _QUALITY_ORDER.copy()
# Start from user's preferred quality
try:
pref_idx = quality_order.index(self._quality)
quality_order = quality_order[pref_idx:] + quality_order[:pref_idx]
@ -712,27 +692,16 @@ class DeezerDownloadClient:
break
if not media_url:
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['error'] = 'No media URL available (may require higher subscription tier)'
self._set_error(download_id, 'No media URL available (may require higher subscription tier)')
return None
if actual_quality != self._quality:
logger.info(f"Quality fallback: {self._quality} → {actual_quality} for {display_name}")
# Determine file extension
ext = '.flac' if actual_quality == 'flac' else '.mp3'
# Sanitize filename
safe_name = self._sanitize_filename(display_name)
out_path = str(self.download_path / f"{safe_name}{ext}")
# Update state
with self._download_lock:
if download_id in self.active_downloads:
dl = self.active_downloads[download_id]
dl['state'] = 'InProgress, Downloading'
# Download and decrypt
try:
bf_key = _get_blowfish_key(track_id)
@ -740,9 +709,8 @@ class DeezerDownloadClient:
resp.raise_for_status()
total_size = int(resp.headers.get('content-length', 0))
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['size'] = total_size
if self._engine is not None:
self._engine.update_record('deezer', download_id, {'size': total_size})
downloaded = 0
chunk_index = 0
@ -755,23 +723,20 @@ class DeezerDownloadClient:
# Check for cancellation/shutdown
if self.shutdown_check and self.shutdown_check():
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['state'] = 'Aborted'
if self._engine is not None:
self._engine.update_record('deezer', download_id, {'state': 'Aborted'})
try:
os.remove(out_path)
except OSError:
pass
return None
with self._download_lock:
if download_id in self.active_downloads:
if self.active_downloads[download_id]['state'] == 'Cancelled':
try:
os.remove(out_path)
except OSError:
pass
return None
if self._is_cancelled(download_id):
try:
os.remove(out_path)
except OSError:
pass
return None
# Decrypt every 3rd chunk (Deezer's encryption pattern)
if chunk_index % 3 == 0 and len(raw_chunk) == _CHUNK_SIZE:
@ -788,12 +753,12 @@ class DeezerDownloadClient:
speed = int(downloaded / elapsed) if elapsed > 0 else 0
progress = (downloaded / total_size * 100) if total_size > 0 else 0
with self._download_lock:
if download_id in self.active_downloads:
dl = self.active_downloads[download_id]
dl['transferred'] = downloaded
dl['progress'] = min(progress, 99.9)
dl['speed'] = speed
if self._engine is not None:
self._engine.update_record('deezer', download_id, {
'transferred': downloaded,
'progress': min(progress, 99.9),
'speed': speed,
})
# Validate file size
file_size = os.path.getsize(out_path)
@ -803,9 +768,7 @@ class DeezerDownloadClient:
os.remove(out_path)
except OSError:
pass
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['error'] = f'File too small ({file_size} bytes)'
self._set_error(download_id, f'File too small ({file_size} bytes)')
return None
logger.info(f"Deezer download complete: {out_path} ({file_size / 1048576:.1f} MB, {actual_quality})")
@ -817,59 +780,58 @@ class DeezerDownloadClient:
os.remove(out_path)
except OSError:
pass
with self._download_lock:
if download_id in self.active_downloads:
self.active_downloads[download_id]['error'] = str(e)
self._set_error(download_id, str(e))
return None
# ─── Download Status ─────────────────────────────────────────
def _record_to_status(self, record: dict) -> DownloadStatus:
return DownloadStatus(
id=record['id'],
filename=record['filename'],
username=record['username'],
state=record['state'],
progress=record['progress'],
size=record.get('size', 0),
transferred=record.get('transferred', 0),
speed=record.get('speed', 0),
file_path=record.get('file_path'),
)
async def get_all_downloads(self) -> List[DownloadStatus]:
"""Return all active downloads."""
with self._download_lock:
return [self._to_status(dl) for dl in self.active_downloads.values()]
if self._engine is None:
return []
return [
self._record_to_status(record)
for record in self._engine.iter_records_for_source('deezer')
]
async def get_download_status(self, download_id: str) -> Optional[DownloadStatus]:
"""Get status of a specific download."""
with self._download_lock:
dl = self.active_downloads.get(download_id)
return self._to_status(dl) if dl else None
if self._engine is None:
return None
record = self._engine.get_record('deezer', download_id)
return self._record_to_status(record) if record is not None else None
async def cancel_download(self, download_id: str, username: str = None,
remove: bool = False) -> bool:
"""Cancel a download."""
with self._download_lock:
dl = self.active_downloads.get(download_id)
if not dl:
return False
dl['state'] = 'Cancelled'
if remove:
del self.active_downloads[download_id]
if self._engine is None:
return False
if self._engine.get_record('deezer', download_id) is None:
return False
self._engine.update_record('deezer', download_id, {'state': 'Cancelled'})
if remove:
self._engine.remove_record('deezer', download_id)
return True
async def clear_all_completed_downloads(self) -> bool:
"""Remove all terminal downloads."""
terminal_states = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
with self._download_lock:
to_remove = [k for k, v in self.active_downloads.items() if v['state'] in terminal_states]
for k in to_remove:
del self.active_downloads[k]
if self._engine is None:
return True
terminal = {'Completed, Succeeded', 'Cancelled', 'Errored', 'Aborted'}
for record in list(self._engine.iter_records_for_source('deezer')):
if record.get('state') in terminal:
self._engine.remove_record('deezer', record['id'])
return True
def _to_status(self, dl: dict) -> DownloadStatus:
"""Convert internal dict to DownloadStatus."""
return DownloadStatus(
id=dl['id'],
filename=dl['filename'],
username=dl['username'],
state=dl['state'],
progress=dl['progress'],
size=dl['size'],
transferred=dl['transferred'],
speed=dl['speed'],
file_path=dl.get('file_path'),
)
# ─── Utilities ───────────────────────────────────────────────
@staticmethod

185
tests/downloads/test_deezer_pinning.py
View file

@ -1,16 +1,11 @@
"""Phase A pinning tests for DeezerDownloadClient's download lifecycle.
"""Phase A pinning tests for DeezerDownloadClient — UPDATED for Phase C6.
Deezer auths via ARL token, fetches Blowfish-encrypted FLAC chunks
from the Deezer GW API, decrypts client-side. Different from
Tidal/Qobuz/HiFi:
- track_id is STRING (not int).
- username is the legacy ``'deezer_dl'`` (not ``'deezer'``).
- Auth gate at the top of `download()` short-circuits when not
authenticated (returns None without spawning a thread).
- Thread is named ``deezer-dl-<track_id>`` for diagnostics.
Engine refactor must preserve all of these.
Deezer has the same engine-driven dispatch as the other streaming
sources, with three Deezer-specific quirks preserved:
- track_id stays as STRING (Deezer GW API uses string IDs).
- Engine record's `username` slot is the legacy `'deezer_dl'`
via worker username_override.
- Worker thread is named `deezer-dl-<track_id>` for diagnostics.
"""
from __future__ import annotations
@ -22,6 +17,7 @@ from unittest.mock import patch
import pytest
from core.download_engine import DownloadEngine
from core.deezer_download_client import DeezerDownloadClient
@ -34,98 +30,119 @@ def _run_async(coro):
@pytest.fixture
def deezer_client():
def deezer_client_with_engine():
client = DeezerDownloadClient.__new__(DeezerDownloadClient)
client.download_path = Path('./test_deezer_downloads')
client.shutdown_check = None
client.active_downloads = {}
client._download_lock = threading.Lock()
client._authenticated = True
return client
client._engine = None
engine = DownloadEngine()
client.set_engine(engine)
return client, engine
def test_download_returns_none_when_not_authenticated(deezer_client):
"""Pinning: unauthenticated client refuses BEFORE any thread is
spawned. The orchestrator's hybrid fallback depends on this
early return — if the auth gate moves into the thread, fallback
behavior changes."""
deezer_client._authenticated = False
result = _run_async(deezer_client.download('deezer_dl', '12345||Some Song', 0))
def test_download_returns_none_when_not_authenticated(deezer_client_with_engine):
client, _ = deezer_client_with_engine
client._authenticated = False
result = _run_async(client.download('deezer_dl', '12345||x', 0))
assert result is None
def test_download_accepts_string_track_id(deezer_client):
"""Pinning: Deezer track_id stays as string — the GW API uses
string IDs. Engine refactor cannot int-coerce on the way through."""
with patch('core.deezer_download_client.threading.Thread') as fake:
fake.return_value.start = lambda: None
download_id = _run_async(
deezer_client.download('deezer_dl', '999||My Deezer Song', 5000)
)
record = deezer_client.active_downloads[download_id]
assert record['track_id'] == '999' # STRING, not int
assert isinstance(record['track_id'], str)
def test_download_returns_none_when_engine_not_wired():
client = DeezerDownloadClient.__new__(DeezerDownloadClient)
client._engine = None
client._authenticated = True
result = _run_async(client.download('deezer_dl', '12345||x', 0))
assert result is None
def test_download_username_field_is_legacy_deezer_dl(deezer_client):
"""Pinning: the `username` slot in the state dict is ``'deezer_dl'``,
not ``'deezer'``. Frontend status indicators + per-source
dispatch strings depend on the legacy form."""
with patch('core.deezer_download_client.threading.Thread') as fake:
fake.return_value.start = lambda: None
download_id = _run_async(
deezer_client.download('deezer_dl', '999||x', 0)
)
def test_download_track_id_stays_as_string(deezer_client_with_engine):
"""Pinning: Deezer GW API uses string IDs — engine record must
keep track_id as str."""
client, engine = deezer_client_with_engine
started = threading.Event()
release = threading.Event()
assert deezer_client.active_downloads[download_id]['username'] == 'deezer_dl'
def slow_impl(*args, **kwargs):
started.set()
release.wait(timeout=1.0)
return '/tmp/done.flac'
with patch.object(client, '_download_sync', side_effect=slow_impl):
download_id = _run_async(client.download('deezer_dl', '999||X', 0))
started.wait(timeout=1.0)
record = engine.get_record('deezer', download_id)
assert record['track_id'] == '999'
assert isinstance(record['track_id'], str)
release.set()
def test_download_handles_missing_display_name_with_fallback(deezer_client):
"""Pinning: filename without `||` produces a synthetic display
name `Track <track_id>`. Other clients return None for missing
`||` — Deezer is more lenient. Engine refactor must NOT change
this defensive fallback."""
with patch('core.deezer_download_client.threading.Thread') as fake:
fake.return_value.start = lambda: None
download_id = _run_async(deezer_client.download('deezer_dl', '12345', 0))
def test_download_username_slot_is_legacy_deezer_dl(deezer_client_with_engine):
"""Pinning: frontend status indicators key off `'deezer_dl'`,
not the canonical `'deezer'`."""
client, engine = deezer_client_with_engine
started = threading.Event()
release = threading.Event()
assert download_id is not None
record = deezer_client.active_downloads[download_id]
assert record['display_name'] == 'Track 12345'
def slow_impl(*args, **kwargs):
started.set()
release.wait(timeout=1.0)
return '/tmp/done.flac'
with patch.object(client, '_download_sync', side_effect=slow_impl):
download_id = _run_async(client.download('deezer_dl', '999||x', 0))
started.wait(timeout=1.0)
assert engine.get_record('deezer', download_id)['username'] == 'deezer_dl'
release.set()
def test_download_populates_active_downloads_with_initial_state(deezer_client):
"""Pinning: per-download record schema. NOTE the extra `error`
slot — Deezer-specific, used for ARL re-auth failure messages."""
with patch('core.deezer_download_client.threading.Thread') as fake:
fake.return_value.start = lambda: None
download_id = _run_async(
deezer_client.download('deezer_dl', '999||My Deezer Song', 1024)
)
def test_download_handles_missing_display_name_with_fallback(deezer_client_with_engine):
"""Pinning: filename without `||` synthesizes display name `Track <id>`."""
client, engine = deezer_client_with_engine
started = threading.Event()
release = threading.Event()
record = deezer_client.active_downloads[download_id]
assert record['id'] == download_id
assert record['filename'] == '999||My Deezer Song'
assert record['username'] == 'deezer_dl'
assert record['state'] == 'Initializing'
assert record['size'] == 1024 # Deezer respects the file_size hint
assert record['file_path'] is None
assert record['error'] is None # Deezer-specific slot
def slow_impl(*args, **kwargs):
started.set()
release.wait(timeout=1.0)
return '/tmp/x.flac'
with patch.object(client, '_download_sync', side_effect=slow_impl):
download_id = _run_async(client.download('deezer_dl', '12345', 0))
started.wait(timeout=1.0)
assert engine.get_record('deezer', download_id)['display_name'] == 'Track 12345'
release.set()
def test_download_thread_is_named_for_diagnostics(deezer_client):
"""Pinning: thread is named `deezer-dl-<track_id>` so multi-thread
debugging shows which download a stuck thread belongs to. Engine
refactor's BackgroundDownloadWorker must preserve diagnostic naming."""
captured_kwargs = {}
def test_download_engine_record_carries_error_slot(deezer_client_with_engine):
"""Pinning: Deezer-specific `error` slot for ARL re-auth failure
messages must be present on init."""
client, engine = deezer_client_with_engine
started = threading.Event()
release = threading.Event()
def capture_thread(*args, **kwargs):
captured_kwargs.update(kwargs)
return type('FakeThread', (), {'start': lambda self: None})()
def slow_impl(*args, **kwargs):
started.set()
release.wait(timeout=1.0)
return '/tmp/x.flac'
with patch('core.deezer_download_client.threading.Thread', side_effect=capture_thread):
_run_async(deezer_client.download('deezer_dl', '777||Title', 0))
with patch.object(client, '_download_sync', side_effect=slow_impl):
download_id = _run_async(client.download('deezer_dl', '999||X', 1024))
started.wait(timeout=1.0)
record = engine.get_record('deezer', download_id)
assert 'error' in record
assert record['error'] is None
assert record['size'] == 1024
release.set()
assert captured_kwargs.get('daemon') is True
assert captured_kwargs.get('name') == 'deezer-dl-777'
def test_get_all_downloads_reads_engine_records(deezer_client_with_engine):
client, engine = deezer_client_with_engine
engine.add_record('deezer', 'dl-1', {
'id': 'dl-1', 'filename': '111||A', 'username': 'deezer_dl',
'state': 'InProgress, Downloading', 'progress': 50.0,
})
result = _run_async(client.get_all_downloads())
assert len(result) == 1
assert result[0].id == 'dl-1'
assert result[0].username == 'deezer_dl'