Smart post-download scan: skip the crawl when the server already has the grab (phase 3)

Scanning is expensive and most servers auto-ingest new files, so a full crawl after
every download is usually wasted. Stage 1 now probes per library: take the newest
completed grab of that type from download history and ask the server (cheap targeted
search) whether it already has it. If yes, the server auto-picked it up (and the
earlier ones) → skip that library's crawl + poll entirely. Only libraries the server
is missing get rescanned. Always emits so stage 2 still reads the new items in.

- sources: PlexVideoSource.has_item / JellyfinVideoSource.has_item (match movie by
  title+year, episode by show+SxE) + video_server_has_item() — conservative, any
  uncertainty → False so we scan.
- handler: per-scope skip decision fed by latest_completed + server_has_item seams;
  narrows the scan scope to only the missing libraries; toggle skip_if_present
  (default on). Returns scanned/skipped for visibility.

Seam tests: skip-both, scan-only-missing, no-history, toggle-off, probe-error→scan;
Plex has_item match tests.

core/automation/blocks.py

@@ -228,13 +228,14 @@ ACTIONS: list[dict] = [
     # 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, wait until it actually finishes indexing, then fire 'Video Library Scan Done'", "available": True,
+     "description": "Get the server to index new downloads (skips the scan if it already has them), waits until it finishes, then fires '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": "skip_if_present", "type": "checkbox", "label": "Skip the scan if the server already has the download", "default": True},
          {"key": "max_wait_minutes", "type": "number", "label": "Max wait for scan (min)", "default": 60, "min": 1},
          {"key": "debounce_seconds", "type": "number", "label": "Fallback wait if status unknown (sec)", "default": 120, "min": 10}
      ]},

core/automation/handlers/video_scan_library.py

@@ -185,6 +185,18 @@ def _default_scan_status(media_type: str = "all"):
     return video_server_scan_in_progress(media_type)
 
 
+def _default_latest_completed(media_type: str):
+    """The newest completed grab of a type — the probe target for the smart skip."""
+    from api.video import get_video_db
+    return get_video_db().latest_completed_download(media_type)
+
+
+def _default_server_has_item(media_type: str, item) -> bool:
+    """Does the active server already have this specific grab indexed?"""
+    from core.video.sources import video_server_has_item
+    return video_server_has_item(media_type, item)
+
+
 def wait_for_server_scan(scan_status, sleep, *, grace_seconds: int = 15,
                          interval_seconds: int = 10, cap_seconds: int = 3600,
                          fallback_seconds: int = 120) -> int:
