video scan: three modes (full refresh / incremental / deep)

Mirrors the music model (full_refresh vs smart incremental, plus deep_scan):
- incremental: only recently-added items from the server (Plex addedAt:desc /
  Jellyfin DateCreated, capped); upsert; no prune.
- full: every item; upsert all (refresh metadata + add new); no prune.
- deep: every item; upsert; prune what the server no longer has (empty-scan
  safety preserved).
scanner.request_scan/scan_sync take mode; /api/video/scan/request reads
{mode} from the body (default full); adapters take incremental=. Tests cover
deep-prunes / full-doesn't / empty-deep-safety / incremental-requests-recent.

api/video/scan.py

@@ -9,7 +9,7 @@ server's own rescan (post-download) is wired separately into the download flow.
 
 from __future__ import annotations
 
-from flask import jsonify
+from flask import jsonify, request
 
 from utils.logging_config import get_logger
 
@@ -22,8 +22,10 @@ def register_routes(bp):
         from . import get_video_db
         from core.video.scanner import get_video_scanner
         from core.video.sources import get_active_video_source
+        body = request.get_json(silent=True) or {}
+        mode = body.get("mode", "full")
         scanner = get_video_scanner(get_video_db())
-        return jsonify(scanner.request_scan(get_active_video_source))
+        return jsonify(scanner.request_scan(get_active_video_source, mode))
 
     @bp.route("/scan/status", methods=["GET"])
     def video_scan_status():

core/video/scanner.py

@@ -8,8 +8,13 @@ Jellyfin adapters live in core/video/sources.py.
 
 A source must provide:
     source.server_name -> 'plex' | 'jellyfin'
