diff --git a/core/automation/handlers/playlist_pipeline.py b/core/automation/handlers/playlist_pipeline.py index 8fc98df2..eacea479 100644 --- a/core/automation/handlers/playlist_pipeline.py +++ b/core/automation/handlers/playlist_pipeline.py @@ -1,227 +1,27 @@ -"""Automation handler: ``playlist_pipeline`` action. +"""Automation adapter for the mirrored playlist pipeline. -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. +The actual all-in-one playlist lifecycle lives in +``core.playlists.pipeline`` so it can be reused by non-automation UI actions. +This module only wires automation-specific dependencies and handlers. """ from __future__ import annotations -import threading -import time from typing import Any, Dict from core.automation.deps import AutomationDeps from core.automation.handlers._pipeline_shared import run_sync_and_wishlist 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. -# Discovery poll cap inside Phase 2. -_DISCOVERY_TIMEOUT_SECONDS = 3600 +from core.playlists.pipeline import run_mirrored_playlist_pipeline 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 + 4: SYNC + WISHLIST (delegated to shared helper) ── - # Each mirrored playlist payload only needs `id` + `name` for - # the helper; `auto_sync_playlist` reads the rest from the - # mirrored DB by id. - sync_summary = run_sync_and_wishlist( - deps, - automation_id, - [pl for pl in playlists if pl.get('id')], - sync_one_fn=lambda pl: auto_sync_playlist( - {'playlist_id': str(pl['id']), '_automation_id': None}, - deps, - ), - sync_id_for_fn=lambda pl: f"auto_mirror_{pl['id']}", - skip_wishlist=skip_wishlist, - progress_start=56, - progress_end=85, - sync_phase_label='Phase 3/4: Syncing to server...', - sync_phase_start_log='Phase 3: Sync', - wishlist_phase_label='Phase 4/4: Processing wishlist...', - wishlist_phase_start_log='Phase 4: Wishlist', - ) - total_synced = sync_summary['synced'] - total_skipped = sync_summary['skipped'] - sync_errors = sync_summary['errors'] - wishlist_queued = sync_summary['wishlist_queued'] - - # ── 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} + """Run REFRESH -> DISCOVER -> SYNC -> WISHLIST for mirrored playlists.""" + return run_mirrored_playlist_pipeline( + config, + deps, + refresh_fn=auto_refresh_mirrored, + sync_one_fn=auto_sync_playlist, + sync_and_wishlist_fn=run_sync_and_wishlist, + ) diff --git a/core/playlists/pipeline.py b/core/playlists/pipeline.py new file mode 100644 index 00000000..9f279118 --- /dev/null +++ b/core/playlists/pipeline.py @@ -0,0 +1,249 @@ +"""Mirrored playlist lifecycle pipeline. + +This module is the playlist-domain home for the all-in-one mirrored +playlist pipeline: + + refresh source -> discover metadata -> sync to server -> process wishlist + +Automation remains one caller, but the orchestration itself lives here so a +future playlist-card "Run Pipeline" button can call the same command. +""" + +from __future__ import annotations + +import threading +import time +from typing import Any, Callable, Dict, List + + +DISCOVERY_TIMEOUT_SECONDS = 3600 + + +RefreshFn = Callable[[Dict[str, Any], Any], Dict[str, Any]] +SyncOneFn = Callable[[Dict[str, Any], Any], Dict[str, Any]] +SyncAndWishlistFn = Callable[..., Dict[str, int]] + + +def run_mirrored_playlist_pipeline( + config: Dict[str, Any], + deps: Any, + *, + refresh_fn: RefreshFn, + sync_one_fn: SyncOneFn, + sync_and_wishlist_fn: SyncAndWishlistFn, +) -> Dict[str, Any]: + """Run REFRESH -> DISCOVER -> SYNC -> WISHLIST in sequence. + + ``deps`` intentionally uses duck typing. Today it is ``AutomationDeps``; + a future web/UI runner can provide the same small surface without becoming + an automation. + """ + 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) + + playlists = _resolve_pipeline_playlists(db, playlist_id, process_all) + if playlists is None: + deps.state.set_pipeline_running(False) + return {'status': 'error', 'error': 'No playlist specified'} + + playlists = _filter_refreshable_playlists(playlists) + if not playlists: + deps.state.set_pipeline_running(False) + return {'status': 'error', 'error': 'No refreshable playlists found'} + + deps.update_progress( + automation_id, + progress=2, + phase=f'Pipeline: {len(playlists)} playlist(s)', + log_line=f'Starting pipeline for: {_summarize_playlist_names(playlists)}', + log_type='info', + ) + + refreshed, refresh_errors = _run_refresh_phase( + config, + deps, + automation_id, + refresh_fn=refresh_fn, + ) + + _run_discovery_phase( + deps, + automation_id, + db=db, + playlist_id=playlist_id, + process_all=process_all, + ) + + sync_summary = sync_and_wishlist_fn( + deps, + automation_id, + [pl for pl in playlists if pl.get('id')], + sync_one_fn=lambda pl: sync_one_fn( + {'playlist_id': str(pl['id']), '_automation_id': None}, + deps, + ), + sync_id_for_fn=lambda pl: f"auto_mirror_{pl['id']}", + skip_wishlist=skip_wishlist, + progress_start=56, + progress_end=85, + sync_phase_label='Phase 3/4: Syncing to server...', + sync_phase_start_log='Phase 3: Sync', + wishlist_phase_label='Phase 4/4: Processing wishlist...', + wishlist_phase_start_log='Phase 4: Wishlist', + ) + + 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(sync_summary['synced']), + 'sync_skipped': str(sync_summary['skipped']), + 'wishlist_queued': str(sync_summary['wishlist_queued']), + 'duration_seconds': str(duration), + } + + except Exception as e: # noqa: BLE001 - pipeline callers should receive status dicts + 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} + + +def _resolve_pipeline_playlists(db: Any, playlist_id: Any, process_all: bool) -> List[Dict[str, Any]] | None: + if process_all: + return db.get_mirrored_playlists() + if playlist_id: + playlist = db.get_mirrored_playlist(int(playlist_id)) + return [playlist] if playlist else [] + return None + + +def _filter_refreshable_playlists(playlists: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + return [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')] + + +def _summarize_playlist_names(playlists: List[Dict[str, Any]]) -> str: + pl_names = ', '.join(p.get('name', '?') for p in playlists[:3]) + if len(playlists) > 3: + pl_names += f' (+{len(playlists) - 3} more)' + return pl_names + + +def _run_refresh_phase( + config: Dict[str, Any], + deps: Any, + automation_id: Any, + *, + refresh_fn: RefreshFn, +) -> tuple[int, int]: + 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 + refresh_result = refresh_fn(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', + ) + return refreshed, refresh_errors + + +def _run_discovery_phase( + deps: Any, + automation_id: Any, + *, + db: Any, + playlist_id: Any, + process_all: bool, +) -> None: + deps.update_progress( + automation_id, + progress=26, + phase='Phase 2/4: Discovering metadata...', + log_line='Phase 2: Discover', + log_type='info', + ) + + 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] + + disc_done = threading.Event() + + def _disc_wrapper(pls): + try: + deps.run_playlist_discovery_worker(pls, automation_id=None) + except Exception as e: # noqa: BLE001 - logged into pipeline progress + 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_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', + )