From f83c671570bd1db41049b3b524f922eee18a0b75 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Sun, 24 May 2026 19:54:04 -0700 Subject: [PATCH] Add direct mirrored playlist pipeline runs Expose playlist-native run and status endpoints that reuse the shared mirrored playlist pipeline engine while routing progress into playlist UI state. Add a Run Pipeline action to mirrored playlist cards and modals with live status polling, and make the shared pipeline lock atomic for manual and scheduled callers. --- core/automation/deps.py | 8 + core/playlists/pipeline.py | 10 +- tests/automation/test_handlers_playlist.py | 13 ++ tests/automation/test_handlers_simple.py | 8 + web_server.py | 207 +++++++++++++++++++++ webui/static/stats-automations.js | 121 +++++++++++- webui/static/style.css | 32 ++++ 7 files changed, 396 insertions(+), 3 deletions(-) diff --git a/core/automation/deps.py b/core/automation/deps.py index a521a496..5811766e 100644 --- a/core/automation/deps.py +++ b/core/automation/deps.py @@ -56,6 +56,14 @@ class AutomationState: with self.lock: return self.pipeline_running + def try_start_pipeline(self) -> bool: + """Atomically mark the shared playlist pipeline as running.""" + with self.lock: + if self.pipeline_running: + return False + self.pipeline_running = True + return True + def set_scan_library_id(self, automation_id: Optional[str]) -> None: with self.lock: self.scan_library_automation_id = automation_id diff --git a/core/playlists/pipeline.py b/core/playlists/pipeline.py index 9f279118..2bf0fc9c 100644 --- a/core/playlists/pipeline.py +++ b/core/playlists/pipeline.py @@ -38,7 +38,15 @@ def run_mirrored_playlist_pipeline( a future web/UI runner can provide the same small surface without becoming an automation. """ - deps.state.set_pipeline_running(True) + if hasattr(deps.state, 'try_start_pipeline'): + if not deps.state.try_start_pipeline(): + return { + 'status': 'skipped', + 'reason': 'playlist_pipeline is already running', + '_manages_own_progress': True, + } + else: + deps.state.set_pipeline_running(True) automation_id = config.get('_automation_id') pipeline_start = time.time() diff --git a/tests/automation/test_handlers_playlist.py b/tests/automation/test_handlers_playlist.py index 0bf7628c..60442bea 100644 --- a/tests/automation/test_handlers_playlist.py +++ b/tests/automation/test_handlers_playlist.py @@ -387,6 +387,19 @@ class TestSyncPlaylist: class TestPlaylistPipeline: + def test_pipeline_skips_when_shared_lock_is_already_running(self): + deps = _build_deps() + deps.state.set_pipeline_running(True) + + result = auto_playlist_pipeline({'all': True}, deps) + + assert result == { + 'status': 'skipped', + 'reason': 'playlist_pipeline is already running', + '_manages_own_progress': True, + } + assert deps.state.pipeline_running is True + def test_no_playlist_specified_returns_error(self): deps = _build_deps() result = auto_playlist_pipeline({}, deps) diff --git a/tests/automation/test_handlers_simple.py b/tests/automation/test_handlers_simple.py index ac8b2ef4..5e535884 100644 --- a/tests/automation/test_handlers_simple.py +++ b/tests/automation/test_handlers_simple.py @@ -311,6 +311,14 @@ class TestAutomationState: s.set_pipeline_running(False) assert s.is_pipeline_running() is False + def test_try_start_pipeline_is_atomic(self): + s = AutomationState() + assert s.try_start_pipeline() is True + assert s.is_pipeline_running() is True + assert s.try_start_pipeline() is False + s.set_pipeline_running(False) + assert s.try_start_pipeline() is True + def test_concurrent_set_safe_via_lock(self): # Smoke test: two threads flipping the same field don't crash. # Lock ensures the final value is consistent. diff --git a/web_server.py b/web_server.py index 2a90c8e1..51303934 100644 --- a/web_server.py +++ b/web_server.py @@ -919,6 +919,12 @@ except Exception as e: # --- Automation Progress Tracking --- _scan_library_automation_id = None +_automation_deps = None + +# Playlist-native manual pipeline runs share the automation dependency +# bundle, but keep their own small progress state for the playlist UI. +playlist_pipeline_progress_states = {} +playlist_pipeline_progress_lock = threading.Lock() def _register_automation_handlers(): @@ -935,6 +941,8 @@ def _register_automation_handlers(): closures still live below until subsequent commits in the same branch finish the lift. """ + global _automation_deps + if not automation_engine: return @@ -32198,6 +32206,7 @@ def get_mirrored_playlists_endpoint(): pl['source_ref_kind'] = source_ref.source_ref_kind pl['source_ref_status'] = source_ref.source_ref_status pl['source_ref_error'] = source_ref.source_ref_error + pl['pipeline_state'] = _snapshot_playlist_pipeline_state(pl['id']) return jsonify(playlists) except Exception as e: logger.error(f"Error getting mirrored playlists: {e}") @@ -32217,6 +32226,7 @@ def get_mirrored_playlist_endpoint(playlist_id): playlist['source_ref_kind'] = source_ref.source_ref_kind playlist['source_ref_status'] = source_ref.source_ref_status playlist['source_ref_error'] = source_ref.source_ref_error + playlist['pipeline_state'] = _snapshot_playlist_pipeline_state(playlist_id) playlist['tracks'] = database.get_mirrored_playlist_tracks(playlist_id) return jsonify(playlist) except Exception as e: @@ -32272,6 +32282,203 @@ def update_mirrored_playlist_source_ref_endpoint(playlist_id): logger.error(f"Error updating mirrored playlist source reference: {e}") return jsonify({"error": str(e)}), 500 + +def _playlist_pipeline_state_key(playlist_id): + return f"mirrored_{int(playlist_id)}" + + +def _snapshot_playlist_pipeline_state(playlist_id): + key = _playlist_pipeline_state_key(playlist_id) + with playlist_pipeline_progress_lock: + state = playlist_pipeline_progress_states.get(key) + return dict(state) if state else None + + +def _replace_playlist_pipeline_state(playlist_id, state): + key = _playlist_pipeline_state_key(playlist_id) + with playlist_pipeline_progress_lock: + playlist_pipeline_progress_states[key] = dict(state) + return dict(playlist_pipeline_progress_states[key]) + + +def _update_playlist_pipeline_progress(playlist_id, **kwargs): + key = _playlist_pipeline_state_key(playlist_id) + with playlist_pipeline_progress_lock: + state = playlist_pipeline_progress_states.setdefault(key, { + 'run_id': key, + 'playlist_id': int(playlist_id), + 'status': 'running', + 'progress': 0, + 'phase': 'Starting pipeline...', + 'log': [], + 'started_at': time.time(), + 'finished_at': None, + 'result': None, + 'error': None, + }) + for field in ('status', 'progress', 'phase', 'result', 'error'): + if field in kwargs: + state[field] = kwargs[field] + if 'log_line' in kwargs and kwargs.get('log_line'): + state.setdefault('log', []).append({ + 'message': kwargs.get('log_line'), + 'type': kwargs.get('log_type') or 'info', + 'timestamp': time.time(), + }) + state['log'] = state['log'][-80:] + if state.get('status') in ('finished', 'error', 'skipped') and not state.get('finished_at'): + state['finished_at'] = time.time() + return dict(state) + + +class _PlaylistPipelineDepsProxy: + """Forward automation deps while routing progress into playlist UI state.""" + + def __init__(self, base_deps, playlist_id): + self._base_deps = base_deps + self._playlist_id = playlist_id + + def __getattr__(self, name): + return getattr(self._base_deps, name) + + def update_progress(self, _automation_id, **kwargs): + _update_playlist_pipeline_progress(self._playlist_id, **kwargs) + + +def _run_mirrored_playlist_pipeline_for_ui(playlist_id, skip_wishlist=False): + try: + if _automation_deps is None: + raise RuntimeError("Automation dependencies are not available") + + from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored + from core.automation.handlers.sync_playlist import auto_sync_playlist + from core.automation.handlers._pipeline_shared import run_sync_and_wishlist + from core.playlists.pipeline import run_mirrored_playlist_pipeline + + deps = _PlaylistPipelineDepsProxy(_automation_deps, playlist_id) + result = run_mirrored_playlist_pipeline( + { + 'playlist_id': str(playlist_id), + 'all': False, + 'skip_wishlist': bool(skip_wishlist), + '_automation_id': _playlist_pipeline_state_key(playlist_id), + }, + deps, + refresh_fn=auto_refresh_mirrored, + sync_one_fn=auto_sync_playlist, + sync_and_wishlist_fn=run_sync_and_wishlist, + ) + + status = result.get('status') + if status == 'completed': + _update_playlist_pipeline_progress( + playlist_id, + status='finished', + progress=100, + phase='Pipeline complete', + result=result, + ) + elif status == 'skipped': + _update_playlist_pipeline_progress( + playlist_id, + status='skipped', + progress=100, + phase='Pipeline already running', + error=result.get('reason') or 'Pipeline already running', + result=result, + log_line=result.get('reason') or 'Pipeline already running', + log_type='warning', + ) + else: + _update_playlist_pipeline_progress( + playlist_id, + status='error', + progress=100, + phase='Pipeline error', + error=result.get('error') or 'Pipeline failed', + result=result, + ) + except Exception as e: + logger.error(f"Manual mirrored playlist pipeline failed for {playlist_id}: {e}") + _update_playlist_pipeline_progress( + playlist_id, + status='error', + progress=100, + phase='Pipeline error', + error=str(e), + log_line=f'Pipeline failed: {e}', + log_type='error', + ) + + +@app.route('/api/mirrored-playlists//pipeline/run', methods=['POST']) +def run_mirrored_playlist_pipeline_endpoint(playlist_id): + """Run the all-in-one mirrored playlist pipeline from the playlist UI.""" + try: + database = get_database() + playlist = database.get_mirrored_playlist(playlist_id) + if not playlist: + return jsonify({"error": "Playlist not found"}), 404 + if playlist.get('source') in ('file', 'beatport'): + return jsonify({"error": "This playlist source cannot be refreshed by the pipeline"}), 400 + if _automation_deps is None: + return jsonify({"error": "Playlist pipeline is not available"}), 503 + if _automation_deps.state.is_pipeline_running(): + return jsonify({"error": "A playlist pipeline is already running"}), 409 + + data = request.get_json(silent=True) or {} + state = _replace_playlist_pipeline_state(playlist_id, { + 'run_id': _playlist_pipeline_state_key(playlist_id), + 'playlist_id': int(playlist_id), + 'playlist_name': playlist.get('name') or '', + 'status': 'running', + 'progress': 0, + 'phase': 'Starting pipeline...', + 'log': [{ + 'message': f"Starting pipeline for {playlist.get('name') or playlist_id}", + 'type': 'info', + 'timestamp': time.time(), + }], + 'started_at': time.time(), + 'finished_at': None, + 'result': None, + 'error': None, + }) + + threading.Thread( + target=_run_mirrored_playlist_pipeline_for_ui, + args=(playlist_id, bool(data.get('skip_wishlist', False))), + daemon=True, + name=f"playlist-pipeline-{playlist_id}", + ).start() + + return jsonify({"success": True, "state": state}) + except Exception as e: + logger.error(f"Error starting mirrored playlist pipeline: {e}") + return jsonify({"error": str(e)}), 500 + + +@app.route('/api/mirrored-playlists//pipeline/status', methods=['GET']) +def get_mirrored_playlist_pipeline_status_endpoint(playlist_id): + """Return the latest manual pipeline progress for a mirrored playlist.""" + try: + state = _snapshot_playlist_pipeline_state(playlist_id) + if not state: + return jsonify({ + "run_id": _playlist_pipeline_state_key(playlist_id), + "playlist_id": int(playlist_id), + "status": "idle", + "progress": 0, + "phase": "Idle", + "log": [], + "result": None, + "error": None, + }) + return jsonify(state) + except Exception as e: + logger.error(f"Error getting mirrored playlist pipeline status: {e}") + return jsonify({"error": str(e)}), 500 + @app.route('/api/mirrored-playlists/', methods=['DELETE']) def delete_mirrored_playlist_endpoint(playlist_id): """Delete a mirrored playlist.""" diff --git a/webui/static/stats-automations.js b/webui/static/stats-automations.js index f7545036..9d25e8bb 100644 --- a/webui/static/stats-automations.js +++ b/webui/static/stats-automations.js @@ -380,6 +380,7 @@ function importFileSubmit() { // ── Mirrored Playlists ──────────────────────────────────────────────── let mirroredPlaylistsLoaded = false; +const mirroredPipelinePollers = {}; /** * Fire-and-forget helper: send parsed playlist data to be mirrored on the backend. @@ -444,13 +445,34 @@ async function loadMirroredPlaylists() { function renderMirroredCard(p, container) { const ago = timeAgo(p.updated_at || p.mirrored_at); const hash = `mirrored_${p.id}`; - const state = youtubePlaylistStates[hash]; + const pipelineState = p.pipeline_state || null; + const pipelinePhase = pipelineState && pipelineState.status === 'running' ? 'pipeline_running' + : pipelineState && pipelineState.status === 'finished' ? 'pipeline_complete' + : pipelineState && (pipelineState.status === 'error' || pipelineState.status === 'skipped') ? 'pipeline_error' + : null; + const state = youtubePlaylistStates[hash] || (pipelinePhase ? { + phase: pipelinePhase, + pipeline_status: pipelineState.status, + pipeline_progress: pipelineState.progress || 0, + pipeline_phase: pipelineState.phase || '', + pipeline_error: pipelineState.error || '', + pipeline_log: pipelineState.log || [], + pipeline_result: pipelineState.result || null, + } : null); const phase = state ? state.phase : null; const sourceRef = getMirroredSourceRef(p); // Build phase indicator let phaseHtml = ''; - if (phase === 'discovering') { + if (phase === 'pipeline_running') { + const pct = state.pipeline_progress || state.progress || 0; + const label = state.pipeline_phase || state.phase_label || 'Pipeline running'; + phaseHtml = `${_esc(label)} ${pct}%`; + } else if (phase === 'pipeline_complete') { + phaseHtml = `Pipeline complete`; + } else if (phase === 'pipeline_error') { + phaseHtml = `Pipeline error`; + } else if (phase === 'discovering') { const pct = state.discoveryProgress || state.discovery_progress || 0; phaseHtml = `Discovering ${pct}%`; } else if (phase === 'discovered') { @@ -491,6 +513,7 @@ function renderMirroredCard(p, container) { Mirrored ${ago} ${ratioHtml} ${phaseHtml} + ${disc > 0 ? `` : ''} @@ -532,6 +555,10 @@ function renderMirroredCard(p, container) { } }); container.appendChild(card); + + if (pipelineState && pipelineState.status === 'running' && !mirroredPipelinePollers[hash]) { + pollMirroredPipelineStatus(p.id, p.name); + } } function getMirroredSourceRef(p) { @@ -577,6 +604,84 @@ async function editMirroredSourceRef(playlistId, name, source, currentRef) { } } +function applyMirroredPipelineState(playlistId, state) { + const hash = `mirrored_${playlistId}`; + const existing = youtubePlaylistStates[hash] || {}; + const status = state.status || 'idle'; + let phase = existing.phase; + if (status === 'running') phase = 'pipeline_running'; + else if (status === 'finished') phase = 'pipeline_complete'; + else if (status === 'error' || status === 'skipped') phase = 'pipeline_error'; + + youtubePlaylistStates[hash] = { + ...existing, + phase, + pipeline_status: status, + pipeline_progress: state.progress || 0, + pipeline_phase: state.phase || '', + pipeline_error: state.error || '', + pipeline_log: state.log || [], + pipeline_result: state.result || null, + }; + + updateMirroredCardPhase(hash, phase); +} + +async function runMirroredPlaylistPipeline(playlistId, name) { + try { + const res = await fetch(`/api/mirrored-playlists/${playlistId}/pipeline/run`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({}) + }); + const data = await res.json(); + if (!res.ok || data.error) { + throw new Error(data.error || 'Failed to start pipeline'); + } + applyMirroredPipelineState(playlistId, data.state || { status: 'running', progress: 0, phase: 'Starting pipeline...' }); + showToast(`Pipeline started for ${name}`, 'success'); + pollMirroredPipelineStatus(playlistId, name); + } catch (err) { + showToast(`Error: ${err.message}`, 'error'); + } +} + +function pollMirroredPipelineStatus(playlistId, name) { + const key = `mirrored_${playlistId}`; + if (mirroredPipelinePollers[key]) clearInterval(mirroredPipelinePollers[key]); + + const tick = async () => { + try { + const res = await fetch(`/api/mirrored-playlists/${playlistId}/pipeline/status`); + const state = await res.json(); + if (!res.ok || state.error) throw new Error(state.error || 'Failed to read pipeline status'); + applyMirroredPipelineState(playlistId, state); + + if (state.status === 'finished') { + clearInterval(mirroredPipelinePollers[key]); + delete mirroredPipelinePollers[key]; + showToast(`Pipeline complete for ${name}`, 'success'); + loadMirroredPlaylists(); + } else if (state.status === 'error' || state.status === 'skipped') { + clearInterval(mirroredPipelinePollers[key]); + delete mirroredPipelinePollers[key]; + showToast(state.error || `Pipeline stopped for ${name}`, 'error'); + loadMirroredPlaylists(); + } else if (state.status === 'idle') { + clearInterval(mirroredPipelinePollers[key]); + delete mirroredPipelinePollers[key]; + } + } catch (err) { + clearInterval(mirroredPipelinePollers[key]); + delete mirroredPipelinePollers[key]; + showToast(`Pipeline status error: ${err.message}`, 'error'); + } + }; + + tick(); + mirroredPipelinePollers[key] = setInterval(tick, 2500); +} + function updateMirroredCardPhase(urlHash, phase) { // Update the state phase (updateYouTubeCardPhase skips this for mirrored playlists due to no cardElement) const state = youtubePlaylistStates[urlHash]; @@ -597,6 +702,17 @@ function updateMirroredCardPhase(urlHash, phase) { // Add new phase indicator let phaseHtml = ''; switch (phase) { + case 'pipeline_running': + const pipelineProgress = state?.pipeline_progress || 0; + const pipelinePhase = state?.pipeline_phase || 'Pipeline running'; + phaseHtml = `${_esc(pipelinePhase)} ${pipelineProgress}%`; + break; + case 'pipeline_complete': + phaseHtml = `Pipeline complete`; + break; + case 'pipeline_error': + phaseHtml = `Pipeline error`; + break; case 'discovering': phaseHtml = `Discovering...`; break; @@ -874,6 +990,7 @@ async function openMirroredPlaylistModal(playlistId) {
+
diff --git a/webui/static/style.css b/webui/static/style.css index 8ab1469c..51c619c6 100644 --- a/webui/static/style.css +++ b/webui/static/style.css @@ -12047,6 +12047,25 @@ body.helper-mode-active #dashboard-activity-feed:hover { letter-spacing: 0.3px; } +.mirrored-card-pipeline { + background: rgba(56, 189, 248, 0.12); + border: 1px solid rgba(56, 189, 248, 0.24); + border-radius: 6px; + color: #7dd3fc; + cursor: pointer; + font-size: 11px; + font-weight: 700; + height: 24px; + padding: 0 10px; + transition: all 0.2s ease; +} + +.mirrored-card-pipeline:hover { + background: rgba(56, 189, 248, 0.2); + border-color: rgba(56, 189, 248, 0.4); + color: #e0f2fe; +} + .mirrored-card-delete, .mirrored-card-clear, .mirrored-card-link { @@ -13226,6 +13245,19 @@ body.helper-mode-active #dashboard-activity-feed:hover { box-shadow: 0 8px 24px rgba(var(--accent-rgb), 0.4); } +.mirrored-btn-pipeline { + background: rgba(56, 189, 248, 0.12); + color: #7dd3fc; + border: 1px solid rgba(56, 189, 248, 0.28) !important; +} + +.mirrored-btn-pipeline:hover { + background: rgba(56, 189, 248, 0.2); + border-color: rgba(56, 189, 248, 0.45) !important; + color: #e0f2fe; + transform: translateY(-1px); +} + .mirrored-btn-delete { background: rgba(244, 67, 54, 0.12); color: #ef5350;