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

Extract automation handlers (2/N): playlist lifecycle group

Browse source 
Continues the lift from `web_server._register_automation_handlers`.
This commit extracts the four playlist-lifecycle closures:

- `refresh_mirrored`   -> core/automation/handlers/refresh_mirrored.py
- `sync_playlist`      -> core/automation/handlers/sync_playlist.py
- `discover_playlist`  -> core/automation/handlers/discover_playlist.py
- `playlist_pipeline`  -> core/automation/handlers/playlist_pipeline.py

The pipeline composes refresh + sync + discover, so all four ship
together. The pipeline imports the other three handler modules
directly (cross-handler call) instead of going through the engine,
preserving the "single trigger from the user's perspective" UX.

`AutomationDeps` grew to cover the new dependency surface:
- run_playlist_discovery_worker, run_sync_task, load_sync_status_file
  (pre-existing background-task entry points)
- get_deezer_client, parse_youtube_playlist (per-source clients)
- get_sync_states (live mutable accessor for the sync UI's state dict)

`web_server._register_automation_handlers` now wires those plus the
existing infrastructure into a single `AutomationDeps` and calls
`register_all`. The 669-line block of closure definitions and engine
register calls (lines 959-1627 pre-edit) is gone -- the file shed
743 lines net on this commit.

`tests/automation/test_handlers_playlist.py` adds 17 new boundary
tests:
- discover_playlist: no_id error, specific_id starts worker, all=True
  enumerates, no playlists in db
- refresh_mirrored: error path, source filter (file/beatport excluded),
  Spotify happy path with auto-discovered marker, per-playlist
  exception captured into errors counter
- sync_playlist: no_id, not_found, no_tracks, no-discovered-tracks
  skip, discovered-track happy path, unchanged-since-last-sync skip
- playlist_pipeline: no_playlist clears running flag, no-refreshable
  clears running flag, exception clears running flag

3223 tests pass. web_server.py: 35,593 -> 34,850 lines (743 removed).
This commit is contained in:
Broque Thomas 2026-05-15 10:47:46 -07:00
parent ea7d5c65bb
commit cde237c7e7
10 changed files with 1304 additions and 670 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
core/automation/deps.py
View file

@ -93,3 +93,11 @@ class AutomationDeps:
is_wishlist_actually_processing: Callable[[], bool]
is_watchlist_actually_scanning: Callable[[], bool]
get_watchlist_scan_state: Callable[[], dict] # accessor returns the live mutable dict
# --- Playlist pipeline entry points ---
run_playlist_discovery_worker: Callable[..., Any]
run_sync_task: Callable[..., Any]
load_sync_status_file: Callable[[], dict]
get_deezer_client: Callable[[], Any]
parse_youtube_playlist: Callable[[str], Any]
get_sync_states: Callable[[], dict] # accessor returns the live dict shared with the sync UI

8
core/automation/handlers/__init__.py
View file

@ -13,11 +13,19 @@ to the engine in one place. ``web_server.py`` calls
from core.automation.handlers.process_wishlist import auto_process_wishlist
from core.automation.handlers.scan_watchlist import auto_scan_watchlist
from core.automation.handlers.scan_library import auto_scan_library
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
from core.automation.handlers.discover_playlist import auto_discover_playlist
from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
from core.automation.handlers.registration import register_all
__all__ = [
'auto_process_wishlist',
'auto_scan_watchlist',
'auto_scan_library',
'auto_refresh_mirrored',
'auto_sync_playlist',
'auto_discover_playlist',
'auto_playlist_pipeline',
'register_all',
]

48
core/automation/handlers/discover_playlist.py Normal file
View file

@ -0,0 +1,48 @@
"""Automation handler: ``discover_playlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_discover_playlist`` closure). Kicks off background discovery
of official Spotify / iTunes metadata for mirrored playlist tracks.
The worker runs in a daemon thread and emits its own progress; this
handler returns immediately after launching it (``_manages_own_progress``).
"""
from __future__ import annotations
import threading
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_discover_playlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Discover official Spotify/iTunes metadata for mirrored
playlist tracks. Runs the worker in a background thread."""
db = deps.get_database()
playlist_id = config.get('playlist_id')
discover_all = config.get('all', False)
if discover_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
if not playlists:
return {'status': 'error', 'reason': 'No playlists found'}
threading.Thread(
target=deps.run_playlist_discovery_worker,
args=(playlists, config.get('_automation_id')),
daemon=True,
name='auto-discover-playlist',
).start()
names = ', '.join(p['name'] for p in playlists[:3])
return {
'status': 'started',
'playlist_count': str(len(playlists)),
'playlists': names,
'_manages_own_progress': True,
}

320
core/automation/handlers/playlist_pipeline.py Normal file
View file

@ -0,0 +1,320 @@
"""Automation handler: ``playlist_pipeline`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_playlist_pipeline`` closure). Runs the full playlist
lifecycle in a single trigger:
Phase 1: REFRESH -- pull fresh track lists from sources
Phase 2: DISCOVER -- look up official Spotify/iTunes metadata
Phase 3: SYNC -- push the result to the active media server
Phase 4: WISHLIST -- queue any missing tracks for download
Each phase emits its own progress range so the trigger card shows
useful per-phase percentages instead of "loading...". Phase 4 is
optional via ``skip_wishlist`` config.
Composition: this handler invokes ``auto_refresh_mirrored`` and
``auto_sync_playlist`` directly (passing ``_automation_id: None`` so
the sub-handlers don't hijack pipeline progress) instead of going
through the engine — keeps the four phases observable as one
trigger from the user's perspective. Pipeline-level guard
(``state.pipeline_running``) prevents overlapping runs.
"""
from __future__ import annotations
import threading
import time
from typing import Any, Dict
from core.automation.deps import AutomationDeps
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
# Per-playlist sync poll cap inside Phase 3.
_SYNC_PER_PLAYLIST_TIMEOUT_SECONDS = 600
# Discovery poll cap inside Phase 2.
_DISCOVERY_TIMEOUT_SECONDS = 3600
def auto_playlist_pipeline(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Run REFRESH → DISCOVER → SYNC → WISHLIST in sequence.
Sets / clears ``deps.state.pipeline_running`` around the whole
run so the registration guard can short-circuit overlapping
triggers.
"""
deps.state.set_pipeline_running(True)
automation_id = config.get('_automation_id')
pipeline_start = time.time()
try:
db = deps.get_database()
playlist_id = config.get('playlist_id')
process_all = config.get('all', False)
skip_wishlist = config.get('skip_wishlist', False)
# Resolve playlists.
if process_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
deps.state.set_pipeline_running(False)
return {'status': 'error', 'error': 'No playlist specified'}
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
if not playlists:
deps.state.set_pipeline_running(False)
return {'status': 'error', 'error': 'No refreshable playlists found'}
pl_names = ', '.join(p.get('name', '?') for p in playlists[:3])
if len(playlists) > 3:
pl_names += f' (+{len(playlists) - 3} more)'
deps.update_progress(
automation_id,
progress=2,
phase=f'Pipeline: {len(playlists)} playlist(s)',
log_line=f'Starting pipeline for: {pl_names}',
log_type='info',
)
# ── PHASE 1: REFRESH ──────────────────────────────────────────
deps.update_progress(
automation_id,
progress=3,
phase='Phase 1/4: Refreshing playlists...',
log_line='Phase 1: Refresh',
log_type='info',
)
refresh_config = dict(config)
refresh_config['_automation_id'] = None # Don't let sub-handler hijack pipeline progress.
refresh_result = auto_refresh_mirrored(refresh_config, deps)
refreshed = int(refresh_result.get('refreshed', 0))
refresh_errors = int(refresh_result.get('errors', 0))
deps.update_progress(
automation_id,
progress=25,
phase='Phase 1/4: Refresh complete',
log_line=f'Phase 1 done: {refreshed} refreshed, {refresh_errors} errors',
log_type='success' if refresh_errors == 0 else 'warning',
)
# ── PHASE 2: DISCOVER ─────────────────────────────────────────
deps.update_progress(
automation_id,
progress=26,
phase='Phase 2/4: Discovering metadata...',
log_line='Phase 2: Discover',
log_type='info',
)
# Reload playlists (refresh may have updated them).
if process_all:
disc_playlists = db.get_mirrored_playlists()
else:
disc_playlists = [db.get_mirrored_playlist(int(playlist_id))]
disc_playlists = [p for p in disc_playlists if p]
# Run discovery in a thread and wait for it.
disc_done = threading.Event()
def _disc_wrapper(pls):
try:
# The worker updates automation_progress internally,
# but we pass None so it doesn't conflict with our
# pipeline progress.
deps.run_playlist_discovery_worker(pls, automation_id=None)
except Exception as e:
deps.logger.error(f"[Pipeline] Discovery error: {e}")
finally:
disc_done.set()
threading.Thread(
target=_disc_wrapper, args=(disc_playlists,),
daemon=True, name='pipeline-discover',
).start()
# Poll for completion with progress updates.
poll_start = time.time()
while not disc_done.wait(timeout=3):
elapsed = int(time.time() - poll_start)
deps.update_progress(
automation_id,
progress=min(26 + elapsed // 4, 54),
phase=f'Phase 2/4: Discovering... ({elapsed}s)',
)
if elapsed > _DISCOVERY_TIMEOUT_SECONDS:
deps.update_progress(
automation_id,
log_line='Discovery timed out after 1 hour',
log_type='warning',
)
break
deps.update_progress(
automation_id,
progress=55,
phase='Phase 2/4: Discovery complete',
log_line='Phase 2 done: discovery complete',
log_type='success',
)
# ── PHASE 3: SYNC ─────────────────────────────────────────────
deps.update_progress(
automation_id,
progress=56,
phase='Phase 3/4: Syncing to server...',
log_line='Phase 3: Sync',
log_type='info',
)
total_synced = 0
total_skipped = 0
sync_errors = 0
sync_states = deps.get_sync_states()
for pl_idx, pl in enumerate(playlists):
pl_id = pl.get('id')
if not pl_id:
continue
sync_config = {
'playlist_id': str(pl_id),
'_automation_id': None, # Don't let sync handler hijack our progress.
}
sync_result = auto_sync_playlist(sync_config, deps)
sync_status = sync_result.get('status', '')
if sync_status == 'started':
# Sync launched a background thread — wait for it.
sync_id = f"auto_mirror_{pl_id}"
sync_poll_start = time.time()
while time.time() - sync_poll_start < _SYNC_PER_PLAYLIST_TIMEOUT_SECONDS:
if (sync_id in sync_states
and sync_states[sync_id].get('status')
in ('finished', 'complete', 'error', 'failed')):
break
time.sleep(2)
elapsed = int(time.time() - sync_poll_start)
sub_progress = 56 + ((pl_idx + 1) / max(1, len(playlists))) * 29
deps.update_progress(
automation_id,
progress=min(int(sub_progress), 84),
phase=f'Phase 3/4: Syncing "{pl.get("name", "")}" ({elapsed}s)',
)
# Check result.
ss = sync_states.get(sync_id, {})
ss_result = ss.get('result', ss.get('progress', {}))
matched = ss_result.get('matched_tracks', 0) if isinstance(ss_result, dict) else 0
total_synced += int(matched) if matched else 0
deps.update_progress(
automation_id,
log_line=f'Synced "{pl.get("name", "")}": {matched} tracks matched',
log_type='success',
)
elif sync_status == 'skipped':
total_skipped += 1
reason = sync_result.get('reason', 'unchanged')
deps.update_progress(
automation_id,
log_line=f'Skipped "{pl.get("name", "")}": {reason}',
log_type='skip',
)
elif sync_status == 'error':
sync_errors += 1
deps.update_progress(
automation_id,
log_line=f'Sync error "{pl.get("name", "")}": {sync_result.get("reason", "unknown")}',
log_type='error',
)
deps.update_progress(
automation_id,
progress=85,
phase='Phase 3/4: Sync complete',
log_line=f'Phase 3 done: {total_synced} matched, {total_skipped} skipped, {sync_errors} errors',
log_type='success' if sync_errors == 0 else 'warning',
)
# ── PHASE 4: WISHLIST ─────────────────────────────────────────
wishlist_queued = 0
if not skip_wishlist:
deps.update_progress(
automation_id,
progress=86,
phase='Phase 4/4: Processing wishlist...',
log_line='Phase 4: Wishlist',
log_type='info',
)
try:
if not deps.is_wishlist_actually_processing():
deps.process_wishlist_automatically(automation_id=None)
deps.update_progress(
automation_id,
log_line='Wishlist processing triggered',
log_type='success',
)
wishlist_queued = 1
else:
deps.update_progress(
automation_id,
log_line='Wishlist already running — skipped',
log_type='skip',
)
except Exception as e:
deps.update_progress(
automation_id,
log_line=f'Wishlist error: {e}',
log_type='warning',
)
else:
deps.update_progress(
automation_id,
progress=86,
log_line='Phase 4: Wishlist skipped (disabled)',
log_type='skip',
)
# ── COMPLETE ──────────────────────────────────────────────────
duration = int(time.time() - pipeline_start)
deps.update_progress(
automation_id,
status='finished',
progress=100,
phase='Pipeline complete',
log_line=f'Pipeline finished in {duration // 60}m {duration % 60}s',
log_type='success',
)
deps.state.set_pipeline_running(False)
return {
'status': 'completed',
'_manages_own_progress': True,
'playlists_refreshed': str(refreshed),
'tracks_discovered': 'completed',
'tracks_synced': str(total_synced),
'sync_skipped': str(total_skipped),
'wishlist_queued': str(wishlist_queued),
'duration_seconds': str(duration),
}
except Exception as e:
deps.state.set_pipeline_running(False)
deps.update_progress(
automation_id,
status='error',
progress=100,
phase='Pipeline error',
log_line=f'Pipeline failed: {e}',
log_type='error',
)
return {'status': 'error', 'error': str(e), '_manages_own_progress': True}

308
core/automation/handlers/refresh_mirrored.py Normal file
View file

@ -0,0 +1,308 @@
"""Automation handler: ``refresh_mirrored`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_refresh_mirrored`` closure). Re-pulls track lists from each
mirrored playlist's source (Spotify / Tidal / Deezer / YouTube),
updates the local mirror DB, and emits a ``playlist_changed``
automation event when the track set actually shifts.
Source-specific branches (Spotify auth + public-embed fallback,
``spotify_public`` URL→ID resolution, Deezer / Tidal / YouTube)
remain identical to the pre-extraction closure — this is a
mechanical lift, not a redesign.
"""
from __future__ import annotations
import json
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_refresh_mirrored(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Refresh mirrored playlist(s) from source.
Returns ``{'status': 'completed', 'refreshed': '<int>',
'errors': '<int>'}`` on success (counts stringified to match the
automation engine's stat-rendering convention).
"""
db = deps.get_database()
playlist_id = config.get('playlist_id')
refresh_all = config.get('all', False)
auto_id = config.get('_automation_id')
if refresh_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
# Filter out sources that can't be refreshed (no external API).
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
refreshed = 0
errors = []
for idx, pl in enumerate(playlists):
try:
source = pl.get('source', '')
source_id = pl.get('source_playlist_id', '')
deps.update_progress(
auto_id,
progress=(idx / max(1, len(playlists))) * 100,
phase=f'Refreshing: "{pl.get("name", "")}"',
current_item=pl.get('name', ''),
)
tracks = None
if source == 'spotify':
# Try authenticated API first, fall back to public embed scraper.
if deps.spotify_client and deps.spotify_client.is_spotify_authenticated():
playlist_obj = deps.spotify_client.get_playlist_by_id(source_id)
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
# Spotify data IS official — auto-mark as discovered.
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth needed).
if tracks is None:
try:
from core.spotify_public_scraper import scrape_spotify_embed
embed_data = scrape_spotify_embed('playlist', source_id)
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
track_dict = {
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
}
# Store Spotify track ID hint but don't mark discovered —
# Discover step needs to run for proper album art.
if t.get('id'):
track_dict['extra_data'] = json.dumps({
'discovered': False,
'spotify_hint': {
'id': t['id'],
'name': t.get('name', ''),
'artists': t.get('artists', []),
}
})
tracks.append(track_dict)
except Exception as e:
deps.logger.warning(f"Spotify public scraper fallback failed for {source_id}: {e}")
elif source == 'spotify_public':
# source_playlist_id is an MD5 hash; extract actual Spotify ID from stored description (URL).
try:
from core.spotify_public_scraper import parse_spotify_url, scrape_spotify_embed
spotify_url = pl.get('description', '')
parsed = parse_spotify_url(spotify_url) if spotify_url else None
# If Spotify is authenticated, use the full API (auto-discovers with album art).
if (parsed and parsed.get('type') == 'playlist'
and deps.spotify_client and deps.spotify_client.is_spotify_authenticated()):
playlist_obj = deps.spotify_client.get_playlist_by_id(parsed['id'])
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth or album-type URL).
if tracks is None and parsed:
embed_data = scrape_spotify_embed(parsed['type'], parsed['id'])
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
# No extra_data — let preservation code keep existing discovery data.
except Exception as e:
deps.logger.warning(f"Spotify public playlist refresh failed for {source_id}: {e}")
elif source == 'deezer':
try:
deezer = deps.get_deezer_client()
playlist_data = deezer.get_playlist(source_id)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': t.get('album', ''),
'duration_ms': t.get('duration_ms', 0),
'source_track_id': str(t.get('id', '')),
})
except Exception as e:
deps.logger.warning(f"Deezer playlist refresh failed for {source_id}: {e}")
elif source == 'tidal':
if not deps.tidal_client or not deps.tidal_client.is_authenticated():
deps.logger.warning(f"Tidal not authenticated — skipping refresh for '{pl.get('name', '')}'")
deps.update_progress(
auto_id,
log_line=f'Skipped "{pl.get("name", "")}" — Tidal not authenticated',
log_type='skip',
)
continue
full_playlist = deps.tidal_client.get_playlist(source_id)
if full_playlist and full_playlist.tracks:
tracks = []
for t in full_playlist.tracks:
artist_name = t.artists[0] if t.artists else ''
tracks.append({
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
})
elif source == 'youtube':
# source_playlist_id is now a deterministic hash; use stored description (original URL) for refresh.
yt_url = pl.get('description', '') or f"https://www.youtube.com/playlist?list={source_id}"
playlist_data = deps.parse_youtube_playlist(yt_url)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': '',
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
if tracks is not None:
# Compare old vs new track IDs to detect changes.
old_tracks = db.get_mirrored_playlist_tracks(pl['id']) if pl.get('id') else []
old_ids = {t.get('source_track_id') for t in old_tracks if t.get('source_track_id')}
new_ids = {t.get('source_track_id') for t in tracks if t.get('source_track_id')}
# Preserve existing discovery extra_data for tracks that still exist.
old_extra_map = db.get_mirrored_tracks_extra_data_map(pl['id']) if pl.get('id') else {}
for t in tracks:
sid = t.get('source_track_id', '')
if sid and sid in old_extra_map and 'extra_data' not in t:
t['extra_data'] = old_extra_map[sid]
db.mirror_playlist(
source=source,
source_playlist_id=source_id,
name=pl['name'],
tracks=tracks,
profile_id=pl.get('profile_id', 1),
owner=pl.get('owner'),
image_url=pl.get('image_url'),
)
refreshed += 1
# Emit playlist_changed if tracks actually changed.
if old_ids != new_ids:
added_count = len(new_ids - old_ids)
removed_count = len(old_ids - new_ids)
deps.logger.info(
f"[AUTOMATION] Playlist changed: '{pl.get('name', '')}' — "
f"{added_count} added, {removed_count} removed (old={len(old_ids)}, new={len(new_ids)})"
)
deps.update_progress(
auto_id,
log_line=f'"{pl.get("name", "")}" — {added_count} added, {removed_count} removed',
log_type='success',
)
try:
if deps.engine:
deps.engine.emit('playlist_changed', {
'playlist_name': pl.get('name', ''),
'playlist_id': str(pl.get('id', '')),
'old_count': str(len(old_ids)),
'new_count': str(len(new_ids)),
'added': str(added_count),
'removed': str(removed_count),
})
except Exception as e:
deps.logger.debug("playlist_synced automation emit failed: %s", e)
else:
deps.logger.warning(f"[AUTOMATION] No changes: '{pl.get('name', '')}' (tracks={len(old_ids)})")
deps.update_progress(
auto_id,
log_line=f'No changes: "{pl.get("name", "")}"',
log_type='skip',
)
except Exception as e:
errors.append(f"{pl.get('name', '?')}: {str(e)}")
deps.update_progress(
auto_id,
log_line=f'Error: {pl.get("name", "?")} — {str(e)}',
log_type='error',
)
return {'status': 'completed', 'refreshed': str(refreshed), 'errors': str(len(errors))}