-    source.iter_movies() -> iterable of normalized movie dicts
+    source.iter_movies(incremental=False) -> iterable of normalized movie dicts
-    source.iter_shows()  -> iterable of normalized show dicts (with seasons/episodes)
+    source.iter_shows(incremental=False)  -> iterable of normalized show dicts
+
+Scan MODES (mirroring the music side's full_refresh / incremental / deep_scan):
+    'incremental' - only recently-added items from the server; upsert; no prune.
+    'full'        - every item; upsert all (refresh metadata + add new); no prune.
+    'deep'        - every item; upsert; PRUNE what the server no longer has.
 
 ISOLATION: imports only video.db + shared infra; music never imports this.
 """
@@ -23,6 +28,8 @@ from utils.logging_config import get_logger
 
 logger = get_logger("video_scanner")
 
+VALID_MODES = ("incremental", "full", "deep")
+
 
 class VideoLibraryScanner:
     """Reads the active media server and upserts movies/shows into video.db."""
@@ -41,30 +48,36 @@ class VideoLibraryScanner:
         with self._lock:
             self._status.update(kw)
 
-    def request_scan(self, source_factory) -> dict:
+    @staticmethod
+    def _norm_mode(mode) -> str:
+        return mode if mode in VALID_MODES else "full"
+
+    def request_scan(self, source_factory, mode: str = "full") -> dict:
         """Kick off a background scan. ``source_factory()`` returns a media
         source (or None if no video-capable server is connected)."""
+        mode = self._norm_mode(mode)
         with self._lock:
             if self._status.get("state") == "scanning":
                 return {"status": "in_progress"}
-            self._status = {"state": "scanning", "phase": "starting",
+            self._status = {"state": "scanning", "phase": "starting", "mode": mode,
                             "started_at": time.time(),
                             "movies": 0, "shows": 0, "episodes": 0}
         self._thread = threading.Thread(
-            target=self._run, args=(source_factory,), daemon=True)
+            target=self._run, args=(source_factory, mode), daemon=True)
         self._thread.start()
-        return {"status": "started"}
+        return {"status": "started", "mode": mode}
 
-    def scan_sync(self, source_factory) -> dict:
+    def scan_sync(self, source_factory, mode: str = "full") -> dict:
         """Run a scan inline (used by tests / callers that want to block)."""
+        mode = self._norm_mode(mode)
         with self._lock:
-            self._status = {"state": "scanning", "phase": "starting",
+            self._status = {"state": "scanning", "phase": "starting", "mode": mode,
                             "started_at": time.time(),
                             "movies": 0, "shows": 0, "episodes": 0}
-        self._run(source_factory)
+        self._run(source_factory, mode)
         return self.get_status()
 
-    def _run(self, source_factory) -> None:
+    def _run(self, source_factory, mode: str = "full") -> None:
         try:
             source = source_factory()
             if source is None:
@@ -72,40 +85,44 @@ class VideoLibraryScanner:
                           error="No connected Plex/Jellyfin video server")
                 return
             server = source.server_name
+            incremental = mode == "incremental"
+            do_prune = mode == "deep"
 
             # ── Movies ──
             self._set(phase="scanning movies")
             seen_movies: set[str] = set()
             movies = 0
-            for item in source.iter_movies():
+            for item in source.iter_movies(incremental=incremental):
                 self.db.upsert_movie(server, item)
                 seen_movies.add(str(item["server_id"]))
                 movies += 1
                 if movies % 25 == 0:
                     self._set(movies=movies)
             self._set(movies=movies)
-            # Prune only when we actually saw items — avoids wiping the library
+            # Prune ONLY on a deep scan, and only when we actually saw items —
-            # if the server returned nothing due to a transient failure.
+            # so a transient empty response can never wipe the library.
-            removed_m = self.db.prune_missing("movies", server, seen_movies) if seen_movies else 0
+            removed_m = (self.db.prune_missing("movies", server, seen_movies)
+                         if do_prune and seen_movies else 0)
 
             # ── Shows ──
             self._set(phase="scanning shows")
             seen_shows: set[str] = set()
             shows = 0
             episodes = 0
-            for show in source.iter_shows():
+            for show in source.iter_shows(incremental=incremental):
                 self.db.upsert_show_tree(server, show)
                 seen_shows.add(str(show["server_id"]))
                 shows += 1
                 episodes += sum(len(s.get("episodes", [])) for s in show.get("seasons", []))
                 self._set(shows=shows, episodes=episodes)
-            removed_s = self.db.prune_missing("shows", server, seen_shows) if seen_shows else 0
+            removed_s = (self.db.prune_missing("shows", server, seen_shows)
+                         if do_prune and seen_shows else 0)
 
             self._set(state="done", phase="complete", finished_at=time.time(),
                       movies=movies, shows=shows, episodes=episodes,
                       removed=removed_m + removed_s)
-            logger.info("Video scan complete: %d movies, %d shows, %d episodes (%d pruned)",
+            logger.info("Video scan (%s) complete: %d movies, %d shows, %d episodes (%d pruned)",
-                        movies, shows, episodes, removed_m + removed_s)
+                        mode, movies, shows, episodes, removed_m + removed_s)
         except Exception as e:  # noqa: BLE001 - report any failure to the UI
             logger.exception("Video library scan failed")
             self._set(state="error", phase="failed", error=str(e))

core/video/sources.py

@@ -59,14 +59,16 @@ class PlexVideoSource:
     def _sections(self, kind: str):
         return [s for s in self._server.library.sections() if s.type == kind]
 
-    def iter_movies(self):
+    def iter_movies(self, incremental=False):
         for section in self._sections("movie"):
-            for m in section.all():
+            items = section.search(sort="addedAt:desc", maxresults=100) if incremental else section.all()
+            for m in items:
                 yield self._movie(m)
 
-    def iter_shows(self):
+    def iter_shows(self, incremental=False):
         for section in self._sections("show"):
-            for sh in section.all():
+            items = section.search(sort="addedAt:desc", maxresults=50) if incremental else section.all()
+            for sh in items:
                 yield self._show(sh)
 
     @staticmethod
@@ -185,11 +187,13 @@ class JellyfinVideoSource:
             "runtime_seconds": JellyfinVideoSource._ticks_to_seconds(item.get("RunTimeTicks")),
         }
 
-    def iter_movies(self):
+    def iter_movies(self, incremental=False):
         for view in self._views("movies"):
-            resp = self._req(f"/Users/{self.uid}/Items", {
+            params = {"ParentId": view["Id"], "IncludeItemTypes": "Movie",
-                "ParentId": view["Id"], "IncludeItemTypes": "Movie",
+                      "Recursive": "true", "Fields": _JF_MOVIE_FIELDS}
-                "Recursive": "true", "Fields": _JF_MOVIE_FIELDS}) or {}
+            if incremental:
+                params.update({"SortBy": "DateCreated", "SortOrder": "Descending", "Limit": "100"})
+            resp = self._req(f"/Users/{self.uid}/Items", params) or {}
             for it in resp.get("Items", []):
                 yield self._movie(it)
 
@@ -208,11 +212,13 @@ class JellyfinVideoSource:
             "file": self._file(it),
         }
 
-    def iter_shows(self):
+    def iter_shows(self, incremental=False):
         for view in self._views("tvshows"):
-            resp = self._req(f"/Users/{self.uid}/Items", {
+            params = {"ParentId": view["Id"], "IncludeItemTypes": "Series",
-                "ParentId": view["Id"], "IncludeItemTypes": "Series",
+                      "Recursive": "true", "Fields": "Overview,ProductionYear,OfficialRating"}
-                "Recursive": "true", "Fields": "Overview,ProductionYear,OfficialRating"}) or {}
+            if incremental:
+                params.update({"SortBy": "DateCreated", "SortOrder": "Descending", "Limit": "50"})
+            resp = self._req(f"/Users/{self.uid}/Items", params) or {}
             for it in resp.get("Items", []):
                 yield self._show(it)
 

tests/test_video_scanner.py

@@ -21,11 +21,14 @@ class FakeSource:
 
     def __init__(self, movies, shows):
         self._movies, self._shows = movies, shows
+        self.incremental_calls = []
 
-    def iter_movies(self):
+    def iter_movies(self, incremental=False):
+        self.incremental_calls.append(("movies", incremental))
         return iter(self._movies)
 
-    def iter_shows(self):
+    def iter_shows(self, incremental=False):
+        self.incremental_calls.append(("shows", incremental))
         return iter(self._shows)
 
 
@@ -46,23 +49,39 @@ def test_scan_sync_populates_library(db):
     assert (lib["movies"], lib["shows"], lib["episodes"]) == (1, 1, 1)
 
 
-def test_scan_sync_prunes_removed_items(db):
+def test_deep_scan_prunes_removed_items(db):
     scanner = VideoLibraryScanner(db)
     scanner.scan_sync(lambda: FakeSource(
-        [{"server_id": "m1", "title": "A"}, {"server_id": "m2", "title": "B"}], []))
+        [{"server_id": "m1", "title": "A"}, {"server_id": "m2", "title": "B"}], []), mode="deep")
     assert db.dashboard_stats()["library"]["movies"] == 2
-    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []))
+    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []), mode="deep")
     assert db.dashboard_stats()["library"]["movies"] == 1
 
 
-def test_empty_scan_does_not_wipe_library(db):
+def test_full_refresh_does_not_prune(db):
-    # Safety: a scan that returns nothing (transient failure) must NOT prune.
+    # 'full' refreshes/adds but never removes — only 'deep' prunes.
     scanner = VideoLibraryScanner(db)
-    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []))
+    scanner.scan_sync(lambda: FakeSource(
-    scanner.scan_sync(lambda: FakeSource([], []))
+        [{"server_id": "m1", "title": "A"}, {"server_id": "m2", "title": "B"}], []), mode="deep")
+    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []), mode="full")
+    assert db.dashboard_stats()["library"]["movies"] == 2  # m2 NOT pruned
+
+
+def test_empty_deep_scan_does_not_wipe_library(db):
+    # Safety: a deep scan that returns nothing (transient failure) must NOT prune.
+    scanner = VideoLibraryScanner(db)
+    scanner.scan_sync(lambda: FakeSource([{"server_id": "m1", "title": "A"}], []), mode="deep")
+    scanner.scan_sync(lambda: FakeSource([], []), mode="deep")
     assert db.dashboard_stats()["library"]["movies"] == 1
 
 
+def test_incremental_mode_requests_incremental_from_source(db):
+    src = FakeSource([{"server_id": "m1", "title": "A"}], [])
+    VideoLibraryScanner(db).scan_sync(lambda: src, mode="incremental")
+    assert ("movies", True) in src.incremental_calls
+    assert ("shows", True) in src.incremental_calls
+
+
 def test_scan_sync_no_source_reports_error(db):
     st = VideoLibraryScanner(db).scan_sync(lambda: None)
     assert st["state"] == "error" and "error" in st