video scan: only FULL resets enrichment; incremental/deep preserve it

A scan wrote server-provided fields straight over the row, so an incremental/deep
re-read wiped the TMDB-backfilled 'status' (Plex returns it blank) — clearing the
airing watchlist. Now matches the intended model: incremental (add recent) and deep
(coverage + prune) PRESERVE enrichment-owned fields the server left blank; only a FULL
scan clobbers them (an explicit reset / fresh start).

- _resilient_upsert gains preserve_enrichment (default True): on a conflict UPDATE,
  enrichment-owned columns (per _ENRICH_META_COLS: status/network/ratings/air dates/…)
  take the server value only when non-blank, else keep what's stored. A real server
  value still wins.
- upsert_movie/upsert_show_tree thread the flag; scanner passes preserve=(mode!='full').

Tests: preserve-on-blank, server-value-wins, full-resets, and the scanner picking the
right mode. 88 DB + 18 scanner tests + isolation green.

core/video/scanner.py

@@ -138,6 +138,10 @@ class VideoLibraryScanner:
             server = source.server_name
             incremental = mode == "incremental"
             do_prune = mode == "deep"
+            # FULL = a clean reset (clobber enrichment-owned fields). Incremental/deep
+            # PRESERVE them, so a routine re-scan never wipes the TMDB-backfilled
+            # `status` the airing watchlist relies on. (Only an explicit full resets.)
+            preserve = mode != "full"
 
             # Incremental on a near-empty library is pointless — fall back to a
             # full pass so the first scan actually populates (music does this).
@@ -178,7 +182,7 @@ class VideoLibraryScanner:
                     continue
                 consec = 0
                 try:
-                    self.db.upsert_movie(server, item)
+                    self.db.upsert_movie(server, item, preserve_enrichment=preserve)
                 except Exception:
                     logger.exception("video scan: skipping movie %s", sid)
                     continue
@@ -215,7 +219,7 @@ class VideoLibraryScanner:
                     continue
                 consec = 0
                 try:
-                    self.db.upsert_show_tree(server, show)
+                    self.db.upsert_show_tree(server, show, preserve_enrichment=preserve)
                 except Exception:
                     logger.exception("video scan: skipping show %s", sid)
                     continue

database/video_database.py

@@ -1381,19 +1381,34 @@ class VideoDatabase:
                          (owner_id, gid))
 
     @staticmethod
-    def _resilient_upsert(conn, table: str, base: dict, id_cols: dict) -> None:
+    def _resilient_upsert(conn, table: str, base: dict, id_cols: dict,
+                          preserve_enrichment: bool = True) -> None:
         """INSERT…ON CONFLICT(server_source, server_id) for a movie/show row.
 
         Resilient to a LEGACY UNIQUE on tmdb_id/tvdb_id/imdb_id (old DBs created
         before those were made non-unique — SQLite can't drop an inline UNIQUE):
         on IntegrityError we retry WITHOUT the id columns, so the row is still
         stored (same film in >1 library) instead of being dropped by the scan.
-        ``base`` holds the always-written cols; ``id_cols`` the droppable ids."""
+        ``base`` holds the always-written cols; ``id_cols`` the droppable ids.
+
+        ``preserve_enrichment`` (default) keeps enrichment-owned fields (``status``,
+        network, ratings, air dates…) that the SERVER left blank — so an incremental
+        or deep re-scan never wipes the TMDB-backfilled ``status`` (which the airing
+        watchlist depends on). A FULL scan passes False = a clean overwrite / reset."""
+        protect = (_ENRICH_META_COLS.get(table, set()) if preserve_enrichment else set())
+
+        def _set(c):
+            # On a conflict UPDATE, an enrichment-owned column only takes the server
+            # value when it's non-blank; otherwise it keeps what's already stored.
+            if c in protect:
+                return f"{c}=COALESCE(NULLIF(excluded.{c}, ''), {table}.{c})"
+            return f"{c}=excluded.{c}"
+
         def run(include_ids):
             cols = list(base.keys()) + (list(id_cols.keys()) if include_ids else [])
             vals = list(base.values()) + (list(id_cols.values()) if include_ids else [])
             updates = [c for c in cols if c not in ("server_source", "server_id")]
-            set_clause = ", ".join(f"{c}=excluded.{c}" for c in updates) + ", updated_at=CURRENT_TIMESTAMP"
+            set_clause = ", ".join(_set(c) for c in updates) + ", updated_at=CURRENT_TIMESTAMP"
             sql = (f"INSERT INTO {table} ({', '.join(cols)}, updated_at) "
                    f"VALUES ({', '.join(['?'] * len(cols))}, CURRENT_TIMESTAMP) "
                    f"ON CONFLICT(server_source, server_id) DO UPDATE SET {set_clause}")
@@ -1453,8 +1468,10 @@ class VideoDatabase:
                 for r in rows if r["department"] == "crew"]
         return {"cast": cast, "crew": crew}
 
-    def upsert_movie(self, server_source: str, item: dict) -> int:
-        """Insert/update one movie (keyed on server id) and its file. Returns row id."""
+    def upsert_movie(self, server_source: str, item: dict, preserve_enrichment: bool = True) -> int:
+        """Insert/update one movie (keyed on server id) and its file. Returns row id.
+        ``preserve_enrichment`` keeps enrichment-owned fields the server left blank
+        (default); a FULL scan passes False for a clean reset."""
         conn = self._get_connection()
         try:
             self._resilient_upsert(conn, "movies", {
@@ -1465,7 +1482,8 @@ class VideoDatabase:
                 "studio": item.get("studio"), "tagline": item.get("tagline"),
                 "rating": item.get("rating"), "rating_critic": item.get("rating_critic"),
                 "poster_url": item.get("poster_url"), "has_file": 1 if item.get("file") else 0,
-            }, {"tmdb_id": item.get("tmdb_id"), "imdb_id": item.get("imdb_id")})
+            }, {"tmdb_id": item.get("tmdb_id"), "imdb_id": item.get("imdb_id")},
+                preserve_enrichment=preserve_enrichment)
             movie_id = conn.execute(
                 "SELECT id FROM movies WHERE server_source=? AND server_id=?",
                 (server_source, item["server_id"]),
@@ -1477,10 +1495,12 @@ class VideoDatabase:
         finally:
             conn.close()
 
-    def upsert_show_tree(self, server_source: str, item: dict) -> int:
+    def upsert_show_tree(self, server_source: str, item: dict, preserve_enrichment: bool = True) -> int:
         """Insert/update a show with its seasons + episodes (and files) in one
         transaction. Episodes/seasons no longer present on the server for this
