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

Add direct mirrored playlist pipeline runs

Browse source 
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.
This commit is contained in:
Broque Thomas 2026-05-24 19:54:04 -07:00
parent bc6bacb7da
commit f83c671570
7 changed files with 396 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
core/automation/deps.py
View file

@ -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

10
core/playlists/pipeline.py
View file

@ -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()

13
tests/automation/test_handlers_playlist.py
View file

@ -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)

8
tests/automation/test_handlers_simple.py
View file

@ -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.

207
web_server.py
View file

@ -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/<int:playlist_id>/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/<int:playlist_id>/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/<int:playlist_id>', methods=['DELETE'])
def delete_mirrored_playlist_endpoint(playlist_id):
"""Delete a mirrored playlist."""

121
webui/static/stats-automations.js
View file

@ -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 = `<span style="color:#38bdf8;">${_esc(label)} ${pct}%</span>`;
} else if (phase === 'pipeline_complete') {
phaseHtml = `<span style="color:#22c55e;">Pipeline complete</span>`;
} else if (phase === 'pipeline_error') {
phaseHtml = `<span style="color:#ef4444;">Pipeline error</span>`;
} else if (phase === 'discovering') {
const pct = state.discoveryProgress || state.discovery_progress || 0;
phaseHtml = `<span style="color:#a78bfa;">Discovering ${pct}%</span>`;
} else if (phase === 'discovered') {
@ -491,6 +513,7 @@ function renderMirroredCard(p, container) {
<span>Mirrored ${ago}</span>
${ratioHtml}
${phaseHtml}
<button class="mirrored-card-pipeline" onclick="event.stopPropagation(); runMirroredPlaylistPipeline(${p.id}, '${_escAttr(p.name)}')" title="Run refresh, discover, sync, and wishlist">Run Pipeline</button>
</div>
</div>
${disc > 0 ? `<button class="mirrored-card-clear" onclick="event.stopPropagation(); clearMirroredDiscovery(${p.id}, '${_escAttr(p.name)}')" title="Clear discovery data">↺</button>` : ''}
@ -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 = `<span style="color:#38bdf8;">${_esc(pipelinePhase)} ${pipelineProgress}%</span>`;
break;
case 'pipeline_complete':
phaseHtml = `<span style="color:#22c55e;">Pipeline complete</span>`;
break;
case 'pipeline_error':
phaseHtml = `<span style="color:#ef4444;">Pipeline error</span>`;
break;
case 'discovering':
phaseHtml = `<span style="color:#a78bfa;">Discovering...</span>`;
break;
@ -874,6 +990,7 @@ async function openMirroredPlaylistModal(playlistId) {
</div>
<div class="mirrored-modal-footer-right" style="display:flex;gap:10px;">
<button class="mirrored-btn-close" onclick="editMirroredSourceRef(${playlistId}, '${_escAttr(data.name)}', '${_escAttr(source)}', '${_escAttr(sourceRef)}')">Edit Source</button>
<button class="mirrored-btn-pipeline" onclick="runMirroredPlaylistPipeline(${playlistId}, '${_escAttr(data.name)}')">Run Pipeline</button>
<button class="mirrored-btn-close" onclick="closeMirroredModal()">Close</button>
<button class="mirrored-btn-discover" onclick="discoverMirroredPlaylist(${playlistId})">Discover</button>
</div>

32
webui/static/style.css
View file

@ -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;