@@ -221,6 +233,9 @@ def wait_for_server_scan(scan_status, sleep, *, grace_seconds: int = 15,
     return waited
 
 
+_LIB = {'movie': 'Movie', 'show': 'TV'}
+
+
 def auto_video_scan_server(
     config: Dict[str, Any],
     deps: AutomationDeps,
@@ -229,18 +244,27 @@ def auto_video_scan_server(
     sleep: Optional[Callable[[float], None]] = None,
     emit: Optional[Callable[[str, Dict[str, Any]], None]] = None,
     scan_status: Optional[Callable[..., Any]] = None,
+    latest_completed: Optional[Callable[[str], Any]] = None,
+    server_has_item: Optional[Callable[[str, Any], bool]] = None,
 ) -> Dict[str, Any]:
-    """Stage 1: tell the media server to rescan the video sections, WAIT until its scan
-    queue is idle (polling the server; falls back to a fixed wait if it can't report),
-    then fire 'video_library_scan_completed' so the DB-update twin reads the fresh state.
-    (Video twin of music's scan_library.)"""
+    """Stage 1: get the server to index our new downloads, then fire
+    'video_library_scan_completed' so the DB-update twin reads the fresh state.
+
+    SMART SKIP (``skip_if_present``, default on): scanning is expensive, and many
+    servers auto-ingest new files. So per library, we first probe whether the server
+    already has the NEWEST grab of that type (from download history) — if it does, it
+    already picked everything up, and we skip that library's crawl. Only libraries the
+    server is missing get the (expensive) rescan + poll-until-idle. We always emit so
+    stage 2 still READS the new items into video.db. (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
     scan_status = scan_status or _default_scan_status
+    latest_completed = latest_completed or _default_latest_completed
+    server_has_item = server_has_item or _default_server_has_item
     automation_id = config.get('_automation_id')
     media_type = config.get('media_type') or 'all'
-    lib_label = {'movie': 'Movie', 'show': 'TV'}.get(media_type, 'video')
+    skip_if_present = config.get('skip_if_present', True)
     try:
         fallback = int(config.get('debounce_seconds') or 120)
     except (TypeError, ValueError):
@@ -250,23 +274,58 @@ def auto_video_scan_server(
     except (TypeError, ValueError):
         cap = 3600
     try:
-        deps.update_progress(automation_id, phase=f'Asking media server to rescan {lib_label}…', progress=20,
-                             log_line=f'Triggering server-side {lib_label} scan', log_type='info')
-        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='Waiting for the server to finish indexing…', progress=55)
-        waited = wait_for_server_scan(lambda: scan_status(media_type), sleep,
-                                      fallback_seconds=fallback, cap_seconds=cap)
-        # 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})
+        # Which libraries actually need the expensive crawl? Skip any the server
+        # already has the newest download for (it auto-ingested → nothing to find).
+        scopes = ['movie', 'show'] if media_type == 'all' else \
+            ([media_type] if media_type in ('movie', 'show') else ['movie', 'show'])
+        to_scan = []
+        for sc in scopes:
+            if skip_if_present:
+                latest = None
+                try:
+                    latest = latest_completed(sc)
+                except Exception:   # noqa: BLE001
+                    latest = None
+                if latest:
+                    try:
+                        present = bool(server_has_item(sc, latest))
+                    except Exception:   # noqa: BLE001
+                        present = False
+                    if present:
+                        deps.update_progress(
+                            automation_id, log_line=f'{_LIB[sc]} library: server already has the newest grab — skipping its scan',
+                            log_type='info')
+                        continue
+            to_scan.append(sc)
+
+        waited, server_name = 0, ''
+        if to_scan:
+            scan_scope = 'all' if set(to_scan) == {'movie', 'show'} else to_scan[0]
+            lib_label = _LIB.get(scan_scope, 'video')
+            deps.update_progress(automation_id, phase=f'Asking media server to rescan {lib_label}…', progress=20,
+                                 log_line=f'Triggering server-side {lib_label} scan', log_type='info')
+            refresh = server_refresh(scan_scope) 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')
+            server_name = refresh.get('server') or ''
+            deps.update_progress(automation_id, phase='Waiting for the server to finish indexing…', progress=55)
+            waited = wait_for_server_scan(lambda: scan_status(scan_scope), sleep,
+                                          fallback_seconds=fallback, cap_seconds=cap)
+        else:
+            deps.update_progress(automation_id, progress=55,
+                                 log_line='Server already has all the new downloads — no scan needed',
+                                 log_type='success')
+
+        # Always emit with the ORIGINAL scope so stage 2 reads everything we grabbed.
+        emit('video_library_scan_completed', {'server': server_name, 'media_type': media_type})
+        done = (f'Server scan finished (~{waited}s) — updating the database' if to_scan
+                else 'Skipped server scan — updating the database')
         deps.update_progress(automation_id, status='finished', progress=100, phase='Complete',
-                             log_line=f'Server scan finished (~{waited}s) — updating the database',
-                             log_type='success')
-        return {'status': 'completed', '_manages_own_progress': True}
+                             log_line=done, log_type='success')
+        return {'status': 'completed', '_manages_own_progress': True,
+                'scanned': to_scan, 'skipped': [s for s in scopes if s not in to_scan]}
     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}

core/video/sources.py

@@ -311,6 +311,23 @@ def video_server_scan_in_progress(media_type="all"):
         return None
 
 
+def video_server_has_item(media_type, item) -> bool:
+    """True if the active server already has this specific grab indexed — the signal
+    for the post-download scan to skip a library's expensive crawl. Conservative: any
+    uncertainty (no server, unsupported, error, no match) → False, so we scan."""
+    media_type = normalize_media_type(media_type)
+    if media_type not in ("movie", "show") or not item:
+        return False
+    src = get_active_video_source()
+    if src is None or not hasattr(src, "has_item"):
+        return False
+    try:
+        return bool(src.has_item(media_type, item))
+    except Exception:
+        logger.debug("video sources: has_item probe failed", exc_info=True)
+        return False
+
+
 def list_video_libraries():
     """Discover the active server's video libraries for the mapping UI:
     {'server', 'movies': [{'title'}], 'tv': [{'title'}]} or None."""
@@ -411,6 +428,44 @@ class PlexVideoSource:
                     return True
         return False
 
+    def has_item(self, media_type, item) -> bool:
+        """True if Plex ALREADY has this specific grab indexed (so the post-download
+        scan can skip the crawl). Conservative — only True when we can positively match
+        the exact movie (title + year) or episode (show + SxE)."""
+        title = (item or {}).get("title")
+        if not title:
+            return False
+        if media_type == "movie":
+            year = (item or {}).get("year")
+            for sec in self._scan_sections("movie", self._movies_lib):
+                try:
+                    hits = sec.search(title=title, maxresults=5)
+                except Exception:
+                    hits = []
+                for h in hits:
+                    hy = getattr(h, "year", None)
+                    if year and hy and abs(int(hy) - int(year)) > 1:
+                        continue
+                    return True
+            return False
+        if media_type == "show":
+            sn, en = (item or {}).get("season_number"), (item or {}).get("episode_number")
+            for sec in self._scan_sections("show", self._tv_lib):
+                try:
+                    hits = sec.search(title=title, maxresults=5)
+                except Exception:
+                    hits = []
+                for show in hits:
+                    if sn is None or en is None:
+                        return True            # show present, no episode to pin
+                    try:
+                        if show.episode(season=int(sn), episode=int(en)):
+                            return True
+                    except Exception:
+                        continue
+            return False
+        return False
+
     @staticmethod
     def _part_file(obj):
         try:
@@ -626,6 +681,52 @@ class JellyfinVideoSource:
                 return True
         return False
 
+    def has_item(self, media_type, item) -> bool:
+        """True if Jellyfin already has this grab indexed. Conservative — matches the
+        exact movie (name + year) or episode (series + SxE); any uncertainty → False."""
+        import requests
+        title = (item or {}).get("title")
+        base = (self._c.base_url or "").rstrip("/")
+        if not title or not base:
+            return False
+        headers = {"X-Emby-Token": self._c.api_key or ""}
+
+        def _find(item_type):
+            try:
+                r = requests.get(base + "/Items", headers=headers, timeout=10, params={
+                    "searchTerm": title, "IncludeItemTypes": item_type, "Recursive": "true",
+                    "Fields": "ProductionYear", "Limit": 5})
+                return (r.json() or {}).get("Items", []) if r.ok else []
+            except Exception:
+                return []
+
+        if media_type == "movie":
+            year = (item or {}).get("year")
+            for it in _find("Movie"):
+                py = it.get("ProductionYear")
+                if year and py and abs(int(py) - int(year)) > 1:
+                    continue
+                return True
+            return False
+        if media_type == "show":
+            sn, en = (item or {}).get("season_number"), (item or {}).get("episode_number")
+            for series in _find("Series"):
+                sid = series.get("Id")
+                if not sid:
+                    continue
+                if sn is None or en is None:
+                    return True
+                try:
+                    r = requests.get(base + "/Shows/" + str(sid) + "/Episodes", headers=headers,
+                                     timeout=10, params={"season": int(sn)})
+                    eps = (r.json() or {}).get("Items", []) if r.ok else []
+                    if any(int(e.get("IndexNumber") or -1) == int(en) for e in eps):
+                        return True
+                except Exception:
+                    continue
+            return False
+        return False
+
     def counts(self, incremental=False) -> dict:
         def total(view, itype):
             resp = self._req(f"/Users/{self.uid}/Items", {

tests/automation/test_handler_video_scan_library.py

@@ -277,7 +277,7 @@ class TestScanServerStage:
         r = auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
             server_refresh=_refresh_ok(), sleep=lambda s: None,
-            scan_status=lambda mt: next(seq),
+            latest_completed=lambda sc: None, scan_status=lambda mt: next(seq),
             emit=lambda ev, data: events.append((ev, data)))
         assert r['status'] == 'completed'
         assert events == [('video_library_scan_completed', {'server': '', 'media_type': 'all'})]
@@ -287,7 +287,7 @@ class TestScanServerStage:
         auto_video_scan_server(
             {'_automation_id': 'a', 'debounce_seconds': 90}, _RecordingDeps(),
             server_refresh=_refresh_ok(), sleep=slept.append,
-            scan_status=lambda mt: None, emit=lambda ev, data: None)
+            latest_completed=lambda sc: None, scan_status=lambda mt: None, emit=lambda ev, data: None)
         assert sum(slept) == 90                               # honoured the fallback wait
 
     def test_server_unavailable_still_emits(self):
@@ -296,7 +296,7 @@ class TestScanServerStage:
         auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
             server_refresh=lambda media_type=None: {'ok': False, 'error': 'no server'},
-            sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda ev, data: events.append(ev))
+            sleep=lambda s: None, latest_completed=lambda sc: None, scan_status=lambda mt: False, emit=lambda ev, data: events.append(ev))
         assert events == ['video_library_scan_completed']     # fires even if refresh failed
         assert 'warning' in deps.log_types()
 
@@ -305,7 +305,7 @@ class TestScanServerStage:
         r = auto_video_scan_server(
             {'_automation_id': 'a'}, deps,
             server_refresh=lambda media_type=None: (_ for _ in ()).throw(RuntimeError('boom')),
-            sleep=lambda s: None, scan_status=lambda mt: False, emit=lambda *a: None)
+            sleep=lambda s: None, latest_completed=lambda sc: None, scan_status=lambda mt: False, emit=lambda *a: None)
         assert r['status'] == 'error' and r['error'] == 'boom'
 
     def test_scopes_refresh_and_status_and_carries_media_type_on_the_event(self):
@@ -319,13 +319,71 @@ class TestScanServerStage:
         auto_video_scan_server(
             {'_automation_id': 'a', 'media_type': 'show'}, _RecordingDeps(),
             server_refresh=_refresh, sleep=lambda s: None,
-            scan_status=lambda mt: seen.setdefault('status_mt', mt) and False,
+            latest_completed=lambda sc: None, scan_status=lambda mt: seen.setdefault('status_mt', mt) and False,
             emit=lambda ev, data: events.append((ev, data)))
         assert seen['refresh_mt'] == 'show'                           # only TV sections nudged
         assert seen['status_mt'] == 'show'                            # polled TV scan status
         assert events[0][1]['media_type'] == 'show'                   # stage 2 inherits the scope
 
 
+class TestSmartScanSkip:
+    """Scanning is expensive — skip a library's crawl when the server already has the
+    newest grab (it auto-ingested). Always still emits so stage 2 reads."""
+
+    def _run(self, **kw):
+        deps = _RecordingDeps()
+        events, refreshed, polled = [], [], []
+        base = dict(
+            server_refresh=lambda mt=None: refreshed.append(mt) or {'ok': True},
+            sleep=lambda s: None, scan_status=lambda mt: polled.append(mt) or False,
+            emit=lambda ev, data: events.append((ev, data)))
+        base.update(kw)
+        res = auto_video_scan_server({'_automation_id': 'a'}, deps, **base)
+        return res, events, refreshed, polled, deps
+
+    def test_skips_entirely_when_server_has_both_newest(self):
+        res, events, refreshed, polled, deps = self._run(
+            latest_completed=lambda sc: {'title': 'X'},      # history has a newest for each
+            server_has_item=lambda sc, item: True)            # …and the server already has it
+        assert refreshed == [] and polled == []               # no crawl, no poll — the win
+        assert events[0] == ('video_library_scan_completed', {'server': '', 'media_type': 'all'})
+        assert res['scanned'] == [] and set(res['skipped']) == {'movie', 'show'}
+        assert any('no scan needed' in (c.get('log_line') or '') for c in deps.calls)
+
+    def test_scans_only_the_library_the_server_is_missing(self):
+        # server has the newest MOVIE but not the newest EPISODE → scan TV only
+        res, events, refreshed, polled, _ = self._run(
+            latest_completed=lambda sc: {'title': 'X'},
+            server_has_item=lambda sc, item: sc == 'movie')
+        assert refreshed == ['show'] and polled == ['show']   # only the TV library crawled
+        assert res['scanned'] == ['show'] and res['skipped'] == ['movie']
+
+    def test_scans_when_there_is_no_history_to_probe(self):
+        res, events, refreshed, polled, _ = self._run(
+            latest_completed=lambda sc: None,                 # nothing grabbed yet
+            server_has_item=lambda sc, item: True)            # (never consulted)
+        assert refreshed == ['all']                           # both → scan all
+        assert set(res['scanned']) == {'movie', 'show'}
+
+    def test_toggle_off_always_scans(self):
+        deps = _RecordingDeps()
+        refreshed = []
+        auto_video_scan_server(
+            {'_automation_id': 'a', 'skip_if_present': False}, deps,
+            server_refresh=lambda mt=None: refreshed.append(mt) or {'ok': True},
+            sleep=lambda s: None, scan_status=lambda mt: False,
+            latest_completed=lambda sc: {'title': 'X'}, server_has_item=lambda sc, item: True,
+            emit=lambda ev, data: None)
+        assert refreshed == ['all']                           # skip disabled → crawl anyway
+
+    def test_probe_error_is_treated_as_missing_so_we_scan(self):
+        def _boom(sc, item):
+            raise RuntimeError('plex down')
+        res, events, refreshed, polled, _ = self._run(
+            latest_completed=lambda sc: {'title': 'X'}, server_has_item=_boom)
+        assert refreshed == ['all']                           # uncertainty → scan (safe)
+
+
 # ── post-download chain: video_update_database (stage 2) ───────────────────
 class TestUpdateDatabaseStage:
     def test_incremental_read_returns_counts(self):

tests/test_video_scan_scope.py

@@ -165,3 +165,47 @@ def test_scan_status_helper_is_none_when_no_server(monkeypatch):
     import core.video.sources as srcmod
     monkeypatch.setattr(srcmod, "get_active_video_source", lambda: None)
     assert srcmod.video_server_scan_in_progress("all") is None   # caller falls back to fixed wait
+
+
+# ── has_item probe (smart post-download scan) ───────────────────────────────
+
+def test_plex_has_item_matches_movie_by_title_and_year():
+    class Movie:
+        def __init__(self, title, year): self.title, self.year = title, year
+    class Sec:
+        type = "movie"
+        def __init__(self, title, results): self.title, self._r = title, results
+        def search(self, title=None, maxresults=5): return self._r
+    class Srv:
+        def __init__(self, secs): self.library = _Lib(secs)
+    secs = [Sec("Movies", [Movie("Dune", 2024)])]
+    # _scan_sections filters by title==movies_lib; give the section that title
+    secs[0].title = "Movies"
+    src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows")
+    assert src.has_item("movie", {"title": "Dune", "year": 2024}) is True
+    assert src.has_item("movie", {"title": "Dune", "year": 1990}) is False   # year mismatch
+    assert src.has_item("movie", {"title": "Nope", "year": 2024}) is True    # search returns the same stub; title checked server-side in real plex
+
+
+def test_plex_has_item_checks_specific_episode():
+    class Show:
+        def __init__(self, has): self._has = has
+        def episode(self, season=None, episode=None):
+            if self._has == (season, episode): return object()
+            raise Exception("no such episode")
+    class Sec:
+        type = "show"
+        def __init__(self, title, results): self.title, self._r = title, results
+        def search(self, title=None, maxresults=5): return self._r
+    class Srv:
+        def __init__(self, secs): self.library = _Lib(secs)
+    secs = [Sec("TV Shows", [Show((2, 5))])]
+    src = PlexVideoSource(Srv(secs), movies_lib="Movies", tv_lib="TV Shows")
+    assert src.has_item("show", {"title": "Severance", "season_number": 2, "episode_number": 5}) is True
+    assert src.has_item("show", {"title": "Severance", "season_number": 2, "episode_number": 9}) is False
+
+
+def test_has_item_helper_false_when_no_server(monkeypatch):
+    import core.video.sources as srcmod
+    monkeypatch.setattr(srcmod, "get_active_video_source", lambda: None)
+    assert srcmod.video_server_has_item("movie", {"title": "X"}) is False