-        show are pruned. Returns the show row id."""
+        show are pruned. Returns the show row id. ``preserve_enrichment`` keeps
+        enrichment-owned fields (status etc.) the server left blank (default); a FULL
+        scan passes False for a clean reset."""
         conn = self._get_connection()
         try:
             self._resilient_upsert(conn, "shows", {
@@ -1492,7 +1512,8 @@ class VideoDatabase:
                 "tagline": item.get("tagline"), "rating": item.get("rating"),
                 "first_air_date": item.get("first_air_date"), "last_air_date": item.get("last_air_date"),
                 "poster_url": item.get("poster_url"),
-            }, {"tvdb_id": item.get("tvdb_id"), "tmdb_id": item.get("tmdb_id"), "imdb_id": item.get("imdb_id")})
+            }, {"tvdb_id": item.get("tvdb_id"), "tmdb_id": item.get("tmdb_id"), "imdb_id": item.get("imdb_id")},
+                preserve_enrichment=preserve_enrichment)
             show_id = conn.execute(
                 "SELECT id FROM shows WHERE server_source=? AND server_id=?",
                 (server_source, item["server_id"]),

tests/test_video_upsert_preserve.py

@@ -0,0 +1,87 @@
+"""Scan must not wipe enrichment-owned fields the server leaves blank.
+
+The media server (Plex) returns a blank `status`, but TMDB enrichment fills it —
+and the airing watchlist depends on it. A routine re-scan (incremental/deep) must
+PRESERVE that backfilled status; only a FULL scan (an explicit reset) clobbers it.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from database.video_database import VideoDatabase
+from core.video.scanner import VideoLibraryScanner
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+def _status(db, show_id):
+    conn = db._get_connection()
+    try:
+        return conn.execute("SELECT status FROM shows WHERE id=?", (show_id,)).fetchone()[0]
+    finally:
+        conn.close()
+
+
+def _set_status(db, show_id, status):
+    conn = db._get_connection()
+    conn.execute("UPDATE shows SET status=? WHERE id=?", (status, show_id))
+    conn.commit()
+    conn.close()
+
+
+def _show(status=None):
+    return {"server_id": "s1", "title": "S", "tmdb_id": 1, "status": status, "seasons": []}
+
+
+def test_preserve_keeps_enriched_status_when_server_is_blank(db):
+    sid = db.upsert_show_tree("plex", _show())
+    _set_status(db, sid, "Returning Series")          # TMDB enrichment fills it
+    db.upsert_show_tree("plex", _show(status=None), preserve_enrichment=True)
+    assert _status(db, sid) == "Returning Series"      # re-scan didn't wipe it
+
+
+def test_server_provided_value_still_wins(db):
+    sid = db.upsert_show_tree("plex", _show())
+    _set_status(db, sid, "Returning Series")
+    db.upsert_show_tree("plex", _show(status="Ended"), preserve_enrichment=True)
+    assert _status(db, sid) == "Ended"                 # a real server value overwrites
+
+
+def test_full_scan_resets_enriched_status(db):
+    sid = db.upsert_show_tree("plex", _show())
+    _set_status(db, sid, "Returning Series")
+    db.upsert_show_tree("plex", _show(status=None), preserve_enrichment=False)
+    assert _status(db, sid) is None                    # full = fresh start, clobbers
+
+
+class _Src:
+    server_name = "plex"
+
+    def __init__(self, shows):
+        self._shows = shows
+
+    def iter_movies(self, incremental=False):
+        return iter([])
+
+    def iter_shows(self, incremental=False):
+        return iter(self._shows)
+
+
+def test_scanner_modes_pick_the_right_preserve(db):
+    # seed a show + enriched status, then re-scan it via each mode
+    sid = db.upsert_show_tree("plex", _show())
+    _set_status(db, sid, "Returning Series")
+    blank = [{"server_id": "s1", "title": "S", "tmdb_id": 1, "status": None, "seasons": []}]
+
+    VideoLibraryScanner(db).scan_sync(lambda: _Src(blank), mode="incremental")
+    assert _status(db, sid) == "Returning Series"      # incremental preserves
+
+    VideoLibraryScanner(db).scan_sync(lambda: _Src(blank), mode="deep")
+    assert _status(db, sid) == "Returning Series"      # deep preserves
+
+    VideoLibraryScanner(db).scan_sync(lambda: _Src(blank), mode="full")
+    assert _status(db, sid) is None                    # full resets