From 017553193f04a2daeb2700ce3c65a950684602f4 Mon Sep 17 00:00:00 2001 From: Broque Thomas <26755000+Nezreka@users.noreply.github.com> Date: Fri, 15 May 2026 11:24:35 -0700 Subject: [PATCH] Extract automation handlers (3/3): maintenance + misc, finishing the lift MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final commit of the automation-handler refactor. With this commit every closure that used to live in `web_server._register_automation_handlers` is now a top-level function in `core/automation/handlers/`. Handlers extracted in this commit: - start_database_update + deep_scan_library -> core/automation/handlers/database_update.py Both share the db_update_state monitoring pattern (poll until status flips, stall detection emits warning at 10 min, 2-hour outer timeout). Lifted into a shared `_run_with_progress` helper inside the module so the per-handler bodies stay tiny. - run_duplicate_cleaner -> core/automation/handlers/duplicate_cleaner.py - start_quality_scan -> core/automation/handlers/quality_scanner.py - clear_quarantine, cleanup_wishlist, update_discovery_pool, backup_database, refresh_beatport_cache -> core/automation/handlers/maintenance.py Grouped because each body is short (~20-50 lines) and they share no state — splitting into per-handler files would just add import noise. - clean_search_history, clean_completed_downloads, full_cleanup -> core/automation/handlers/download_cleanup.py Grouped because all three reach the download orchestrator, tasks_lock, and download_batches/download_tasks accessors. The full_cleanup multi-step orchestration shares phase-detection logic with clean_completed_downloads. - run_script -> core/automation/handlers/run_script.py - search_and_download -> core/automation/handlers/search_and_download.py `AutomationDeps` grew with the new dependency surface: - get_db_update_state + db_update_lock + db_update_executor + run_db_update_task + run_deep_scan_task - get_duplicate_cleaner_state + duplicate_cleaner_lock + duplicate_cleaner_executor + run_duplicate_cleaner - get_quality_scanner_state + quality_scanner_lock + quality_scanner_executor + run_quality_scanner - download_orchestrator + run_async + tasks_lock + get_download_batches + get_download_tasks + sweep_empty_download_directories + get_staging_path - docker_resolve_path + get_current_profile_id + get_watchlist_scanner + get_app + get_beatport_data_cache - set_db_update_automation_id (writes the legacy global so the live DB-update progress callbacks still living in web_server.py keep emitting against the active automation card) `web_server._register_automation_handlers` is now ~50 lines: build deps once, call register_all. The 667-line block of remaining closure definitions and engine register calls is gone. The final orphan was the `_db_update_automation_id` module global — the DB-update progress callbacks at line ~14080 still read it directly, so the extracted database_update handler propagates the automation id through `deps.set_db_update_automation_id` (a closure in web_server that writes the global). When the legacy callbacks get extracted in a future PR the setter goes away. Tests: - tests/automation/test_handlers_maintenance.py adds 21 boundary tests covering every newly-extracted handler shape: guard short-circuits (already-running returns skipped), deps wiring (set_db_update_automation_id called with the right id), exception swallow contract, status returns, path-traversal blocked in run_script, source-mode skip in clean_search_history, active-batch skip in clean_completed_downloads, etc. - 3244 tests pass (was 3223 — 21 new), no regression. web_server.py: 35,593 -> 34,220 lines (-1,373 net across 3 commits). Issue #1 from the extraction punch list is now COMPLETE. --- core/automation/deps.py | 32 + core/automation/handlers/__init__.py | 31 + core/automation/handlers/database_update.py | 136 ++++ core/automation/handlers/download_cleanup.py | 267 +++++++ core/automation/handlers/duplicate_cleaner.py | 87 +++ core/automation/handlers/maintenance.py | 213 ++++++ core/automation/handlers/quality_scanner.py | 83 +++ core/automation/handlers/registration.py | 82 ++ core/automation/handlers/run_script.py | 103 +++ .../handlers/search_and_download.py | 57 ++ tests/automation/test_handlers_maintenance.py | 390 ++++++++++ tests/automation/test_handlers_playlist.py | 26 + tests/automation/test_handlers_simple.py | 26 + web_server.py | 704 +----------------- 14 files changed, 1570 insertions(+), 667 deletions(-) create mode 100644 core/automation/handlers/database_update.py create mode 100644 core/automation/handlers/download_cleanup.py create mode 100644 core/automation/handlers/duplicate_cleaner.py create mode 100644 core/automation/handlers/maintenance.py create mode 100644 core/automation/handlers/quality_scanner.py create mode 100644 core/automation/handlers/run_script.py create mode 100644 core/automation/handlers/search_and_download.py create mode 100644 tests/automation/test_handlers_maintenance.py diff --git a/core/automation/deps.py b/core/automation/deps.py index bf5a6a4a..83b60fb0 100644 --- a/core/automation/deps.py +++ b/core/automation/deps.py @@ -101,3 +101,35 @@ class AutomationDeps: 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 + + # --- Database update + quality scanner (shared state + executors) --- + set_db_update_automation_id: Callable[[Optional[str]], None] # syncs the legacy `_db_update_automation_id` global so the live DB-update progress callbacks (which still read the global directly) keep firing for the active automation + get_db_update_state: Callable[[], dict] + db_update_lock: Any # threading.Lock + db_update_executor: Any # ThreadPoolExecutor + run_db_update_task: Callable[..., Any] + run_deep_scan_task: Callable[..., Any] + get_duplicate_cleaner_state: Callable[[], dict] + duplicate_cleaner_lock: Any + duplicate_cleaner_executor: Any + run_duplicate_cleaner: Callable[..., Any] + get_quality_scanner_state: Callable[[], dict] + quality_scanner_lock: Any + quality_scanner_executor: Any + run_quality_scanner: Callable[..., Any] + + # --- Download orchestrator + queue accessors --- + download_orchestrator: Any + run_async: Callable[..., Any] + tasks_lock: Any + get_download_batches: Callable[[], dict] + get_download_tasks: Callable[[], dict] + sweep_empty_download_directories: Callable[[], int] + get_staging_path: Callable[[], str] + + # --- Maintenance helpers --- + docker_resolve_path: Callable[[str], str] + get_current_profile_id: Callable[[], int] + get_watchlist_scanner: Callable[[Any], Any] + get_app: Callable[[], Any] # Flask app for test_client (beatport refresh) + get_beatport_data_cache: Callable[[], dict] diff --git a/core/automation/handlers/__init__.py b/core/automation/handlers/__init__.py index 5f086115..95f357a4 100644 --- a/core/automation/handlers/__init__.py +++ b/core/automation/handlers/__init__.py @@ -17,6 +17,23 @@ 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.database_update import auto_start_database_update, auto_deep_scan_library +from core.automation.handlers.duplicate_cleaner import auto_run_duplicate_cleaner +from core.automation.handlers.quality_scanner import auto_start_quality_scan +from core.automation.handlers.maintenance import ( + auto_clear_quarantine, + auto_cleanup_wishlist, + auto_update_discovery_pool, + auto_backup_database, + auto_refresh_beatport_cache, +) +from core.automation.handlers.download_cleanup import ( + auto_clean_search_history, + auto_clean_completed_downloads, + auto_full_cleanup, +) +from core.automation.handlers.run_script import auto_run_script +from core.automation.handlers.search_and_download import auto_search_and_download from core.automation.handlers.registration import register_all __all__ = [ @@ -27,5 +44,19 @@ __all__ = [ 'auto_sync_playlist', 'auto_discover_playlist', 'auto_playlist_pipeline', + 'auto_start_database_update', + 'auto_deep_scan_library', + 'auto_run_duplicate_cleaner', + 'auto_start_quality_scan', + 'auto_clear_quarantine', + 'auto_cleanup_wishlist', + 'auto_update_discovery_pool', + 'auto_backup_database', + 'auto_refresh_beatport_cache', + 'auto_clean_search_history', + 'auto_clean_completed_downloads', + 'auto_full_cleanup', + 'auto_run_script', + 'auto_search_and_download', 'register_all', ] diff --git a/core/automation/handlers/database_update.py b/core/automation/handlers/database_update.py new file mode 100644 index 00000000..a9d51036 --- /dev/null +++ b/core/automation/handlers/database_update.py @@ -0,0 +1,136 @@ +"""Automation handlers: ``start_database_update`` and +``deep_scan_library`` actions. + +Lifted from ``web_server._register_automation_handlers`` (the +``_auto_start_database_update`` and ``_auto_deep_scan_library`` +closures). Both share the same ``db_update_state`` / executor / lock +infrastructure -- the only difference is which task they submit +(``run_db_update_task`` vs ``run_deep_scan_task``). + +Pattern: pre-set state to running, submit task to executor, then +poll the state dict until it transitions away from ``running``. +Stall-detection emits a warning every 10 minutes when progress +hasn't budged. 2-hour outer timeout caps the worst case. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +_TIMEOUT_SECONDS = 7200 # 2 hours — covers the worst large-library case +_STALL_WARNING_SECONDS = 600 # 10 minutes without progress = stall +_POLL_INTERVAL_SECONDS = 3 +_INITIAL_DELAY_SECONDS = 1 + + +def auto_start_database_update(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Run a full or incremental DB update via ``run_db_update_task``.""" + return _run_with_progress( + config, deps, + task=deps.run_db_update_task, + task_args=(config.get('full_refresh', False), deps.config_manager.get_active_media_server()), + initial_phase='Initializing...', + stall_label='Database update', + finished_extras=lambda: {'full_refresh': str(config.get('full_refresh', False))}, + timeout_label='Database update timed out after 2 hours', + ) + + +def auto_deep_scan_library(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Run a deep library scan via ``run_deep_scan_task``.""" + return _run_with_progress( + config, deps, + task=deps.run_deep_scan_task, + task_args=(deps.config_manager.get_active_media_server(),), + initial_phase='Deep scan: Initializing...', + stall_label='Deep scan', + finished_extras=lambda: {}, + timeout_label='Deep scan timed out after 2 hours', + ) + + +def _run_with_progress( + config: Dict[str, Any], + deps: AutomationDeps, + *, + task, + task_args: tuple, + initial_phase: str, + stall_label: str, + finished_extras, + timeout_label: str, +) -> Dict[str, Any]: + """Shared poll-and-wait body for both DB-update handlers.""" + automation_id = config.get('_automation_id') + state = deps.get_db_update_state() + if state.get('status') == 'running': + return {'status': 'skipped', 'reason': 'Database update already running'} + deps.state.db_update_automation_id = automation_id + # Sync legacy module global so the DB-update progress callbacks + # (still living in web_server.py) emit against this automation. + deps.set_db_update_automation_id(automation_id) + + with deps.db_update_lock: + state.update({ + 'status': 'running', 'phase': initial_phase, + 'progress': 0, 'current_item': '', 'processed': 0, 'total': 0, + 'error_message': '', + }) + deps.db_update_executor.submit(task, *task_args) + + # Monitor progress (callbacks handle card updates, we just block until done). + time.sleep(_INITIAL_DELAY_SECONDS) + poll_start = time.time() + last_progress_time = time.time() + last_progress_val = 0 + while time.time() - poll_start < _TIMEOUT_SECONDS: + time.sleep(_POLL_INTERVAL_SECONDS) + with deps.db_update_lock: + current_status = state.get('status', 'idle') + current_progress = state.get('progress', 0) + if current_status != 'running': + break + # 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 > _STALL_WARNING_SECONDS: + deps.update_progress( + automation_id, + log_line=f'{stall_label} appears stalled — waiting...', + log_type='warning', + ) + last_progress_time = time.time() # Reset so warning repeats every 10 min. + else: + # 2-hour timeout reached. + deps.update_progress( + automation_id, status='error', + phase='Timed out', log_line=timeout_label, log_type='error', + ) + return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} + + # Finished/error callback already updated the card — return matching status. + with deps.db_update_lock: + final_status = state.get('status', 'unknown') + if final_status == 'error': + return { + 'status': 'error', + 'reason': state.get('error_message', 'Unknown error'), + '_manages_own_progress': True, + } + with deps.db_update_lock: + stats = { + 'status': 'completed', '_manages_own_progress': True, + 'artists': state.get('total', 0), + 'albums': state.get('total_albums', 0), + 'tracks': state.get('total_tracks', 0), + 'removed_artists': state.get('removed_artists', 0), + 'removed_albums': state.get('removed_albums', 0), + 'removed_tracks': state.get('removed_tracks', 0), + } + stats.update(finished_extras()) + return stats diff --git a/core/automation/handlers/download_cleanup.py b/core/automation/handlers/download_cleanup.py new file mode 100644 index 00000000..4aba711a --- /dev/null +++ b/core/automation/handlers/download_cleanup.py @@ -0,0 +1,267 @@ +"""Automation handlers: download-queue cleanup actions. + +Lifted from ``web_server._register_automation_handlers``: +- ``clean_search_history`` → :func:`auto_clean_search_history` +- ``clean_completed_downloads`` → :func:`auto_clean_completed_downloads` +- ``full_cleanup`` → :func:`auto_full_cleanup` + +All three share the download-orchestrator + tasks_lock / +download_batches / download_tasks accessors. ``full_cleanup`` is a +multi-step orchestration that pulls in quarantine purge + staging +sweep on top of the queue cleanup -- kept as one big handler since +its phases share state-detection logic. +""" + +from __future__ import annotations + +import os +import shutil as _shutil +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +# ─── clean_search_history ──────────────────────────────────────────── + + +def auto_clean_search_history(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Remove old searches from Soulseek when configured.""" + automation_id = config.get('_automation_id') + # Skip if soulseek is not the active download source or in hybrid order. + dl_mode = deps.config_manager.get('download_source.mode', 'hybrid') + hybrid_order = deps.config_manager.get( + 'download_source.hybrid_order', ['hifi', 'youtube', 'soulseek'], + ) + soulseek_active = ( + dl_mode == 'soulseek' + or (dl_mode == 'hybrid' and 'soulseek' in hybrid_order) + ) + # Reach the underlying SoulseekClient via the orchestrator's + # generic accessor. + slskd = deps.download_orchestrator.client('soulseek') if deps.download_orchestrator else None + if not soulseek_active or not slskd or not slskd.base_url: + deps.update_progress(automation_id, log_line='Soulseek not active — skipped', log_type='skip') + return {'status': 'skipped'} + if not deps.config_manager.get('soulseek.auto_clear_searches', True): + deps.update_progress( + automation_id, log_line='Auto-clear disabled in settings', log_type='skip', + ) + return {'status': 'skipped'} + try: + success = deps.run_async(deps.download_orchestrator.maintain_search_history_with_buffer( + keep_searches=50, trigger_threshold=200, + )) + if success: + deps.update_progress( + automation_id, + log_line='Search history maintenance completed', + log_type='success', + ) + return {'status': 'completed'} + else: + deps.update_progress(automation_id, log_line='No cleanup needed', log_type='skip') + return {'status': 'completed'} + except Exception as e: # noqa: BLE001 — automation handlers must never raise + return {'status': 'error', 'error': str(e)} + + +# ─── clean_completed_downloads ─────────────────────────────────────── + + +def auto_clean_completed_downloads(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Clear completed downloads + sweep empty download directories. + Skips when active batches or post-processing is in flight.""" + automation_id = config.get('_automation_id') + try: + has_active_batches = False + has_post_processing = False + with deps.tasks_lock: + batches = deps.get_download_batches() + for batch_data in batches.values(): + if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]: + has_active_batches = True + break + if not has_active_batches: + tasks = deps.get_download_tasks() + for task_data in tasks.values(): + if task_data.get('status') == 'post_processing': + has_post_processing = True + break + + if has_active_batches: + deps.update_progress( + automation_id, log_line='Skipped — downloads active', log_type='skip', + ) + return {'status': 'completed'} + + deps.run_async(deps.download_orchestrator.clear_all_completed_downloads()) + if not has_post_processing: + deps.sweep_empty_download_directories() + deps.update_progress( + automation_id, log_line='Download cleanup completed', log_type='success', + ) + return {'status': 'completed'} + except Exception as e: # noqa: BLE001 — automation handlers must never raise + return {'status': 'error', 'reason': str(e)} + + +# ─── full_cleanup ──────────────────────────────────────────────────── + + +def auto_full_cleanup(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Run all cleanup tasks: quarantine purge → download queue clear + → empty-dir sweep → staging sweep → search history.""" + automation_id = config.get('_automation_id') + steps = [] + + # --- 1. Clear quarantine --- + deps.update_progress(automation_id, phase='Clearing quarantine...', progress=0) + quarantine_path = os.path.join( + deps.docker_resolve_path(deps.config_manager.get('soulseek.download_path', './downloads')), + 'ss_quarantine', + ) + q_removed = 0 + if os.path.exists(quarantine_path): + for f in os.listdir(quarantine_path): + fp = os.path.join(quarantine_path, f) + try: + if os.path.isfile(fp): + os.remove(fp) + q_removed += 1 + elif os.path.isdir(fp): + _shutil.rmtree(fp) + q_removed += 1 + except Exception as e: # noqa: BLE001 — best-effort purge + deps.logger.debug("quarantine entry purge failed: %s", e) + steps.append(f'Quarantine: removed {q_removed} items') + deps.update_progress( + automation_id, + log_line=f'Quarantine: removed {q_removed} items', + log_type='success' if q_removed else 'info', + ) + + # --- 2. Clear completed/errored/cancelled downloads from Soulseek queue --- + deps.update_progress(automation_id, phase='Clearing download queue...', progress=20) + has_active_batches = False + has_post_processing = False + with deps.tasks_lock: + batches = deps.get_download_batches() + for batch_data in batches.values(): + if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]: + has_active_batches = True + break + if not has_active_batches: + tasks = deps.get_download_tasks() + for task_data in tasks.values(): + if task_data.get('status') == 'post_processing': + has_post_processing = True + break + if has_active_batches: + steps.append('Download queue: skipped (active batches)') + deps.update_progress( + automation_id, + log_line='Download queue: skipped (active batches)', + log_type='skip', + ) + else: + try: + deps.run_async(deps.download_orchestrator.clear_all_completed_downloads()) + steps.append('Download queue: cleared') + deps.update_progress( + automation_id, log_line='Download queue: cleared', log_type='success', + ) + except Exception as e: # noqa: BLE001 — per-step best-effort + steps.append(f'Download queue: error ({e})') + deps.update_progress( + automation_id, + log_line=f'Download queue: error ({e})', + log_type='error', + ) + + # --- 3. Sweep empty download directories --- + deps.update_progress(automation_id, phase='Sweeping empty directories...', progress=40) + if has_active_batches or has_post_processing: + reason = 'active batches' if has_active_batches else 'post-processing active' + steps.append(f'Empty directories: skipped ({reason})') + deps.update_progress( + automation_id, + log_line=f'Empty directories: skipped ({reason})', + log_type='skip', + ) + else: + dirs_removed = deps.sweep_empty_download_directories() + steps.append(f'Empty directories: removed {dirs_removed}') + deps.update_progress( + automation_id, + log_line=f'Empty directories: removed {dirs_removed}', + log_type='success' if dirs_removed else 'info', + ) + + # --- 4. Sweep empty staging directories --- + deps.update_progress(automation_id, phase='Sweeping import folder...', progress=60) + staging_path = deps.get_staging_path() + s_removed = 0 + if os.path.isdir(staging_path): + for dirpath, _dirnames, _filenames in os.walk(staging_path, topdown=False): + if os.path.normpath(dirpath) == os.path.normpath(staging_path): + continue + try: + entries = os.listdir(dirpath) + except OSError: + continue + visible = [e for e in entries if not e.startswith('.')] + if not visible: + for hidden in entries: + try: + os.remove(os.path.join(dirpath, hidden)) + except Exception as e: # noqa: BLE001 — best-effort + deps.logger.debug("hidden file cleanup failed: %s", e) + try: + os.rmdir(dirpath) + s_removed += 1 + except OSError: + pass + steps.append(f'Staging: removed {s_removed} empty directories') + deps.update_progress( + automation_id, + log_line=f'Staging: removed {s_removed} empty directories', + log_type='success' if s_removed else 'info', + ) + + # --- 5. Clean search history (if enabled) --- + deps.update_progress(automation_id, phase='Cleaning search history...', progress=80) + try: + if not deps.config_manager.get('soulseek.auto_clear_searches', True): + steps.append('Search cleanup: disabled in settings') + deps.update_progress( + automation_id, log_line='Search cleanup: disabled in settings', log_type='skip', + ) + else: + deps.run_async(deps.download_orchestrator.maintain_search_history_with_buffer( + keep_searches=50, trigger_threshold=200, + )) + steps.append('Search history: cleaned') + deps.update_progress( + automation_id, log_line='Search history: cleaned', log_type='success', + ) + except Exception as e: # noqa: BLE001 — per-step best-effort + steps.append(f'Search history: error ({e})') + deps.update_progress( + automation_id, log_line=f'Search history: error ({e})', log_type='error', + ) + + total_removed = q_removed + s_removed + deps.update_progress( + automation_id, status='finished', progress=100, + phase='Complete', + log_line=f'Full cleanup complete — {total_removed} items removed', + log_type='success', + ) + return { + 'status': 'completed', + 'quarantine_removed': str(q_removed), + 'staging_removed': str(s_removed), + 'total_removed': str(total_removed), + 'steps': steps, + '_manages_own_progress': True, + } diff --git a/core/automation/handlers/duplicate_cleaner.py b/core/automation/handlers/duplicate_cleaner.py new file mode 100644 index 00000000..a0531622 --- /dev/null +++ b/core/automation/handlers/duplicate_cleaner.py @@ -0,0 +1,87 @@ +"""Automation handler: ``run_duplicate_cleaner`` action. + +Lifted from ``web_server._register_automation_handlers`` (the +``_auto_run_duplicate_cleaner`` closure). Submits the duplicate +cleaner to its executor, then polls the shared state dict until +the worker transitions away from ``running``. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +_TIMEOUT_SECONDS = 7200 # 2 hours +_POLL_INTERVAL_SECONDS = 3 +_INITIAL_DELAY_SECONDS = 1 + + +def auto_run_duplicate_cleaner(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Kick off the duplicate cleaner and report final stats.""" + automation_id = config.get('_automation_id') + state = deps.get_duplicate_cleaner_state() + if state.get('status') == 'running': + return {'status': 'skipped', 'reason': 'Duplicate cleaner already running'} + + # Pre-set status before submit so the polling loop doesn't see a + # stale 'finished' from a previous run. + with deps.duplicate_cleaner_lock: + state['status'] = 'running' + deps.duplicate_cleaner_executor.submit(deps.run_duplicate_cleaner) + deps.update_progress(automation_id, log_line='Duplicate cleaner started', log_type='info') + + # Monitor progress (max 2 hours). + time.sleep(_INITIAL_DELAY_SECONDS) + poll_start = time.time() + while time.time() - poll_start < _TIMEOUT_SECONDS: + time.sleep(_POLL_INTERVAL_SECONDS) + current_status = state.get('status', 'idle') + if current_status not in ('running',): + break + deps.update_progress( + automation_id, + phase=state.get('phase', 'Scanning...'), + progress=state.get('progress', 0), + processed=state.get('files_scanned', 0), + total=state.get('total_files', 0), + ) + else: + # 2-hour timeout reached. + deps.update_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 = state.get('status', 'idle') + if final_status == 'error': + err = state.get('error_message', 'Unknown error') + deps.update_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 = state.get('duplicates_found', 0) + removed = state.get('deleted', 0) + space_freed = state.get('space_freed', 0) + scanned = state.get('files_scanned', 0) + deps.update_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, + 'files_scanned': scanned, + 'duplicates_found': dupes, + 'files_deleted': removed, + 'space_freed_mb': round(space_freed / (1024 * 1024), 1), + } diff --git a/core/automation/handlers/maintenance.py b/core/automation/handlers/maintenance.py new file mode 100644 index 00000000..dacd13c8 --- /dev/null +++ b/core/automation/handlers/maintenance.py @@ -0,0 +1,213 @@ +"""Automation handlers: short maintenance actions. + +Lifted from ``web_server._register_automation_handlers``: +- ``clear_quarantine`` → :func:`auto_clear_quarantine` +- ``cleanup_wishlist`` → :func:`auto_cleanup_wishlist` +- ``update_discovery_pool`` → :func:`auto_update_discovery_pool` +- ``backup_database`` → :func:`auto_backup_database` +- ``refresh_beatport_cache`` → :func:`auto_refresh_beatport_cache` + +Each is a thin wrapper around an existing service / helper. Grouped +in one module because every body is short and they share no state +between them — splitting into per-handler files would just add +import noise. +""" + +from __future__ import annotations + +import glob as _glob +import os +import shutil as _shutil +import sqlite3 +import time +from datetime import datetime +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +# ─── clear_quarantine ──────────────────────────────────────────────── + + +def auto_clear_quarantine(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Purge every file/folder under the configured ss_quarantine path.""" + automation_id = config.get('_automation_id') + quarantine_path = os.path.join( + deps.docker_resolve_path(deps.config_manager.get('soulseek.download_path', './downloads')), + 'ss_quarantine', + ) + if not os.path.exists(quarantine_path): + deps.update_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): + fp = os.path.join(quarantine_path, f) + try: + if os.path.isfile(fp): + os.remove(fp) + removed += 1 + elif os.path.isdir(fp): + _shutil.rmtree(fp) + removed += 1 + except Exception as e: # noqa: BLE001 — best-effort purge + deps.logger.debug("quarantine entry purge failed: %s", e) + deps.update_progress( + automation_id, + log_line=f'Removed {removed} quarantined items', + log_type='success' if removed > 0 else 'info', + ) + return {'status': 'completed', 'removed': str(removed)} + + +# ─── cleanup_wishlist ──────────────────────────────────────────────── + + +def auto_cleanup_wishlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Drop duplicate entries from the wishlist for the active profile.""" + automation_id = config.get('_automation_id') + db = deps.get_database() + removed = db.remove_wishlist_duplicates(deps.get_current_profile_id()) + deps.update_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)} + + +# ─── update_discovery_pool ─────────────────────────────────────────── + + +def auto_update_discovery_pool(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Run an incremental refresh of the discovery pool via the + watchlist scanner.""" + automation_id = config.get('_automation_id') + try: + scanner = deps.get_watchlist_scanner(deps.spotify_client) + deps.update_progress(automation_id, log_line='Updating discovery pool...', log_type='info') + scanner.update_discovery_pool_incremental(deps.get_current_profile_id()) + deps.update_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: # noqa: BLE001 — automation handlers must never raise + deps.update_progress( + automation_id, status='error', + phase='Error', log_line=str(e), log_type='error', + ) + return {'status': 'error', 'reason': str(e), '_manages_own_progress': True} + + +# ─── backup_database ───────────────────────────────────────────────── + + +_MAX_BACKUPS = 5 + + +def auto_backup_database(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Create a hot SQLite backup, then prune old backups so only the + newest ``_MAX_BACKUPS`` remain.""" + 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'} + + timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') + backup_path = f"{db_path}.backup_{timestamp}" + # Use SQLite backup API for a safe hot-copy of an active database. + src = sqlite3.connect(db_path) + dst = sqlite3.connect(backup_path) + src.backup(dst) + dst.close() + src.close() + size_mb = round(os.path.getsize(backup_path) / (1024 * 1024), 1) + + # Rolling cleanup — keep only the newest N backups. + existing = sorted(_glob.glob(f"{db_path}.backup_*"), key=os.path.getmtime) + while len(existing) > _MAX_BACKUPS: + try: + os.remove(existing.pop(0)) + except Exception as e: # noqa: BLE001 — best-effort cleanup + deps.logger.debug("rolling backup cleanup failed: %s", e) + deps.update_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)} + + +# ─── refresh_beatport_cache ────────────────────────────────────────── + + +_BEATPORT_SECTIONS = ( + ('hero_tracks', '/api/beatport/hero-tracks', 'Hero Tracks'), + ('new_releases', '/api/beatport/new-releases', 'New Releases'), + ('featured_charts', '/api/beatport/featured-charts', 'Featured Charts'), + ('dj_charts', '/api/beatport/dj-charts', 'DJ Charts'), + ('top_10_lists', '/api/beatport/homepage/top-10-lists', 'Top 10 Lists'), + ('top_10_releases', '/api/beatport/homepage/top-10-releases-cards', 'Top 10 Releases'), + ('hype_picks', '/api/beatport/hype-picks', 'Hype Picks'), +) + + +def auto_refresh_beatport_cache(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + """Refresh Beatport homepage cache by calling each endpoint internally + via Flask's ``test_client``. Invalidates the homepage cache first + so endpoints re-scrape rather than returning stale data.""" + automation_id = config.get('_automation_id') + cache = deps.get_beatport_data_cache() + # Invalidate all homepage cache timestamps so endpoints re-scrape. + with cache['cache_lock']: + for key in cache['homepage']: + cache['homepage'][key]['timestamp'] = 0 + cache['homepage'][key]['data'] = None + + refreshed = 0 + errors = [] + app = deps.get_app() + with app.test_client() as client: + for idx, (_, endpoint, label) in enumerate(_BEATPORT_SECTIONS): + deps.update_progress( + automation_id, + progress=(idx / len(_BEATPORT_SECTIONS)) * 100, + phase=f'Scraping: {label}', + current_item=label, + ) + try: + resp = client.get(endpoint) + if resp.status_code == 200: + refreshed += 1 + deps.update_progress( + automation_id, log_line=f'{label}: cached', log_type='success', + ) + else: + errors.append(label) + deps.update_progress( + automation_id, + log_line=f'{label}: HTTP {resp.status_code}', + log_type='error', + ) + except Exception as e: # noqa: BLE001 — per-section best-effort + errors.append(label) + deps.update_progress( + automation_id, + log_line=f'{label}: {str(e)}', + log_type='error', + ) + if idx < len(_BEATPORT_SECTIONS) - 1: + time.sleep(2) + + deps.update_progress( + automation_id, status='finished', progress=100, + phase='Complete', + log_line=f'Refreshed {refreshed}/{len(_BEATPORT_SECTIONS)} sections', + log_type='success', + ) + return { + 'status': 'completed', + 'refreshed': str(refreshed), + 'errors': str(len(errors)), + '_manages_own_progress': True, + } diff --git a/core/automation/handlers/quality_scanner.py b/core/automation/handlers/quality_scanner.py new file mode 100644 index 00000000..69ec3f02 --- /dev/null +++ b/core/automation/handlers/quality_scanner.py @@ -0,0 +1,83 @@ +"""Automation handler: ``start_quality_scan`` action. + +Lifted from ``web_server._register_automation_handlers`` (the +``_auto_start_quality_scan`` closure). Submits the quality scanner +to its executor with the configured scope (default: ``watchlist``) +then polls the shared state dict. +""" + +from __future__ import annotations + +import time +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +_TIMEOUT_SECONDS = 7200 # 2 hours +_POLL_INTERVAL_SECONDS = 3 +_INITIAL_DELAY_SECONDS = 1 + + +def auto_start_quality_scan(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + automation_id = config.get('_automation_id') + state = deps.get_quality_scanner_state() + if state.get('status') == 'running': + return {'status': 'skipped', 'reason': 'Quality scan already running'} + + scope = config.get('scope', 'watchlist') + # Pre-set status before submit so the polling loop doesn't see a + # stale 'finished' from a previous run. + with deps.quality_scanner_lock: + state['status'] = 'running' + deps.quality_scanner_executor.submit(deps.run_quality_scanner, scope, deps.get_current_profile_id()) + deps.update_progress( + automation_id, log_line=f'Quality scan started (scope: {scope})', log_type='info', + ) + + # Monitor progress (max 2 hours). + time.sleep(_INITIAL_DELAY_SECONDS) + poll_start = time.time() + while time.time() - poll_start < _TIMEOUT_SECONDS: + time.sleep(_POLL_INTERVAL_SECONDS) + current_status = state.get('status', 'idle') + if current_status not in ('running',): + break + deps.update_progress( + automation_id, + phase=state.get('phase', 'Scanning...'), + progress=state.get('progress', 0), + processed=state.get('processed', 0), + total=state.get('total', 0), + ) + else: + deps.update_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} + + final_status = state.get('status', 'idle') + if final_status == 'error': + err = state.get('error_message', 'Unknown error') + deps.update_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 = state.get('low_quality', 0) + deps.update_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, + 'tracks_scanned': state.get('processed', 0), + 'quality_met': state.get('quality_met', 0), + 'low_quality': issues, + 'matched': state.get('matched', 0), + } diff --git a/core/automation/handlers/registration.py b/core/automation/handlers/registration.py index 80611a57..7044b75a 100644 --- a/core/automation/handlers/registration.py +++ b/core/automation/handlers/registration.py @@ -15,6 +15,25 @@ 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.database_update import ( + auto_start_database_update, auto_deep_scan_library, +) +from core.automation.handlers.duplicate_cleaner import auto_run_duplicate_cleaner +from core.automation.handlers.quality_scanner import auto_start_quality_scan +from core.automation.handlers.maintenance import ( + auto_clear_quarantine, + auto_cleanup_wishlist, + auto_update_discovery_pool, + auto_backup_database, + auto_refresh_beatport_cache, +) +from core.automation.handlers.download_cleanup import ( + auto_clean_search_history, + auto_clean_completed_downloads, + auto_full_cleanup, +) +from core.automation.handlers.run_script import auto_run_script +from core.automation.handlers.search_and_download import auto_search_and_download def register_all(deps: AutomationDeps) -> None: @@ -69,3 +88,66 @@ def register_all(deps: AutomationDeps) -> None: lambda config: auto_playlist_pipeline(config, deps), deps.state.is_pipeline_running, ) + + # Database update + deep scan share the db_update_state guard — + # only one operation can mutate that state at a time. + engine.register_action_handler( + 'start_database_update', + lambda config: auto_start_database_update(config, deps), + lambda: deps.get_db_update_state().get('status') == 'running', + ) + engine.register_action_handler( + 'deep_scan_library', + lambda config: auto_deep_scan_library(config, deps), + lambda: deps.get_db_update_state().get('status') == 'running', + ) + engine.register_action_handler( + 'run_duplicate_cleaner', + lambda config: auto_run_duplicate_cleaner(config, deps), + lambda: deps.get_duplicate_cleaner_state().get('status') == 'running', + ) + engine.register_action_handler( + 'clear_quarantine', + lambda config: auto_clear_quarantine(config, deps), + ) + engine.register_action_handler( + 'cleanup_wishlist', + lambda config: auto_cleanup_wishlist(config, deps), + ) + engine.register_action_handler( + 'update_discovery_pool', + lambda config: auto_update_discovery_pool(config, deps), + ) + engine.register_action_handler( + 'start_quality_scan', + lambda config: auto_start_quality_scan(config, deps), + lambda: deps.get_quality_scanner_state().get('status') == 'running', + ) + engine.register_action_handler( + 'backup_database', + lambda config: auto_backup_database(config, deps), + ) + engine.register_action_handler( + 'refresh_beatport_cache', + lambda config: auto_refresh_beatport_cache(config, deps), + ) + engine.register_action_handler( + 'clean_search_history', + lambda config: auto_clean_search_history(config, deps), + ) + engine.register_action_handler( + 'clean_completed_downloads', + lambda config: auto_clean_completed_downloads(config, deps), + ) + engine.register_action_handler( + 'full_cleanup', + lambda config: auto_full_cleanup(config, deps), + ) + engine.register_action_handler( + 'run_script', + lambda config: auto_run_script(config, deps), + ) + engine.register_action_handler( + 'search_and_download', + lambda config: auto_search_and_download(config, deps), + ) diff --git a/core/automation/handlers/run_script.py b/core/automation/handlers/run_script.py new file mode 100644 index 00000000..2d10a6fb --- /dev/null +++ b/core/automation/handlers/run_script.py @@ -0,0 +1,103 @@ +"""Automation handler: ``run_script`` action. + +Lifted from ``web_server._register_automation_handlers`` (the +``_auto_run_script`` closure). Runs a user-provided shell or Python +script from the configured scripts directory with bounded timeout + +captured stdout/stderr. Path-traversal guard ensures users can't +escape the scripts directory. + +Environment variables exposed to the script: +- ``SOULSYNC_EVENT``: triggering event type (when fired by an event) +- ``SOULSYNC_AUTOMATION``: automation name +- ``SOULSYNC_SCRIPTS_DIR``: absolute path to the scripts dir +""" + +from __future__ import annotations + +import os +import subprocess as _sp +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +_MAX_TIMEOUT_SECONDS = 300 # Hard cap on user-supplied timeout config. + + +def auto_run_script(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + script_name = config.get('script_name', '') + timeout = min(int(config.get('timeout', 60)), _MAX_TIMEOUT_SECONDS) + automation_id = config.get('_automation_id') + + if not script_name: + return {'status': 'error', 'error': 'No script selected'} + + scripts_dir = deps.docker_resolve_path(deps.config_manager.get('scripts.path', './scripts')) + if not scripts_dir or not os.path.isdir(scripts_dir): + os.makedirs(scripts_dir, exist_ok=True) + return { + 'status': 'error', + 'error': 'Scripts directory is empty. Add scripts to the scripts/ folder.', + } + + script_path = os.path.join(scripts_dir, script_name) + script_path = os.path.realpath(script_path) + + # Security: block path traversal — script must resolve under + # the scripts dir, no symlinks/.. tricks allowed out. + if not script_path.startswith(os.path.realpath(scripts_dir)): + return {'status': 'error', 'error': 'Script path traversal blocked'} + + if not os.path.isfile(script_path): + return {'status': 'error', 'error': f'Script not found: {script_name}'} + + deps.update_progress(automation_id, phase=f'Running {script_name}...', progress=10) + + # Build environment with SoulSync context. + env = os.environ.copy() + event_data = config.get('_event_data') or {} + env['SOULSYNC_EVENT'] = str(event_data.get('type', '')) + env['SOULSYNC_AUTOMATION'] = config.get('_automation_name', '') + env['SOULSYNC_SCRIPTS_DIR'] = scripts_dir + + try: + # Determine how to run the script. + if script_path.endswith('.py'): + cmd = ['python', script_path] + elif script_path.endswith('.sh'): + cmd = ['bash', script_path] + else: + cmd = [script_path] + + result = _sp.run( + cmd, + capture_output=True, text=True, timeout=timeout, + cwd=scripts_dir, env=env, + ) + + deps.update_progress(automation_id, phase='Script completed', progress=100) + + stdout = result.stdout[:2000] if result.stdout else '' + stderr = result.stderr[:1000] if result.stderr else '' + + if result.returncode == 0: + deps.logger.info(f"Script '{script_name}' completed (exit 0)") + else: + deps.logger.warning(f"Script '{script_name}' exited with code {result.returncode}") + + return { + 'status': 'completed' if result.returncode == 0 else 'error', + 'exit_code': str(result.returncode), + 'stdout': stdout, + 'stderr': stderr, + 'script': script_name, + } + except _sp.TimeoutExpired: + deps.update_progress(automation_id, phase='Script timed out', progress=100) + return { + 'status': 'error', + 'error': f'Script timed out after {timeout}s', + 'script': script_name, + } + except Exception as e: # noqa: BLE001 — automation handlers must never raise + return {'status': 'error', 'error': str(e), 'script': script_name} diff --git a/core/automation/handlers/search_and_download.py b/core/automation/handlers/search_and_download.py new file mode 100644 index 00000000..fa49c534 --- /dev/null +++ b/core/automation/handlers/search_and_download.py @@ -0,0 +1,57 @@ +"""Automation handler: ``search_and_download`` action. + +Lifted from ``web_server._register_automation_handlers`` (the +``_auto_search_and_download`` closure). Searches for a track by +name/artist string and dispatches the best match through the +download orchestrator. Query can come from the trigger config +(direct value) or from event data (e.g. webhook payload). +""" + +from __future__ import annotations + +from typing import Any, Dict + +from core.automation.deps import AutomationDeps + + +def auto_search_and_download(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]: + automation_id = config.get('_automation_id') + query = config.get('query', '').strip() + # Event-triggered: pull query from event data (e.g. webhook_received). + if not query: + event_data = config.get('_event_data', {}) + query = (event_data.get('query', '') or '').strip() + if not query: + if automation_id: + deps.update_progress( + automation_id, log_line='No search query provided', log_type='error', + ) + return {'status': 'error', 'error': 'No search query provided'} + try: + if automation_id: + deps.update_progress( + automation_id, phase='Searching', + log_line=f'Searching: {query}', log_type='info', + ) + result = deps.run_async(deps.download_orchestrator.search_and_download_best(query)) + if result: + if automation_id: + deps.update_progress( + automation_id, + log_line=f'Download started for: {query}', + log_type='success', + ) + return {'status': 'completed', 'query': query, 'download_id': result} + if automation_id: + deps.update_progress( + automation_id, + log_line=f'No match found for: {query}', + log_type='warning', + ) + return {'status': 'not_found', 'query': query, 'error': 'No match found'} + except Exception as e: # noqa: BLE001 — automation handlers must never raise + if automation_id: + deps.update_progress( + automation_id, log_line=f'Error: {e}', log_type='error', + ) + return {'status': 'error', 'query': query, 'error': str(e)} diff --git a/tests/automation/test_handlers_maintenance.py b/tests/automation/test_handlers_maintenance.py new file mode 100644 index 00000000..280ee3d6 --- /dev/null +++ b/tests/automation/test_handlers_maintenance.py @@ -0,0 +1,390 @@ +"""Boundary tests for the maintenance + misc automation handlers +(database_update / deep_scan_library / duplicate_cleaner / +quality_scanner / clear_quarantine / cleanup_wishlist / +update_discovery_pool / backup_database / refresh_beatport_cache / +clean_search_history / clean_completed_downloads / full_cleanup / +run_script / search_and_download). + +Each handler is tested as a pure function via stub deps. The bodies +are mechanical lifts of the closures that used to live in +``web_server._register_automation_handlers`` — these tests pin the +seam (deps wiring, exception swallow contract, return shapes, +guard short-circuits) so future drift fails here, not at runtime +against real executors / clients.""" + +from __future__ import annotations + +import os +import threading +from typing import Any, Dict, List + +import pytest + +from core.automation.deps import AutomationDeps, AutomationState +from core.automation.handlers.database_update import ( + auto_start_database_update, auto_deep_scan_library, +) +from core.automation.handlers.duplicate_cleaner import auto_run_duplicate_cleaner +from core.automation.handlers.quality_scanner import auto_start_quality_scan +from core.automation.handlers.maintenance import ( + auto_clear_quarantine, auto_cleanup_wishlist, + auto_update_discovery_pool, auto_backup_database, +) +from core.automation.handlers.download_cleanup import ( + auto_clean_search_history, auto_clean_completed_downloads, +) +from core.automation.handlers.run_script import auto_run_script +from core.automation.handlers.search_and_download import auto_search_and_download + + +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 + + +class _StubConfig: + def __init__(self, values=None): + self._values = values or {} + + def get(self, key, default=None): + return self._values.get(key, default) + + def get_active_media_server(self): + return 'plex' + + +def _build_deps(**overrides) -> AutomationDeps: + defaults = dict( + engine=object(), + state=AutomationState(), + config_manager=_StubConfig(), + update_progress=lambda *a, **k: None, + logger=_StubLogger(), + get_database=lambda: object(), + 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: {}, + set_db_update_automation_id=lambda v: None, + get_db_update_state=lambda: {}, + db_update_lock=threading.Lock(), + db_update_executor=None, + run_db_update_task=lambda *a, **k: None, + run_deep_scan_task=lambda *a, **k: None, + get_duplicate_cleaner_state=lambda: {}, + duplicate_cleaner_lock=threading.Lock(), + duplicate_cleaner_executor=None, + run_duplicate_cleaner=lambda: None, + get_quality_scanner_state=lambda: {}, + quality_scanner_lock=threading.Lock(), + quality_scanner_executor=None, + run_quality_scanner=lambda *a, **k: None, + download_orchestrator=None, + run_async=lambda coro: None, + tasks_lock=threading.Lock(), + get_download_batches=lambda: {}, + get_download_tasks=lambda: {}, + sweep_empty_download_directories=lambda: 0, + get_staging_path=lambda: '/staging', + docker_resolve_path=lambda p: p, + get_current_profile_id=lambda: 1, + get_watchlist_scanner=lambda spc: None, + get_app=lambda: None, + get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}}, + ) + defaults.update(overrides) + return AutomationDeps(**defaults) # type: ignore[arg-type] + + +# ─── database_update / deep_scan ────────────────────────────────────── + + +class _StubExecutor: + def __init__(self): + self.submits: List[tuple] = [] + + def submit(self, fn, *args, **kwargs): + self.submits.append((fn, args, kwargs)) + + +class TestDatabaseUpdate: + def test_already_running_returns_skipped(self): + state = {'status': 'running'} + deps = _build_deps(get_db_update_state=lambda: state) + result = auto_start_database_update({}, deps) + assert result == {'status': 'skipped', 'reason': 'Database update already running'} + + def test_set_db_update_automation_id_called(self): + # Handler must propagate the automation id through the deps + # setter so the legacy global stays in sync. + captured: List[Any] = [] + state = {'status': 'idle'} + # Make the polling loop terminate immediately by flipping + # the status as soon as it's set to 'running'. + executor = _StubExecutor() + + def fake_task(*_a, **_k): + state['status'] = 'finished' + + executor.submit = lambda fn, *a, **k: fake_task() + deps = _build_deps( + get_db_update_state=lambda: state, + db_update_executor=executor, + set_db_update_automation_id=lambda v: captured.append(v), + run_db_update_task=fake_task, + ) + # Replace time.sleep so we don't actually wait + import core.automation.handlers.database_update as module + original = module.time.sleep + module.time.sleep = lambda _: None + try: + result = auto_start_database_update({'_automation_id': 'auto-1'}, deps) + finally: + module.time.sleep = original + assert captured == ['auto-1'] + assert result['status'] == 'completed' + + +class TestDeepScan: + def test_already_running_returns_skipped(self): + state = {'status': 'running'} + deps = _build_deps(get_db_update_state=lambda: state) + result = auto_deep_scan_library({}, deps) + assert result == {'status': 'skipped', 'reason': 'Database update already running'} + + +# ─── duplicate_cleaner ──────────────────────────────────────────────── + + +class TestDuplicateCleaner: + def test_already_running_returns_skipped(self): + state = {'status': 'running'} + deps = _build_deps(get_duplicate_cleaner_state=lambda: state) + result = auto_run_duplicate_cleaner({}, deps) + assert result == {'status': 'skipped', 'reason': 'Duplicate cleaner already running'} + + +# ─── quality_scanner ────────────────────────────────────────────────── + + +class TestQualityScanner: + def test_already_running_returns_skipped(self): + state = {'status': 'running'} + deps = _build_deps(get_quality_scanner_state=lambda: state) + result = auto_start_quality_scan({}, deps) + assert result == {'status': 'skipped', 'reason': 'Quality scan already running'} + + +# ─── clear_quarantine ──────────────────────────────────────────────── + + +class TestClearQuarantine: + def test_no_quarantine_folder_returns_zero(self, tmp_path): + # Point at a non-existent path. + deps = _build_deps( + config_manager=_StubConfig({'soulseek.download_path': str(tmp_path / 'nonexistent')}), + docker_resolve_path=lambda p: p, + ) + result = auto_clear_quarantine({}, deps) + assert result == {'status': 'completed', 'removed': '0'} + + def test_clears_files_from_quarantine(self, tmp_path): + download_path = tmp_path + quarantine_path = download_path / 'ss_quarantine' + quarantine_path.mkdir() + (quarantine_path / 'a.flac').write_bytes(b'') + (quarantine_path / 'b.flac').write_bytes(b'') + deps = _build_deps( + config_manager=_StubConfig({'soulseek.download_path': str(download_path)}), + docker_resolve_path=lambda p: p, + ) + result = auto_clear_quarantine({}, deps) + assert result == {'status': 'completed', 'removed': '2'} + + +# ─── cleanup_wishlist ──────────────────────────────────────────────── + + +class TestCleanupWishlist: + def test_returns_count_from_db(self): + class _DB: + def remove_wishlist_duplicates(self, profile_id): + assert profile_id == 1 + return 7 + + deps = _build_deps(get_database=lambda: _DB()) + result = auto_cleanup_wishlist({}, deps) + assert result == {'status': 'completed', 'removed': '7'} + + def test_returns_zero_when_db_returns_falsey(self): + class _DB: + def remove_wishlist_duplicates(self, profile_id): + return None + + deps = _build_deps(get_database=lambda: _DB()) + result = auto_cleanup_wishlist({}, deps) + assert result == {'status': 'completed', 'removed': '0'} + + +# ─── update_discovery_pool ─────────────────────────────────────────── + + +class TestUpdateDiscoveryPool: + def test_success(self): + called: List[Any] = [] + + class _Scanner: + def update_discovery_pool_incremental(self, profile_id): + called.append(profile_id) + + deps = _build_deps( + get_watchlist_scanner=lambda spc: _Scanner(), + get_current_profile_id=lambda: 7, + ) + result = auto_update_discovery_pool({}, deps) + assert result['status'] == 'completed' + assert result['_manages_own_progress'] is True + assert called == [7] + + def test_exception_swallowed(self): + def boom(_): + raise RuntimeError('scanner down') + + deps = _build_deps(get_watchlist_scanner=boom) + result = auto_update_discovery_pool({}, deps) + assert result['status'] == 'error' + assert 'scanner down' in result['reason'] + + +# ─── backup_database ───────────────────────────────────────────────── + + +class TestBackupDatabase: + def test_missing_database_returns_error(self, tmp_path, monkeypatch): + monkeypatch.setenv('DATABASE_PATH', str(tmp_path / 'no.db')) + deps = _build_deps() + result = auto_backup_database({}, deps) + assert result == {'status': 'error', 'reason': 'Database file not found'} + + +# ─── clean_search_history ──────────────────────────────────────────── + + +class TestCleanSearchHistory: + def test_soulseek_inactive_skips(self): + deps = _build_deps( + config_manager=_StubConfig({'download_source.mode': 'tidal'}), + ) + result = auto_clean_search_history({}, deps) + assert result == {'status': 'skipped'} + + def test_no_orchestrator_skips(self): + deps = _build_deps( + config_manager=_StubConfig({'download_source.mode': 'soulseek'}), + download_orchestrator=None, + ) + result = auto_clean_search_history({}, deps) + assert result == {'status': 'skipped'} + + +# ─── clean_completed_downloads ─────────────────────────────────────── + + +class TestCleanCompletedDownloads: + def test_active_batches_skip(self): + # Active batch present → handler returns 'completed' without doing anything. + deps = _build_deps( + get_download_batches=lambda: {'b1': {'phase': 'downloading'}}, + get_download_tasks=lambda: {}, + ) + result = auto_clean_completed_downloads({}, deps) + assert result == {'status': 'completed'} + + +# ─── run_script ────────────────────────────────────────────────────── + + +class TestRunScript: + def test_no_script_name_returns_error(self): + deps = _build_deps() + result = auto_run_script({}, deps) + assert result == {'status': 'error', 'error': 'No script selected'} + + def test_path_traversal_blocked(self, tmp_path): + scripts_dir = tmp_path / 'scripts' + scripts_dir.mkdir() + # Place a script OUTSIDE the scripts dir + try to reach it + # via ../ traversal. + evil = tmp_path / 'evil.sh' + evil.write_text('#!/bin/bash\necho evil') + deps = _build_deps( + config_manager=_StubConfig({'scripts.path': str(scripts_dir)}), + docker_resolve_path=lambda p: p, + ) + result = auto_run_script({'script_name': '../evil.sh'}, deps) + assert result['status'] == 'error' + assert 'path traversal' in result['error'].lower() + + def test_missing_script_returns_error(self, tmp_path): + scripts_dir = tmp_path / 'scripts' + scripts_dir.mkdir() + deps = _build_deps( + config_manager=_StubConfig({'scripts.path': str(scripts_dir)}), + ) + result = auto_run_script({'script_name': 'no_such_script.sh'}, deps) + assert result['status'] == 'error' + assert 'not found' in result['error'].lower() + + +# ─── search_and_download ───────────────────────────────────────────── + + +class TestSearchAndDownload: + def test_no_query_returns_error(self): + deps = _build_deps() + result = auto_search_and_download({}, deps) + assert result == {'status': 'error', 'error': 'No search query provided'} + + def test_query_from_event_data_used(self): + captured_queries: List[str] = [] + + class _Orchestrator: + async def search_and_download_best(self, q): + captured_queries.append(q) + return 'dl-id-123' + + deps = _build_deps( + download_orchestrator=_Orchestrator(), + run_async=lambda coro: 'dl-id-123', + ) + result = auto_search_and_download( + {'_event_data': {'query': 'Adele Hello'}}, deps, + ) + assert result['status'] == 'completed' + assert result['query'] == 'Adele Hello' + assert result['download_id'] == 'dl-id-123' + + def test_no_match_returns_not_found(self): + class _Orchestrator: + async def search_and_download_best(self, q): + return None + + deps = _build_deps( + download_orchestrator=_Orchestrator(), + run_async=lambda coro: None, + ) + result = auto_search_and_download({'query': 'xyz'}, deps) + assert result['status'] == 'not_found' + assert result['query'] == 'xyz' diff --git a/tests/automation/test_handlers_playlist.py b/tests/automation/test_handlers_playlist.py index 83c2596b..b831b6fd 100644 --- a/tests/automation/test_handlers_playlist.py +++ b/tests/automation/test_handlers_playlist.py @@ -94,6 +94,32 @@ def _build_deps(**overrides) -> AutomationDeps: get_deezer_client=lambda: None, parse_youtube_playlist=lambda url: None, get_sync_states=lambda: {}, + set_db_update_automation_id=lambda v: None, + get_db_update_state=lambda: {}, + db_update_lock=threading.Lock(), + db_update_executor=None, + run_db_update_task=lambda *a, **k: None, + run_deep_scan_task=lambda *a, **k: None, + get_duplicate_cleaner_state=lambda: {}, + duplicate_cleaner_lock=threading.Lock(), + duplicate_cleaner_executor=None, + run_duplicate_cleaner=lambda: None, + get_quality_scanner_state=lambda: {}, + quality_scanner_lock=threading.Lock(), + quality_scanner_executor=None, + run_quality_scanner=lambda *a, **k: None, + download_orchestrator=None, + run_async=lambda coro: None, + tasks_lock=threading.Lock(), + get_download_batches=lambda: {}, + get_download_tasks=lambda: {}, + sweep_empty_download_directories=lambda: 0, + get_staging_path=lambda: '/staging', + docker_resolve_path=lambda p: p, + get_current_profile_id=lambda: 1, + get_watchlist_scanner=lambda spc: None, + get_app=lambda: None, + get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}}, ) defaults.update(overrides) return AutomationDeps(**defaults) # type: ignore[arg-type] diff --git a/tests/automation/test_handlers_simple.py b/tests/automation/test_handlers_simple.py index a8aa8834..30b220ad 100644 --- a/tests/automation/test_handlers_simple.py +++ b/tests/automation/test_handlers_simple.py @@ -64,6 +64,32 @@ def _build_deps(**overrides: Any) -> AutomationDeps: get_deezer_client=lambda: None, parse_youtube_playlist=lambda url: None, get_sync_states=lambda: {}, + set_db_update_automation_id=lambda v: None, + get_db_update_state=lambda: {}, + db_update_lock=threading.Lock(), + db_update_executor=None, + run_db_update_task=lambda *a, **k: None, + run_deep_scan_task=lambda *a, **k: None, + get_duplicate_cleaner_state=lambda: {}, + duplicate_cleaner_lock=threading.Lock(), + duplicate_cleaner_executor=None, + run_duplicate_cleaner=lambda: None, + get_quality_scanner_state=lambda: {}, + quality_scanner_lock=threading.Lock(), + quality_scanner_executor=None, + run_quality_scanner=lambda *a, **k: None, + download_orchestrator=None, + run_async=lambda coro: None, + tasks_lock=threading.Lock(), + get_download_batches=lambda: {}, + get_download_tasks=lambda: {}, + sweep_empty_download_directories=lambda: 0, + get_staging_path=lambda: '/staging', + docker_resolve_path=lambda p: p, + get_current_profile_id=lambda: 1, + get_watchlist_scanner=lambda spc: None, + get_app=lambda: None, + get_beatport_data_cache=lambda: {'cache_lock': threading.Lock(), 'homepage': {}}, ) defaults.update(overrides) return AutomationDeps(**defaults) # type: ignore[arg-type] diff --git a/web_server.py b/web_server.py index c0511e8b..d5965540 100644 --- a/web_server.py +++ b/web_server.py @@ -767,6 +767,15 @@ db_update_state = { _db_update_automation_id = None # Set when automation triggers DB update, used by callbacks db_update_lock = threading.Lock() + +def _set_db_update_automation_id(value): + """Setter exposed to extracted automation handlers — keeps the + legacy `_db_update_automation_id` global in sync so the live + DB-update progress callbacks below (which still read the global + directly) emit against the right automation card.""" + global _db_update_automation_id + _db_update_automation_id = value + # Quality Scanner state quality_scanner_state = { "status": "idle", # idle, running, finished, error @@ -932,6 +941,8 @@ def _register_automation_handlers(): # work; they'll be removed once the rest of the lift is done. _automation_state = AutomationState() + from core.watchlist_scanner import get_watchlist_scanner as _get_watchlist_scanner_fn + _automation_deps = AutomationDeps( engine=automation_engine, state=_automation_state, @@ -953,679 +964,38 @@ def _register_automation_handlers(): get_deezer_client=_get_deezer_client, parse_youtube_playlist=parse_youtube_playlist, get_sync_states=lambda: sync_states, + set_db_update_automation_id=_set_db_update_automation_id, + get_db_update_state=lambda: db_update_state, + db_update_lock=db_update_lock, + db_update_executor=db_update_executor, + run_db_update_task=_run_db_update_task, + run_deep_scan_task=_run_deep_scan_task, + get_duplicate_cleaner_state=lambda: duplicate_cleaner_state, + duplicate_cleaner_lock=duplicate_cleaner_lock, + duplicate_cleaner_executor=duplicate_cleaner_executor, + run_duplicate_cleaner=_run_duplicate_cleaner, + get_quality_scanner_state=lambda: quality_scanner_state, + quality_scanner_lock=quality_scanner_lock, + quality_scanner_executor=quality_scanner_executor, + run_quality_scanner=_run_quality_scanner, + download_orchestrator=download_orchestrator, + run_async=run_async, + tasks_lock=tasks_lock, + get_download_batches=lambda: download_batches, + get_download_tasks=lambda: download_tasks, + sweep_empty_download_directories=_sweep_empty_download_directories, + get_staging_path=get_staging_path, + docker_resolve_path=docker_resolve_path, + get_current_profile_id=get_current_profile_id, + get_watchlist_scanner=_get_watchlist_scanner_fn, + get_app=lambda: app, + get_beatport_data_cache=lambda: beatport_data_cache, ) _register_extracted_handlers(_automation_deps) # --- Phase 3 action 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 = automation_id - full = config.get('full_refresh', False) - active_server = config_manager.get_active_media_server() - with db_update_lock: - db_update_state.update({ - "status": "running", "phase": "Initializing...", - "progress": 0, "current_item": "", "processed": 0, "total": 0, "error_message": "" - }) - db_update_executor.submit(_run_db_update_task, full, active_server) - - # 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} - with db_update_lock: - stats = { - 'status': 'completed', 'full_refresh': str(full), '_manages_own_progress': True, - 'artists': db_update_state.get('total', 0), - 'albums': db_update_state.get('total_albums', 0), - 'tracks': db_update_state.get('total_tracks', 0), - 'removed_artists': db_update_state.get('removed_artists', 0), - 'removed_albums': db_update_state.get('removed_albums', 0), - 'removed_tracks': db_update_state.get('removed_tracks', 0), - } - return stats - - def _auto_deep_scan_library(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 = automation_id - active_server = config_manager.get_active_media_server() - with db_update_lock: - db_update_state.update({ - "status": "running", "phase": "Deep scan: Initializing...", - "progress": 0, "current_item": "", "processed": 0, "total": 0, "error_message": "" - }) - db_update_executor.submit(_run_deep_scan_task, active_server) - - # Monitor 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 - 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='Deep scan appears stalled — waiting...', log_type='warning') - last_progress_time = time.time() - else: - _update_automation_progress(automation_id, status='error', - phase='Timed out', log_line='Deep scan timed out after 2 hours', log_type='error') - return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True} - - 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} - with db_update_lock: - stats = { - 'status': 'completed', '_manages_own_progress': True, - 'artists': db_update_state.get('total', 0), - 'albums': db_update_state.get('total_albums', 0), - 'tracks': db_update_state.get('total_tracks', 0), - 'removed_artists': db_update_state.get('removed_artists', 0), - 'removed_albums': db_update_state.get('removed_albums', 0), - 'removed_tracks': db_update_state.get('removed_tracks', 0), - } - return stats - - 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) - _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) - space_freed = duplicate_cleaner_state.get('space_freed', 0) - scanned = duplicate_cleaner_state.get('files_scanned', 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, - 'files_scanned': scanned, - 'duplicates_found': dupes, - 'files_deleted': removed, - 'space_freed_mb': round(space_freed / (1024 * 1024), 1), - } - - 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): - fp = os.path.join(quarantine_path, f) - try: - if os.path.isfile(fp): - os.remove(fp) - removed += 1 - elif os.path.isdir(fp): - _shutil.rmtree(fp) - removed += 1 - except Exception as e: - logger.debug("quarantine entry purge failed: %s", e) - _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) - _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: - _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()) - _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, - 'tracks_scanned': quality_scanner_state.get('processed', 0), - 'quality_met': quality_scanner_state.get('quality_met', 0), - 'low_quality': issues, - 'matched': quality_scanner_state.get('matched', 0), - } - - 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'} - max_backups = 5 - timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') - backup_path = f"{db_path}.backup_{timestamp}" - # Use SQLite backup API for safe hot-copy of active database - src = sqlite3.connect(db_path) - dst = sqlite3.connect(backup_path) - src.backup(dst) - dst.close() - src.close() - size_mb = round(os.path.getsize(backup_path) / (1024 * 1024), 1) - # Rolling cleanup — keep only the newest N backups - existing = sorted(_glob.glob(f"{db_path}.backup_*"), key=os.path.getmtime) - while len(existing) > max_backups: - try: - os.remove(existing.pop(0)) - except Exception as e: - logger.debug("rolling backup cleanup failed: %s", e) - _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)} - - def _auto_refresh_beatport_cache(config): - """Refresh Beatport homepage cache by calling each endpoint internally.""" - automation_id = config.get('_automation_id') - sections = [ - ('hero_tracks', '/api/beatport/hero-tracks', 'Hero Tracks'), - ('new_releases', '/api/beatport/new-releases', 'New Releases'), - ('featured_charts', '/api/beatport/featured-charts', 'Featured Charts'), - ('dj_charts', '/api/beatport/dj-charts', 'DJ Charts'), - ('top_10_lists', '/api/beatport/homepage/top-10-lists', 'Top 10 Lists'), - ('top_10_releases', '/api/beatport/homepage/top-10-releases-cards', 'Top 10 Releases'), - ('hype_picks', '/api/beatport/hype-picks', 'Hype Picks'), - ] - # Invalidate all homepage cache timestamps so endpoints re-scrape - with beatport_data_cache['cache_lock']: - for key in beatport_data_cache['homepage']: - beatport_data_cache['homepage'][key]['timestamp'] = 0 - beatport_data_cache['homepage'][key]['data'] = None - - refreshed = 0 - errors = [] - with app.test_client() as client: - for idx, (_, endpoint, label) in enumerate(sections): - _update_automation_progress(automation_id, - progress=(idx / len(sections)) * 100, - phase=f'Scraping: {label}', - current_item=label) - try: - resp = client.get(endpoint) - if resp.status_code == 200: - refreshed += 1 - _update_automation_progress(automation_id, - log_line=f'{label}: cached', log_type='success') - else: - errors.append(label) - _update_automation_progress(automation_id, - log_line=f'{label}: HTTP {resp.status_code}', log_type='error') - except Exception as e: - errors.append(label) - _update_automation_progress(automation_id, - log_line=f'{label}: {str(e)}', log_type='error') - if idx < len(sections) - 1: - time.sleep(2) - - _update_automation_progress(automation_id, status='finished', progress=100, - phase='Complete', - log_line=f'Refreshed {refreshed}/{len(sections)} sections', log_type='success') - return {'status': 'completed', 'refreshed': str(refreshed), 'errors': str(len(errors)), - '_manages_own_progress': True} - - def _auto_clean_search_history(config): - """Remove old searches from Soulseek.""" - automation_id = config.get('_automation_id') - # Skip if soulseek is not the active download source or in hybrid order - dl_mode = config_manager.get('download_source.mode', 'hybrid') - hybrid_order = config_manager.get('download_source.hybrid_order', ['hifi', 'youtube', 'soulseek']) - soulseek_active = (dl_mode == 'soulseek' or - (dl_mode == 'hybrid' and 'soulseek' in hybrid_order)) - # Reach the underlying SoulseekClient via the orchestrator's - # generic accessor. - slskd = download_orchestrator.client('soulseek') if download_orchestrator else None - if not soulseek_active or not slskd or not slskd.base_url: - _update_automation_progress(automation_id, - log_line='Soulseek not active — skipped', log_type='skip') - return {'status': 'skipped'} - if not config_manager.get('soulseek.auto_clear_searches', True): - _update_automation_progress(automation_id, - log_line='Auto-clear disabled in settings', log_type='skip') - return {'status': 'skipped'} - try: - success = run_async(download_orchestrator.maintain_search_history_with_buffer( - keep_searches=50, trigger_threshold=200 - )) - if success: - _update_automation_progress(automation_id, - log_line='Search history maintenance completed', log_type='success') - return {'status': 'completed'} - else: - _update_automation_progress(automation_id, - log_line='No cleanup needed', log_type='skip') - return {'status': 'completed'} - except Exception as e: - return {'status': 'error', 'error': str(e)} - - def _auto_clean_completed_downloads(config): - """Clear completed downloads and empty directories.""" - automation_id = config.get('_automation_id') - try: - has_active_batches = False - has_post_processing = False - with tasks_lock: - for batch_data in download_batches.values(): - if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]: - has_active_batches = True - break - if not has_active_batches: - for task_data in download_tasks.values(): - if task_data.get('status') == 'post_processing': - has_post_processing = True - break - - if has_active_batches: - _update_automation_progress(automation_id, - log_line='Skipped — downloads active', log_type='skip') - return {'status': 'completed'} - - run_async(download_orchestrator.clear_all_completed_downloads()) - if not has_post_processing: - _sweep_empty_download_directories() - _update_automation_progress(automation_id, - log_line='Download cleanup completed', log_type='success') - return {'status': 'completed'} - except Exception as e: - return {'status': 'error', 'reason': str(e)} - - def _auto_full_cleanup(config): - """Run all cleanup tasks: quarantine, download queue, empty dirs, staging, search history.""" - import shutil as _shutil - automation_id = config.get('_automation_id') - steps = [] - - # --- 1. Clear quarantine --- - _update_automation_progress(automation_id, phase='Clearing quarantine...', progress=0) - quarantine_path = os.path.join(docker_resolve_path(config_manager.get('soulseek.download_path', './downloads')), 'ss_quarantine') - q_removed = 0 - if os.path.exists(quarantine_path): - for f in os.listdir(quarantine_path): - fp = os.path.join(quarantine_path, f) - try: - if os.path.isfile(fp): - os.remove(fp) - q_removed += 1 - elif os.path.isdir(fp): - _shutil.rmtree(fp) - q_removed += 1 - except Exception as e: - logger.debug("quarantine entry purge failed: %s", e) - steps.append(f'Quarantine: removed {q_removed} items') - _update_automation_progress(automation_id, - log_line=f'Quarantine: removed {q_removed} items', log_type='success' if q_removed else 'info') - - # --- 2. Clear completed/errored/cancelled downloads from Soulseek queue --- - _update_automation_progress(automation_id, phase='Clearing download queue...', progress=20) - has_active_batches = False - has_post_processing = False - with tasks_lock: - for batch_data in download_batches.values(): - if batch_data.get('phase') not in ['complete', 'error', 'cancelled', None]: - has_active_batches = True - break - if not has_active_batches: - for task_data in download_tasks.values(): - if task_data.get('status') == 'post_processing': - has_post_processing = True - break - if has_active_batches: - steps.append('Download queue: skipped (active batches)') - _update_automation_progress(automation_id, - log_line='Download queue: skipped (active batches)', log_type='skip') - else: - try: - run_async(download_orchestrator.clear_all_completed_downloads()) - steps.append('Download queue: cleared') - _update_automation_progress(automation_id, - log_line='Download queue: cleared', log_type='success') - except Exception as e: - steps.append(f'Download queue: error ({e})') - _update_automation_progress(automation_id, - log_line=f'Download queue: error ({e})', log_type='error') - - # --- 3. Sweep empty download directories --- - _update_automation_progress(automation_id, phase='Sweeping empty directories...', progress=40) - if has_active_batches or has_post_processing: - reason = 'active batches' if has_active_batches else 'post-processing active' - steps.append(f'Empty directories: skipped ({reason})') - _update_automation_progress(automation_id, - log_line=f'Empty directories: skipped ({reason})', log_type='skip') - else: - dirs_removed = _sweep_empty_download_directories() - steps.append(f'Empty directories: removed {dirs_removed}') - _update_automation_progress(automation_id, - log_line=f'Empty directories: removed {dirs_removed}', log_type='success' if dirs_removed else 'info') - - # --- 4. Sweep empty staging directories --- - _update_automation_progress(automation_id, phase='Sweeping import folder...', progress=60) - staging_path = get_staging_path() - s_removed = 0 - if os.path.isdir(staging_path): - for dirpath, _dirnames, _filenames in os.walk(staging_path, topdown=False): - if os.path.normpath(dirpath) == os.path.normpath(staging_path): - continue - try: - entries = os.listdir(dirpath) - except OSError: - continue - visible = [e for e in entries if not e.startswith('.')] - if not visible: - for hidden in entries: - try: - os.remove(os.path.join(dirpath, hidden)) - except Exception as e: - logger.debug("hidden file cleanup failed: %s", e) - try: - os.rmdir(dirpath) - s_removed += 1 - except OSError: - pass - steps.append(f'Staging: removed {s_removed} empty directories') - _update_automation_progress(automation_id, - log_line=f'Staging: removed {s_removed} empty directories', log_type='success' if s_removed else 'info') - - # --- 5. Clean search history (if enabled) --- - _update_automation_progress(automation_id, phase='Cleaning search history...', progress=80) - try: - if not config_manager.get('soulseek.auto_clear_searches', True): - steps.append('Search cleanup: disabled in settings') - _update_automation_progress(automation_id, - log_line='Search cleanup: disabled in settings', log_type='skip') - else: - run_async(download_orchestrator.maintain_search_history_with_buffer( - keep_searches=50, trigger_threshold=200 - )) - steps.append('Search history: cleaned') - _update_automation_progress(automation_id, - log_line='Search history: cleaned', log_type='success') - except Exception as e: - steps.append(f'Search history: error ({e})') - _update_automation_progress(automation_id, - log_line=f'Search history: error ({e})', log_type='error') - - total_removed = q_removed + s_removed - _update_automation_progress(automation_id, status='finished', progress=100, - phase='Complete', - log_line=f'Full cleanup complete — {total_removed} items removed', log_type='success') - return { - 'status': 'completed', - 'quarantine_removed': str(q_removed), - 'staging_removed': str(s_removed), - 'total_removed': str(total_removed), - 'steps': steps, - '_manages_own_progress': True, - } - - def _auto_run_script(config): - """Execute a user script from the scripts directory.""" - import subprocess as _sp - script_name = config.get('script_name', '') - timeout = min(int(config.get('timeout', 60)), 300) - automation_id = config.get('_automation_id') - - if not script_name: - return {'status': 'error', 'error': 'No script selected'} - - scripts_dir = docker_resolve_path(config_manager.get('scripts.path', './scripts')) - if not scripts_dir or not os.path.isdir(scripts_dir): - os.makedirs(scripts_dir, exist_ok=True) - return {'status': 'error', 'error': 'Scripts directory is empty. Add scripts to the scripts/ folder.'} - - script_path = os.path.join(scripts_dir, script_name) - script_path = os.path.realpath(script_path) - - # Security: block path traversal - if not script_path.startswith(os.path.realpath(scripts_dir)): - return {'status': 'error', 'error': 'Script path traversal blocked'} - - if not os.path.isfile(script_path): - return {'status': 'error', 'error': f'Script not found: {script_name}'} - - _update_automation_progress(automation_id, phase=f'Running {script_name}...', progress=10) - - # Build environment with SoulSync context - env = os.environ.copy() - event_data = config.get('_event_data') or {} - env['SOULSYNC_EVENT'] = str(event_data.get('type', '')) - env['SOULSYNC_AUTOMATION'] = config.get('_automation_name', '') - env['SOULSYNC_SCRIPTS_DIR'] = scripts_dir - - try: - # Determine how to run the script - if script_path.endswith('.py'): - cmd = ['python', script_path] - elif script_path.endswith('.sh'): - cmd = ['bash', script_path] - else: - cmd = [script_path] - - result = _sp.run( - cmd, - capture_output=True, text=True, timeout=timeout, - cwd=scripts_dir, env=env - ) - - _update_automation_progress(automation_id, phase='Script completed', progress=100) - - stdout = result.stdout[:2000] if result.stdout else '' - stderr = result.stderr[:1000] if result.stderr else '' - - if result.returncode == 0: - logger.info(f"Script '{script_name}' completed (exit 0)") - else: - logger.warning(f"Script '{script_name}' exited with code {result.returncode}") - - return { - 'status': 'completed' if result.returncode == 0 else 'error', - 'exit_code': str(result.returncode), - 'stdout': stdout, - 'stderr': stderr, - 'script': script_name, - } - except _sp.TimeoutExpired: - _update_automation_progress(automation_id, phase='Script timed out', progress=100) - return {'status': 'error', 'error': f'Script timed out after {timeout}s', 'script': script_name} - except Exception as e: - return {'status': 'error', 'error': str(e), 'script': script_name} - - automation_engine.register_action_handler('run_script', _auto_run_script) - automation_engine.register_action_handler('full_cleanup', _auto_full_cleanup) - - automation_engine.register_action_handler('start_database_update', _auto_start_database_update, - lambda: db_update_state.get('status') == 'running') - automation_engine.register_action_handler('deep_scan_library', _auto_deep_scan_library, - lambda: db_update_state.get('status') == 'running') - automation_engine.register_action_handler('run_duplicate_cleaner', _auto_run_duplicate_cleaner, - lambda: duplicate_cleaner_state.get('status') == 'running') - automation_engine.register_action_handler('clear_quarantine', _auto_clear_quarantine) - automation_engine.register_action_handler('cleanup_wishlist', _auto_cleanup_wishlist) - automation_engine.register_action_handler('update_discovery_pool', _auto_update_discovery_pool) - automation_engine.register_action_handler('start_quality_scan', _auto_start_quality_scan, - lambda: quality_scanner_state.get('status') == 'running') - automation_engine.register_action_handler('backup_database', _auto_backup_database) - automation_engine.register_action_handler('refresh_beatport_cache', _auto_refresh_beatport_cache) - automation_engine.register_action_handler('clean_search_history', _auto_clean_search_history) - automation_engine.register_action_handler('clean_completed_downloads', _auto_clean_completed_downloads) - - def _auto_search_and_download(config): - """Search for a track and download the best match.""" - automation_id = config.get('_automation_id') - query = config.get('query', '').strip() - # Event-triggered: pull query from event data (e.g. webhook_received) - if not query: - event_data = config.get('_event_data', {}) - query = (event_data.get('query', '') or '').strip() - if not query: - if automation_id: - _update_automation_progress(automation_id, - log_line='No search query provided', log_type='error') - return {'status': 'error', 'error': 'No search query provided'} - try: - if automation_id: - _update_automation_progress(automation_id, - phase='Searching', log_line=f'Searching: {query}', log_type='info') - result = run_async(download_orchestrator.search_and_download_best(query)) - if result: - if automation_id: - _update_automation_progress(automation_id, - log_line=f'Download started for: {query}', log_type='success') - return {'status': 'completed', 'query': query, 'download_id': result} - else: - if automation_id: - _update_automation_progress(automation_id, - log_line=f'No match found for: {query}', log_type='warning') - return {'status': 'not_found', 'query': query, 'error': 'No match found'} - except Exception as e: - if automation_id: - _update_automation_progress(automation_id, - log_line=f'Error: {e}', log_type='error') - return {'status': 'error', 'query': query, 'error': str(e)} - - automation_engine.register_action_handler('search_and_download', _auto_search_and_download) - # Register progress tracking callbacks def _progress_init(aid, name, action_type): _init_automation_progress(aid, name, action_type)