Video scan family: make every action movie/TV-aware + deep scans are real actions

The deep-scan action types weren't selectable builder actions, and Scan Video Server
/ Update Video Database had no movie-vs-TV dimension — inconsistent with the rest.

- video_deep_scan_tv / video_deep_scan_movies are now proper builder blocks
  (Deep Scan TV/Movie Library), not just system-automation action types.
- video_scan_server + video_update_database gain a media_type ('all'|'movie'|'show')
  config + selector, threaded through. The post-download chain carries the scope on
  the scan-done event, so a TV-only rescan updates only TV (stage 2 inherits it).
- refresh_video_server_sections / Plex+Jellyfin refresh_sections scope the server
  nudge to the chosen library; auto_video_scan_library now nudges only its library.
- shared normalize_media_type() in sources; update_database skips cleanly when the
  singleton scanner is busy. Defaults stay 'all' so existing chains are unchanged.

Seam tests for refresh scoping, scan-server scope+event, update-db scope/inherit/skip.

core/automation/blocks.py

@@ -230,6 +230,11 @@ ACTIONS: list[dict] = [
     {"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": "media_type", "type": "select", "label": "Library",
+          "options": [{"value": "all", "label": "Movies + TV"},
+                      {"value": "movie", "label": "Movies only"},
+                      {"value": "show", "label": "TV only"}],
+          "default": "all"},
          {"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",
@@ -238,8 +243,20 @@ ACTIONS: list[dict] = [
          {"key": "mode", "type": "select", "label": "Mode",
           "options": [{"value": "incremental", "label": "Incremental (recent only)"},
                       {"value": "full", "label": "Full (add + refresh)"}],
-          "default": "incremental"}
+          "default": "incremental"},
+         {"key": "media_type", "type": "select", "label": "Library",
+          "options": [{"value": "all", "label": "Movies + TV"},
+                      {"value": "movie", "label": "Movies only"},
+                      {"value": "show", "label": "TV only"}],
+          "default": "all"}
      ]},
+    # Per-library deep-scan presets (the system 'Auto-Deep Scan TV/Movie Library' run
+    # these). Scope + deep mode are baked in by the registration wrapper, so no config
+    # fields — drag one in and it just deep-scans that library.
+    {"type": "video_deep_scan_tv", "label": "Deep Scan TV Library", "icon": "search", "scope": "video",
+     "description": "Deep-rescan the TV library: re-read every show and prune ones the server no longer has (never touches movies)", "available": True},
+    {"type": "video_deep_scan_movies", "label": "Deep Scan Movie Library", "icon": "search", "scope": "video",
+     "description": "Deep-rescan the Movie library: re-read every movie and prune ones the server no longer has (never touches TV)", "available": True},
     {"type": "video_add_airing_episodes", "label": "Wishlist Today's Airings", "icon": "calendar", "scope": "video",
      "description": "Sonarr-style: add every episode airing today (for shows you follow) to the wishlist, skipping ones you already own", "available": True},
 ]

core/automation/handlers/video_scan_library.py

@@ -31,10 +31,11 @@ from typing import Any, Callable, Dict, Optional
 from core.automation.deps import AutomationDeps
 
 
-def _default_server_refresh() -> Dict[str, Any]:
+def _default_server_refresh(media_type: str = "all") -> Dict[str, Any]:
-    """Production wiring: nudge the media server to rescan its video sections."""
+    """Production wiring: nudge the media server to rescan its video sections.
+    ``media_type`` scopes it to the Movie or TV section; 'all' nudges both."""
     from core.video.sources import refresh_video_server_sections
-    return refresh_video_server_sections()
+    return refresh_video_server_sections(media_type)
 
 
 def _default_run_video_scan(mode: str, media_type: str = "all") -> Dict[str, Any]:
@@ -83,11 +84,11 @@ def auto_video_scan_library(
             log_type='info',
         )
 
