From ba6aea5435f8b5869e99777d598f07b1a9c21660 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Sat, 7 Mar 2026 00:57:03 -0800 Subject: [PATCH] Add rich progress tracking to all automation cards with stall detection and timeout handling --- core/automation_engine.py | 60 ++++++++-- web_server.py | 235 +++++++++++++++++++++++++++++++++++--- 2 files changed, 271 insertions(+), 24 deletions(-) diff --git a/core/automation_engine.py b/core/automation_engine.py index 753d43be..c4019daa 100644 --- a/core/automation_engine.py +++ b/core/automation_engine.py @@ -67,6 +67,7 @@ class AutomationEngine: # Progress tracking callbacks (registered by web_server.py) self._progress_init_fn = None self._progress_finish_fn = None + self._progress_update_fn = None # Event trigger cache: trigger_type → [automation_id, ...] self._event_automations = {} @@ -97,10 +98,11 @@ class AutomationEngine: } logger.debug(f"Registered action handler: {action_type}") - def register_progress_callbacks(self, init_fn, finish_fn): + def register_progress_callbacks(self, init_fn, finish_fn, update_fn=None): """Register callbacks for live progress tracking from web_server.py.""" self._progress_init_fn = init_fn self._progress_finish_fn = finish_fn + self._progress_update_fn = update_fn @staticmethod def _sanitize_signal_name(name): @@ -352,11 +354,25 @@ class AutomationEngine: action_config['_automation_name'] = auto.get('name', '') delay_minutes = action_config.get('delay', 0) + _delay_already_inited = False if delay_minutes and delay_minutes > 0: + # Initialize progress BEFORE delay so card glows during wait + if self._progress_init_fn: + try: self._progress_init_fn(automation_id, auto.get('name', ''), action_type) + except Exception: pass + _delay_already_inited = True + + delay_seconds = int(delay_minutes) * 60 logger.info(f"Event automation '{auto.get('name')}' delaying {delay_minutes}m before action") - time.sleep(int(delay_minutes) * 60) - if not self._running: - return + for remaining in range(delay_seconds, 0, -1): + if not self._running: + return + if self._progress_update_fn and remaining % 5 == 0: + mins, secs = divmod(remaining, 60) + self._progress_update_fn(automation_id, + phase=f'Delay: {mins}m {secs}s remaining', + progress=int((delay_seconds - remaining) / delay_seconds * 10)) + time.sleep(1) # notify_only = no action, just send notification with event data if action_type == 'notify_only': @@ -371,9 +387,13 @@ class AutomationEngine: if guard_fn and guard_fn(): result = {'status': 'skipped', 'reason': f'{action_type} already running'} logger.info(f"Event automation '{auto.get('name')}' skipped — {action_type} busy") + # If progress was initialized during delay, finalize it + if _delay_already_inited and self._progress_finish_fn: + try: self._progress_finish_fn(automation_id, result) + except Exception: pass else: - # Initialize progress tracking - if self._progress_init_fn: + # Initialize progress tracking (skip if already done during delay) + if not _delay_already_inited and self._progress_init_fn: try: self._progress_init_fn(automation_id, auto.get('name', ''), action_type) except Exception: pass try: @@ -441,22 +461,40 @@ class AutomationEngine: # Action delay (skipped for manual run_now) delay_minutes = action_config.get('delay', 0) + _delay_already_inited = False if not skip_delay and delay_minutes and delay_minutes > 0: + # Initialize progress BEFORE delay so card glows during wait + if self._progress_init_fn: + try: self._progress_init_fn(automation_id, auto.get('name', ''), action_type) + except Exception: pass + _delay_already_inited = True + + delay_seconds = int(delay_minutes) * 60 logger.info(f"Automation '{auto['name']}' delaying {delay_minutes}m before action") - time.sleep(int(delay_minutes) * 60) - if not self._running: - return + for remaining in range(delay_seconds, 0, -1): + if not self._running: + return + if self._progress_update_fn and remaining % 5 == 0: + mins, secs = divmod(remaining, 60) + self._progress_update_fn(automation_id, + phase=f'Delay: {mins}m {secs}s remaining', + progress=int((delay_seconds - remaining) / delay_seconds * 10)) + time.sleep(1) # Check guard (is the operation already running?) guard_fn = handler_info.get('guard') if guard_fn and guard_fn(): result = {'status': 'skipped', 'reason': f'{action_type} is already running'} logger.info(f"Automation '{auto['name']}' skipped — {action_type} already running") + # If progress was initialized during delay, finalize it + if _delay_already_inited and self._progress_finish_fn: + try: self._progress_finish_fn(automation_id, result) + except Exception: pass self._finish_run(auto, automation_id, result, error=None) return - # Initialize progress tracking - if self._progress_init_fn: + # Initialize progress tracking (skip if already done during delay) + if not _delay_already_inited and self._progress_init_fn: try: self._progress_init_fn(automation_id, auto.get('name', ''), action_type) except Exception: pass diff --git a/web_server.py b/web_server.py index 286fd5f8..8bbe46d3 100644 --- a/web_server.py +++ b/web_server.py @@ -283,15 +283,89 @@ def _register_automation_handlers(): return {'status': 'error', 'error': str(e)} def _auto_scan_library(config): - if web_scan_manager: + global _scan_library_automation_id + automation_id = config.get('_automation_id') + + if not web_scan_manager: + return {'status': 'error', 'reason': 'Scan manager not available'} + + # If another automation is already tracking the scan, just forward the request + if _scan_library_automation_id is not None: + web_scan_manager.request_scan('Automation trigger (additional batch)') + return {'status': 'skipped', 'reason': 'Scan already being tracked'} + + _scan_library_automation_id = automation_id + + try: result = web_scan_manager.request_scan('Automation trigger') - return {'status': result.get('status', 'unknown')} - return {'status': 'error', 'reason': 'Scan manager not available'} + scan_status_val = result.get('status', 'unknown') + + if scan_status_val == 'queued': + _update_automation_progress(automation_id, + log_line='Scan already in progress — waiting for completion', log_type='info') + else: + delay = result.get('delay_seconds', 60) + _update_automation_progress(automation_id, + log_line=f'Scan scheduled (debounce: {delay}s)', log_type='info') + + # Unified polling loop — handles debounce → scanning → idle transitions + poll_start = time.time() + scan_started = (scan_status_val == 'queued') # Already scanning if queued + while time.time() - poll_start < 1800: # Max 30 min overall + status = web_scan_manager.get_scan_status() + st = status.get('status') + + if st == 'idle': + break # Scan completed (or finished before we started polling) + + elif st == 'scheduled': + elapsed = int(time.time() - poll_start) + _update_automation_progress(automation_id, + phase=f'Waiting for scan to start... ({elapsed}s)', + progress=min(int(elapsed / 60 * 10), 14)) + time.sleep(2) + + elif st == 'scanning': + if not scan_started: + scan_started = True + _update_automation_progress(automation_id, progress=15, + log_line='Scan triggered on media server', log_type='success') + elapsed = status.get('elapsed_seconds', 0) + max_time = status.get('max_time_seconds', 300) + pct = min(15 + int(elapsed / max_time * 80), 95) + mins, secs = divmod(elapsed, 60) + _update_automation_progress(automation_id, + phase=f'Library scan in progress... ({mins}m {secs}s)', + progress=pct) + time.sleep(5) + + else: + time.sleep(2) # Unknown status, avoid tight loop + else: + # 30-min timeout reached + _update_automation_progress(automation_id, status='error', + phase='Timed out', log_line='Library scan timed out after 30 minutes', log_type='error') + return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} + + _update_automation_progress(automation_id, status='finished', progress=100, + phase='Complete', + log_line='Library scan completed', log_type='success') + + return {'status': 'completed', '_manages_own_progress': True} + + except Exception as e: + _update_automation_progress(automation_id, status='error', + phase='Error', log_line=str(e), log_type='error') + return {'status': 'error', 'error': str(e), '_manages_own_progress': True} + + finally: + _scan_library_automation_id = None cross_guard = lambda: is_wishlist_actually_processing() or is_watchlist_actually_scanning() automation_engine.register_action_handler('process_wishlist', _auto_process_wishlist, cross_guard) automation_engine.register_action_handler('scan_watchlist', _auto_scan_watchlist, cross_guard) - automation_engine.register_action_handler('scan_library', _auto_scan_library) + automation_engine.register_action_handler('scan_library', _auto_scan_library, + lambda: _scan_library_automation_id is not None) def _auto_refresh_mirrored(config): """Refresh mirrored playlist(s) from source.""" @@ -541,9 +615,10 @@ def _register_automation_handlers(): def _auto_start_database_update(config): global _db_update_automation_id + automation_id = config.get('_automation_id') if db_update_state.get('status') == 'running': return {'status': 'skipped', 'reason': 'Database update already running'} - _db_update_automation_id = config.get('_automation_id') + _db_update_automation_id = automation_id full = config.get('full_refresh', False) active_server = config_manager.get_active_media_server() with db_update_lock: @@ -552,18 +627,95 @@ def _register_automation_handlers(): "progress": 0, "current_item": "", "processed": 0, "total": 0, "error_message": "" }) db_update_executor.submit(_run_db_update_task, full, active_server) - return {'status': 'started', 'full_refresh': str(full), '_manages_own_progress': True} + + # Monitor DB update progress (callbacks handle card updates, we just block until done) + time.sleep(1) + poll_start = time.time() + last_progress_time = time.time() + last_progress_val = 0 + while time.time() - poll_start < 7200: # Max 2 hours + time.sleep(3) + with db_update_lock: + status = db_update_state.get('status', 'idle') + current_progress = db_update_state.get('progress', 0) + if status != 'running': + break + # Track stall detection — if no progress change in 10 minutes, warn + if current_progress != last_progress_val: + last_progress_val = current_progress + last_progress_time = time.time() + elif time.time() - last_progress_time > 600: + _update_automation_progress(automation_id, + log_line='Database update appears stalled — waiting...', log_type='warning') + last_progress_time = time.time() # Reset so warning repeats every 10 min + else: + # 2-hour timeout reached + _update_automation_progress(automation_id, status='error', + phase='Timed out', log_line='Database update timed out after 2 hours', log_type='error') + return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} + + # Finished/error callback already updated the card — return matching status + with db_update_lock: + final_status = db_update_state.get('status', 'unknown') + if final_status == 'error': + return {'status': 'error', 'reason': db_update_state.get('error_message', 'Unknown error'), '_manages_own_progress': True} + return {'status': 'completed', 'full_refresh': str(full), '_manages_own_progress': True} def _auto_run_duplicate_cleaner(config): + automation_id = config.get('_automation_id') if duplicate_cleaner_state.get('status') == 'running': return {'status': 'skipped', 'reason': 'Duplicate cleaner already running'} + + # Pre-set status before submit so polling loop doesn't see stale 'finished' from last run + with duplicate_cleaner_lock: + duplicate_cleaner_state["status"] = "running" duplicate_cleaner_executor.submit(_run_duplicate_cleaner) - return {'status': 'started'} + _update_automation_progress(automation_id, + log_line='Duplicate cleaner started', log_type='info') + + # Monitor duplicate cleaner progress (max 2 hours) + time.sleep(1) # Brief pause for executor to start + poll_start = time.time() + while time.time() - poll_start < 7200: + time.sleep(3) + status = duplicate_cleaner_state.get('status', 'idle') + if status not in ('running',): + break + phase = duplicate_cleaner_state.get('phase', 'Scanning...') + progress = duplicate_cleaner_state.get('progress', 0) + scanned = duplicate_cleaner_state.get('files_scanned', 0) + total = duplicate_cleaner_state.get('total_files', 0) + _update_automation_progress(automation_id, + phase=phase, progress=progress, + processed=scanned, total=total) + else: + # 2-hour timeout reached + _update_automation_progress(automation_id, status='error', + phase='Timed out', log_line='Duplicate cleaner timed out after 2 hours', log_type='error') + return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} + + # Check actual exit status (could be 'finished' or 'error') + final_status = duplicate_cleaner_state.get('status', 'idle') + if final_status == 'error': + err = duplicate_cleaner_state.get('error_message', 'Unknown error') + _update_automation_progress(automation_id, status='error', progress=100, + phase='Error', log_line=err, log_type='error') + return {'status': 'error', 'reason': err, '_manages_own_progress': True} + + dupes = duplicate_cleaner_state.get('duplicates_found', 0) + removed = duplicate_cleaner_state.get('deleted', 0) + _update_automation_progress(automation_id, status='finished', progress=100, + phase='Complete', + log_line=f'Found {dupes} duplicates, removed {removed} files', log_type='success') + return {'status': 'completed', '_manages_own_progress': True} def _auto_clear_quarantine(config): import shutil as _shutil + automation_id = config.get('_automation_id') quarantine_path = os.path.join(docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')), 'ss_quarantine') if not os.path.exists(quarantine_path): + _update_automation_progress(automation_id, + log_line='No quarantine folder found', log_type='info') return {'status': 'completed', 'removed': '0'} removed = 0 for f in os.listdir(quarantine_path): @@ -577,32 +729,86 @@ def _register_automation_handlers(): removed += 1 except Exception: pass + _update_automation_progress(automation_id, + log_line=f'Removed {removed} quarantined items', log_type='success' if removed > 0 else 'info') return {'status': 'completed', 'removed': str(removed)} def _auto_cleanup_wishlist(config): + automation_id = config.get('_automation_id') db = get_database() removed = db.remove_wishlist_duplicates(get_current_profile_id()) + _update_automation_progress(automation_id, + log_line=f'Removed {removed or 0} duplicate wishlist entries', log_type='success' if removed else 'info') return {'status': 'completed', 'removed': str(removed or 0)} def _auto_update_discovery_pool(config): + automation_id = config.get('_automation_id') try: from core.watchlist_scanner import get_watchlist_scanner scanner = get_watchlist_scanner(spotify_client) - threading.Thread(target=scanner.update_discovery_pool_incremental, - args=(get_current_profile_id(),), daemon=True).start() - return {'status': 'started'} + _update_automation_progress(automation_id, + log_line='Updating discovery pool...', log_type='info') + scanner.update_discovery_pool_incremental(get_current_profile_id()) + _update_automation_progress(automation_id, status='finished', progress=100, + phase='Complete', + log_line='Discovery pool updated', log_type='success') + return {'status': 'completed', '_manages_own_progress': True} except Exception as e: - return {'status': 'error', 'reason': str(e)} + _update_automation_progress(automation_id, status='error', + phase='Error', log_line=str(e), log_type='error') + return {'status': 'error', 'reason': str(e), '_manages_own_progress': True} def _auto_start_quality_scan(config): + automation_id = config.get('_automation_id') if quality_scanner_state.get('status') == 'running': return {'status': 'skipped', 'reason': 'Quality scan already running'} + scope = config.get('scope', 'watchlist') + # Pre-set status before submit so polling loop doesn't see stale 'finished' from last run + with quality_scanner_lock: + quality_scanner_state["status"] = "running" quality_scanner_executor.submit(_run_quality_scanner, scope, get_current_profile_id()) - return {'status': 'started', 'scope': scope} + _update_automation_progress(automation_id, + log_line=f'Quality scan started (scope: {scope})', log_type='info') + + # Monitor quality scanner progress (max 2 hours) + time.sleep(1) # Brief pause for executor to start + poll_start = time.time() + while time.time() - poll_start < 7200: + time.sleep(3) + status = quality_scanner_state.get('status', 'idle') + if status not in ('running',): + break + phase = quality_scanner_state.get('phase', 'Scanning...') + progress = quality_scanner_state.get('progress', 0) + processed = quality_scanner_state.get('processed', 0) + total = quality_scanner_state.get('total', 0) + _update_automation_progress(automation_id, + phase=phase, progress=progress, + processed=processed, total=total) + else: + # 2-hour timeout reached + _update_automation_progress(automation_id, status='error', + phase='Timed out', log_line='Quality scan timed out after 2 hours', log_type='error') + return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} + + # Check actual exit status (could be 'finished' or 'error') + final_status = quality_scanner_state.get('status', 'idle') + if final_status == 'error': + err = quality_scanner_state.get('error_message', 'Unknown error') + _update_automation_progress(automation_id, status='error', progress=100, + phase='Error', log_line=err, log_type='error') + return {'status': 'error', 'reason': err, '_manages_own_progress': True} + + issues = quality_scanner_state.get('low_quality', 0) + _update_automation_progress(automation_id, status='finished', progress=100, + phase='Complete', + log_line=f'Quality scan complete — {issues} issues found', log_type='success') + return {'status': 'completed', 'scope': scope, '_manages_own_progress': True} def _auto_backup_database(config): import sqlite3, glob as _glob + automation_id = config.get('_automation_id') db_path = os.environ.get('DATABASE_PATH', 'database/music_library.db') if not os.path.exists(db_path): return {'status': 'error', 'reason': 'Database file not found'} @@ -623,6 +829,8 @@ def _register_automation_handlers(): os.remove(existing.pop(0)) except Exception: pass + _update_automation_progress(automation_id, + log_line=f'Backup created: {size_mb}MB ({os.path.basename(backup_path)})', log_type='success') return {'status': 'completed', 'backup_path': backup_path, 'size_mb': str(size_mb)} automation_engine.register_action_handler('start_database_update', _auto_start_database_update, @@ -652,7 +860,7 @@ def _register_automation_handlers(): phase='Error' if status == 'error' else 'Complete', log_line=msg, log_type='error' if status == 'error' else 'success') - automation_engine.register_progress_callbacks(_progress_init, _progress_finish) + automation_engine.register_progress_callbacks(_progress_init, _progress_finish, _update_automation_progress) # Register permanent callback: when any scan completes, emit library_scan_completed event # This replaces the hardcoded scan_completion_callback → trigger_automatic_database_update chain @@ -743,6 +951,7 @@ db_update_state = { "removed_tracks": 0 } _db_update_automation_id = None # Set when automation triggers DB update, used by callbacks +_scan_library_automation_id = None # Set when automation triggers library scan, used for tracking # Quality Scanner state quality_scanner_state = {