video scan: align incremental + deep with music's logic

- Incremental now does smart early-stopping like music: skips already-known
  items and stops after 25 consecutive known (server lists recent first),
  instead of a blind fixed cap. Falls back to a full pass when the library is
  near-empty (<50), matching music's small-DB behavior.
- Deep scan gains music's 50% safety threshold: if removal would wipe >50% of a
  >100-row library, it skips (assumes a partial server response, not a real
  emptying) — prevents catastrophic deletion.
- Full Refresh already matched (re-read all, upsert, no removal).
Added DB helpers (server_ids, table_count). Tests: early-stop skips known,
small-lib fallback, 50% prune safety. 122 tests green.

core/video/scanner.py

@@ -30,6 +30,13 @@ logger = get_logger("video_scanner")
 
 VALID_MODES = ("incremental", "full", "deep")
 
+# Incremental stops after this many consecutive already-known items (recent
+# first), mirroring music's "25 consecutive complete albums" early-stop.
+INCREMENTAL_STOP_AFTER = 25
+# Below this library size, an incremental scan falls back to a full pass (music
+# does the same when the DB is too small to be worth an incremental).
+INCREMENTAL_MIN_LIBRARY = 50
+
 
 class VideoLibraryScanner:
     """Reads the active media server and upserts movies/shows into video.db."""
@@ -105,6 +112,11 @@ class VideoLibraryScanner:
             incremental = mode == "incremental"
             do_prune = mode == "deep"
 
+            # Incremental on a near-empty library is pointless — fall back to a
+            # full pass so the first scan actually populates (music does this).
+            if incremental and (self.db.table_count("movies") + self.db.table_count("shows")) < INCREMENTAL_MIN_LIBRARY:
+                incremental = False
+
             # Totals up front so the progress bar shows a REAL percentage
             # (movies + shows are the unit; episodes ride along under each show).
             total = 0
@@ -118,15 +130,28 @@ class VideoLibraryScanner:
             def pct():
                 return round(processed / total * 100) if total else None
 
+            known_movies = self.db.server_ids("movies", server) if incremental else set()
+            known_shows = self.db.server_ids("shows", server) if incremental else set()
+
             # ── Movies ──
             self._set(phase="scanning movies", total=total, percent=pct())
             seen_movies: set[str] = set()
             movies = 0
+            consec = 0
             for item in source.iter_movies(incremental=incremental):
                 if self._cancel:
                     return self._finish_cancelled(movies, 0, 0)
+                sid = str(item["server_id"])
+                # Incremental early-stop: skip already-known items and bail after
+                # a run of consecutive known ones (server lists recent first).
+                if incremental and sid in known_movies:
+                    consec += 1
+                    if consec >= INCREMENTAL_STOP_AFTER:
+                        break
+                    continue
+                consec = 0
                 self.db.upsert_movie(server, item)
-                seen_movies.add(str(item["server_id"]))
+                seen_movies.add(sid)
                 movies += 1
                 processed += 1
                 if movies % 10 == 0:
@@ -142,11 +167,19 @@ class VideoLibraryScanner:
             seen_shows: set[str] = set()
             shows = 0
             episodes = 0
+            consec = 0
             for show in source.iter_shows(incremental=incremental):
                 if self._cancel:
                     return self._finish_cancelled(movies, shows, episodes)
+                sid = str(show["server_id"])
+                if incremental and sid in known_shows:
+                    consec += 1
+                    if consec >= INCREMENTAL_STOP_AFTER:
+                        break
+                    continue
+                consec = 0
                 self.db.upsert_show_tree(server, show)
-                seen_shows.add(str(show["server_id"]))
+                seen_shows.add(sid)
                 shows += 1
                 episodes += sum(len(s.get("episodes", [])) for s in show.get("seasons", []))
                 processed += 1

database/video_database.py

