diff --git a/core/automation/blocks.py b/core/automation/blocks.py index de106ddb..0d4f1afb 100644 --- a/core/automation/blocks.py +++ b/core/automation/blocks.py @@ -123,6 +123,14 @@ TRIGGERS: list[dict] = [ {"type": "webhook_received", "label": "Webhook Received", "icon": "globe", "scope": "both", "description": "When an external API request is received (POST /api/v1/request)", "available": True, "variables": ["query", "request_id", "source"]}, + + # ── Video side (scope='video') — the post-download scan chain triggers ── + {"type": "video_batch_complete", "label": "Video Download Batch Done", "icon": "check-circle", "scope": "video", + "description": "When a batch of video downloads finishes", "available": True, + "variables": ["completed"]}, + {"type": "video_library_scan_completed", "label": "Video Library Scan Done", "icon": "hard-drive", "scope": "video", + "description": "When the media server finishes rescanning your video sections", "available": True, + "variables": ["server"]}, ] @@ -212,6 +220,21 @@ ACTIONS: list[dict] = [ {"value": "deep", "label": "Deep (also remove missing)"}], "default": "full"} ]}, + # Post-download chain actions (two stages, like music's scan_library + + # start_database_update). Stage 1 nudges the server; stage 2 reads it in. + {"type": "video_scan_server", "label": "Scan Video Server", "icon": "refresh", "scope": "video", + "description": "Tell the media server to rescan your video sections, then fire 'Video Library Scan Done'", "available": True, + "config_fields": [ + {"key": "debounce_seconds", "type": "number", "label": "Wait for indexing (sec)", "default": 120, "min": 10} + ]}, + {"type": "video_update_database", "label": "Update Video Database", "icon": "database", "scope": "video", + "description": "Read newly-indexed media from the server into SoulSync (incremental)", "available": True, + "config_fields": [ + {"key": "mode", "type": "select", "label": "Mode", + "options": [{"value": "incremental", "label": "Incremental (recent only)"}, + {"value": "full", "label": "Full (add + refresh)"}], + "default": "incremental"} + ]}, ] diff --git a/core/automation/handlers/registration.py b/core/automation/handlers/registration.py index 65c641f9..9aed539f 100644 --- a/core/automation/handlers/registration.py +++ b/core/automation/handlers/registration.py @@ -35,7 +35,9 @@ from core.automation.handlers.download_cleanup import ( ) 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.video_scan_library import auto_video_scan_library +from core.automation.handlers.video_scan_library import ( + auto_video_scan_library, auto_video_scan_server, auto_video_update_database, +) from core.automation.handlers.progress_callbacks import ( progress_init, progress_finish, @@ -175,6 +177,15 @@ def register_all(deps: AutomationDeps) -> None: 'video_scan_library', lambda config: auto_video_scan_library(config, deps), ) + # Post-download chain: scan the server, then (on the scan-done event) update the DB. + engine.register_action_handler( + 'video_scan_server', + lambda config: auto_video_scan_server(config, deps), + ) + engine.register_action_handler( + 'video_update_database', + lambda config: auto_video_update_database(config, deps), + ) # Progress + history callbacks: the engine invokes these around # each handler run. Lift the closures from diff --git a/core/automation/handlers/video_scan_library.py b/core/automation/handlers/video_scan_library.py index fe9ccde9..ea7f7703 100644 --- a/core/automation/handlers/video_scan_library.py +++ b/core/automation/handlers/video_scan_library.py @@ -25,6 +25,7 @@ phase string with a generic 'completed' label. from __future__ import annotations +import time from typing import Any, Callable, Dict, Optional from core.automation.deps import AutomationDeps @@ -143,3 +144,78 @@ def auto_video_scan_library( log_type='error', ) return {'status': 'error', 'error': str(e), '_manages_own_progress': True} + + +# ── post-download chain (video twin of music's scan_library → start_database_update) ── +# Split into two stages so the calendar/library reflect a download promptly, mirroring +# the music side: tell the server to rescan, then (after it has had time to index) read +# the new state into video.db. Stage 1 emits 'video_library_scan_completed' on a debounce +# (like music's web_scan_manager time-based completion); stage 2 listens for that. + +def auto_video_scan_server( + config: Dict[str, Any], + deps: AutomationDeps, + *, + server_refresh: Optional[Callable[[], Dict[str, Any]]] = None, + sleep: Optional[Callable[[float], None]] = None, + emit: Optional[Callable[[str, Dict[str, Any]], None]] = None, +) -> Dict[str, Any]: + """Stage 1: tell the media server to rescan the video sections, wait a debounce + for it to index, then fire 'video_library_scan_completed' so the DB-update twin + reads the fresh state. (Video twin of music's scan_library.)""" + server_refresh = server_refresh or _default_server_refresh + sleep = sleep or time.sleep + emit = emit or deps.engine.emit + automation_id = config.get('_automation_id') + try: + debounce = int(config.get('debounce_seconds') or 120) + except (TypeError, ValueError): + debounce = 120 + try: + deps.update_progress(automation_id, phase='Asking media server to rescan…', progress=20, + log_line='Triggering server-side video scan', log_type='info') + refresh = server_refresh() or {} + if not refresh.get('ok'): + deps.update_progress(automation_id, + log_line='Server scan trigger unavailable: ' + str(refresh.get('error') or 'unknown'), + log_type='warning') + deps.update_progress(automation_id, phase=f'Waiting for the server to index ({debounce}s)…', progress=55) + sleep(debounce) + emit('video_library_scan_completed', {'server': refresh.get('server') or ''}) + deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', + log_line='Server scan done — updating the database', log_type='success') + return {'status': 'completed', '_manages_own_progress': True} + except Exception as e: # noqa: BLE001 + 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} + + +def auto_video_update_database( + config: Dict[str, Any], + deps: AutomationDeps, + *, + run_video_scan: Optional[Callable[[str], Dict[str, Any]]] = None, +) -> Dict[str, Any]: + """Stage 2: read the (now-rescanned) server into video.db — INCREMENTAL by default + (newest-first, stop after N consecutive known). Video twin of start_database_update.""" + run_video_scan = run_video_scan or _default_run_video_scan + automation_id = config.get('_automation_id') + mode = config.get('mode') or 'incremental' + try: + deps.update_progress(automation_id, phase='Reading new media into SoulSync…', progress=40) + result = run_video_scan(mode) or {} + if result.get('state') == 'error': + err = result.get('error') or 'Video database update failed' + deps.update_progress(automation_id, status='error', phase='Error', log_line=err, log_type='error') + return {'status': 'error', 'error': err, '_manages_own_progress': True} + movies = int(result.get('movies', 0) or 0) + shows = int(result.get('shows', 0) or 0) + episodes = int(result.get('episodes', 0) or 0) + deps.update_progress(automation_id, status='finished', progress=100, phase='Complete', + log_line=f'Video database updated: {movies} movies, {shows} shows, {episodes} episodes', + log_type='success') + return {'status': 'completed', '_manages_own_progress': True, + 'movies': movies, 'shows': shows, 'episodes': episodes} + except Exception as e: # noqa: BLE001 + 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} diff --git a/core/automation_engine.py b/core/automation_engine.py index 0e1aa477..4d83defa 100644 --- a/core/automation_engine.py +++ b/core/automation_engine.py @@ -161,6 +161,24 @@ SYSTEM_AUTOMATIONS = [ # app startup; post-download freshness is handled by the event chain instead. 'initial_delay': 6 * 60 * 60, }, + # Post-download chain (video twin of music's batch_complete → scan_library → + # library_scan_completed → start_database_update). Event-based, so a finished + # video download refreshes the server then pulls the new media into video.db. + { + 'name': 'Auto-Scan Video After Downloads', + 'trigger_type': 'video_batch_complete', + 'trigger_config': {}, + 'action_type': 'video_scan_server', + 'owned_by': 'video', + }, + { + 'name': 'Auto-Update Video Database After Scan', + 'trigger_type': 'video_library_scan_completed', + 'trigger_config': {}, + 'action_type': 'video_update_database', + 'action_config': {'mode': 'incremental'}, + 'owned_by': 'video', + }, ] diff --git a/core/video/download_events.py b/core/video/download_events.py new file mode 100644 index 00000000..035d007a --- /dev/null +++ b/core/video/download_events.py @@ -0,0 +1,40 @@ +"""Video download events — a tiny publish/subscribe bridge. + +ISOLATION: ``core/video`` must not import the automation engine (that's music-side +shared infra). So when a batch of video downloads finishes, the monitor publishes +to this callback registry instead of touching the engine directly. The shared side +(web_server) registers a callback that forwards to +``automation_engine.emit('video_batch_complete', …)`` — the same one-way bridge the +music ``web_scan_manager`` uses for ``library_scan_completed``. +""" + +from __future__ import annotations + +from typing import Callable + +from utils.logging_config import get_logger + +logger = get_logger("video_download_events") + +_batch_complete_callbacks: list[Callable[[dict], None]] = [] + + +def register_batch_complete_callback(cb: Callable[[dict], None]) -> None: + """Subscribe to 'a batch of video downloads just finished'. Idempotent.""" + if cb not in _batch_complete_callbacks: + _batch_complete_callbacks.append(cb) + + +def notify_batch_complete(data: dict | None = None) -> None: + """Fire every registered batch-complete callback (best-effort; one failing + subscriber never blocks the others or the monitor).""" + payload = data or {} + for cb in list(_batch_complete_callbacks): + try: + cb(payload) + except Exception: + logger.exception("video batch-complete callback failed") + + +def _reset_for_tests() -> None: + _batch_complete_callbacks.clear() diff --git a/core/video/download_monitor.py b/core/video/download_monitor.py index 34275054..ce0abf99 100644 --- a/core/video/download_monitor.py +++ b/core/video/download_monitor.py @@ -234,11 +234,14 @@ def _tick(db) -> None: download_dir = str(config_manager.get("soulseek.download_path", "") or "") transfers = list_downloads() live_ids = set() + completed_now = 0 for dl in active: live_ids.add(dl["id"]) upd = process_download(dl, transfers, download_dir, lister=_walk, mover=_move) if not upd: continue + if upd.get("status") == "completed": + completed_now += 1 if upd.get("_missing"): n = _misses.get(dl["id"], 0) + 1 _misses[dl["id"]] = n @@ -258,6 +261,16 @@ def _tick(db) -> None: logger.exception("video download %s: failed to persist update", dl.get("id")) for k in [k for k in _misses if k not in live_ids]: _misses.pop(k, None) + # Batch complete: we finished ≥1 download this tick AND nothing is left in + # flight (queued/downloading/searching). Fires once, on the transition to + # empty — the next tick early-returns. Publishes to the event bridge so the + # 'Auto-Scan Video After Downloads' automation can refresh the server. + if completed_now and not db.get_active_video_downloads(): + try: + from core.video.download_events import notify_batch_complete + notify_batch_complete({"completed": completed_now}) + except Exception: + logger.exception("video monitor: batch-complete notify failed") def _run(db_provider) -> None: diff --git a/tests/automation/test_automation_blocks.py b/tests/automation/test_automation_blocks.py index 3cd5122b..742f4c9d 100644 --- a/tests/automation/test_automation_blocks.py +++ b/tests/automation/test_automation_blocks.py @@ -125,6 +125,19 @@ def test_music_scope_matches_legacy_full_lists_minus_video(): assert {b['type'] for b in music[key]} == expected +def test_post_download_chain_blocks_are_video_scoped(): + music = blocks.blocks_for_scope('music') + video = blocks.blocks_for_scope('video') + trig = {t['type'] for t in video['triggers']} + act = {a['type'] for a in video['actions']} + assert {'video_batch_complete', 'video_library_scan_completed'} <= trig + assert {'video_scan_server', 'video_update_database'} <= act + mtrig = {t['type'] for t in music['triggers']} + mact = {a['type'] for a in music['actions']} + assert not ({'video_batch_complete', 'video_library_scan_completed'} & mtrig) + assert not ({'video_scan_server', 'video_update_database'} & mact) + + def test_video_scan_library_block_shape(): action = next(a for a in blocks.ACTIONS if a['type'] == 'video_scan_library') assert action['scope'] == 'video' diff --git a/tests/automation/test_handler_registration.py b/tests/automation/test_handler_registration.py index 23441a05..ef18334c 100644 --- a/tests/automation/test_handler_registration.py +++ b/tests/automation/test_handler_registration.py @@ -55,6 +55,8 @@ EXPECTED_ACTION_NAMES = frozenset({ 'search_and_download', # Video side (isolated app, shared engine). 'video_scan_library', + 'video_scan_server', + 'video_update_database', }) # Action names that MUST register a guard (duplicate-run prevention). diff --git a/tests/automation/test_handler_video_scan_library.py b/tests/automation/test_handler_video_scan_library.py index 8411a3d9..c39065b0 100644 --- a/tests/automation/test_handler_video_scan_library.py +++ b/tests/automation/test_handler_video_scan_library.py @@ -173,3 +173,65 @@ class TestHandlerNeverRaises: {}, deps, server_refresh=_refresh_ok(), run_video_scan=_scan_done(), ) assert result['status'] == 'completed' + + +# ── post-download chain: video_scan_server (stage 1) ─────────────────────── +from core.automation.handlers.video_scan_library import ( # noqa: E402 + auto_video_scan_server, auto_video_update_database) + + +class TestScanServerStage: + def test_refreshes_waits_then_emits_scan_done(self): + deps = _RecordingDeps() + events = [] + slept = [] + r = auto_video_scan_server( + {'_automation_id': 'a', 'debounce_seconds': 90}, deps, + server_refresh=_refresh_ok(), sleep=lambda s: slept.append(s), + emit=lambda ev, data: events.append((ev, data))) + assert r['status'] == 'completed' + assert slept == [90] # waited the debounce + assert events == [('video_library_scan_completed', {'server': ''})] + + def test_default_debounce_and_server_unavailable_still_emits(self): + deps = _RecordingDeps() + events = [] + auto_video_scan_server( + {'_automation_id': 'a'}, deps, + server_refresh=lambda: {'ok': False, 'error': 'no server'}, + sleep=lambda s: None, emit=lambda ev, data: events.append(ev)) + assert events == ['video_library_scan_completed'] # fires even if refresh failed + assert 'warning' in deps.log_types() + + def test_never_raises(self): + deps = _RecordingDeps() + r = auto_video_scan_server( + {'_automation_id': 'a'}, deps, + server_refresh=lambda: (_ for _ in ()).throw(RuntimeError('boom')), + sleep=lambda s: None, emit=lambda *a: None) + assert r['status'] == 'error' and r['error'] == 'boom' + + +# ── post-download chain: video_update_database (stage 2) ─────────────────── +class TestUpdateDatabaseStage: + def test_incremental_read_returns_counts(self): + deps = _RecordingDeps() + r = auto_video_update_database({'_automation_id': 'a'}, deps, run_video_scan=_scan_done(2, 1, 5)) + assert r == {'status': 'completed', '_manages_own_progress': True, + 'movies': 2, 'shows': 1, 'episodes': 5} + + def test_defaults_to_incremental_mode(self): + seen = {} + + def _scan(mode): + seen['mode'] = mode + return {'state': 'done'} + + auto_video_update_database({'_automation_id': 'a'}, _RecordingDeps(), run_video_scan=_scan) + assert seen['mode'] == 'incremental' + + def test_scan_error_propagates(self): + deps = _RecordingDeps() + r = auto_video_update_database({'_automation_id': 'a'}, deps, + run_video_scan=lambda m: {'state': 'error', 'error': 'no server'}) + assert r['status'] == 'error' and r['error'] == 'no server' diff --git a/tests/test_video_download_events.py b/tests/test_video_download_events.py new file mode 100644 index 00000000..e5cbe5cc --- /dev/null +++ b/tests/test_video_download_events.py @@ -0,0 +1,79 @@ +"""Video download → batch-complete event bridge + the monitor detection that fires it. + +The isolated monitor (core/video) publishes 'batch complete' to a callback registry; +web_server bridges that to automation_engine.emit('video_batch_complete', …). These +pin the publish side: the registry, and the monitor firing exactly once when the last +download finishes (never while work is still in flight). +""" + +from __future__ import annotations + +import core.video.download_monitor as mon +import core.video.download_events as events + + +def setup_function(_): + events._reset_for_tests() + + +# ── the event registry ───────────────────────────────────────────────────── + +def test_register_and_notify_fires_callbacks(): + got = [] + events.register_batch_complete_callback(lambda d: got.append(d)) + events.notify_batch_complete({"completed": 3}) + assert got == [{"completed": 3}] + + +def test_register_is_idempotent_and_one_failure_is_isolated(): + got = [] + cb = lambda d: got.append(d) + events.register_batch_complete_callback(cb) + events.register_batch_complete_callback(cb) # dup → ignored + events.register_batch_complete_callback(lambda d: (_ for _ in ()).throw(RuntimeError("boom"))) + events.notify_batch_complete({}) # must not raise despite the bad cb + assert got == [{}] # the good cb fired exactly once + + +# ── the monitor fires it on the last completion ──────────────────────────── + +class _FakeDb: + def __init__(self, active): + self._active = list(active) + + def get_active_video_downloads(self): + return list(self._active) + + def update_video_download(self, dl_id, **kw): + if kw.get("status") in ("completed", "failed", "cancelled"): + self._active = [d for d in self._active if d["id"] != dl_id] + + +def _patch(monkeypatch, result): + monkeypatch.setattr(mon, "list_downloads", lambda: []) + monkeypatch.setattr(mon, "process_download", lambda dl, *a, **k: dict(result)) + + +def test_tick_fires_batch_complete_when_last_download_finishes(monkeypatch): + fired = [] + events.register_batch_complete_callback(lambda d: fired.append(d)) + _patch(monkeypatch, {"status": "completed", "progress": 100.0}) + db = _FakeDb([{"id": 1, "status": "downloading", "filename": "a.mkv"}]) + mon._tick(db) + assert fired == [{"completed": 1}] # one download done, none left → fired once + + +def test_tick_does_not_fire_while_a_download_is_still_active(monkeypatch): + fired = [] + events.register_batch_complete_callback(lambda d: fired.append(d)) + # one completes, one is still downloading → batch NOT done yet + monkeypatch.setattr(mon, "list_downloads", lambda: []) + + def _proc(dl, *a, **k): + return {"status": "completed", "progress": 100.0} if dl["id"] == 1 else {"progress": 50.0} + + monkeypatch.setattr(mon, "process_download", _proc) + db = _FakeDb([{"id": 1, "status": "downloading", "filename": "a.mkv"}, + {"id": 2, "status": "downloading", "filename": "b.mkv"}]) + mon._tick(db) + assert fired == [] # 2 still active → no batch-complete diff --git a/web_server.py b/web_server.py index c9faf469..0b14d5a3 100644 --- a/web_server.py +++ b/web_server.py @@ -1259,6 +1259,17 @@ def _register_automation_handlers(): ) _register_extracted_handlers(_automation_deps) + # Bridge the isolated video download monitor's batch-complete signal into the + # automation engine (core/video can't import the engine). Mirrors how the music + # web_scan_manager forwards library_scan_completed. Fires the 'Auto-Scan Video + # After Downloads' automation when a batch of video downloads finishes. + if automation_engine is not None: + try: + from core.video.download_events import register_batch_complete_callback + register_batch_complete_callback( + lambda data: automation_engine.emit('video_batch_complete', data or {})) + except Exception: + logger.exception("Could not wire video batch-complete -> automation engine") logger.info("Automation action handlers registered")