-        # Step 1 — best-effort server nudge. A server that can't be triggered
+        # Step 1 — best-effort server nudge, scoped to the same library we're about
-        # (none configured, or an adapter without refresh support) is surfaced
+        # to read. A server that can't be triggered (none configured, or an adapter
-        # as a warning, NOT a hard failure: the read below still mirrors whatever
+        # without refresh support) is surfaced as a warning, NOT a hard failure: the
-        # the server currently reports, so the automation stays useful.
+        # read below still mirrors whatever the server currently reports.
-        refresh = server_refresh() or {}
+        refresh = server_refresh(media_type) or {}
         if not refresh.get('ok'):
             deps.update_progress(
                 automation_id,
@@ -192,21 +193,25 @@ def auto_video_scan_server(
     sleep = sleep or time.sleep
     emit = emit or deps.engine.emit
     automation_id = config.get('_automation_id')
+    media_type = config.get('media_type') or 'all'
+    lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video')
     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,
+        deps.update_progress(automation_id, phase=f'Asking media server to rescan {lib_label}…', progress=20,
-                             log_line='Triggering server-side video scan', log_type='info')
+                             log_line=f'Triggering server-side {lib_label} scan', log_type='info')
-        refresh = server_refresh() or {}
+        refresh = server_refresh(media_type) 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 ''})
+        # Carry the scope so stage 2 (Update DB) reads only the library we rescanned.
+        emit('video_library_scan_completed',
+             {'server': refresh.get('server') or '', 'media_type': media_type})
         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}