26
core/automation/handlers/registration.py
View file

@ -11,6 +11,10 @@ from core.automation.deps import AutomationDeps
from core.automation.handlers.process_wishlist import auto_process_wishlist
from core.automation.handlers.scan_watchlist import auto_scan_watchlist
from core.automation.handlers.scan_library import auto_scan_library
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
from core.automation.handlers.discover_playlist import auto_discover_playlist
from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
def register_all(deps: AutomationDeps) -> None:
@ -43,3 +47,25 @@ def register_all(deps: AutomationDeps) -> None:
lambda config: auto_scan_library(config, deps),
deps.state.is_scan_library_active,
)
# Playlist lifecycle handlers. The pipeline composes refresh +
# sync + discover (it imports them directly), so all four ship
# together. The pipeline guard prevents an in-flight pipeline
# from being re-triggered mid-run.
engine.register_action_handler(
'refresh_mirrored',
lambda config: auto_refresh_mirrored(config, deps),
)
engine.register_action_handler(
'sync_playlist',
lambda config: auto_sync_playlist(config, deps),
)
engine.register_action_handler(
'discover_playlist',
lambda config: auto_discover_playlist(config, deps),
)
engine.register_action_handler(
'playlist_pipeline',
lambda config: auto_playlist_pipeline(config, deps),
deps.state.is_pipeline_running,
)

