Merge pull request #611 from Nezreka/refactor/extract-automation-handlers

Refactor/extract automation handlers
This commit is contained in:
BoulderBadgeDad 2026-05-15 13:31:12 -07:00 committed by GitHub
commit 6a8870f070
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
25 changed files with 4358 additions and 1496 deletions

View file

@ -1,7 +1,11 @@
"""Automation API + progress tracking helpers package.
"""Automation API + progress + handlers package.
Lifted from web_server.py /api/automations/* routes and progress
emitters. The action handler registration (`_register_automation_handlers`)
stays in web_server.py because each handler closure is tightly coupled
to other application features.
Lifted from web_server.py:
- `/api/automations/*` route helpers `api.py`
- block library used by the trigger/action UI `blocks.py`
- progress tracker (init / update / finish) `progress.py`
- cross-handler signal bus `signals.py`
- per-action handler functions `handlers/` subpackage (with
`deps.py` defining the dependency-injection surface so handlers
stay testable in isolation)
"""

140
core/automation/deps.py Normal file
View file

@ -0,0 +1,140 @@
"""Dependency-injection surface for automation handlers.
Each handler in ``core.automation.handlers`` is a top-level pure
function that accepts ``(config: dict, deps: AutomationDeps)`` instead
of reaching for module-level globals in ``web_server``. The deps
namespace bundles every callable, client, and mutable-state container
the handlers need.
Construction happens once at app startup in ``web_server.py``:
from core.automation.deps import AutomationDeps, AutomationState
state = AutomationState()
deps = AutomationDeps(
engine=automation_engine,
state=state,
get_database=get_database,
spotify_client=spotify_client,
...
)
register_all(deps)
Tests construct a fake ``AutomationDeps`` with stub callables every
handler is then exercisable without spinning up Flask, the DB, or
the real media-server clients.
"""
from __future__ import annotations
import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Optional
@dataclass
class AutomationState:
"""Mutable flags shared across handler invocations.
Pre-refactor each was a ``global`` or ``nonlocal`` variable inside
the registration closure. Lifted here so handlers + their guards
can read/write a single object instead of importing globals.
All mutations should hold ``lock``; the helper methods below do
so for the common get/set patterns.
"""
scan_library_automation_id: Optional[str] = None
db_update_automation_id: Optional[str] = None
pipeline_running: bool = False
lock: threading.Lock = field(default_factory=threading.Lock)
def is_scan_library_active(self) -> bool:
with self.lock:
return self.scan_library_automation_id is not None
def is_pipeline_running(self) -> bool:
with self.lock:
return self.pipeline_running
def set_scan_library_id(self, automation_id: Optional[str]) -> None:
with self.lock:
self.scan_library_automation_id = automation_id
def set_pipeline_running(self, value: bool) -> None:
with self.lock:
self.pipeline_running = value
@dataclass
class AutomationDeps:
"""Bundle of every callable + client an automation handler may need.
Add fields as new handlers are extracted. Every field is required
at construction (no defaults) so a missing dep fails loudly at
startup, not silently mid-handler.
"""
# --- Engine + shared state ---
engine: Any # AutomationEngine instance
state: AutomationState
config_manager: Any # config.settings.ConfigManager singleton
update_progress: Callable[..., None] # _update_automation_progress
logger: Any # module-level logger from utils.logging_config
# --- Service clients (each may be None depending on user config) ---
get_database: Callable[[], Any] # late-binding so tests don't need DB
spotify_client: Any
tidal_client: Any
web_scan_manager: Any
# --- Background-task entry points ---
process_wishlist_automatically: Callable[..., Any]
process_watchlist_scan_automatically: Callable[..., Any]
is_wishlist_actually_processing: Callable[[], bool]
is_watchlist_actually_scanning: Callable[[], bool]
get_watchlist_scan_state: Callable[[], dict] # accessor returns the live mutable dict
# --- Playlist pipeline entry points ---
run_playlist_discovery_worker: Callable[..., Any]
run_sync_task: Callable[..., Any]
load_sync_status_file: Callable[[], dict]
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]
# --- Progress + history callbacks (used by register_all to wire
# the engine's progress callback hooks). ---
init_automation_progress: Callable[..., Any]
record_progress_history: Callable[..., Any]

View file

@ -0,0 +1,62 @@
"""Per-action automation handlers.
Each module in this subpackage exposes one top-level handler function
(or a small cluster of related handlers) of the form::
def auto_<action_name>(config: dict, deps: AutomationDeps) -> dict
The ``register_all`` helper in :mod:`registration` wires every handler
to the engine in one place. ``web_server.py`` calls
``register_all(deps)`` once at startup.
"""
from core.automation.handlers.process_wishlist import auto_process_wishlist
from core.automation.handlers.scan_watchlist import auto_scan_watchlist
from core.automation.handlers.scan_library import auto_scan_library
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__ = [
'auto_process_wishlist',
'auto_scan_watchlist',
'auto_scan_library',
'auto_refresh_mirrored',
'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',
]

View file

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

View file

@ -0,0 +1,48 @@
"""Automation handler: ``discover_playlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_discover_playlist`` closure). Kicks off background discovery
of official Spotify / iTunes metadata for mirrored playlist tracks.
The worker runs in a daemon thread and emits its own progress; this
handler returns immediately after launching it (``_manages_own_progress``).
"""
from __future__ import annotations
import threading
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_discover_playlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Discover official Spotify/iTunes metadata for mirrored
playlist tracks. Runs the worker in a background thread."""
db = deps.get_database()
playlist_id = config.get('playlist_id')
discover_all = config.get('all', False)
if discover_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
if not playlists:
return {'status': 'error', 'reason': 'No playlists found'}
threading.Thread(
target=deps.run_playlist_discovery_worker,
args=(playlists, config.get('_automation_id')),
daemon=True,
name='auto-discover-playlist',
).start()
names = ', '.join(p['name'] for p in playlists[:3])
return {
'status': 'started',
'playlist_count': str(len(playlists)),
'playlists': names,
'_manages_own_progress': True,
}

View file

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

View file

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

View file

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

View file

@ -0,0 +1,320 @@
"""Automation handler: ``playlist_pipeline`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_playlist_pipeline`` closure). Runs the full playlist
lifecycle in a single trigger:
Phase 1: REFRESH -- pull fresh track lists from sources
Phase 2: DISCOVER -- look up official Spotify/iTunes metadata
Phase 3: SYNC -- push the result to the active media server
Phase 4: WISHLIST -- queue any missing tracks for download
Each phase emits its own progress range so the trigger card shows
useful per-phase percentages instead of "loading...". Phase 4 is
optional via ``skip_wishlist`` config.
Composition: this handler invokes ``auto_refresh_mirrored`` and
``auto_sync_playlist`` directly (passing ``_automation_id: None`` so
the sub-handlers don't hijack pipeline progress) instead of going
through the engine keeps the four phases observable as one
trigger from the user's perspective. Pipeline-level guard
(``state.pipeline_running``) prevents overlapping runs.
"""
from __future__ import annotations
import threading
import time
from typing import Any, Dict
from core.automation.deps import AutomationDeps
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
# Per-playlist sync poll cap inside Phase 3.
_SYNC_PER_PLAYLIST_TIMEOUT_SECONDS = 600
# Discovery poll cap inside Phase 2.
_DISCOVERY_TIMEOUT_SECONDS = 3600
def auto_playlist_pipeline(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Run REFRESH → DISCOVER → SYNC → WISHLIST in sequence.
Sets / clears ``deps.state.pipeline_running`` around the whole
run so the registration guard can short-circuit overlapping
triggers.
"""
deps.state.set_pipeline_running(True)
automation_id = config.get('_automation_id')
pipeline_start = time.time()
try:
db = deps.get_database()
playlist_id = config.get('playlist_id')
process_all = config.get('all', False)
skip_wishlist = config.get('skip_wishlist', False)
# Resolve playlists.
if process_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
deps.state.set_pipeline_running(False)
return {'status': 'error', 'error': 'No playlist specified'}
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
if not playlists:
deps.state.set_pipeline_running(False)
return {'status': 'error', 'error': 'No refreshable playlists found'}
pl_names = ', '.join(p.get('name', '?') for p in playlists[:3])
if len(playlists) > 3:
pl_names += f' (+{len(playlists) - 3} more)'
deps.update_progress(
automation_id,
progress=2,
phase=f'Pipeline: {len(playlists)} playlist(s)',
log_line=f'Starting pipeline for: {pl_names}',
log_type='info',
)
# ── PHASE 1: REFRESH ──────────────────────────────────────────
deps.update_progress(
automation_id,
progress=3,
phase='Phase 1/4: Refreshing playlists...',
log_line='Phase 1: Refresh',
log_type='info',
)
refresh_config = dict(config)
refresh_config['_automation_id'] = None # Don't let sub-handler hijack pipeline progress.
refresh_result = auto_refresh_mirrored(refresh_config, deps)
refreshed = int(refresh_result.get('refreshed', 0))
refresh_errors = int(refresh_result.get('errors', 0))
deps.update_progress(
automation_id,
progress=25,
phase='Phase 1/4: Refresh complete',
log_line=f'Phase 1 done: {refreshed} refreshed, {refresh_errors} errors',
log_type='success' if refresh_errors == 0 else 'warning',
)
# ── PHASE 2: DISCOVER ─────────────────────────────────────────
deps.update_progress(
automation_id,
progress=26,
phase='Phase 2/4: Discovering metadata...',
log_line='Phase 2: Discover',
log_type='info',
)
# Reload playlists (refresh may have updated them).
if process_all:
disc_playlists = db.get_mirrored_playlists()
else:
disc_playlists = [db.get_mirrored_playlist(int(playlist_id))]
disc_playlists = [p for p in disc_playlists if p]
# Run discovery in a thread and wait for it.
disc_done = threading.Event()
def _disc_wrapper(pls):
try:
# The worker updates automation_progress internally,
# but we pass None so it doesn't conflict with our
# pipeline progress.
deps.run_playlist_discovery_worker(pls, automation_id=None)
except Exception as e:
deps.logger.error(f"[Pipeline] Discovery error: {e}")
finally:
disc_done.set()
threading.Thread(
target=_disc_wrapper, args=(disc_playlists,),
daemon=True, name='pipeline-discover',
).start()
# Poll for completion with progress updates.
poll_start = time.time()
while not disc_done.wait(timeout=3):
elapsed = int(time.time() - poll_start)
deps.update_progress(
automation_id,
progress=min(26 + elapsed // 4, 54),
phase=f'Phase 2/4: Discovering... ({elapsed}s)',
)
if elapsed > _DISCOVERY_TIMEOUT_SECONDS:
deps.update_progress(
automation_id,
log_line='Discovery timed out after 1 hour',
log_type='warning',
)
break
deps.update_progress(
automation_id,
progress=55,
phase='Phase 2/4: Discovery complete',
log_line='Phase 2 done: discovery complete',
log_type='success',
)
# ── PHASE 3: SYNC ─────────────────────────────────────────────
deps.update_progress(
automation_id,
progress=56,
phase='Phase 3/4: Syncing to server...',
log_line='Phase 3: Sync',
log_type='info',
)
total_synced = 0
total_skipped = 0
sync_errors = 0
sync_states = deps.get_sync_states()
for pl_idx, pl in enumerate(playlists):
pl_id = pl.get('id')
if not pl_id:
continue
sync_config = {
'playlist_id': str(pl_id),
'_automation_id': None, # Don't let sync handler hijack our progress.
}
sync_result = auto_sync_playlist(sync_config, deps)
sync_status = sync_result.get('status', '')
if sync_status == 'started':
# Sync launched a background thread — wait for it.
sync_id = f"auto_mirror_{pl_id}"
sync_poll_start = time.time()
while time.time() - sync_poll_start < _SYNC_PER_PLAYLIST_TIMEOUT_SECONDS:
if (sync_id in sync_states
and sync_states[sync_id].get('status')
in ('finished', 'complete', 'error', 'failed')):
break
time.sleep(2)
elapsed = int(time.time() - sync_poll_start)
sub_progress = 56 + ((pl_idx + 1) / max(1, len(playlists))) * 29
deps.update_progress(
automation_id,
progress=min(int(sub_progress), 84),
phase=f'Phase 3/4: Syncing "{pl.get("name", "")}" ({elapsed}s)',
)
# Check result.
ss = sync_states.get(sync_id, {})
ss_result = ss.get('result', ss.get('progress', {}))
matched = ss_result.get('matched_tracks', 0) if isinstance(ss_result, dict) else 0
total_synced += int(matched) if matched else 0
deps.update_progress(
automation_id,
log_line=f'Synced "{pl.get("name", "")}": {matched} tracks matched',
log_type='success',
)
elif sync_status == 'skipped':
total_skipped += 1
reason = sync_result.get('reason', 'unchanged')
deps.update_progress(
automation_id,
log_line=f'Skipped "{pl.get("name", "")}": {reason}',
log_type='skip',
)
elif sync_status == 'error':
sync_errors += 1
deps.update_progress(
automation_id,
log_line=f'Sync error "{pl.get("name", "")}": {sync_result.get("reason", "unknown")}',
log_type='error',
)
deps.update_progress(
automation_id,
progress=85,
phase='Phase 3/4: Sync complete',
log_line=f'Phase 3 done: {total_synced} matched, {total_skipped} skipped, {sync_errors} errors',
log_type='success' if sync_errors == 0 else 'warning',
)
# ── PHASE 4: WISHLIST ─────────────────────────────────────────
wishlist_queued = 0
if not skip_wishlist:
deps.update_progress(
automation_id,
progress=86,
phase='Phase 4/4: Processing wishlist...',
log_line='Phase 4: Wishlist',
log_type='info',
)
try:
if not deps.is_wishlist_actually_processing():
deps.process_wishlist_automatically(automation_id=None)
deps.update_progress(
automation_id,
log_line='Wishlist processing triggered',
log_type='success',
)
wishlist_queued = 1
else:
deps.update_progress(
automation_id,
log_line='Wishlist already running — skipped',
log_type='skip',
)
except Exception as e:
deps.update_progress(
automation_id,
log_line=f'Wishlist error: {e}',
log_type='warning',
)
else:
deps.update_progress(
automation_id,
progress=86,
log_line='Phase 4: Wishlist skipped (disabled)',
log_type='skip',
)
# ── COMPLETE ──────────────────────────────────────────────────
duration = int(time.time() - pipeline_start)
deps.update_progress(
automation_id,
status='finished',
progress=100,
phase='Pipeline complete',
log_line=f'Pipeline finished in {duration // 60}m {duration % 60}s',
log_type='success',
)
deps.state.set_pipeline_running(False)
return {
'status': 'completed',
'_manages_own_progress': True,
'playlists_refreshed': str(refreshed),
'tracks_discovered': 'completed',
'tracks_synced': str(total_synced),
'sync_skipped': str(total_skipped),
'wishlist_queued': str(wishlist_queued),
'duration_seconds': str(duration),
}
except Exception as e:
deps.state.set_pipeline_running(False)
deps.update_progress(
automation_id,
status='error',
progress=100,
phase='Pipeline error',
log_line=f'Pipeline failed: {e}',
log_type='error',
)
return {'status': 'error', 'error': str(e), '_manages_own_progress': True}

View file

@ -0,0 +1,27 @@
"""Automation handler: ``process_wishlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_process_wishlist`` closure). Wishlist processing is async
the helper submits a batch to an executor and returns immediately;
per-track stats arrive later via batch-completion callbacks.
"""
from __future__ import annotations
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_process_wishlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Kick off the wishlist processor for an automation trigger.
Returns immediately after submission; the wishlist worker emits
per-batch progress via its own callbacks. We only report
``status: completed`` to mark the trigger fired successfully.
"""
try:
deps.process_wishlist_automatically(automation_id=config.get('_automation_id'))
return {'status': 'completed'}
except Exception as e: # noqa: BLE001 — automation handlers must never raise into the engine
return {'status': 'error', 'error': str(e)}

View file

@ -0,0 +1,89 @@
"""Progress + history callbacks the automation engine invokes around
each handler run.
Lifted from the closures at the bottom of
``web_server._register_automation_handlers``:
- ``_progress_init`` :func:`progress_init`
- ``_progress_finish`` :func:`progress_finish`
- ``_record_automation_history`` :func:`record_history`
- ``_on_library_scan_completed`` :func:`on_library_scan_completed`
The engine accepts four callables via
``register_progress_callbacks(init, finish, update, history)``;
``registration.register_all`` wires these here. The
``library_scan_completed`` callback is registered separately on the
``web_scan_manager`` (when one is available) -- see
``register_library_scan_completed_emitter``.
"""
from __future__ import annotations
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def progress_init(aid: Any, name: str, action_type: str, deps: AutomationDeps) -> None:
"""Initialize per-automation progress state when the engine starts
a handler. Thin wrapper so the engine receives a closure that
delegates into the live progress tracker."""
deps.init_automation_progress(aid, name, action_type)
def progress_finish(aid: Any, result: Dict[str, Any], deps: AutomationDeps) -> None:
"""Emit the final progress update when a handler returns.
Skipped for handlers that manage their own progress lifecycle
(they call ``update_progress(status='finished')`` themselves and
set ``_manages_own_progress: True`` in the returned dict).
Otherwise translates the handler's status into a finished/error
progress emit with a status-appropriate phase + log line.
"""
if result.get('_manages_own_progress'):
return
result_status = result.get('status', '')
status = 'error' if result_status == 'error' else 'finished'
msg = result.get('error', result.get('reason', result_status or 'done'))
deps.update_progress(
aid,
status=status,
progress=100,
phase='Error' if status == 'error' else 'Complete',
log_line=msg,
log_type='error' if status == 'error' else 'success',
)
def record_history(aid: Any, result: Dict[str, Any], deps: AutomationDeps) -> None:
"""Capture progress state into run history before the engine's
cleanup pass clears it. Thin wrapper so the engine sees a stable
callable."""
deps.record_progress_history(aid, result, deps.get_database())
def on_library_scan_completed(deps: AutomationDeps) -> None:
"""Emit the ``library_scan_completed`` automation event with the
active media-server type. Replaces the hard-coded
``scan_completion_callback trigger_automatic_database_update``
chain so any automation can listen for scan completion as a
trigger."""
if not deps.engine:
return
server_type = (
getattr(deps.web_scan_manager, '_current_server_type', None)
or 'unknown'
)
deps.engine.emit('library_scan_completed', {
'server_type': server_type,
})
def register_library_scan_completed_emitter(deps: AutomationDeps) -> None:
"""Wire :func:`on_library_scan_completed` to the
``web_scan_manager``'s scan-completion callback list. No-op when
no scan manager is configured (e.g. headless / test contexts)."""
if not deps.web_scan_manager:
return
deps.web_scan_manager.add_scan_completion_callback(
lambda: on_library_scan_completed(deps),
)

View file

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

View file

@ -0,0 +1,308 @@
"""Automation handler: ``refresh_mirrored`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_refresh_mirrored`` closure). Re-pulls track lists from each
mirrored playlist's source (Spotify / Tidal / Deezer / YouTube),
updates the local mirror DB, and emits a ``playlist_changed``
automation event when the track set actually shifts.
Source-specific branches (Spotify auth + public-embed fallback,
``spotify_public`` URLID resolution, Deezer / Tidal / YouTube)
remain identical to the pre-extraction closure this is a
mechanical lift, not a redesign.
"""
from __future__ import annotations
import json
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_refresh_mirrored(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Refresh mirrored playlist(s) from source.
Returns ``{'status': 'completed', 'refreshed': '<int>',
'errors': '<int>'}`` on success (counts stringified to match the
automation engine's stat-rendering convention).
"""
db = deps.get_database()
playlist_id = config.get('playlist_id')
refresh_all = config.get('all', False)
auto_id = config.get('_automation_id')
if refresh_all:
playlists = db.get_mirrored_playlists()
elif playlist_id:
p = db.get_mirrored_playlist(int(playlist_id))
playlists = [p] if p else []
else:
return {'status': 'error', 'reason': 'No playlist specified'}
# Filter out sources that can't be refreshed (no external API).
playlists = [pl for pl in playlists if pl.get('source', '') not in ('file', 'beatport')]
refreshed = 0
errors = []
for idx, pl in enumerate(playlists):
try:
source = pl.get('source', '')
source_id = pl.get('source_playlist_id', '')
deps.update_progress(
auto_id,
progress=(idx / max(1, len(playlists))) * 100,
phase=f'Refreshing: "{pl.get("name", "")}"',
current_item=pl.get('name', ''),
)
tracks = None
if source == 'spotify':
# Try authenticated API first, fall back to public embed scraper.
if deps.spotify_client and deps.spotify_client.is_spotify_authenticated():
playlist_obj = deps.spotify_client.get_playlist_by_id(source_id)
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
# Spotify data IS official — auto-mark as discovered.
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth needed).
if tracks is None:
try:
from core.spotify_public_scraper import scrape_spotify_embed
embed_data = scrape_spotify_embed('playlist', source_id)
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
track_dict = {
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
}
# Store Spotify track ID hint but don't mark discovered —
# Discover step needs to run for proper album art.
if t.get('id'):
track_dict['extra_data'] = json.dumps({
'discovered': False,
'spotify_hint': {
'id': t['id'],
'name': t.get('name', ''),
'artists': t.get('artists', []),
}
})
tracks.append(track_dict)
except Exception as e:
deps.logger.warning(f"Spotify public scraper fallback failed for {source_id}: {e}")
elif source == 'spotify_public':
# source_playlist_id is an MD5 hash; extract actual Spotify ID from stored description (URL).
try:
from core.spotify_public_scraper import parse_spotify_url, scrape_spotify_embed
spotify_url = pl.get('description', '')
parsed = parse_spotify_url(spotify_url) if spotify_url else None
# If Spotify is authenticated, use the full API (auto-discovers with album art).
if (parsed and parsed.get('type') == 'playlist'
and deps.spotify_client and deps.spotify_client.is_spotify_authenticated()):
playlist_obj = deps.spotify_client.get_playlist_by_id(parsed['id'])
if playlist_obj and playlist_obj.tracks:
tracks = []
for t in playlist_obj.tracks:
artist_name = t.artists[0] if t.artists else ''
track_dict = {
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
}
if t.id:
_album_obj = {'name': t.album or ''}
if getattr(t, 'image_url', None):
_album_obj['images'] = [{'url': t.image_url, 'height': 600, 'width': 600}]
track_dict['extra_data'] = json.dumps({
'discovered': True,
'provider': 'spotify',
'confidence': 1.0,
'matched_data': {
'id': t.id,
'name': t.name or '',
'artists': [{'name': str(a)} for a in (t.artists or [])],
'album': _album_obj,
'duration_ms': t.duration_ms or 0,
'image_url': getattr(t, 'image_url', None),
}
})
tracks.append(track_dict)
# Fallback: public embed scraper (no auth or album-type URL).
if tracks is None and parsed:
embed_data = scrape_spotify_embed(parsed['type'], parsed['id'])
if embed_data and not embed_data.get('error') and embed_data.get('tracks'):
embed_album = embed_data.get('name', '') if embed_data.get('type') == 'album' else ''
tracks = []
for t in embed_data['tracks']:
artist_names = [a['name'] for a in t.get('artists', [])]
artist_name = artist_names[0] if artist_names else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': artist_name,
'album_name': embed_album,
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
# No extra_data — let preservation code keep existing discovery data.
except Exception as e:
deps.logger.warning(f"Spotify public playlist refresh failed for {source_id}: {e}")
elif source == 'deezer':
try:
deezer = deps.get_deezer_client()
playlist_data = deezer.get_playlist(source_id)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': t.get('album', ''),
'duration_ms': t.get('duration_ms', 0),
'source_track_id': str(t.get('id', '')),
})
except Exception as e:
deps.logger.warning(f"Deezer playlist refresh failed for {source_id}: {e}")
elif source == 'tidal':
if not deps.tidal_client or not deps.tidal_client.is_authenticated():
deps.logger.warning(f"Tidal not authenticated — skipping refresh for '{pl.get('name', '')}'")
deps.update_progress(
auto_id,
log_line=f'Skipped "{pl.get("name", "")}" — Tidal not authenticated',
log_type='skip',
)
continue
full_playlist = deps.tidal_client.get_playlist(source_id)
if full_playlist and full_playlist.tracks:
tracks = []
for t in full_playlist.tracks:
artist_name = t.artists[0] if t.artists else ''
tracks.append({
'track_name': t.name or '',
'artist_name': str(artist_name),
'album_name': t.album or '',
'duration_ms': t.duration_ms or 0,
'source_track_id': t.id or '',
})
elif source == 'youtube':
# source_playlist_id is now a deterministic hash; use stored description (original URL) for refresh.
yt_url = pl.get('description', '') or f"https://www.youtube.com/playlist?list={source_id}"
playlist_data = deps.parse_youtube_playlist(yt_url)
if playlist_data and playlist_data.get('tracks'):
tracks = []
for t in playlist_data['tracks']:
artist_name = t['artists'][0] if t.get('artists') else ''
tracks.append({
'track_name': t.get('name', ''),
'artist_name': str(artist_name),
'album_name': '',
'duration_ms': t.get('duration_ms', 0),
'source_track_id': t.get('id', ''),
})
if tracks is not None:
# Compare old vs new track IDs to detect changes.
old_tracks = db.get_mirrored_playlist_tracks(pl['id']) if pl.get('id') else []
old_ids = {t.get('source_track_id') for t in old_tracks if t.get('source_track_id')}
new_ids = {t.get('source_track_id') for t in tracks if t.get('source_track_id')}
# Preserve existing discovery extra_data for tracks that still exist.
old_extra_map = db.get_mirrored_tracks_extra_data_map(pl['id']) if pl.get('id') else {}
for t in tracks:
sid = t.get('source_track_id', '')
if sid and sid in old_extra_map and 'extra_data' not in t:
t['extra_data'] = old_extra_map[sid]
db.mirror_playlist(
source=source,
source_playlist_id=source_id,
name=pl['name'],
tracks=tracks,
profile_id=pl.get('profile_id', 1),
owner=pl.get('owner'),
image_url=pl.get('image_url'),
)
refreshed += 1
# Emit playlist_changed if tracks actually changed.
if old_ids != new_ids:
added_count = len(new_ids - old_ids)
removed_count = len(old_ids - new_ids)
deps.logger.info(
f"[AUTOMATION] Playlist changed: '{pl.get('name', '')}'"
f"{added_count} added, {removed_count} removed (old={len(old_ids)}, new={len(new_ids)})"
)
deps.update_progress(
auto_id,
log_line=f'"{pl.get("name", "")}"{added_count} added, {removed_count} removed',
log_type='success',
)
try:
if deps.engine:
deps.engine.emit('playlist_changed', {
'playlist_name': pl.get('name', ''),
'playlist_id': str(pl.get('id', '')),
'old_count': str(len(old_ids)),
'new_count': str(len(new_ids)),
'added': str(added_count),
'removed': str(removed_count),
})
except Exception as e:
deps.logger.debug("playlist_synced automation emit failed: %s", e)
else:
deps.logger.warning(f"[AUTOMATION] No changes: '{pl.get('name', '')}' (tracks={len(old_ids)})")
deps.update_progress(
auto_id,
log_line=f'No changes: "{pl.get("name", "")}"',
log_type='skip',
)
except Exception as e:
errors.append(f"{pl.get('name', '?')}: {str(e)}")
deps.update_progress(
auto_id,
log_line=f'Error: {pl.get("name", "?")}{str(e)}',
log_type='error',
)
return {'status': 'completed', 'refreshed': str(refreshed), 'errors': str(len(errors))}

View file

@ -0,0 +1,175 @@
"""One-stop registration of every extracted automation handler.
``web_server`` builds the deps once at startup and calls
:func:`register_all` here. Each new handler module gets one line in
this file when it lands.
"""
from __future__ import annotations
from core.automation.deps import AutomationDeps
from core.automation.handlers.process_wishlist import auto_process_wishlist
from core.automation.handlers.scan_watchlist import auto_scan_watchlist
from core.automation.handlers.scan_library import auto_scan_library
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.progress_callbacks import (
progress_init,
progress_finish,
record_history,
register_library_scan_completed_emitter,
)
def register_all(deps: AutomationDeps) -> None:
"""Wire every extracted handler to the engine.
Each ``register_action_handler`` call binds the action name (the
string the trigger uses to look up its action) to a thin lambda
that injects ``deps`` and forwards the engine-supplied config.
Guards stay alongside their handler so duplicate-run prevention
behaves identically to the pre-extraction code.
"""
engine = deps.engine
# Self-guards prevent duplicate runs of the SAME operation, but
# different operations can run concurrently — wishlist downloads
# use bandwidth, watchlist scans use API calls, library scans use
# media-server CPU. Different resources, no contention.
engine.register_action_handler(
'process_wishlist',
lambda config: auto_process_wishlist(config, deps),
guard_fn=deps.is_wishlist_actually_processing,
)
engine.register_action_handler(
'scan_watchlist',
lambda config: auto_scan_watchlist(config, deps),
guard_fn=deps.is_watchlist_actually_scanning,
)
engine.register_action_handler(
'scan_library',
lambda config: auto_scan_library(config, deps),
deps.state.is_scan_library_active,
)
# Playlist lifecycle handlers. The pipeline composes refresh +
# sync + discover (it imports them directly), so all four ship
# together. The pipeline guard prevents an in-flight pipeline
# from being re-triggered mid-run.
engine.register_action_handler(
'refresh_mirrored',
lambda config: auto_refresh_mirrored(config, deps),
)
engine.register_action_handler(
'sync_playlist',
lambda config: auto_sync_playlist(config, deps),
)
engine.register_action_handler(
'discover_playlist',
lambda config: auto_discover_playlist(config, deps),
)
engine.register_action_handler(
'playlist_pipeline',
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),
)
# Progress + history callbacks: the engine invokes these around
# each handler run. Lift the closures from
# `web_server._register_automation_handlers` into thin lambdas
# that delegate into the extracted top-level functions.
engine.register_progress_callbacks(
lambda aid, name, action_type: progress_init(aid, name, action_type, deps),
lambda aid, result: progress_finish(aid, result, deps),
deps.update_progress,
lambda aid, result: record_history(aid, result, deps),
)
# `library_scan_completed` event: when the media-server scan
# manager finishes a scan, emit the event so any automation can
# trigger off it. No-op when no scan manager is configured.
register_library_scan_completed_emitter(deps)

View file

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

View file

@ -0,0 +1,158 @@
"""Automation handler: ``scan_library`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_scan_library`` closure). The handler triggers a media-server
scan via ``web_scan_manager``, then polls the manager's status until
the scan completes (or a 30-minute timeout fires). Progress phases
are emitted via :func:`AutomationDeps.update_progress` so the
trigger card stays current throughout the run.
The handler manages its own progress reporting (it sets
``_manages_own_progress: True`` in the result) so the engine doesn't
overwrite the live phase string with a generic 'completed' label.
"""
from __future__ import annotations
import time
from typing import Any, Dict
from core.automation.deps import AutomationDeps
# Outer poll cap — covers extreme worst case (long Plex scans on
# huge libraries). Past this point we surface a clear timeout error
# so users notice rather than letting the trigger hang forever.
_SCAN_TIMEOUT_SECONDS = 1800
# Per-phase poll intervals.
_POLL_SCHEDULED_SECONDS = 2
_POLL_SCANNING_SECONDS = 5
_POLL_UNKNOWN_SECONDS = 2
# Progress percentage waypoints.
_PROGRESS_SCHEDULED_MAX = 14
_PROGRESS_SCAN_START = 15
_PROGRESS_SCAN_MAX = 95
def auto_scan_library(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Run a media-server library scan and stream progress to the
trigger card.
Returns one of:
- ``{'status': 'completed', '_manages_own_progress': True, ...}``
- ``{'status': 'skipped', 'reason': 'Scan already being tracked'}``
- ``{'status': 'error', 'reason': '...', '_manages_own_progress': True}``
"""
automation_id = config.get('_automation_id')
if not deps.web_scan_manager:
return {'status': 'error', 'reason': 'Scan manager not available'}
# If another automation is already tracking the scan, just forward
# the request — the original tracker keeps emitting progress.
if deps.state.is_scan_library_active():
deps.web_scan_manager.request_scan('Automation trigger (additional batch)')
return {'status': 'skipped', 'reason': 'Scan already being tracked'}
deps.state.set_scan_library_id(automation_id)
try:
result = deps.web_scan_manager.request_scan('Automation trigger')
scan_status_val = result.get('status', 'unknown')
if scan_status_val == 'queued':
deps.update_progress(
automation_id,
log_line='Scan already in progress — waiting for completion',
log_type='info',
)
else:
delay = result.get('delay_seconds', 60)
deps.update_progress(
automation_id,
log_line=f'Scan scheduled (debounce: {delay}s)',
log_type='info',
)
# Unified polling loop — handles debounce → scanning → idle.
poll_start = time.time()
scan_started = (scan_status_val == 'queued')
while time.time() - poll_start < _SCAN_TIMEOUT_SECONDS:
status = deps.web_scan_manager.get_scan_status()
st = status.get('status')
if st == 'idle':
break # Scan completed (or finished before we polled)
if st == 'scheduled':
elapsed = int(time.time() - poll_start)
deps.update_progress(
automation_id,
phase=f'Waiting for scan to start... ({elapsed}s)',
progress=min(int(elapsed / 60 * 10), _PROGRESS_SCHEDULED_MAX),
)
time.sleep(_POLL_SCHEDULED_SECONDS)
continue
if st == 'scanning':
if not scan_started:
scan_started = True
deps.update_progress(
automation_id,
progress=_PROGRESS_SCAN_START,
log_line='Scan triggered on media server',
log_type='success',
)
elapsed = status.get('elapsed_seconds', 0)
max_time = status.get('max_time_seconds', 300)
pct = min(_PROGRESS_SCAN_START + int(elapsed / max_time * 80), _PROGRESS_SCAN_MAX)
mins, secs = divmod(elapsed, 60)
deps.update_progress(
automation_id,
phase=f'Library scan in progress... ({mins}m {secs}s)',
progress=pct,
)
time.sleep(_POLL_SCANNING_SECONDS)
continue
time.sleep(_POLL_UNKNOWN_SECONDS)
else:
# 30-min timeout reached
deps.update_progress(
automation_id,
status='error',
phase='Timed out',
log_line='Library scan timed out after 30 minutes',
log_type='error',
)
return {'status': 'error', 'reason': 'Timed out', '_manages_own_progress': True}
elapsed = round(time.time() - poll_start, 1)
deps.update_progress(
automation_id,
status='finished',
progress=100,
phase='Complete',
log_line='Library scan completed',
log_type='success',
)
return {
'status': 'completed',
'_manages_own_progress': True,
'scan_duration_seconds': elapsed,
}
except Exception as e: # noqa: BLE001 — automation handlers must never raise into the engine
deps.update_progress(
automation_id,
status='error',
phase='Error',
log_line=str(e),
log_type='error',
)
return {'status': 'error', 'error': str(e), '_manages_own_progress': True}
finally:
deps.state.set_scan_library_id(None)

View file

@ -0,0 +1,46 @@
"""Automation handler: ``scan_watchlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_scan_watchlist`` closure). The watchlist scanner returns
summary stats for the trigger card only when a fresh scan actually
ran detected by snapshotting ``id(state_dict)`` before/after, since
the live processor reassigns the dict on each new scan.
"""
from __future__ import annotations
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_scan_watchlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Run a watchlist scan when the automation triggers.
Pre-scan we capture ``id(watchlist_scan_state)`` so we can tell
afterwards whether the worker ran (and reassigned the state dict)
or short-circuited (kept the same dict). Only fresh scans report
summary stats repeat triggers without an intervening run return
a bare ``completed``.
"""
try:
pre_state = deps.get_watchlist_scan_state()
pre_state_id = id(pre_state)
deps.process_watchlist_scan_automatically(
automation_id=config.get('_automation_id'),
profile_id=config.get('_profile_id'),
)
post_state = deps.get_watchlist_scan_state()
# Fresh scan = state dict was reassigned mid-run.
if id(post_state) != pre_state_id:
summary = post_state.get('summary', {}) if isinstance(post_state, dict) else {}
return {
'status': 'completed',
'artists_scanned': summary.get('total_artists', 0),
'successful_scans': summary.get('successful_scans', 0),
'new_tracks_found': summary.get('new_tracks_found', 0),
'tracks_added_to_wishlist': summary.get('tracks_added_to_wishlist', 0),
}
return {'status': 'completed'}
except Exception as e: # noqa: BLE001 — automation handlers must never raise into the engine
return {'status': 'error', 'error': str(e)}

View file

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

View file

@ -0,0 +1,195 @@
"""Automation handler: ``sync_playlist`` action.
Lifted from ``web_server._register_automation_handlers`` (the
``_auto_sync_playlist`` closure). Syncs a mirrored playlist to the
configured media server, using discovered metadata when available
and skipping undiscovered tracks. When triggered on a schedule with
no track changes since the last sync, short-circuits with
``status: skipped`` (saves Plex / Jellyfin / Navidrome from
needless rewrites)."""
from __future__ import annotations
import hashlib
import json
import threading
from typing import Any, Dict
from core.automation.deps import AutomationDeps
def auto_sync_playlist(config: Dict[str, Any], deps: AutomationDeps) -> Dict[str, Any]:
"""Sync a mirrored playlist to the active media server.
Behavior:
- Tracks with discovered metadata (extra_data.discovered + matched_data)
are routed via the official metadata.
- Tracks with a Spotify hint (real Spotify ID from the embed
scraper) are included so they can still hit Soulseek + the
wishlist.
- Tracks with neither are counted as ``skipped_tracks``.
- Empty result ``status: skipped`` with the skipped count.
- Same track set as last sync (matched_tracks unchanged)
``status: skipped`` (no-op).
- Otherwise spawns a daemon thread running ``run_sync_task`` and
returns ``status: started`` with ``_manages_own_progress: True``.
"""
auto_id = config.get('_automation_id')
playlist_id = config.get('playlist_id')
if not playlist_id:
return {'status': 'error', 'reason': 'No playlist specified'}
db = deps.get_database()
pl = db.get_mirrored_playlist(int(playlist_id))
if not pl:
return {'status': 'error', 'reason': 'Playlist not found'}
tracks = db.get_mirrored_playlist_tracks(int(playlist_id))
if not tracks:
return {'status': 'error', 'reason': 'No tracks in playlist'}
# Convert mirrored tracks to format expected by run_sync_task.
# Use discovered metadata when available, fall back to Spotify
# hint or raw playlist fields when not.
tracks_json = []
skipped_count = 0
for t in tracks:
# Parse extra_data for discovery info.
extra = {}
if t.get('extra_data'):
try:
extra = json.loads(t['extra_data']) if isinstance(t['extra_data'], str) else t['extra_data']
except (json.JSONDecodeError, TypeError):
pass
if extra.get('discovered') and extra.get('matched_data'):
# Use official discovered metadata.
md = extra['matched_data']
album_raw = md.get('album', '')
album_obj = album_raw if isinstance(album_raw, dict) else {'name': album_raw or ''}
_track_entry = {
'name': md.get('name', ''),
'artists': md.get('artists', [{'name': t.get('artist_name', '')}]),
'album': album_obj,
'duration_ms': md.get('duration_ms', 0),
'id': md.get('id', ''),
}
if md.get('track_number'):
_track_entry['track_number'] = md['track_number']
if md.get('disc_number'):
_track_entry['disc_number'] = md['disc_number']
tracks_json.append(_track_entry)
else:
# NOT discovered — try to include using available metadata so
# the track can still be searched on Soulseek and added to
# wishlist. Without this, failed discovery blocks the entire
# download pipeline.
#
# Priority: spotify_hint (has real Spotify ID from embed
# scraper) > raw playlist fields (only if source_track_id
# is valid).
hint = extra.get('spotify_hint', {})
# Build album object with cover art from the mirrored playlist track.
track_image = (t.get('image_url') or '').strip()
album_obj = {
'name': (t.get('album_name') or '').strip(),
'images': [{'url': track_image, 'height': 300, 'width': 300}] if track_image else [],
}
if hint.get('id') and hint.get('name'):
# spotify_hint has proper Spotify track ID + metadata from embed scraper.
hint_artists = hint.get('artists', [])
if hint_artists and isinstance(hint_artists[0], str):
hint_artists = [{'name': a} for a in hint_artists]
elif hint_artists and isinstance(hint_artists[0], dict):
pass # Already in correct format
else:
hint_artists = [{'name': t.get('artist_name', '')}]
tracks_json.append({
'name': hint['name'],
'artists': hint_artists,
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': hint['id'],
})
elif t.get('source_track_id') and (t.get('track_name') or '').strip():
# Has a valid source ID and track name — usable for wishlist.
tracks_json.append({
'name': t['track_name'].strip(),
'artists': [{'name': (t.get('artist_name') or '').strip() or 'Unknown Artist'}],
'album': album_obj,
'duration_ms': t.get('duration_ms', 0),
'id': t['source_track_id'],
})
else:
skipped_count += 1 # No usable ID or name — truly can't process.
if not tracks_json:
deps.update_progress(
auto_id,
log_line=f'No discovered tracks — {skipped_count} need discovery first',
log_type='skip',
)
return {
'status': 'skipped',
'reason': f'No discovered tracks to sync ({skipped_count} tracks need discovery first)',
'skipped_tracks': str(skipped_count),
}
# Preflight: hash the track list and compare against last sync.
# Skip if the exact same set of tracks was already synced and
# everything matched (no-op preserves Plex / Jellyfin / Navidrome
# from needless rewrites).
track_ids_str = ','.join(sorted(t.get('id', '') for t in tracks_json))
tracks_hash = hashlib.md5(track_ids_str.encode()).hexdigest()
sync_id_key = f"auto_mirror_{playlist_id}"
try:
sync_statuses = deps.load_sync_status_file()
last_status = sync_statuses.get(sync_id_key, {})
last_hash = last_status.get('tracks_hash', '')
last_matched = last_status.get('matched_tracks', -1)
if last_hash == tracks_hash and last_matched >= len(tracks_json):
# Exact same tracks, all matched last time — nothing to do.
deps.update_progress(
auto_id,
log_line=f'All {len(tracks_json)} tracks unchanged since last sync — skipping',
log_type='skip',
)
return {
'status': 'skipped',
'reason': f'All {len(tracks_json)} tracks unchanged since last sync',
}
except Exception as e:
deps.logger.debug("mirror sync last-status read: %s", e)
deps.update_progress(
auto_id,
progress=50,
phase=f'Syncing "{pl["name"]}"',
log_line=f'{len(tracks_json)} discovered, {skipped_count} skipped',
log_type='info',
)
sync_id = f"auto_mirror_{playlist_id}"
deps.update_progress(
auto_id,
progress=90,
log_line=f'Starting sync: {len(tracks_json)} tracks',
log_type='success',
)
threading.Thread(
target=deps.run_sync_task,
args=(sync_id, pl['name'], tracks_json, auto_id, 1, pl.get('image_url', '')),
daemon=True,
name=f'auto-sync-{playlist_id}',
).start()
return {
'status': 'started',
'playlist_name': pl['name'],
'discovered_tracks': str(len(tracks_json)),
'skipped_tracks': str(skipped_count),
'_manages_own_progress': True,
}

View file

@ -0,0 +1,388 @@
"""Engine-boundary tests for the automation handler registration.
Per-handler boundary tests in the sibling test files prove each
handler's body works in isolation. These tests prove the
**registration layer** wires every handler to the right action name,
attaches the right guard, and registers the four progress callbacks
in the slots the engine expects.
The kettui standard for refactor PRs: don't ship a "behavior
preserved" claim that's only validated at the function boundary.
Wire the seam engine + register_all + deps and exercise it.
These tests use a minimal recording engine that captures every
``register_action_handler`` / ``register_progress_callbacks`` call,
plus a no-op ``add_scan_completion_callback`` on a fake scan
manager. No real AutomationEngine, no real DB, no real Flask app.
"""
from __future__ import annotations
import threading
from typing import Any, Dict, List, Tuple
import pytest
from core.automation.deps import AutomationDeps, AutomationState
from core.automation.handlers import register_all
# Every action name `register_all` is expected to register. Drift
# (rename / new handler / removed handler) fails this test
# immediately so refactor PRs can't quietly drop a handler.
EXPECTED_ACTION_NAMES = frozenset({
'process_wishlist',
'scan_watchlist',
'scan_library',
'refresh_mirrored',
'sync_playlist',
'discover_playlist',
'playlist_pipeline',
'start_database_update',
'deep_scan_library',
'run_duplicate_cleaner',
'clear_quarantine',
'cleanup_wishlist',
'update_discovery_pool',
'start_quality_scan',
'backup_database',
'refresh_beatport_cache',
'clean_search_history',
'clean_completed_downloads',
'full_cleanup',
'run_script',
'search_and_download',
})
# Action names that MUST register a guard (duplicate-run prevention).
EXPECTED_GUARDED_ACTIONS = frozenset({
'process_wishlist',
'scan_watchlist',
'scan_library',
'playlist_pipeline',
'start_database_update',
'deep_scan_library',
'run_duplicate_cleaner',
'start_quality_scan',
})
class _RecordingEngine:
"""Minimal AutomationEngine stand-in. Captures everything
register_all does so tests can assert on it."""
def __init__(self):
self.action_handlers: Dict[str, Dict[str, Any]] = {}
self.progress_callbacks: Tuple = ()
self.emits: List[Tuple[str, dict]] = []
def register_action_handler(self, action_type, handler_fn, guard_fn=None):
self.action_handlers[action_type] = {'handler': handler_fn, 'guard': guard_fn}
def register_progress_callbacks(self, init_fn, finish_fn, update_fn=None, history_fn=None):
self.progress_callbacks = (init_fn, finish_fn, update_fn, history_fn)
def emit(self, event_name, payload):
self.emits.append((event_name, payload))
class _RecordingScanMgr:
_current_server_type = 'plex'
def __init__(self):
self.callbacks: List = []
def add_scan_completion_callback(self, cb):
self.callbacks.append(cb)
def _build_deps(engine, scan_mgr=None) -> AutomationDeps:
class _Logger:
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 _Cfg:
def get(self, key, default=None): return default
def get_active_media_server(self): return 'plex'
return AutomationDeps(
engine=engine,
state=AutomationState(),
config_manager=_Cfg(),
update_progress=lambda *a, **k: None,
logger=_Logger(),
get_database=lambda: object(),
spotify_client=None,
tidal_client=None,
web_scan_manager=scan_mgr,
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': {}},
init_automation_progress=lambda *a, **k: None,
record_progress_history=lambda *a, **k: None,
)
# ─── action handler registration ─────────────────────────────────────
class TestActionHandlerRegistration:
def test_every_expected_action_name_registered(self):
engine = _RecordingEngine()
register_all(_build_deps(engine))
registered = set(engine.action_handlers.keys())
missing = EXPECTED_ACTION_NAMES - registered
extra = registered - EXPECTED_ACTION_NAMES
assert not missing, f"register_all dropped: {missing}"
assert not extra, f"register_all added unexpected: {extra}"
def test_guarded_actions_have_a_guard(self):
engine = _RecordingEngine()
register_all(_build_deps(engine))
for name in EXPECTED_GUARDED_ACTIONS:
assert engine.action_handlers[name]['guard'] is not None, (
f"action {name!r} expected to register a guard but didn't"
)
def test_unguarded_actions_have_no_guard(self):
engine = _RecordingEngine()
register_all(_build_deps(engine))
unguarded = EXPECTED_ACTION_NAMES - EXPECTED_GUARDED_ACTIONS
for name in unguarded:
assert engine.action_handlers[name]['guard'] is None, (
f"action {name!r} unexpectedly registered a guard"
)
def test_every_handler_callable(self):
# Every registered handler must be callable with a config dict.
engine = _RecordingEngine()
register_all(_build_deps(engine))
for name, entry in engine.action_handlers.items():
handler = entry['handler']
assert callable(handler), f"{name} handler is not callable"
def test_every_guard_callable_returns_bool(self):
engine = _RecordingEngine()
register_all(_build_deps(engine))
for name, entry in engine.action_handlers.items():
guard = entry['guard']
if guard is None:
continue
value = guard()
assert isinstance(value, bool), (
f"{name} guard returned non-bool: {type(value).__name__}"
)
# ─── progress callback registration ──────────────────────────────────
class TestProgressCallbackRegistration:
def test_all_four_callbacks_registered(self):
engine = _RecordingEngine()
register_all(_build_deps(engine))
init_fn, finish_fn, update_fn, history_fn = engine.progress_callbacks
assert callable(init_fn)
assert callable(finish_fn)
assert callable(update_fn)
assert callable(history_fn)
def test_progress_init_callback_invocable(self):
# Engine signature: init_fn(aid, name, action_type)
engine = _RecordingEngine()
captured: List[Tuple] = []
deps = _build_deps(engine)
deps = AutomationDeps(**{
**{f.name: getattr(deps, f.name) for f in deps.__dataclass_fields__.values()},
'init_automation_progress': lambda aid, name, at: captured.append((aid, name, at)),
})
register_all(deps)
init_fn, _, _, _ = engine.progress_callbacks
init_fn('auto-1', 'My Auto', 'wishlist')
assert captured == [('auto-1', 'My Auto', 'wishlist')]
def test_progress_finish_callback_invocable(self):
# Engine signature: finish_fn(aid, result)
engine = _RecordingEngine()
captured: List[Tuple] = []
deps = _build_deps(engine)
deps = AutomationDeps(**{
**{f.name: getattr(deps, f.name) for f in deps.__dataclass_fields__.values()},
'update_progress': lambda aid, **kw: captured.append((aid, kw)),
})
register_all(deps)
_, finish_fn, _, _ = engine.progress_callbacks
# Non-_manages_own_progress result triggers update_progress emit.
finish_fn('auto-1', {'status': 'completed'})
assert len(captured) == 1
assert captured[0][0] == 'auto-1'
assert captured[0][1]['status'] == 'finished'
def test_progress_finish_skips_self_managed(self):
engine = _RecordingEngine()
captured: List[Tuple] = []
deps = _build_deps(engine)
deps = AutomationDeps(**{
**{f.name: getattr(deps, f.name) for f in deps.__dataclass_fields__.values()},
'update_progress': lambda aid, **kw: captured.append((aid, kw)),
})
register_all(deps)
_, finish_fn, _, _ = engine.progress_callbacks
finish_fn('auto-1', {'_manages_own_progress': True, 'status': 'completed'})
assert captured == []
def test_history_callback_invocable_with_db(self):
# Engine signature: history_fn(aid, result)
engine = _RecordingEngine()
captured: List[Tuple] = []
db_obj = object()
deps = _build_deps(engine)
deps = AutomationDeps(**{
**{f.name: getattr(deps, f.name) for f in deps.__dataclass_fields__.values()},
'get_database': lambda: db_obj,
'record_progress_history': lambda aid, result, db: captured.append((aid, result, db)),
})
register_all(deps)
_, _, _, history_fn = engine.progress_callbacks
history_fn('auto-1', {'status': 'completed'})
assert captured == [('auto-1', {'status': 'completed'}, db_obj)]
# ─── library_scan_completed wiring ───────────────────────────────────
class TestLibraryScanCompletedEmitter:
def test_no_scan_manager_safe(self):
# Should not raise when scan manager is absent (test/headless mode).
engine = _RecordingEngine()
register_all(_build_deps(engine, scan_mgr=None))
# No callbacks captured (no scan manager to register against),
# but engine still has all the handlers.
assert engine.action_handlers
def test_scan_completion_callback_registered(self):
engine = _RecordingEngine()
scan_mgr = _RecordingScanMgr()
register_all(_build_deps(engine, scan_mgr=scan_mgr))
assert len(scan_mgr.callbacks) == 1
# Invoking the callback should fire engine.emit('library_scan_completed', ...).
scan_mgr.callbacks[0]()
assert engine.emits == [('library_scan_completed', {'server_type': 'plex'})]
# ─── handler invocation through the engine boundary ─────────────────
class TestHandlerInvocation:
"""For each registered handler, exercise the lambda the engine
would call. Verifies the deps closure is captured correctly + the
handler returns a result dict.
Forces every long-running handler down its guard short-circuit
path by pre-setting the relevant `*_state` dicts to ``running``
(database_update, deep_scan, duplicate_cleaner, quality_scanner)
or by pre-occupying the state flags (scan_library, playlist_pipeline).
Other handlers either return error early (no playlist specified,
no scan manager, etc) or run cleanly against the no-op stub deps.
"""
def test_every_handler_returns_dict(self):
engine = _RecordingEngine()
# Pre-set state dicts so guarded handlers skip-return cleanly.
running_state = {'status': 'running'}
active_batches = {'b1': {'phase': 'downloading'}}
# Stub DB that satisfies the handlers reaching for it on
# short paths. cleanup_wishlist calls remove_wishlist_duplicates.
# update_discovery_pool's exception path swallows missing
# scanner. Other handlers either short-circuit or use stub
# callables.
class _StubDB:
def remove_wishlist_duplicates(self, profile_id): return 0
def get_mirrored_playlists(self): return []
def get_mirrored_playlist(self, _id): return None
def get_mirrored_playlist_tracks(self, _id): return []
# Stub Flask app so refresh_beatport_cache's test_client call
# doesn't crash. Use a minimal context manager.
class _FakeResponse:
status_code = 200
class _FakeClient:
def __enter__(self): return self
def __exit__(self, *a): return False
def get(self, _path): return _FakeResponse()
class _StubApp:
def test_client(self): return _FakeClient()
deps = _build_deps(engine)
deps = AutomationDeps(**{
**{f.name: getattr(deps, f.name) for f in deps.__dataclass_fields__.values()},
'get_db_update_state': lambda: running_state,
'get_duplicate_cleaner_state': lambda: running_state,
'get_quality_scanner_state': lambda: running_state,
'get_download_batches': lambda: active_batches, # forces clean_completed_downloads to skip
'get_database': lambda: _StubDB(),
'get_app': lambda: _StubApp(),
})
# Pre-set state flags too so scan_library + playlist_pipeline guards fire.
deps.state.scan_library_automation_id = 'someone-else'
deps.state.pipeline_running = True
# Patch time.sleep across handler modules so refresh_beatport_cache
# (which sleeps 2s between sections) doesn't extend the test.
import core.automation.handlers.maintenance as maint_mod
original_sleep = maint_mod.time.sleep
maint_mod.time.sleep = lambda _: None
register_all(deps)
try:
for name, entry in engine.action_handlers.items():
handler = entry['handler']
# Minimal config: trigger the natural error/skip paths
# for handlers that need a playlist_id, query, script_name etc.
result = handler({'_automation_id': 'test', 'all': False})
assert isinstance(result, dict), (
f"handler {name!r} returned {type(result).__name__}, expected dict"
)
assert 'status' in result, (
f"handler {name!r} returned dict without 'status': {list(result.keys())}"
)
finally:
maint_mod.time.sleep = original_sleep

View file

@ -0,0 +1,392 @@
"""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': {}},
init_automation_progress=lambda *a, **k: None,
record_progress_history=lambda *a, **k: None,
)
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'

View file

@ -0,0 +1,399 @@
"""Boundary tests for the playlist-lifecycle automation handlers
(``refresh_mirrored``, ``sync_playlist``, ``discover_playlist``,
``playlist_pipeline``).
The handlers themselves are mechanical lifts of the closures that
used to live in ``web_server._register_automation_handlers`` these
tests pin the seam so the wiring stays correct (deps are read from
the deps object, not module-level globals; cross-handler calls in
the pipeline still compose; failure paths still return clear status
shapes).
Source-specific branches inside ``refresh_mirrored`` (Spotify auth
+ public-embed fallback, Deezer / Tidal / YouTube) are validated
end-to-end via fake clients in ``deps`` rather than per-source
because they're a verbatim lift — drift would show up here as a
behavior change."""
from __future__ import annotations
import json
import threading
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
import pytest
from core.automation.deps import AutomationDeps, AutomationState
from core.automation.handlers.discover_playlist import auto_discover_playlist
from core.automation.handlers.playlist_pipeline import auto_playlist_pipeline
from core.automation.handlers.refresh_mirrored import auto_refresh_mirrored
from core.automation.handlers.sync_playlist import auto_sync_playlist
# ─── shared scaffolding ──────────────────────────────────────────────
class _StubLogger:
def __init__(self):
self.messages: List[tuple] = []
def debug(self, *a, **k): self.messages.append(('debug', a))
def info(self, *a, **k): self.messages.append(('info', a))
def warning(self, *a, **k): self.messages.append(('warning', a))
def error(self, *a, **k): self.messages.append(('error', a))
@dataclass
class _StubDB:
"""Fake MusicDatabase — minimal surface used by the playlist handlers."""
playlists: List[dict] = field(default_factory=list)
playlist_tracks: Dict[int, List[dict]] = field(default_factory=dict)
extra_data_maps: Dict[int, Dict[str, str]] = field(default_factory=dict)
mirror_calls: List[dict] = field(default_factory=list)
def get_mirrored_playlists(self) -> list:
return list(self.playlists)
def get_mirrored_playlist(self, playlist_id: int) -> Optional[dict]:
for p in self.playlists:
if int(p.get('id', -1)) == int(playlist_id):
return p
return None
def get_mirrored_playlist_tracks(self, playlist_id: int) -> list:
return list(self.playlist_tracks.get(int(playlist_id), []))
def get_mirrored_tracks_extra_data_map(self, playlist_id: int) -> dict:
return dict(self.extra_data_maps.get(int(playlist_id), {}))
def mirror_playlist(self, **kwargs) -> None:
self.mirror_calls.append(kwargs)
def _build_deps(**overrides) -> AutomationDeps:
defaults = dict(
engine=object(),
state=AutomationState(),
config_manager=object(),
update_progress=lambda *a, **k: None,
logger=_StubLogger(),
get_database=lambda: _StubDB(),
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': {}},
init_automation_progress=lambda *a, **k: None,
record_progress_history=lambda *a, **k: None,
)
defaults.update(overrides)
return AutomationDeps(**defaults) # type: ignore[arg-type]
# ─── discover_playlist ───────────────────────────────────────────────
class TestDiscoverPlaylist:
def test_no_playlist_id_returns_error(self):
deps = _build_deps()
result = auto_discover_playlist({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_specific_playlist_id_starts_worker(self):
db = _StubDB(playlists=[{'id': 42, 'name': 'Test Playlist'}])
called: List[Any] = []
deps = _build_deps(
get_database=lambda: db,
run_playlist_discovery_worker=lambda *a, **k: called.append((a, k)),
)
result = auto_discover_playlist({'playlist_id': '42', '_automation_id': 'auto-1'}, deps)
assert result['status'] == 'started'
assert result['_manages_own_progress'] is True
assert result['playlist_count'] == '1'
# Worker spawned on a thread; give it a moment.
for _ in range(50):
if called:
break
import time
time.sleep(0.01)
assert len(called) == 1
def test_all_playlists_includes_every_one(self):
db = _StubDB(playlists=[
{'id': 1, 'name': 'A'}, {'id': 2, 'name': 'B'}, {'id': 3, 'name': 'C'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_discover_playlist({'all': True}, deps)
assert result['playlist_count'] == '3'
assert 'A' in result['playlists']
assert 'B' in result['playlists']
assert 'C' in result['playlists']
def test_no_playlists_in_db_returns_error(self):
deps = _build_deps(get_database=lambda: _StubDB(playlists=[]))
result = auto_discover_playlist({'all': True}, deps)
assert result == {'status': 'error', 'reason': 'No playlists found'}
# ─── refresh_mirrored ────────────────────────────────────────────────
@dataclass
class _StubSpotifyTrack:
id: str
name: str
artists: list
album: str
duration_ms: int
image_url: Optional[str] = None
@dataclass
class _StubSpotifyPlaylist:
tracks: list
class _StubSpotifyClient:
def __init__(self, playlist):
self._playlist = playlist
self._authenticated = True
def is_spotify_authenticated(self) -> bool:
return self._authenticated
def get_playlist_by_id(self, _source_id):
return self._playlist
class TestRefreshMirrored:
def test_no_playlist_specified_returns_error(self):
deps = _build_deps()
result = auto_refresh_mirrored({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_filters_unrefreshable_sources(self):
# Sources 'file' and 'beatport' have no API to refresh from.
db = _StubDB(playlists=[
{'id': 1, 'name': 'F', 'source': 'file', 'source_playlist_id': '1'},
{'id': 2, 'name': 'B', 'source': 'beatport', 'source_playlist_id': '2'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_refresh_mirrored({'all': True}, deps)
assert result['status'] == 'completed'
assert result['refreshed'] == '0'
assert db.mirror_calls == [] # nothing got pushed to DB
def test_spotify_refresh_writes_to_db(self):
track = _StubSpotifyTrack(
id='track123', name='Hello', artists=['Adele'],
album='25', duration_ms=295000,
)
playlist = _StubSpotifyPlaylist(tracks=[track])
spotify = _StubSpotifyClient(playlist)
db = _StubDB(playlists=[
{'id': 5, 'name': 'My Spot', 'source': 'spotify',
'source_playlist_id': 'spot-id', 'profile_id': 1},
])
deps = _build_deps(get_database=lambda: db, spotify_client=spotify)
result = auto_refresh_mirrored({'playlist_id': '5'}, deps)
assert result['status'] == 'completed'
assert result['refreshed'] == '1'
assert len(db.mirror_calls) == 1
call = db.mirror_calls[0]
assert call['source'] == 'spotify'
assert call['source_playlist_id'] == 'spot-id'
assert call['name'] == 'My Spot'
assert len(call['tracks']) == 1
# Spotify-source tracks should be auto-marked discovered.
extra = json.loads(call['tracks'][0]['extra_data'])
assert extra['discovered'] is True
assert extra['provider'] == 'spotify'
assert extra['matched_data']['id'] == 'track123'
def test_per_playlist_exception_collected_into_errors(self):
# Force an exception by making the DB blow up on mirror_playlist.
class _ExplodingDB(_StubDB):
def mirror_playlist(self, **kwargs):
raise RuntimeError('db disk full')
track = _StubSpotifyTrack(id='t', name='t', artists=['a'], album='a', duration_ms=0)
spotify = _StubSpotifyClient(_StubSpotifyPlaylist(tracks=[track]))
db = _ExplodingDB(playlists=[
{'id': 1, 'name': 'X', 'source': 'spotify', 'source_playlist_id': 'spot'},
])
deps = _build_deps(get_database=lambda: db, spotify_client=spotify)
result = auto_refresh_mirrored({'all': True}, deps)
# Error captured, status still 'completed' (handler returns counts).
assert result['status'] == 'completed'
assert result['errors'] == '1'
assert result['refreshed'] == '0'
# ─── sync_playlist ───────────────────────────────────────────────────
class TestSyncPlaylist:
def test_no_playlist_id_returns_error(self):
deps = _build_deps()
result = auto_sync_playlist({}, deps)
assert result == {'status': 'error', 'reason': 'No playlist specified'}
def test_playlist_not_found_returns_error(self):
deps = _build_deps(get_database=lambda: _StubDB(playlists=[]))
result = auto_sync_playlist({'playlist_id': '99'}, deps)
assert result == {'status': 'error', 'reason': 'Playlist not found'}
def test_no_tracks_returns_error(self):
db = _StubDB(playlists=[{'id': 1, 'name': 'P'}], playlist_tracks={1: []})
deps = _build_deps(get_database=lambda: db)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result == {'status': 'error', 'reason': 'No tracks in playlist'}
def test_no_discovered_tracks_skips(self):
# All tracks lack discovery + spotify_hint + valid IDs.
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [{}, {}]}, # empty tracks → nothing usable
)
deps = _build_deps(get_database=lambda: db)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'skipped'
assert 'No discovered tracks' in result['reason']
assert result['skipped_tracks'] == '2'
def test_discovered_track_starts_sync_thread(self):
discovered_track = {
'extra_data': json.dumps({
'discovered': True,
'matched_data': {
'id': 'spot-1', 'name': 'Track', 'artists': [{'name': 'X'}],
'album': {'name': 'Album'}, 'duration_ms': 200000,
},
}),
'artist_name': 'X',
}
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [discovered_track]},
)
sync_calls: List[tuple] = []
deps = _build_deps(
get_database=lambda: db,
run_sync_task=lambda *a, **k: sync_calls.append((a, k)),
)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'started'
assert result['_manages_own_progress'] is True
assert result['discovered_tracks'] == '1'
# Wait for thread to fire run_sync_task
for _ in range(50):
if sync_calls:
break
import time
time.sleep(0.01)
assert len(sync_calls) == 1
def test_unchanged_since_last_sync_returns_skipped(self):
discovered_track = {
'extra_data': json.dumps({
'discovered': True,
'matched_data': {
'id': 'spot-1', 'name': 'T', 'artists': [{'name': 'X'}],
'album': {'name': 'A'}, 'duration_ms': 0,
},
}),
'artist_name': 'X',
}
db = _StubDB(
playlists=[{'id': 1, 'name': 'P'}],
playlist_tracks={1: [discovered_track]},
)
# Pre-populate the sync-status file with the EXPECTED hash so the
# preflight short-circuit fires.
import hashlib
expected_hash = hashlib.md5('spot-1'.encode()).hexdigest()
sync_statuses = {
'auto_mirror_1': {'tracks_hash': expected_hash, 'matched_tracks': 1}
}
deps = _build_deps(
get_database=lambda: db,
load_sync_status_file=lambda: sync_statuses,
)
result = auto_sync_playlist({'playlist_id': '1'}, deps)
assert result['status'] == 'skipped'
assert 'unchanged' in result['reason']
# ─── playlist_pipeline ───────────────────────────────────────────────
class TestPlaylistPipeline:
def test_no_playlist_specified_returns_error(self):
deps = _build_deps()
result = auto_playlist_pipeline({}, deps)
assert result == {'status': 'error', 'error': 'No playlist specified'}
# Pipeline-running flag MUST be cleared on error so the guard
# doesn't block subsequent triggers.
assert deps.state.pipeline_running is False
def test_no_refreshable_playlists_clears_running_flag(self):
db = _StubDB(playlists=[
{'id': 1, 'name': 'F', 'source': 'file'},
{'id': 2, 'name': 'B', 'source': 'beatport'},
])
deps = _build_deps(get_database=lambda: db)
result = auto_playlist_pipeline({'all': True}, deps)
assert result == {'status': 'error', 'error': 'No refreshable playlists found'}
assert deps.state.pipeline_running is False
def test_pipeline_clears_running_on_unhandled_exception(self):
# Force the database accessor to blow up after the early checks.
class _ExplodingDB(_StubDB):
def get_mirrored_playlists(self):
raise RuntimeError('db down')
db = _ExplodingDB(playlists=[])
deps = _build_deps(get_database=lambda: db)
result = auto_playlist_pipeline({'all': True}, deps)
assert result['status'] == 'error'
assert result['_manages_own_progress'] is True
assert deps.state.pipeline_running is False

View file

@ -0,0 +1,328 @@
"""Boundary tests for the simple extracted automation handlers
(``process_wishlist``, ``scan_watchlist``, ``scan_library``).
Each handler is tested as a pure function: real ``AutomationDeps``
constructed with stub callables, no Flask, no DB, no media-server
clients. The tests exercise the success path, the guard paths
(handler short-circuits when another instance is running), the
exception-swallowing contract (handlers must NEVER raise into the
engine), and the mutable-state machinery for handlers that own a
flag in ``AutomationState``.
Pre-extraction these closures lived inside
``web_server._register_automation_handlers`` and were essentially
un-testable every test would have needed to spin up the whole
Flask app and stub a dozen module-level globals."""
from __future__ import annotations
import threading
import time
from dataclasses import dataclass, field
from typing import Any, Callable, List, Optional
import pytest
from core.automation.deps import AutomationDeps, AutomationState
from core.automation.handlers.process_wishlist import auto_process_wishlist
from core.automation.handlers.scan_watchlist import auto_scan_watchlist
from core.automation.handlers.scan_library import auto_scan_library
# ─── shared test scaffolding ──────────────────────────────────────────
def _build_deps(**overrides: Any) -> AutomationDeps:
"""Return a default `AutomationDeps` with no-op callables. Tests
pass ``overrides`` to install behaviour on the specific deps they
care about."""
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
defaults = dict(
engine=object(),
state=AutomationState(),
config_manager=object(),
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': {}},
init_automation_progress=lambda *a, **k: None,
record_progress_history=lambda *a, **k: None,
)
defaults.update(overrides)
return AutomationDeps(**defaults) # type: ignore[arg-type]
# ─── process_wishlist ─────────────────────────────────────────────────
class TestProcessWishlist:
def test_success_returns_completed_status(self):
called: List[Any] = []
def stub(automation_id=None):
called.append(automation_id)
deps = _build_deps(process_wishlist_automatically=stub)
result = auto_process_wishlist({'_automation_id': 'auto-1'}, deps)
assert result == {'status': 'completed'}
assert called == ['auto-1']
def test_passes_none_when_no_automation_id(self):
called: List[Any] = []
def stub(automation_id=None):
called.append(automation_id)
deps = _build_deps(process_wishlist_automatically=stub)
result = auto_process_wishlist({}, deps)
assert result == {'status': 'completed'}
assert called == [None]
def test_handler_swallows_exceptions(self):
def stub(**_kwargs):
raise RuntimeError('boom')
deps = _build_deps(process_wishlist_automatically=stub)
result = auto_process_wishlist({'_automation_id': 'a'}, deps)
assert result == {'status': 'error', 'error': 'boom'}
# ─── scan_watchlist ──────────────────────────────────────────────────
class TestScanWatchlist:
def test_fresh_scan_reports_summary_stats(self):
# Worker reassigns the state dict mid-run — handler detects
# via id() change and reports stats.
states = [
{'summary': {}},
{'summary': {
'total_artists': 5,
'successful_scans': 4,
'new_tracks_found': 12,
'tracks_added_to_wishlist': 8,
}},
]
idx = {'i': 0}
def get_state():
return states[idx['i']]
def stub(**_kwargs):
idx['i'] = 1 # simulate the worker swapping the dict
deps = _build_deps(
process_watchlist_scan_automatically=stub,
get_watchlist_scan_state=get_state,
)
result = auto_scan_watchlist({}, deps)
assert result == {
'status': 'completed',
'artists_scanned': 5,
'successful_scans': 4,
'new_tracks_found': 12,
'tracks_added_to_wishlist': 8,
}
def test_no_fresh_scan_returns_bare_completed(self):
# Same dict identity before and after = no fresh scan ran.
same_dict = {'summary': {'total_artists': 999}}
deps = _build_deps(
process_watchlist_scan_automatically=lambda **_k: None,
get_watchlist_scan_state=lambda: same_dict,
)
result = auto_scan_watchlist({}, deps)
assert result == {'status': 'completed'}
def test_handler_swallows_exceptions(self):
def stub(**_kwargs):
raise ValueError('no scanner')
deps = _build_deps(process_watchlist_scan_automatically=stub)
result = auto_scan_watchlist({}, deps)
assert result == {'status': 'error', 'error': 'no scanner'}
# ─── scan_library ────────────────────────────────────────────────────
@dataclass
class _StubScanManager:
"""Minimal fake of ``web_scan_manager`` — records calls + lets
tests script its responses."""
request_responses: List[dict] = field(default_factory=list)
status_responses: List[dict] = field(default_factory=list)
request_calls: List[str] = field(default_factory=list)
def request_scan(self, label: str) -> dict:
self.request_calls.append(label)
return self.request_responses.pop(0) if self.request_responses else {'status': 'queued'}
def get_scan_status(self) -> dict:
return self.status_responses.pop(0) if self.status_responses else {'status': 'idle'}
class TestScanLibrary:
def test_no_scan_manager_returns_error(self):
deps = _build_deps(web_scan_manager=None)
result = auto_scan_library({'_automation_id': 'a'}, deps)
assert result == {'status': 'error', 'reason': 'Scan manager not available'}
def test_already_tracked_returns_skipped(self):
# Pre-set the state flag — handler should short-circuit.
state = AutomationState()
state.scan_library_automation_id = 'someone-else'
scanner = _StubScanManager(request_responses=[{'status': 'queued'}])
deps = _build_deps(state=state, web_scan_manager=scanner)
result = auto_scan_library({'_automation_id': 'a'}, deps)
assert result == {'status': 'skipped', 'reason': 'Scan already being tracked'}
assert scanner.request_calls == ['Automation trigger (additional batch)']
def test_scan_completes_normally(self):
# request_scan returns scheduled; first poll = scheduled;
# second poll = scanning; third poll = idle.
scanner = _StubScanManager(
request_responses=[{'status': 'scheduled', 'delay_seconds': 5}],
status_responses=[
{'status': 'scheduled'},
{'status': 'scanning', 'elapsed_seconds': 10, 'max_time_seconds': 100},
{'status': 'idle'},
],
)
progress: List[dict] = []
def stub_progress(automation_id, **kwargs):
progress.append({'aid': automation_id, **kwargs})
deps = _build_deps(
web_scan_manager=scanner,
update_progress=stub_progress,
)
# Patch time.sleep so the test runs instantly.
import core.automation.handlers.scan_library as module
original = module.time.sleep
module.time.sleep = lambda _: None
try:
result = auto_scan_library({'_automation_id': 'auto-1'}, deps)
finally:
module.time.sleep = original
assert result['status'] == 'completed'
assert result.get('_manages_own_progress') is True
# State flag cleaned up after run
assert deps.state.scan_library_automation_id is None
# Progress phases emitted: scheduled, scan-start, scanning, completed
statuses = [p.get('status') for p in progress]
assert 'finished' in statuses
def test_state_cleanup_on_exception(self):
class ExplodingScanner:
def request_scan(self, _):
raise RuntimeError('boom')
def get_scan_status(self):
return {'status': 'idle'}
progress: List[dict] = []
deps = _build_deps(
web_scan_manager=ExplodingScanner(),
update_progress=lambda aid, **kw: progress.append({'aid': aid, **kw}),
)
result = auto_scan_library({'_automation_id': 'auto-x'}, deps)
assert result['status'] == 'error'
assert result['_manages_own_progress'] is True
# State flag still cleaned up
assert deps.state.scan_library_automation_id is None
# Error progress emitted
assert any(p.get('status') == 'error' for p in progress)
# ─── AutomationState ──────────────────────────────────────────────────
class TestAutomationState:
def test_default_state(self):
s = AutomationState()
assert s.scan_library_automation_id is None
assert s.db_update_automation_id is None
assert s.pipeline_running is False
assert s.is_scan_library_active() is False
assert s.is_pipeline_running() is False
def test_set_scan_library_id(self):
s = AutomationState()
s.set_scan_library_id('auto-1')
assert s.scan_library_automation_id == 'auto-1'
assert s.is_scan_library_active() is True
s.set_scan_library_id(None)
assert s.is_scan_library_active() is False
def test_set_pipeline_running(self):
s = AutomationState()
s.set_pipeline_running(True)
assert s.is_pipeline_running() is True
s.set_pipeline_running(False)
assert s.is_pipeline_running() is False
def test_concurrent_set_safe_via_lock(self):
# Smoke test: two threads flipping the same field don't crash.
# Lock ensures the final value is consistent.
s = AutomationState()
def worker():
for _ in range(100):
s.set_pipeline_running(True)
s.set_pipeline_running(False)
threads = [threading.Thread(target=worker) for _ in range(4)]
for t in threads:
t.start()
for t in threads:
t.join()
assert s.pipeline_running is False

View file

@ -0,0 +1,243 @@
"""Boundary tests for the progress + history callbacks extracted
from ``web_server._register_automation_handlers``.
The callbacks are wired by the engine via ``register_progress_callbacks``;
each test invokes the extracted top-level function with stub deps
and verifies the right downstream call fires."""
from __future__ import annotations
import threading
from typing import Any, Dict, List, Tuple
import pytest
from core.automation.deps import AutomationDeps, AutomationState
from core.automation.handlers.progress_callbacks import (
progress_init,
progress_finish,
record_history,
on_library_scan_completed,
register_library_scan_completed_emitter,
)
def _build_deps(**overrides) -> AutomationDeps:
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
defaults = dict(
engine=object(),
state=AutomationState(),
config_manager=object(),
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': {}},
init_automation_progress=lambda *a, **k: None,
record_progress_history=lambda *a, **k: None,
)
defaults.update(overrides)
return AutomationDeps(**defaults) # type: ignore[arg-type]
# ─── progress_init ───────────────────────────────────────────────────
class TestProgressInit:
def test_forwards_to_init_automation_progress(self):
captured: List[Tuple] = []
def fake(aid, name, action_type):
captured.append((aid, name, action_type))
deps = _build_deps(init_automation_progress=fake)
progress_init('auto-1', 'My Auto', 'wishlist', deps)
assert captured == [('auto-1', 'My Auto', 'wishlist')]
# ─── progress_finish ─────────────────────────────────────────────────
class TestProgressFinish:
def test_skips_when_handler_manages_own_progress(self):
# Handler set the flag — engine callback must NOT emit a
# second 'finished' over the top of the handler's own.
calls: List[Dict] = []
deps = _build_deps(update_progress=lambda *a, **k: calls.append({'a': a, 'k': k}))
progress_finish('auto-1', {'_manages_own_progress': True, 'status': 'completed'}, deps)
assert calls == []
def test_completed_emits_finished_status(self):
calls: List[Dict] = []
deps = _build_deps(update_progress=lambda aid, **kw: calls.append({'aid': aid, **kw}))
progress_finish('auto-1', {'status': 'completed'}, deps)
assert len(calls) == 1
assert calls[0]['aid'] == 'auto-1'
assert calls[0]['status'] == 'finished'
assert calls[0]['progress'] == 100
assert calls[0]['phase'] == 'Complete'
assert calls[0]['log_type'] == 'success'
def test_error_status_emits_error_phase(self):
calls: List[Dict] = []
deps = _build_deps(update_progress=lambda aid, **kw: calls.append({'aid': aid, **kw}))
progress_finish('auto-1', {'status': 'error', 'error': 'boom'}, deps)
assert calls[0]['status'] == 'error'
assert calls[0]['phase'] == 'Error'
assert calls[0]['log_line'] == 'boom'
assert calls[0]['log_type'] == 'error'
def test_msg_falls_back_through_keys(self):
# error -> reason -> status -> 'done'
calls: List[Dict] = []
deps = _build_deps(update_progress=lambda aid, **kw: calls.append({'aid': aid, **kw}))
progress_finish('auto-1', {'status': 'completed', 'reason': 'all good'}, deps)
assert calls[0]['log_line'] == 'all good'
def test_msg_default_done(self):
calls: List[Dict] = []
deps = _build_deps(update_progress=lambda aid, **kw: calls.append({'aid': aid, **kw}))
progress_finish('auto-1', {}, deps)
assert calls[0]['log_line'] == 'done'
# ─── record_history ──────────────────────────────────────────────────
class TestRecordHistory:
def test_passes_db_to_recorder(self):
captured: List[Tuple] = []
db_obj = object()
deps = _build_deps(
get_database=lambda: db_obj,
record_progress_history=lambda aid, result, db: captured.append((aid, result, db)),
)
record_history('auto-1', {'status': 'completed'}, deps)
assert captured == [('auto-1', {'status': 'completed'}, db_obj)]
# ─── on_library_scan_completed ───────────────────────────────────────
class TestOnLibraryScanCompleted:
def test_no_engine_skips(self):
deps = _build_deps(engine=None)
# Should not raise.
on_library_scan_completed(deps)
def test_emits_event_with_server_type(self):
emits: List[Tuple] = []
class _Engine:
def emit(self, name, payload):
emits.append((name, payload))
class _ScanMgr:
_current_server_type = 'plex'
deps = _build_deps(engine=_Engine(), web_scan_manager=_ScanMgr())
on_library_scan_completed(deps)
assert emits == [('library_scan_completed', {'server_type': 'plex'})]
def test_unknown_server_type_when_attr_missing(self):
emits: List[Tuple] = []
class _Engine:
def emit(self, name, payload):
emits.append((name, payload))
deps = _build_deps(engine=_Engine(), web_scan_manager=object())
on_library_scan_completed(deps)
assert emits[0][1] == {'server_type': 'unknown'}
# ─── register_library_scan_completed_emitter ─────────────────────────
class TestRegisterEmitter:
def test_no_scan_manager_noop(self):
# No web_scan_manager → no callback registered, no error.
deps = _build_deps(web_scan_manager=None)
register_library_scan_completed_emitter(deps)
def test_registers_callback_with_scan_manager(self):
callbacks: List = []
class _ScanMgr:
_current_server_type = 'plex'
def add_scan_completion_callback(self, cb):
callbacks.append(cb)
deps = _build_deps(web_scan_manager=_ScanMgr())
register_library_scan_completed_emitter(deps)
assert len(callbacks) == 1
# The registered callback must invoke without args (web_scan_manager
# calls completion callbacks with no params).
# Verify it does fire on_library_scan_completed when invoked.
emits: List = []
class _Engine:
def emit(self, name, payload):
emits.append((name, payload))
deps2 = _build_deps(engine=_Engine(), web_scan_manager=_ScanMgr())
register_library_scan_completed_emitter(deps2)
# The lambda captured deps2; we need to grab the registered
# callback to invoke it. Re-register and capture.
captured = []
class _Mgr2:
_current_server_type = 'jellyfin'
def add_scan_completion_callback(self, cb):
captured.append(cb)
deps3 = _build_deps(engine=_Engine(), web_scan_manager=_Mgr2())
emits3 = []
deps3 = _build_deps(
engine=type('E', (), {'emit': lambda self, n, p: emits3.append((n, p))})(),
web_scan_manager=_Mgr2(),
)
register_library_scan_completed_emitter(deps3)
captured[0]() # invoke the registered callback
assert emits3 == [('library_scan_completed', {'server_type': 'jellyfin'})]

File diff suppressed because it is too large Load diff