@@ -219,16 +224,30 @@ def auto_video_update_database(
     config: Dict[str, Any],
     deps: AutomationDeps,
     *,
-    run_video_scan: Optional[Callable[[str], Dict[str, Any]]] = None,
+    run_video_scan: Optional[Callable[..., 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."""
+    (newest-first, stop after N consecutive known). Video twin of start_database_update.
+
+    ``media_type`` scopes it to the Movie or TV library ('all' does both). When fired by
+    the post-download chain it inherits the scope of the scan that ran (carried on the
+    event), so a TV-only rescan updates only TV."""
     run_video_scan = run_video_scan or _default_run_video_scan
     automation_id = config.get('_automation_id')
     mode = config.get('mode') or 'incremental'
+    media_type = (config.get('media_type')
+                  or (config.get('_event_data') or {}).get('media_type')
+                  or 'all')
+    lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video')
     try:
-        deps.update_progress(automation_id, phase='Reading new media into SoulSync…', progress=40)
+        deps.update_progress(automation_id, phase=f'Reading new {lib_label} media into SoulSync…', progress=40)
-        result = run_video_scan(mode) or {}
+        result = run_video_scan(mode, media_type) or {}
+        if result.get('state') == 'in_progress':
+            deps.update_progress(automation_id, status='finished', phase='Skipped',
+                                 log_line='Another video scan is already running — skipping this run',
+                                 log_type='info')
+            return {'status': 'skipped', 'reason': 'a video scan is already running',
+                    '_manages_own_progress': True}
         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')
@@ -236,9 +255,14 @@ def auto_video_update_database(
         movies = int(result.get('movies', 0) or 0)
         shows = int(result.get('shows', 0) or 0)
         episodes = int(result.get('episodes', 0) or 0)
+        if media_type == 'movie':
+            summary = f'Video database updated: {movies} movies'
+        elif media_type == 'show':
+            summary = f'Video database updated: {shows} shows, {episodes} episodes'
+        else:
+            summary = f'Video database updated: {movies} movies, {shows} shows, {episodes} episodes'
         deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
-                             log_line=f'Video database updated: {movies} movies, {shows} shows, {episodes} episodes',
+                             log_line=summary, log_type='success')
-                             log_type='success')
         return {'status': 'completed', '_manages_own_progress': True,
                 'movies': movies, 'shows': shows, 'episodes': episodes}
     except Exception as e:  # noqa: BLE001

core/video/sources.py

@@ -266,17 +266,31 @@ def get_active_video_source():
     return _build_source(sel.get("movies") or None, sel.get("tv") or None)
 
 
-def refresh_video_server_sections():
+def normalize_media_type(media_type) -> str:
+    """'movie'|'show'|'all' — accepts the friendly aliases (movies/tv/series/…) the
+    UI and automation configs use. Movies and TV are independent libraries, so the
+    scan family is scoped by this everywhere."""
+    m = str(media_type or "all").lower()
+    if m in ("movie", "movies", "film", "films"):
+        return "movie"
+    if m in ("show", "shows", "tv", "series", "episode", "episodes"):
+        return "show"
+    return "all"
+
+
+def refresh_video_server_sections(media_type="all"):
     """Tell the active media server to rescan its selected VIDEO sections (so newly
-    downloaded files get indexed) — the video twin of music's 'Scan Library'. Returns
+    downloaded files get indexed) — the video twin of music's 'Scan Library'.
-    {ok, sections} or {ok: False, error}."""
+    ``media_type`` scopes it to one library ('movie' / 'show'); 'all' (default) nudges
+    both. Returns {ok, sections} or {ok: False, error}."""
+    media_type = normalize_media_type(media_type)
     src = get_active_video_source()
     if src is None:
         return {"ok": False, "error": "No video server configured"}
     if not hasattr(src, "refresh_sections"):
         return {"ok": False, "error": "This server doesn't support a scan trigger"}
     try:
-        return src.refresh_sections()
+        return src.refresh_sections(media_type)
     except Exception as e:   # noqa: BLE001 - surface any server error to the automation
         logger.exception("video sources: refresh failed")
         return {"ok": False, "error": str(e)}
@@ -346,11 +360,14 @@ class PlexVideoSource:
                 except Exception:
                     logger.exception("Plex: skipping show %s", getattr(sh, "title", "?"))
 
-    def refresh_sections(self) -> dict:
+    def refresh_sections(self, media_type="all") -> dict:
         """Tell Plex to rescan the selected video sections so freshly-downloaded files
-        get indexed. (plexapi ``section.update()`` triggers the library scan.)"""
+        get indexed. (plexapi ``section.update()`` triggers the library scan.)
+        ``media_type`` scopes it to the Movie or TV section; 'all' does both."""
         n = 0
         for kind, name in (("movie", self._movies_lib), ("show", self._tv_lib)):
+            if media_type != "all" and media_type != kind:
+                continue
             for s in self._scan_sections(kind, name):
                 try:
                     s.update()
@@ -526,15 +543,20 @@ class JellyfinVideoSource:
             "tv": [{"title": v.get("Name")} for v in self._views("tvshows")],
         }
 
-    def refresh_sections(self) -> dict:
+    def refresh_sections(self, media_type="all") -> dict:
         """Ask Jellyfin to rescan the selected video libraries (POST /Items/{id}/Refresh)
-        so freshly-downloaded files get indexed. _make_request is GET-only, so POST direct."""
+        so freshly-downloaded files get indexed. _make_request is GET-only, so POST direct.
+        ``media_type`` scopes it to the Movie or TV view; 'all' does both."""
         import requests
         base = (self._c.base_url or "").rstrip("/")
         if not base:
             return {"ok": False, "sections": 0}
         headers = {"X-Emby-Token": self._c.api_key or ""}
-        views = list(self._scan_views("movies", self._movies_lib)) + list(self._scan_views("tvshows", self._tv_lib))
+        views = []
+        if media_type in ("all", "movie"):
+            views += list(self._scan_views("movies", self._movies_lib))
+        if media_type in ("all", "show"):
+            views += list(self._scan_views("tvshows", self._tv_lib))
         n = 0
         for v in views:
             vid = v.get("Id")

tests/automation/test_handler_video_scan_library.py

@@ -34,7 +34,7 @@ class _RecordingDeps:
 
 
 def _refresh_ok(sections: int = 2):
-    return lambda: {'ok': True, 'sections': sections}
+    return lambda media_type=None: {'ok': True, 'sections': sections}
 
 
 def _scan_done(movies: int = 3, shows: int = 1, episodes: int = 9):
@@ -158,7 +158,7 @@ class TestServerUnavailable:
         deps = _RecordingDeps()
         result = auto_video_scan_library(
             {'_automation_id': 'a'}, deps,
-            server_refresh=lambda: {'ok': False, 'error': 'No video server configured'},
+            server_refresh=lambda media_type=None: {'ok': False, 'error': 'No video server configured'},
             run_video_scan=_scan,
         )
         assert scanned.get('ran') is True
@@ -169,7 +169,7 @@ class TestServerUnavailable:
         deps = _RecordingDeps()
         result = auto_video_scan_library(
             {'_automation_id': 'a'}, deps,
-            server_refresh=lambda: None, run_video_scan=_scan_done(),
+            server_refresh=lambda media_type=None: None, run_video_scan=_scan_done(),
         )
         assert result['status'] == 'completed'
 
@@ -203,7 +203,7 @@ class TestHandlerNeverRaises:
         deps = _RecordingDeps()
         result = auto_video_scan_library(
             {'_automation_id': 'a'}, deps,
-            server_refresh=lambda: (_ for _ in ()).throw(RuntimeError('boom')),
+            server_refresh=lambda media_type=None: (_ for _ in ()).throw(RuntimeError('boom')),
         )
         assert result['status'] == 'error'
         assert result['error'] == 'boom'
@@ -243,14 +243,14 @@ class TestScanServerStage:
             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': ''})]
+        assert events == [('video_library_scan_completed', {'server': '', 'media_type': 'all'})]
 
     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'},
+            server_refresh=lambda media_type=None: {'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()
@@ -259,10 +259,25 @@ class TestScanServerStage:
         deps = _RecordingDeps()
         r = auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
-            server_refresh=lambda: (_ for _ in ()).throw(RuntimeError('boom')),
+            server_refresh=lambda media_type=None: (_ for _ in ()).throw(RuntimeError('boom')),
             sleep=lambda s: None, emit=lambda *a: None)
         assert r['status'] == 'error' and r['error'] == 'boom'
 
+    def test_scopes_refresh_and_carries_media_type_on_the_event(self):
+        seen = {}
+        events = []
+
+        def _refresh(media_type=None):
+            seen['mt'] = media_type
+            return {'ok': True}
+
+        auto_video_scan_server(
+            {'_automation_id': 'a', 'media_type': 'show'}, _RecordingDeps(),
+            server_refresh=_refresh, sleep=lambda s: None,
+            emit=lambda ev, data: events.append((ev, data)))
+        assert seen['mt'] == 'show'                                   # only TV sections nudged
+        assert events[0][1]['media_type'] == 'show'                   # stage 2 inherits the scope
+
 
 # ── post-download chain: video_update_database (stage 2) ───────────────────
 class TestUpdateDatabaseStage:
@@ -285,5 +300,33 @@ class TestUpdateDatabaseStage:
     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'})
+                                       run_video_scan=lambda m, media_type=None: {'state': 'error', 'error': 'no server'})
         assert r['status'] == 'error' and r['error'] == 'no server'
+
+    def test_inherits_media_type_from_the_scan_event(self):
+        # The post-download chain carries the scope: a TV-only rescan updates only TV.
+        seen = {}
+
+        def _scan(mode, media_type=None):
+            seen['media_type'] = media_type
+            return {'state': 'done', 'shows': 3, 'episodes': 12}
+
+        deps = _RecordingDeps()
+        auto_video_update_database(
+            {'_automation_id': 'a', '_event_data': {'media_type': 'show'}}, deps, run_video_scan=_scan)
+        assert seen['media_type'] == 'show'
+        assert deps.calls[-1]['log_line'] == 'Video database updated: 3 shows, 12 episodes'
+
+    def test_explicit_media_type_beats_event(self):
+        seen = {}
+        auto_video_update_database(
+            {'_automation_id': 'a', 'media_type': 'movie', '_event_data': {'media_type': 'show'}},
+            _RecordingDeps(),
+            run_video_scan=lambda mode, media_type=None: seen.setdefault('mt', media_type) or {'state': 'done'})
+        assert seen['mt'] == 'movie'
+
+    def test_busy_scanner_skips_cleanly(self):
+        r = auto_video_update_database(
+            {'_automation_id': 'a'}, _RecordingDeps(),
+            run_video_scan=lambda mode, media_type=None: {'state': 'in_progress'})
+        assert r['status'] == 'skipped'

tests/test_video_scan_scope.py

@@ -101,3 +101,28 @@ def test_scan_never_resumes_a_manually_paused_youtube_enricher(engine, monkeypat
     assert "youtube" not in engine._scan_paused   # we didn't touch it
     engine.resume_after_scan()
     assert fake._paused is True                    # still paused after the scan
+
+
+# ── refresh_sections is scoped by media_type too (server nudge) ─────────────
+
+def test_refresh_sections_scopes_by_media_type():
+    class SecU:
+        def __init__(self, type_, title):
+            self.type, self.title, self.updated = type_, title, False
+
+        def update(self):
+            self.updated = True
+
+    secs = [SecU("movie", "Movies"), SecU("show", "TV Shows")]
+    src = PlexVideoSource(_Server(secs), movies_lib="Movies", tv_lib="TV Shows")
+
+    src.refresh_sections("movie")
+    assert secs[0].updated and not secs[1].updated       # only the Movie section nudged
+    secs[0].updated = False
+
+    src.refresh_sections("show")
+    assert secs[1].updated and not secs[0].updated       # only the TV section nudged
+    secs[1].updated = False
+
+    res = src.refresh_sections("all")
+    assert secs[0].updated and secs[1].updated and res["sections"] == 2   # both