@@ -307,9 +307,32 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def server_ids(self, table: str, server_source: str) -> set:
+        """All server_ids already stored for a server (for incremental early-stop)."""
+        if table not in ("movies", "shows"):
+            return set()
+        conn = self._get_connection()
+        try:
+            return {str(r[0]) for r in conn.execute(
+                f"SELECT server_id FROM {table} WHERE server_source=?", (server_source,)).fetchall()}
+        finally:
+            conn.close()
+
+    def table_count(self, table: str) -> int:
+        if table not in ("movies", "shows", "episodes"):
+            return 0
+        conn = self._get_connection()
+        try:
+            return conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
+        finally:
+            conn.close()
+
     def prune_missing(self, table: str, server_source: str, seen_ids) -> int:
         """Delete top-level rows for a server that the scan no longer saw.
-        ``table`` is internal ('movies'|'shows'); cascades clean children."""
+        ``table`` is internal ('movies'|'shows'); cascades clean children.
+
+        Safety (mirrors music's deep scan): if removal would wipe >50% of a
+        >100-row library, assume a partial server failure and skip it."""
         if table not in ("movies", "shows"):
             raise ValueError(f"prune_missing: unexpected table {table!r}")
         seen = {str(s) for s in seen_ids}
@@ -319,6 +342,11 @@ class VideoDatabase:
                 f"SELECT server_id FROM {table} WHERE server_source=?", (server_source,)
             ).fetchall()]
             stale = [sid for sid in existing if str(sid) not in seen]
+            if len(stale) > len(existing) * 0.5 and len(existing) > 100:
+                logger.warning(
+                    "Video deep scan: %d/%d %s stale (>50%%) — skipping removal (likely a "
+                    "partial server response)", len(stale), len(existing), table)
+                return 0
             for sid in stale:
                 conn.execute(f"DELETE FROM {table} WHERE server_source=? AND server_id=?",
                              (server_source, sid))

tests/test_video_database.py

@@ -224,6 +224,15 @@ def test_upsert_show_tree_builds_seasons_episodes_and_prunes(db):
     assert eps == [1]
 
 
+def test_prune_missing_skips_when_over_half_would_be_removed(db):
+    # >100 movies; a scan that "sees" only a couple must NOT wipe the rest
+    # (mirrors music's deep-scan 50% safety against partial server failures).
+    for i in range(150):
+        db.upsert_movie("plex", {"server_id": "m%d" % i, "title": "M%d" % i})
+    assert db.prune_missing("movies", "plex", {"m0", "m1"}) == 0
+    assert db.table_count("movies") == 150
+
+
 def test_upsert_show_tree_skips_episodes_without_number(db):
     sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
         {"season_number": 1, "episodes": [

tests/test_video_scanner.py

@@ -76,6 +76,9 @@ def test_empty_deep_scan_does_not_wipe_library(db):
 
 
 def test_incremental_mode_requests_incremental_from_source(db):
+    # Populate past the small-library fallback so incremental stays incremental.
+    for i in range(60):
+        db.upsert_movie("plex", {"server_id": "p%d" % i, "title": str(i)})
     src = FakeSource([{"server_id": "m1", "title": "A"}], [])
     VideoLibraryScanner(db).scan_sync(lambda: src, mode="incremental")
     assert ("movies", True) in src.incremental_calls
@@ -123,6 +126,47 @@ def test_scan_cancel_stops_midway(db):
     assert db.dashboard_stats()["library"]["movies"] == 1   # only the first was saved
 
 
+def test_incremental_skips_known_and_early_stops(db):
+    # Past the small-library fallback, with everything already known.
+    for i in range(60):
+        db.upsert_movie("plex", {"server_id": "k%d" % i, "title": "K%d" % i})
+    new_item = {"server_id": "new1", "title": "New"}
+    known = [{"server_id": "k%d" % i, "title": "K%d" % i} for i in range(60)]
+
+    class S:
+        server_name = "plex"
+        def counts(self, incremental=False):
+            return {"movies": 61, "shows": 0}
+        def iter_movies(self, incremental=False):
+            assert incremental is True          # NOT fallen back (library big enough)
+            return iter([new_item] + known)     # one new, then a long run of known
+        def iter_shows(self, incremental=False):
+            return iter([])
+
+    st = VideoLibraryScanner(db).scan_sync(lambda: S(), mode="incremental")
+    assert st["state"] == "done"
+    assert st["movies"] == 1                     # only the new one; known skipped
+    assert db.table_count("movies") == 61
+
+
+def test_incremental_falls_back_to_full_on_small_library(db):
+    captured = {}
+
+    class S:
+        server_name = "plex"
+        def counts(self, incremental=False):
+            return {"movies": 3, "shows": 0}
+        def iter_movies(self, incremental=False):
+            captured["incremental"] = incremental
+            return iter([{"server_id": "m%d" % i, "title": str(i)} for i in range(3)])
+        def iter_shows(self, incremental=False):
+            return iter([])
+
+    st = VideoLibraryScanner(db).scan_sync(lambda: S(), mode="incremental")
+    assert captured["incremental"] is False      # empty DB (<50) -> full pass
+    assert st["movies"] == 3
+
+
 def test_core_video_imports_nothing_from_music():
     base = Path(__file__).resolve().parent.parent / "core" / "video"
     for py in base.glob("*.py"):