diff --git a/core/automation/deps.py b/core/automation/deps.py index 58b8c9aa..bf5a6a4a 100644 --- a/core/automation/deps.py +++ b/core/automation/deps.py @@ -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 diff --git a/core/automation/handlers/__init__.py b/core/automation/handlers/__init__.py index 8c8120f2..5f086115 100644 --- a/core/automation/handlers/__init__.py +++ b/core/automation/handlers/__init__.py @@ -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', ] diff --git a/core/automation/handlers/discover_playlist.py b/core/automation/handlers/discover_playlist.py new file mode 100644 index 00000000..90edd2a2 --- /dev/null +++ b/core/automation/handlers/discover_playlist.py @@ -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, + } diff --git a/core/automation/handlers/playlist_pipeline.py b/core/automation/handlers/playlist_pipeline.py new file mode 100644 index 00000000..2ef18669 --- /dev/null +++ b/core/automation/handlers/playlist_pipeline.py @@ -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} diff --git a/core/automation/handlers/refresh_mirrored.py b/core/automation/handlers/refresh_mirrored.py new file mode 100644 index 00000000..76ea32ce --- /dev/null +++ b/core/automation/handlers/refresh_mirrored.py @@ -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': '', + 'errors': ''}`` 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))} diff --git a/core/automation/handlers/registration.py b/core/automation/handlers/registration.py index 6fcf5f9c..80611a57 100644 --- a/core/automation/handlers/registration.py +++ b/core/automation/handlers/registration.py @@ -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, + ) diff --git a/core/automation/handlers/sync_playlist.py b/core/automation/handlers/sync_playlist.py new file mode 100644 index 00000000..5734b48f --- /dev/null +++ b/core/automation/handlers/sync_playlist.py @@ -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, + } diff --git a/tests/automation/test_handlers_playlist.py b/tests/automation/test_handlers_playlist.py new file mode 100644 index 00000000..83c2596b --- /dev/null +++ b/tests/automation/test_handlers_playlist.py @@ -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 diff --git a/tests/automation/test_handlers_simple.py b/tests/automation/test_handlers_simple.py index 3c5d0bc9..a8aa8834 100644 --- a/tests/automation/test_handlers_simple.py +++ b/tests/automation/test_handlers_simple.py @@ -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] diff --git a/web_server.py b/web_server.py index aedbfd9d..c0511e8b 100644 --- a/web_server.py +++ b/web_server.py @@ -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 ---