195
core/automation/handlers/sync_playlist.py Normal file
View file

@ -0,0 +1,195 @@
"""Automation handler: ``sync_playlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_sync_playlist`` closure). Syncs a mirrored playlist to the
configured media server, using discovered metadata when available
and skipping undiscovered tracks. When triggered on a schedule with
no track changes since the last sync, short-circuits with
``status: skipped`` (saves Plex / Jellyfin / Navidrome from
needless rewrites)."""
from __future__ import annotations
import hashlib
import json
import threading
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_sync_playlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Sync a mirrored playlist to the active media server.
Behavior:
- Tracks with discovered metadata (extra_data.discovered + matched_data)
are routed via the official metadata.
- Tracks with a Spotify hint (real Spotify ID from the embed
scraper) are included so they can still hit Soulseek + the
wishlist.
- Tracks with neither are counted as ``skipped_tracks``.
- Empty result → ``status: skipped`` with the skipped count.
- Same track set as last sync (matched_tracks unchanged) →
``status: skipped`` (no-op).
- Otherwise spawns a daemon thread running ``run_sync_task`` and
returns ``status: started`` with ``_manages_own_progress: True``.
"""
auto_id = config.get('_automation_id')
playlist_id = config.get('playlist_id')
if not playlist_id:
return {'status': 'error', 'reason': 'No playlist specified'}
db = deps.get_database()
pl = db.get_mirrored_playlist(int(playlist_id))
if not pl:
return {'status': 'error', 'reason': 'Playlist not found'}
tracks = db.get_mirrored_playlist_tracks(int(playlist_id))
if not tracks:
return {'status': 'error', 'reason': 'No tracks in playlist'}
# Convert mirrored tracks to format expected by run_sync_task.
# Use discovered metadata when available, fall back to Spotify
# hint or raw playlist fields when not.
tracks_json = []
skipped_count = 0
for t in tracks:
# Parse extra_data for discovery info.
extra = {}
if t.get('extra_data'):
try:
extra = json.loads(t['extra_data']) if isinstance(t['extra_data'], str) else t['extra_data']
except (json.JSONDecodeError, TypeError):
pass
if extra.get('discovered') and extra.get('matched_data'):
# Use official discovered metadata.
md = extra['matched_data']
album_raw = md.get('album', '')
album_obj = album_raw if isinstance(album_raw, dict) else {'name': album_raw or ''}
_track_entry = {
'name': md.get('name', ''),
'artists': md.get('artists', [{'name': t.get('artist_name', '')}]),
'album': album_obj,
'duration_ms': md.get('duration_ms', 0),
'id': md.get('id', ''),
}
if md.get('track_number'):
_track_entry['track_number'] = md['track_number']
if md.get('disc_number'):
_track_entry['disc_number'] = md['disc_number']
tracks_json.append(_track_entry)
else:
# NOT discovered — try to include using available metadata so
# the track can still be searched on Soulseek and added to
# wishlist. Without this, failed discovery blocks the entire
# download pipeline.
#
# Priority: spotify_hint (has real Spotify ID from embed
# scraper) > raw playlist fields (only if source_track_id
# is valid).
hint = extra.get('spotify_hint', {})
# Build album object with cover art from the mirrored playlist track.
track_image = (t.get('image_url') or '').strip()
album_obj = {
'name': (t.get('album_name') or '').strip(),
'images': [{'url': track_image, 'height': 300, 'width': 300}] if track_image else [],
}
if hint.get('id') and hint.get('name'):
# spotify_hint has proper Spotify track ID + metadata from embed scraper.
hint_artists = hint.get('artists', [])
if hint_artists and isinstance(hint_artists[0], str):
hint_artists = [{'name': a} for a in hint_artists]
elif hint_artists and isinstance(hint_artists[0], dict):
pass # Already in correct format
else:
hint_artists = [{'name': t.get('artist_name', '')}]
tracks_json.append({
'name': hint['name'],
'artists': hint_artists,
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': hint['id'],
})
elif t.get('source_track_id') and (t.get('track_name') or '').strip():
# Has a valid source ID and track name — usable for wishlist.
tracks_json.append({
'name': t['track_name'].strip(),
'artists': [{'name': (t.get('artist_name') or '').strip() or 'Unknown Artist'}],
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': t['source_track_id'],
})
else:
skipped_count += 1 # No usable ID or name — truly can't process.
if not tracks_json:
deps.update_progress(
auto_id,
log_line=f'No discovered tracks — {skipped_count} need discovery first',
log_type='skip',
)
return {
'status': 'skipped',
'reason': f'No discovered tracks to sync ({skipped_count} tracks need discovery first)',
'skipped_tracks': str(skipped_count),
}
# Preflight: hash the track list and compare against last sync.
# Skip if the exact same set of tracks was already synced and
# everything matched (no-op preserves Plex / Jellyfin / Navidrome
# from needless rewrites).
track_ids_str = ','.join(sorted(t.get('id', '') for t in tracks_json))
tracks_hash = hashlib.md5(track_ids_str.encode()).hexdigest()
sync_id_key = f"auto_mirror_{playlist_id}"
try:
sync_statuses = deps.load_sync_status_file()
last_status = sync_statuses.get(sync_id_key, {})
last_hash = last_status.get('tracks_hash', '')
last_matched = last_status.get('matched_tracks', -1)
if last_hash == tracks_hash and last_matched >= len(tracks_json):
# Exact same tracks, all matched last time — nothing to do.
deps.update_progress(
auto_id,
log_line=f'All {len(tracks_json)} tracks unchanged since last sync — skipping',
log_type='skip',
)
return {
'status': 'skipped',
'reason': f'All {len(tracks_json)} tracks unchanged since last sync',
}
except Exception as e:
deps.logger.debug("mirror sync last-status read: %s", e)
deps.update_progress(
auto_id,
progress=50,
phase=f'Syncing "{pl["name"]}"',
log_line=f'{len(tracks_json)} discovered, {skipped_count} skipped',
log_type='info',
)
sync_id = f"auto_mirror_{playlist_id}"
deps.update_progress(
auto_id,
progress=90,
log_line=f'Starting sync: {len(tracks_json)} tracks',
log_type='success',
)
threading.Thread(
target=deps.run_sync_task,
args=(sync_id, pl['name'], tracks_json, auto_id, 1, pl.get('image_url', '')),
daemon=True,
name=f'auto-sync-{playlist_id}',
).start()
return {
'status': 'started',
'playlist_name': pl['name'],
'discovered_tracks': str(len(tracks_json)),
'skipped_tracks': str(skipped_count),
'_manages_own_progress': True,
}

371
tests/automation/test_handlers_playlist.py Normal file
View file

@ -0,0 +1,371 @@
"""Boundary tests for the playlist-lifecycle automation handlers
(``refresh_mirrored``, ``sync_playlist``, ``discover_playlist``,
``playlist_pipeline``).
The handlers themselves are mechanical lifts of the closures that
used to live in ``web_server._register_automation_handlers`` — these
tests pin the seam so the wiring stays correct (deps are read from
the deps object, not module-level globals; cross-handler calls in
the pipeline still compose; failure paths still return clear status
shapes).
Source-specific branches inside ``refresh_mirrored`` (Spotify auth
+ public-embed fallback, Deezer / Tidal / YouTube) are validated
end-to-end via fake clients in ``deps`` rather than per-source
because they're a verbatim lift — drift would show up here as a
behavior change."""
from __future__ import annotations
import json
import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
import pytest
from core.automation.deps import AutomationDeps, AutomationState
from core.automation.handlers.discover_playlist import auto_discover_playlist
from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
# ─── shared scaffolding ──────────────────────────────────────────────
class _StubLogger:
def __init__(self):
self.messages: List[tuple] = []
def debug(self, *a, **k): self.messages.append(('debug', a))
def info(self, *a, **k): self.messages.append(('info', a))
def warning(self, *a, **k): self.messages.append(('warning', a))
def error(self, *a, **k): self.messages.append(('error', a))
@dataclass
class _StubDB:
"""Fake MusicDatabase — minimal surface used by the playlist handlers."""
playlists: List[dict] = field(default_factory=list)
playlist_tracks: Dict[int, List[dict]] = field(default_factory=dict)
extra_data_maps: Dict[int, Dict[str, str]] = field(default_factory=dict)
mirror_calls: List[dict] = field(default_factory=list)
def get_mirrored_playlists(self) -> list:
return list(self.playlists)
def get_mirrored_playlist(self, playlist_id: int) -> Optional[dict]:
for p in self.playlists:
if int(p.get('id', -1)) == int(playlist_id):
return p
return None
def get_mirrored_playlist_tracks(self, playlist_id: int) -> list:
return list(self.playlist_tracks.get(int(playlist_id), []))
def get_mirrored_tracks_extra_data_map(self, playlist_id: int) -> dict:
return dict(self.extra_data_maps.get(int(playlist_id), {}))
def mirror_playlist(self, **kwargs) -> None:
self.mirror_calls.append(kwargs)
def _build_deps(**overrides) -> AutomationDeps:
defaults = dict(
engine=object(),
state=AutomationState(),
config_manager=object(),
update_progress=lambda *a, **k: None,
logger=_StubLogger(),
get_database=lambda: _StubDB(),
spotify_client=None,
tidal_client=None,
web_scan_manager=None,
process_wishlist_automatically=lambda **k: None,
process_watchlist_scan_automatically=lambda **k: None,
is_wishlist_actually_processing=lambda: False,
is_watchlist_actually_scanning=lambda: False,
get_watchlist_scan_state=lambda: {},
run_playlist_discovery_worker=lambda *a, **k: None,
run_sync_task=lambda *a, **k: None,
load_sync_status_file=lambda: {},
get_deezer_client=lambda: None,
parse_youtube_playlist=lambda url: None,
get_sync_states=lambda: {},
)
defaults.update(overrides)
return AutomationDeps(**defaults) # type: ignore[arg-type]
# ─── discover_playlist ───────────────────────────────────────────────
class TestDiscoverPlaylist:
def test_no_playlist_id_returns_error(self):
deps = _build_deps()
result = auto_discover_playlist({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_specific_playlist_id_starts_worker(self):
db = _StubDB(playlists=[{'id': 42, 'name': 'Test Playlist'}])
called: List[Any] = []
deps = _build_deps(
get_database=lambda: db,
run_playlist_discovery_worker=lambda *a, **k: called.append((a, k)),
)
result = auto_discover_playlist({'playlist_id': '42', '_automation_id': 'auto-1'}, deps)
assert result['status'] == 'started'
assert result['_manages_own_progress'] is True
assert result['playlist_count'] == '1'
# Worker spawned on a thread; give it a moment.
for _ in range(50):
if called:
break
import time
time.sleep(0.01)
assert len(called) == 1
def test_all_playlists_includes_every_one(self):
db = _StubDB(playlists=[
{'id': 1, 'name': 'A'}, {'id': 2, 'name': 'B'}, {'id': 3, 'name': 'C'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_discover_playlist({'all': True}, deps)
assert result['playlist_count'] == '3'
assert 'A' in result['playlists']
assert 'B' in result['playlists']
assert 'C' in result['playlists']
def test_no_playlists_in_db_returns_error(self):
deps = _build_deps(get_database=lambda: _StubDB(playlists=[]))
result = auto_discover_playlist({'all': True}, deps)
assert result == {'status': 'error', 'reason': 'No playlists found'}
# ─── refresh_mirrored ────────────────────────────────────────────────
@dataclass
class _StubSpotifyTrack:
id: str
name: str
artists: list
album: str
duration_ms: int
image_url: Optional[str] = None
@dataclass
class _StubSpotifyPlaylist:
tracks: list
class _StubSpotifyClient:
def __init__(self, playlist):
self._playlist = playlist
self._authenticated = True
def is_spotify_authenticated(self) -> bool:
return self._authenticated
def get_playlist_by_id(self, _source_id):
return self._playlist
class TestRefreshMirrored:
def test_no_playlist_specified_returns_error(self):
deps = _build_deps()
result = auto_refresh_mirrored({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_filters_unrefreshable_sources(self):
# Sources 'file' and 'beatport' have no API to refresh from.
db = _StubDB(playlists=[
{'id': 1, 'name': 'F', 'source': 'file', 'source_playlist_id': '1'},
{'id': 2, 'name': 'B', 'source': 'beatport', 'source_playlist_id': '2'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_refresh_mirrored({'all': True}, deps)
assert result['status'] == 'completed'
assert result['refreshed'] == '0'
assert db.mirror_calls == [] # nothing got pushed to DB
def test_spotify_refresh_writes_to_db(self):
track = _StubSpotifyTrack(
id='track123', name='Hello', artists=['Adele'],
album='25', duration_ms=295000,
)
playlist = _StubSpotifyPlaylist(tracks=[track])
spotify = _StubSpotifyClient(playlist)
db = _StubDB(playlists=[
{'id': 5, 'name': 'My Spot', 'source': 'spotify',
'source_playlist_id': 'spot-id', 'profile_id': 1},
])
deps = _build_deps(get_database=lambda: db, spotify_client=spotify)
result = auto_refresh_mirrored({'playlist_id': '5'}, deps)
assert result['status'] == 'completed'
assert result['refreshed'] == '1'
assert len(db.mirror_calls) == 1
call = db.mirror_calls[0]
assert call['source'] == 'spotify'
assert call['source_playlist_id'] == 'spot-id'
assert call['name'] == 'My Spot'
assert len(call['tracks']) == 1
# Spotify-source tracks should be auto-marked discovered.
extra = json.loads(call['tracks'][0]['extra_data'])
assert extra['discovered'] is True
assert extra['provider'] == 'spotify'
assert extra['matched_data']['id'] == 'track123'
def test_per_playlist_exception_collected_into_errors(self):
# Force an exception by making the DB blow up on mirror_playlist.
class _ExplodingDB(_StubDB):
def mirror_playlist(self, **kwargs):
raise RuntimeError('db disk full')
track = _StubSpotifyTrack(id='t', name='t', artists=['a'], album='a', duration_ms=0)
spotify = _StubSpotifyClient(_StubSpotifyPlaylist(tracks=[track]))
db = _ExplodingDB(playlists=[
{'id': 1, 'name': 'X', 'source': 'spotify', 'source_playlist_id': 'spot'},
])
deps = _build_deps(get_database=lambda: db, spotify_client=spotify)
result = auto_refresh_mirrored({'all': True}, deps)
# Error captured, status still 'completed' (handler returns counts).
assert result['status'] == 'completed'
assert result['errors'] == '1'
assert result['refreshed'] == '0'
# ─── sync_playlist ───────────────────────────────────────────────────
class TestSyncPlaylist:
def test_no_playlist_id_returns_error(self):
deps = _build_deps()
result = auto_sync_playlist({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_playlist_not_found_returns_error(self):
deps = _build_deps(get_database=lambda: _StubDB(playlists=[]))
result = auto_sync_playlist({'playlist_id': '99'}, deps)
assert result == {'status': 'error', 'reason': 'Playlist not found'}
def test_no_tracks_returns_error(self):
db = _StubDB(playlists=[{'id': 1, 'name': 'P'}], playlist_tracks={1: []})
deps = _build_deps(get_database=lambda: db)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result == {'status': 'error', 'reason': 'No tracks in playlist'}
def test_no_discovered_tracks_skips(self):
# All tracks lack discovery + spotify_hint + valid IDs.
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [{}, {}]}, # empty tracks → nothing usable
)
deps = _build_deps(get_database=lambda: db)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'skipped'
assert 'No discovered tracks' in result['reason']
assert result['skipped_tracks'] == '2'
def test_discovered_track_starts_sync_thread(self):
discovered_track = {
'extra_data': json.dumps({
'discovered': True,
'matched_data': {
'id': 'spot-1', 'name': 'Track', 'artists': [{'name': 'X'}],
'album': {'name': 'Album'}, 'duration_ms': 200000,
},
}),
'artist_name': 'X',
}
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [discovered_track]},
)
sync_calls: List[tuple] = []
deps = _build_deps(
get_database=lambda: db,
run_sync_task=lambda *a, **k: sync_calls.append((a, k)),
)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'started'
assert result['_manages_own_progress'] is True
assert result['discovered_tracks'] == '1'
# Wait for thread to fire run_sync_task
for _ in range(50):
if sync_calls:
break
import time
time.sleep(0.01)
assert len(sync_calls) == 1
def test_unchanged_since_last_sync_returns_skipped(self):
discovered_track = {
'extra_data': json.dumps({
'discovered': True,
'matched_data': {
'id': 'spot-1', 'name': 'T', 'artists': [{'name': 'X'}],
'album': {'name': 'A'}, 'duration_ms': 0,
},
}),
'artist_name': 'X',
}
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [discovered_track]},
)
# Pre-populate the sync-status file with the EXPECTED hash so the
# preflight short-circuit fires.
import hashlib
expected_hash = hashlib.md5('spot-1'.encode()).hexdigest()
sync_statuses = {
'auto_mirror_1': {'tracks_hash': expected_hash, 'matched_tracks': 1}
}
deps = _build_deps(
get_database=lambda: db,
load_sync_status_file=lambda: sync_statuses,
)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'skipped'
assert 'unchanged' in result['reason']
# ─── playlist_pipeline ───────────────────────────────────────────────
class TestPlaylistPipeline:
def test_no_playlist_specified_returns_error(self):
deps = _build_deps()
result = auto_playlist_pipeline({}, deps)
assert result == {'status': 'error', 'error': 'No playlist specified'}
# Pipeline-running flag MUST be cleared on error so the guard
# doesn't block subsequent triggers.
assert deps.state.pipeline_running is False
def test_no_refreshable_playlists_clears_running_flag(self):
db = _StubDB(playlists=[
{'id': 1, 'name': 'F', 'source': 'file'},
{'id': 2, 'name': 'B', 'source': 'beatport'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_playlist_pipeline({'all': True}, deps)
assert result == {'status': 'error', 'error': 'No refreshable playlists found'}
assert deps.state.pipeline_running is False
def test_pipeline_clears_running_on_unhandled_exception(self):
# Force the database accessor to blow up after the early checks.
class _ExplodingDB(_StubDB):
def get_mirrored_playlists(self):
raise RuntimeError('db down')
db = _ExplodingDB(playlists=[])
deps = _build_deps(get_database=lambda: db)
result = auto_playlist_pipeline({'all': True}, deps)
assert result['status'] == 'error'
assert result['_manages_own_progress'] is True
assert deps.state.pipeline_running is False

15
tests/automation/test_handlers_simple.py
View file

@ -36,12 +36,19 @@ def _build_deps(**overrides: Any) -> AutomationDeps:
"""Return a default `AutomationDeps` with no-op callables. Tests
pass ``overrides`` to install behaviour on the specific deps they
care about."""
class _StubLogger:
def debug(self, *_a, **_k): pass
def info(self, *_a, **_k): pass
def warning(self, *_a, **_k): pass
def error(self, *_a, **_k): pass
defaults = dict(
engine=object(),
state=AutomationState(),
config_manager=object(),
update_progress=lambda *a, **k: None,
logger=object(),
logger=_StubLogger(),
get_database=lambda: object(),
spotify_client=None,
tidal_client=None,
@ -51,6 +58,12 @@ def _build_deps(**overrides: Any) -> AutomationDeps:
is_wishlist_actually_processing=lambda: False,
is_watchlist_actually_scanning=lambda: False,
get_watchlist_scan_state=lambda: {},
run_playlist_discovery_worker=lambda *a, **k: None,
run_sync_task=lambda *a, **k: None,
load_sync_status_file=lambda: {},
get_deezer_client=lambda: None,
parse_youtube_playlist=lambda url: None,
get_sync_states=lambda: {},
)
defaults.update(overrides)
return AutomationDeps(**defaults) # type: ignore[arg-type]

675
web_server.py
View file

@ -947,678 +947,15 @@ def _register_automation_handlers():
is_wishlist_actually_processing=is_wishlist_actually_processing,
is_watchlist_actually_scanning=is_watchlist_actually_scanning,
get_watchlist_scan_state=lambda: watchlist_scan_state,
run_playlist_discovery_worker=_run_playlist_discovery_worker,
run_sync_task=_run_sync_task,
load_sync_status_file=_load_sync_status_file,
get_deezer_client=_get_deezer_client,
parse_youtube_playlist=parse_youtube_playlist,
get_sync_states=lambda: sync_states,
)
_register_extracted_handlers(_automation_deps)
def _auto_refresh_mirrored(config):
"""Refresh mirrored playlist(s) from source."""
db = get_database()
playlist_id = config.get('playlist_id')
refresh_all = config.get('all', False)
auto_id = config.get('_automation_id')
if refresh_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
# Filter out sources that can't be refreshed (no external API)
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
refreshed = 0
errors = []
for idx, pl in enumerate(playlists):
try:
source = pl.get('source', '')
source_id = pl.get('source_playlist_id', '')
_update_automation_progress(auto_id,
progress=(idx / max(1, len(playlists))) * 100,
phase=f'Refreshing: "{pl.get("name", "")}"',
current_item=pl.get('name', ''))
tracks = None
if source == 'spotify':
# Try authenticated API first, fall back to public embed scraper
if spotify_client and spotify_client.is_spotify_authenticated():
playlist_obj = spotify_client.get_playlist_by_id(source_id)
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
# Spotify data IS official — auto-mark as discovered
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth needed)
if tracks is None:
try:
from core.spotify_public_scraper import scrape_spotify_embed
embed_data = scrape_spotify_embed('playlist', source_id)
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
track_dict = {
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
}
# Store Spotify track ID hint but don't mark discovered —
# Discover step needs to run for proper album art
if t.get('id'):
track_dict['extra_data'] = json.dumps({
'discovered': False,
'spotify_hint': {
'id': t['id'],
'name': t.get('name', ''),
'artists': t.get('artists', []),
}
})
tracks.append(track_dict)
except Exception as e:
logger.warning(f"Spotify public scraper fallback failed for {source_id}: {e}")
elif source == 'spotify_public':
# source_playlist_id is an MD5 hash; extract actual Spotify ID from stored description (URL)
try:
from core.spotify_public_scraper import parse_spotify_url, scrape_spotify_embed
spotify_url = pl.get('description', '')
parsed = parse_spotify_url(spotify_url) if spotify_url else None
# If Spotify is authenticated, use the full API (auto-discovers with album art)
if parsed and parsed.get('type') == 'playlist' and spotify_client and spotify_client.is_spotify_authenticated():
playlist_obj = spotify_client.get_playlist_by_id(parsed['id'])
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth or album-type URL)
if tracks is None and parsed:
embed_data = scrape_spotify_embed(parsed['type'], parsed['id'])
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
# No extra_data — let preservation code keep existing discovery data
except Exception as e:
logger.warning(f"Spotify public playlist refresh failed for {source_id}: {e}")
elif source == 'deezer':
try:
deezer = _get_deezer_client()
playlist_data = deezer.get_playlist(source_id)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': t.get('album', ''),
'duration_ms': t.get('duration_ms', 0),
'source_track_id': str(t.get('id', '')),
})
except Exception as e:
logger.warning(f"Deezer playlist refresh failed for {source_id}: {e}")
elif source == 'tidal':
if not tidal_client or not tidal_client.is_authenticated():
logger.warning(f"Tidal not authenticated — skipping refresh for '{pl.get('name', '')}'")
_update_automation_progress(auto_id,
log_line=f'Skipped "{pl.get("name", "")}" — Tidal not authenticated', log_type='skip')
continue
full_playlist = tidal_client.get_playlist(source_id)
if full_playlist and full_playlist.tracks:
tracks = []
for t in full_playlist.tracks:
artist_name = t.artists[0] if t.artists else ''
tracks.append({
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
})
elif source == 'youtube':
# source_playlist_id is now a deterministic hash; use stored description (original URL) for refresh
yt_url = pl.get('description', '') or f"https://www.youtube.com/playlist?list={source_id}"
playlist_data = parse_youtube_playlist(yt_url)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': '',
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
if tracks is not None:
# Compare old vs new track IDs to detect changes
old_tracks = db.get_mirrored_playlist_tracks(pl['id']) if pl.get('id') else []
old_ids = {t.get('source_track_id') for t in old_tracks if t.get('source_track_id')}
new_ids = {t.get('source_track_id') for t in tracks if t.get('source_track_id')}
# Preserve existing discovery extra_data for tracks that still exist
old_extra_map = db.get_mirrored_tracks_extra_data_map(pl['id']) if pl.get('id') else {}
for t in tracks:
sid = t.get('source_track_id', '')
if sid and sid in old_extra_map and 'extra_data' not in t:
t['extra_data'] = old_extra_map[sid]
db.mirror_playlist(
source=source,
source_playlist_id=source_id,
name=pl['name'],
tracks=tracks,
profile_id=pl.get('profile_id', 1),
owner=pl.get('owner'),
image_url=pl.get('image_url'),
)
refreshed += 1
# Emit playlist_changed if tracks actually changed
if old_ids != new_ids:
added_count = len(new_ids - old_ids)
removed_count = len(old_ids - new_ids)
logger.info(f"[AUTOMATION] Playlist changed: '{pl.get('name', '')}' — {added_count} added, {removed_count} removed (old={len(old_ids)}, new={len(new_ids)})")
_update_automation_progress(auto_id,
log_line=f'"{pl.get("name", "")}" — {added_count} added, {removed_count} removed', log_type='success')
try:
if automation_engine:
automation_engine.emit('playlist_changed', {
'playlist_name': pl.get('name', ''),
'playlist_id': str(pl.get('id', '')),
'old_count': str(len(old_ids)),
'new_count': str(len(new_ids)),
'added': str(added_count),
'removed': str(removed_count),
})
except Exception as e:
logger.debug("playlist_synced automation emit failed: %s", e)
else:
logger.warning(f"[AUTOMATION] No changes: '{pl.get('name', '')}' (tracks={len(old_ids)})")
_update_automation_progress(auto_id,
log_line=f'No changes: "{pl.get("name", "")}"', log_type='skip')
except Exception as e:
errors.append(f"{pl.get('name', '?')}: {str(e)}")
_update_automation_progress(auto_id,
log_line=f'Error: {pl.get("name", "?")} — {str(e)}', log_type='error')
return {'status': 'completed', 'refreshed': str(refreshed), 'errors': str(len(errors))}
def _auto_sync_playlist(config):
"""Sync a mirrored playlist to media server.
Uses discovered metadata when available, skips undiscovered tracks.
When triggered on a schedule, skips if nothing changed since last sync."""
auto_id = config.get('_automation_id')
playlist_id = config.get('playlist_id')
if not playlist_id:
return {'status': 'error', 'reason': 'No playlist specified'}
db = get_database()
pl = db.get_mirrored_playlist(int(playlist_id))
if not pl:
return {'status': 'error', 'reason': 'Playlist not found'}
tracks = db.get_mirrored_playlist_tracks(int(playlist_id))
if not tracks:
return {'status': 'error', 'reason': 'No tracks in playlist'}
# Count currently discovered tracks for smart-skip check
current_discovered = 0
for t in tracks:
extra = {}
if t.get('extra_data'):
try:
extra = json.loads(t['extra_data']) if isinstance(t['extra_data'], str) else t['extra_data']
except (json.JSONDecodeError, TypeError):
pass
if extra.get('discovered') and extra.get('matched_data'):
current_discovered += 1
# Convert mirrored tracks to format expected by _run_sync_task
# Use discovered metadata when available, skip undiscovered tracks
tracks_json = []
skipped_count = 0
for t in tracks:
# Parse extra_data for discovery info
extra = {}
if t.get('extra_data'):
try:
extra = json.loads(t['extra_data']) if isinstance(t['extra_data'], str) else t['extra_data']
except (json.JSONDecodeError, TypeError):
pass
if extra.get('discovered') and extra.get('matched_data'):
# Use official discovered metadata
md = extra['matched_data']
album_raw = md.get('album', '')
album_obj = album_raw if isinstance(album_raw, dict) else {'name': album_raw or ''}
_track_entry = {
'name': md.get('name', ''),
'artists': md.get('artists', [{'name': t.get('artist_name', '')}]),
'album': album_obj,
'duration_ms': md.get('duration_ms', 0),
'id': md.get('id', ''),
}
if md.get('track_number'):
_track_entry['track_number'] = md['track_number']
if md.get('disc_number'):
_track_entry['disc_number'] = md['disc_number']
tracks_json.append(_track_entry)
else:
# NOT discovered — try to include using available metadata so the
# track can still be searched on Soulseek and added to wishlist.
# Without this, failed discovery blocks the entire download pipeline.
#
# Priority: spotify_hint (has real Spotify ID from embed scraper)
# > raw playlist fields (only if source_track_id is valid)
hint = extra.get('spotify_hint', {})
# Build album object with cover art from the mirrored playlist track
track_image = (t.get('image_url') or '').strip()
album_obj = {
'name': (t.get('album_name') or '').strip(),
'images': [{'url': track_image, 'height': 300, 'width': 300}] if track_image else [],
}
if hint.get('id') and hint.get('name'):
# spotify_hint has proper Spotify track ID + metadata from embed scraper
hint_artists = hint.get('artists', [])
if hint_artists and isinstance(hint_artists[0], str):
hint_artists = [{'name': a} for a in hint_artists]
elif hint_artists and isinstance(hint_artists[0], dict):
pass # Already in correct format
else:
hint_artists = [{'name': t.get('artist_name', '')}]
tracks_json.append({
'name': hint['name'],
'artists': hint_artists,
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': hint['id'],
})
elif t.get('source_track_id') and (t.get('track_name') or '').strip():
# Has a valid source ID and track name — usable for wishlist
tracks_json.append({
'name': t['track_name'].strip(),
'artists': [{'name': (t.get('artist_name') or '').strip() or 'Unknown Artist'}],
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': t['source_track_id'],
})
else:
skipped_count += 1 # No usable ID or name — truly can't process
if not tracks_json:
_update_automation_progress(auto_id,
log_line=f'No discovered tracks — {skipped_count} need discovery first', log_type='skip')
return {
'status': 'skipped',
'reason': f'No discovered tracks to sync ({skipped_count} tracks need discovery first)',
'skipped_tracks': str(skipped_count),
}
# Preflight: hash the track list and compare against last sync
# Skip if the exact same set of tracks was already synced and all matched
import hashlib
track_ids_str = ','.join(sorted(t.get('id', '') for t in tracks_json))
tracks_hash = hashlib.md5(track_ids_str.encode()).hexdigest()
sync_id_key = f"auto_mirror_{playlist_id}"
try:
sync_statuses = _load_sync_status_file()
last_status = sync_statuses.get(sync_id_key, {})
last_hash = last_status.get('tracks_hash', '')
last_matched = last_status.get('matched_tracks', -1)
if (last_hash == tracks_hash and
last_matched >= len(tracks_json)):
# Exact same tracks, all matched last time — nothing to do
_update_automation_progress(auto_id,
log_line=f'All {len(tracks_json)} tracks unchanged since last sync — skipping',
log_type='skip')
return {
'status': 'skipped',
'reason': f'All {len(tracks_json)} tracks unchanged since last sync',
}
except Exception as e:
logger.debug("mirror sync last-status read: %s", e)
_update_automation_progress(auto_id, progress=50,
phase=f'Syncing "{pl["name"]}"',
log_line=f'{len(tracks_json)} discovered, {skipped_count} skipped',
log_type='info')
sync_id = f"auto_mirror_{playlist_id}"
_update_automation_progress(auto_id, progress=90,
log_line=f'Starting sync: {len(tracks_json)} tracks', log_type='success')
threading.Thread(
target=_run_sync_task,
args=(sync_id, pl['name'], tracks_json, auto_id, 1, pl.get('image_url', '')),
daemon=True,
name=f'auto-sync-{playlist_id}'
).start()
return {
'status': 'started',
'playlist_name': pl['name'],
'discovered_tracks': str(len(tracks_json)),
'skipped_tracks': str(skipped_count),
'_manages_own_progress': True,
}
def _auto_discover_playlist(config):
"""Discover official Spotify/iTunes metadata for mirrored playlist tracks."""
db = get_database()
playlist_id = config.get('playlist_id')
discover_all = config.get('all', False)
if discover_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
if not playlists:
return {'status': 'error', 'reason': 'No playlists found'}
threading.Thread(
target=_run_playlist_discovery_worker,
args=(playlists, config.get('_automation_id')),
daemon=True,
name='auto-discover-playlist'
).start()
names = ', '.join(p['name'] for p in playlists[:3])
return {'status': 'started', 'playlist_count': str(len(playlists)), 'playlists': names,
'_manages_own_progress': True}
# --- Playlist Pipeline: single automation for full lifecycle ---
_pipeline_running = False
def _pipeline_guard():
return _pipeline_running
def _auto_playlist_pipeline(config):
"""Full playlist lifecycle: refresh → discover → sync → wishlist.
Runs all 4 phases sequentially in one automation, reporting progress throughout."""
nonlocal _pipeline_running
_pipeline_running = True
automation_id = config.get('_automation_id')
pipeline_start = time.time()
try:
db = get_database()
playlist_id = config.get('playlist_id')
process_all = config.get('all', False)
skip_wishlist = config.get('skip_wishlist', False)
# Resolve playlists
if process_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
_pipeline_running = False
return {'status': 'error', 'error': 'No playlist specified'}
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
if not playlists:
_pipeline_running = False
return {'status': 'error', 'error': 'No refreshable playlists found'}
pl_names = ', '.join(p.get('name', '?') for p in playlists[:3])
if len(playlists) > 3:
pl_names += f' (+{len(playlists) - 3} more)'
_update_automation_progress(automation_id, progress=2,
phase=f'Pipeline: {len(playlists)} playlist(s)',
log_line=f'Starting pipeline for: {pl_names}', log_type='info')
# ── PHASE 1: REFRESH ──────────────────────────────────────────
_update_automation_progress(automation_id, progress=3,
phase='Phase 1/4: Refreshing playlists...',
log_line='Phase 1: Refresh', log_type='info')
refresh_config = dict(config)
refresh_config['_automation_id'] = None # Don't let sub-handler hijack pipeline progress
refresh_result = _auto_refresh_mirrored(refresh_config)
refreshed = int(refresh_result.get('refreshed', 0))
refresh_errors = int(refresh_result.get('errors', 0))
_update_automation_progress(automation_id, progress=25,
phase='Phase 1/4: Refresh complete',
log_line=f'Phase 1 done: {refreshed} refreshed, {refresh_errors} errors',
log_type='success' if refresh_errors == 0 else 'warning')
# ── PHASE 2: DISCOVER ─────────────────────────────────────────
_update_automation_progress(automation_id, progress=26,
phase='Phase 2/4: Discovering metadata...',
log_line='Phase 2: Discover', log_type='info')
# Reload playlists (refresh may have updated them)
if process_all:
disc_playlists = db.get_mirrored_playlists()
else:
disc_playlists = [db.get_mirrored_playlist(int(playlist_id))]
disc_playlists = [p for p in disc_playlists if p]
# Run discovery in a thread and wait for it
disc_done = threading.Event()
disc_result = {'discovered': 0, 'failed': 0, 'skipped': 0, 'total': 0}
def _disc_wrapper(pls):
try:
# The worker updates automation_progress internally,
# but we pass None so it doesn't conflict with our pipeline progress
_run_playlist_discovery_worker(pls, automation_id=None)
except Exception as e:
logger.error(f"[Pipeline] Discovery error: {e}")
finally:
disc_done.set()
threading.Thread(target=_disc_wrapper, args=(disc_playlists,), daemon=True,
name='pipeline-discover').start()
# Poll for completion with progress updates
poll_start = time.time()
while not disc_done.wait(timeout=3):
elapsed = int(time.time() - poll_start)
_update_automation_progress(automation_id, progress=min(26 + elapsed // 4, 54),
phase=f'Phase 2/4: Discovering... ({elapsed}s)')
if elapsed > 3600: # 1hr safety timeout
_update_automation_progress(automation_id,
log_line='Discovery timed out after 1 hour', log_type='warning')
break
_update_automation_progress(automation_id, progress=55,
phase='Phase 2/4: Discovery complete',
log_line='Phase 2 done: discovery complete', log_type='success')
# ── PHASE 3: SYNC ─────────────────────────────────────────────
_update_automation_progress(automation_id, progress=56,
phase='Phase 3/4: Syncing to server...',
log_line='Phase 3: Sync', log_type='info')
total_synced = 0
total_skipped = 0
sync_errors = 0
for pl_idx, pl in enumerate(playlists):
pl_id = pl.get('id')
if not pl_id:
continue
# Build sync config for this playlist (reuse existing sync handler)
sync_config = {
'playlist_id': str(pl_id),
'_automation_id': None, # Don't let sync handler hijack our progress
}
sync_result = _auto_sync_playlist(sync_config)
sync_status = sync_result.get('status', '')
if sync_status == 'started':
# Sync launched a background thread — wait for it
sync_id = f"auto_mirror_{pl_id}"
sync_poll_start = time.time()
while time.time() - sync_poll_start < 600: # 10 min per playlist max
if sync_id in sync_states and sync_states[sync_id].get('status') in ('finished', 'complete', 'error', 'failed'):
break
time.sleep(2)
elapsed = int(time.time() - sync_poll_start)
sub_progress = 56 + ((pl_idx + 1) / max(1, len(playlists))) * 29
_update_automation_progress(automation_id, progress=min(int(sub_progress), 84),
phase=f'Phase 3/4: Syncing "{pl.get("name", "")}" ({elapsed}s)')
# Check result
ss = sync_states.get(sync_id, {})
ss_result = ss.get('result', ss.get('progress', {}))
matched = ss_result.get('matched_tracks', 0) if isinstance(ss_result, dict) else 0
total_synced += int(matched) if matched else 0
_update_automation_progress(automation_id,
log_line=f'Synced "{pl.get("name", "")}": {matched} tracks matched',
log_type='success')
elif sync_status == 'skipped':
total_skipped += 1
reason = sync_result.get('reason', 'unchanged')
_update_automation_progress(automation_id,
log_line=f'Skipped "{pl.get("name", "")}": {reason}',
log_type='skip')
elif sync_status == 'error':
sync_errors += 1
_update_automation_progress(automation_id,
log_line=f'Sync error "{pl.get("name", "")}": {sync_result.get("reason", "unknown")}',
log_type='error')
_update_automation_progress(automation_id, progress=85,
phase='Phase 3/4: Sync complete',
log_line=f'Phase 3 done: {total_synced} matched, {total_skipped} skipped, {sync_errors} errors',
log_type='success' if sync_errors == 0 else 'warning')
# ── PHASE 4: WISHLIST ─────────────────────────────────────────
wishlist_queued = 0
if not skip_wishlist:
_update_automation_progress(automation_id, progress=86,
phase='Phase 4/4: Processing wishlist...',
log_line='Phase 4: Wishlist', log_type='info')
try:
if not is_wishlist_actually_processing():
_process_wishlist_automatically(automation_id=None)
_update_automation_progress(automation_id,
log_line='Wishlist processing triggered', log_type='success')
wishlist_queued = 1
else:
_update_automation_progress(automation_id,
log_line='Wishlist already running — skipped', log_type='skip')
except Exception as e:
_update_automation_progress(automation_id,
log_line=f'Wishlist error: {e}', log_type='warning')
else:
_update_automation_progress(automation_id, progress=86,
log_line='Phase 4: Wishlist skipped (disabled)', log_type='skip')
# ── COMPLETE ──────────────────────────────────────────────────
duration = int(time.time() - pipeline_start)
_update_automation_progress(automation_id, status='finished', progress=100,
phase='Pipeline complete',
log_line=f'Pipeline finished in {duration // 60}m {duration % 60}s',
log_type='success')
_pipeline_running = False
return {
'status': 'completed',
'_manages_own_progress': True,
'playlists_refreshed': str(refreshed),
'tracks_discovered': 'completed',
'tracks_synced': str(total_synced),
'sync_skipped': str(total_skipped),
'wishlist_queued': str(wishlist_queued),
'duration_seconds': str(duration),
}
except Exception as e:
_pipeline_running = False
_update_automation_progress(automation_id, status='error', progress=100,
phase='Pipeline error',
log_line=f'Pipeline failed: {e}', log_type='error')
return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
automation_engine.register_action_handler('refresh_mirrored', _auto_refresh_mirrored)
automation_engine.register_action_handler('sync_playlist', _auto_sync_playlist)
automation_engine.register_action_handler('discover_playlist', _auto_discover_playlist)
automation_engine.register_action_handler('playlist_pipeline', _auto_playlist_pipeline, _pipeline_guard)
# --- Phase 3 action handlers ---