video discover phase 1b-d: gap-engine queries, engine collection fetch, /discover/gaps API

- video_database: owned_movie_tmdb_ids (diff set), owned_movie_collections (franchises you've
  started, most-invested first), top_owned_people (directors/creators you own the most).
- engine.collection(id): cached + owned-annotated franchise film list (person_detail already
  gives owned-annotated filmography).
- /api/video/discover/gaps: builds 'Complete the <franchise>' rails (collection_gaps) + 'More
  from <person>' rails (filmography_gaps, movies, vote-filtered) — the 'what am I missing' section.
All additive; gap diffs are the pure tested core.

api/video/discover.py

@@ -85,6 +85,45 @@ def register_routes(bp):
             logger.exception("discover morelike failed")
             return jsonify({"rails": []})
 
+    @bp.route("/discover/gaps", methods=["GET"])
+    def video_discover_gaps():
+        """'What am I missing?' rails — franchises you've started but not finished, and
+        more from the directors/creators you own the most. Powered by the gap engine."""
+        from . import get_video_db
+        from core.video.enrichment.engine import get_video_enrichment_engine
+        from core.video.discovery_gaps import collection_gaps, filmography_gaps
+        try:
+            from core.video.sources import resolve_video_server
+            srv = resolve_video_server()
+        except Exception:
+            srv = None
+        db = get_video_db()
+        eng = get_video_enrichment_engine()
+        try:
+            owned = db.owned_movie_tmdb_ids(srv)
+            rails = []
+            # Complete your collections — top franchises you've started, missing entries.
+            for coll in db.owned_movie_collections(srv, limit=8):
+                missing = collection_gaps(owned, eng.collection(coll["collection_id"]))
+                if missing:
+                    name = (coll.get("name") or "Collection").strip()
+                    rails.append({"title": "Complete the " + name, "kind": "collection",
+                                  "items": missing[:30]})
+            # More from the people you own the most (directors / creators).
+            for person in db.top_owned_people(min_titles=2, limit=6, server_source=srv):
+                p = eng.person_detail(person["tmdb_id"])
+                if not p:
+                    continue
+                missing = filmography_gaps(owned, p.get("credits") or [],
+                                           kinds=("movie",), min_vote_count=50, limit=30)
+                if len(missing) >= 3:
+                    rails.append({"title": "More from " + person["name"], "kind": "person",
+                                  "items": missing})
+            return jsonify({"rails": rails})
+        except Exception:
+            logger.exception("discover gaps failed")
+            return jsonify({"rails": []})
+
     @bp.route("/discover/trailer", methods=["GET"])
     def video_discover_trailer():
         """Best YouTube trailer {key,name} for a tmdb title (hero 'Trailer' button)."""

core/video/enrichment/engine.py

@@ -419,6 +419,23 @@ class VideoEnrichmentEngine:
                 return []
         return self._stamp_owned(items)
 
+    def collection(self, collection_id) -> list:
+        """The films of a TMDB collection (franchise) — cached + owned-annotated.
+        Drives the 'complete your collections' gap rails."""
+        w = self.workers.get("tmdb")
+        if not w or not w.enabled or not collection_id:
+            return []
+        ck = ("collection", collection_id)
+        items = self._cache_get(ck)
+        if items is None:
+            try:
+                items = w.client.collection(collection_id) or []
+                self._cache_put(ck, items, ttl=86400)   # franchises rarely change
+            except Exception:
+                logger.exception("collection failed (%s)", collection_id)
+                return []
+        return self._stamp_owned(items)
+
     def trailer(self, kind, tmdb_id) -> dict | None:
         """Best YouTube trailer for a title (cached a day — trailers don't move)."""
         w = self.workers.get("tmdb")

database/video_database.py

@@ -796,6 +796,77 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def owned_movie_tmdb_ids(self, server_source=None) -> set:
+        """Set of TMDB ids the user OWNS (movies with a file) — for diffing against
+        franchise/filmography lists in the gap engine."""
+        sql = "SELECT DISTINCT tmdb_id FROM movies WHERE has_file=1 AND tmdb_id IS NOT NULL"
+        args: list = []
+        if server_source:
+            sql += " AND server_source=?"
+            args.append(server_source)
+        conn = self._get_connection()
+        try:
+            return {r["tmdb_id"] for r in conn.execute(sql, args)}
+        except sqlite3.Error:
+            return set()
+        finally:
+            conn.close()
+
+    def owned_movie_collections(self, server_source=None, limit: int = 12) -> list:
+        """Franchises the user has STARTED (owns >=1 movie in), most-invested first —
+        drives the 'Complete your collections' gap rails. Returns
+        [{collection_id, name, owned_count}]."""
+        sql = ("SELECT tmdb_collection_id AS cid, "
+               "MAX(tmdb_collection_name) AS name, COUNT(*) AS c "
+               "FROM movies WHERE has_file=1 AND tmdb_collection_id IS NOT NULL")
+        args: list = []
+        if server_source:
+            sql += " AND server_source=?"
+            args.append(server_source)
+        sql += " GROUP BY tmdb_collection_id ORDER BY c DESC, name LIMIT ?"
+        args.append(int(limit))
+        conn = self._get_connection()
+        try:
+            return [{"collection_id": r["cid"], "name": r["name"], "owned_count": r["c"]}
+                    for r in conn.execute(sql, args)]
+        except sqlite3.Error:
+            return []
+        finally:
+            conn.close()
+
+    def top_owned_people(self, jobs=("Director", "Creator"), min_titles: int = 2,
+                         limit: int = 8, server_source=None) -> list:
+        """People the user owns the most titles from (e.g. directors), busiest first —
+        drives the 'More from <person>' gap rails. Returns
+        [{person_id, tmdb_id, name, owned_count}] for people with a TMDB id and at
+        least ``min_titles`` owned movies in the given crew ``jobs``."""
+        job_list = [j for j in (jobs or []) if j]
+        if not job_list:
+            return []
+        placeholders = ",".join("?" for _ in job_list)
+        sql = (f"SELECT p.id AS pid, p.tmdb_id AS tmdb_id, p.name AS name, "
+               f"COUNT(DISTINCT c.movie_id) AS c "
+               f"FROM credits c JOIN people p ON p.id = c.person_id "
+               f"JOIN movies m ON m.id = c.movie_id "
+               f"WHERE m.has_file=1 AND c.department='crew' AND c.job IN ({placeholders}) "
+               f"AND p.tmdb_id IS NOT NULL")
+        args: list = list(job_list)
+        if server_source:
+            sql += " AND m.server_source=?"
+            args.append(server_source)
+        sql += " GROUP BY p.id HAVING c >= ? ORDER BY c DESC, p.name LIMIT ?"
+        args.append(int(min_titles))
+        args.append(int(limit))
+        conn = self._get_connection()
+        try:
+            return [{"person_id": r["pid"], "tmdb_id": r["tmdb_id"],
+                     "name": r["name"], "owned_count": r["c"]}
+                    for r in conn.execute(sql, args)]
+        except sqlite3.Error:
+            return []
+        finally:
+            conn.close()
+
     def apply_ratings(self, kind: str, item_id: int, ratings: dict) -> None:
         """Store IMDb / RT / Metacritic scores (from OMDb) + mark ratings_synced.
         Ratings are dynamic, so these overwrite (unlike gap-only metadata)."""