video: detail-page data layer (show tree + movie) + backdrop proxy

Backend for the upcoming TV/movie detail pages, isolated to video.db:
- show_detail(id): show + seasons->episodes tree with owned/total roll-ups
  (season 0 -> 'Specials', missing-season-row episodes still grouped).
- movie_detail(id): movie + owned flag + best media-file (resolution/quality).
- get_art_ref generalizes the poster ref to poster|backdrop; new
  /api/video/backdrop/<kind>/<id> streams the hero art server-side (Jellyfin
  Backdrop vs Primary handled).
- /api/video/detail/{show,movie}/<id> endpoints.

Seam tests for the tree roll-ups, owned/file, art ref, and both endpoints.

api/video/__init__.py

@@ -43,11 +43,13 @@ def create_video_blueprint() -> Blueprint:
     from .libraries import register_routes as reg_libraries
     from .poster import register_routes as reg_poster
     from .enrichment import register_routes as reg_enrichment
+    from .detail import register_routes as reg_detail
     reg_dashboard(bp)
     reg_scan(bp)
     reg_library(bp)
     reg_libraries(bp)
     reg_poster(bp)
     reg_enrichment(bp)
+    reg_detail(bp)
 
     return bp

api/video/detail.py

@@ -0,0 +1,33 @@
+"""Video detail payloads (drill-in pages).
+
+GET /api/video/detail/show/<id>   → show + seasons→episodes tree (owned roll-ups)
+GET /api/video/detail/movie/<id>  → movie + owned/file info
+
+Reads only video.db; isolated from the music API.
+"""
+
+from __future__ import annotations
+
+from flask import jsonify
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_api.detail")
+
+
+def register_routes(bp):
+    @bp.route("/detail/show/<int:show_id>", methods=["GET"])
+    def video_show_detail(show_id):
+        from . import get_video_db
+        data = get_video_db().show_detail(show_id)
+        if not data:
+            return jsonify({"error": "not found"}), 404
+        return jsonify(data)
+
+    @bp.route("/detail/movie/<int:movie_id>", methods=["GET"])
+    def video_movie_detail(movie_id):
+        from . import get_video_db
+        data = get_video_db().movie_detail(movie_id)
+        if not data:
+            return jsonify({"error": "not found"}), 404
+        return jsonify(data)

api/video/poster.py

@@ -15,10 +15,9 @@ logger = get_logger("video_api.poster")
 
 
 def register_routes(bp):
-    @bp.route("/poster/<kind>/<int:item_id>", methods=["GET"])
-    def video_poster(kind, item_id):
+    def _stream_art(kind, item_id, art):
         from . import get_video_db
-        ref = get_video_db().get_poster_ref(kind, item_id)
+        ref = get_video_db().get_art_ref(kind, item_id, art)
         if not ref or not ref.get("poster_url"):
             abort(404)
         try:
@@ -37,7 +36,8 @@ def register_routes(bp):
                 base, key = cfg.get("base_url"), cfg.get("api_key")
                 if not base:
                     abort(404)
-                url = base.rstrip("/") + f"/Items/{ref['server_id']}/Images/Primary"
+                image = "Backdrop" if art == "backdrop" else "Primary"
+                url = base.rstrip("/") + f"/Items/{ref['server_id']}/Images/{image}"
                 params = {"api_key": key} if key else {}
             else:
                 abort(404)
@@ -50,5 +50,13 @@ def register_routes(bp):
             resp.headers["Cache-Control"] = "public, max-age=86400"
             return resp
         except Exception:
-            logger.exception("video poster proxy failed for %s/%s", kind, item_id)
+            logger.exception("video %s proxy failed for %s/%s", art, kind, item_id)
             abort(404)
+
+    @bp.route("/poster/<kind>/<int:item_id>", methods=["GET"])
+    def video_poster(kind, item_id):
+        return _stream_art(kind, item_id, "poster")
+
+    @bp.route("/backdrop/<kind>/<int:item_id>", methods=["GET"])
+    def video_backdrop(kind, item_id):
+        return _stream_art(kind, item_id, "backdrop")

database/video_database.py

@@ -600,18 +600,115 @@ class VideoDatabase:
 
     def get_poster_ref(self, kind: str, item_id: int) -> dict | None:
         """Server source/id/poster path for one movie or show, for the poster proxy."""
+        return self.get_art_ref(kind, item_id, "poster")
+
+    def get_art_ref(self, kind: str, item_id: int, art: str = "poster") -> dict | None:
+        """Server source/id + artwork path for one movie/show, for the image proxy.
+        ``art`` is 'poster' or 'backdrop'. Returns the path under 'poster_url' so
+        the proxy is artwork-agnostic."""
         table = {"movie": "movies", "show": "shows"}.get(kind)
-        if not table:
+        col = {"poster": "poster_url", "backdrop": "backdrop_url"}.get(art)
+        if not table or not col:
             return None
         conn = self._get_connection()
         try:
             row = conn.execute(
-                f"SELECT server_source, server_id, poster_url FROM {table} WHERE id=?",
+                f"SELECT server_source, server_id, {col} AS poster_url FROM {table} WHERE id=?",
                 (item_id,)).fetchone()
             return dict(row) if row else None
         finally:
             conn.close()
 
+    # ── detail payloads (drill-in pages) ──────────────────────────────────────
+    def show_detail(self, show_id: int) -> dict | None:
+        """Full TV-show detail: the show + its seasons → episodes tree, with
+        owned/total roll-ups. Drives the (isolated) video show-detail page."""
+        conn = self._get_connection()
+        try:
+            show = conn.execute("SELECT * FROM shows WHERE id=?", (show_id,)).fetchone()
+            if not show:
+                return None
+            seasons = conn.execute(
+                "SELECT id, season_number, title, overview, "
+                "(poster_url IS NOT NULL AND poster_url<>'') AS has_poster "
+                "FROM seasons WHERE show_id=? ORDER BY season_number", (show_id,)).fetchall()
+            eps = conn.execute(
+                "SELECT id, season_number, episode_number, title, overview, air_date, "
+                "runtime_minutes, monitored, has_file FROM episodes WHERE show_id=? "
+                "ORDER BY season_number, episode_number", (show_id,)).fetchall()
+        finally:
+            conn.close()
+
+        by_season: dict = {}
+        for e in eps:
+            by_season.setdefault(e["season_number"], []).append({
+                "id": e["id"], "episode_number": e["episode_number"],
+                "title": e["title"], "overview": e["overview"], "air_date": e["air_date"],
+                "runtime_minutes": e["runtime_minutes"],
+                "monitored": bool(e["monitored"]), "owned": bool(e["has_file"]),
+            })
+
+        # Seasons declared in the seasons table, plus any season numbers that only
+        # exist via episodes (defensive — a show with episodes but no season row).
+        season_nums = [s["season_number"] for s in seasons]
+        season_meta = {s["season_number"]: s for s in seasons}
+        for num in by_season:
+            if num not in season_meta:
+                season_nums.append(num)
+        out_seasons = []
+        for num in sorted(set(season_nums)):
+            ep_list = by_season.get(num, [])
+            owned = sum(1 for e in ep_list if e["owned"])
+            meta = season_meta.get(num)
+            out_seasons.append({
+                "season_number": num,
+                "title": (meta["title"] if meta else None) or (
+                    "Specials" if num == 0 else "Season %d" % num),
+                "overview": meta["overview"] if meta else None,
+                "has_poster": bool(meta["has_poster"]) if meta else False,
+                "episode_total": len(ep_list),
+                "episode_owned": owned,
+                "episodes": ep_list,
+            })
+
+        total = len(eps)
+        owned_total = sum(1 for e in eps if e["has_file"])
+        return {
+            "kind": "show", "id": show["id"], "title": show["title"], "year": show["year"],
+            "overview": show["overview"], "status": show["status"], "network": show["network"],
+            "content_rating": show["content_rating"], "runtime_minutes": show["runtime_minutes"],
+            "tmdb_id": show["tmdb_id"], "tvdb_id": show["tvdb_id"], "imdb_id": show["imdb_id"],
+            "has_poster": bool(show["poster_url"]), "has_backdrop": bool(show["backdrop_url"]),
+            "season_count": len(out_seasons),
+            "episode_total": total, "episode_owned": owned_total,
+            "seasons": out_seasons,
+        }
+
+    def movie_detail(self, movie_id: int) -> dict | None:
+        """Full movie detail: the movie + owned/file info. Drives the (isolated)
+        video movie-detail page."""
+        conn = self._get_connection()
+        try:
+            m = conn.execute("SELECT * FROM movies WHERE id=?", (movie_id,)).fetchone()
+            if not m:
+                return None
+            f = conn.execute(
+                "SELECT resolution, quality, video_codec, audio_codec, size_bytes "
+                "FROM media_files WHERE movie_id=? ORDER BY size_bytes DESC LIMIT 1",
+                (movie_id,)).fetchone()
+        finally:
+            conn.close()
+        return {
+            "kind": "movie", "id": m["id"], "title": m["title"], "year": m["year"],
+            "overview": m["overview"], "status": m["status"], "studio": m["studio"],
+            "release_date": m["release_date"], "runtime_minutes": m["runtime_minutes"],
+            "content_rating": m["content_rating"],
+            "tmdb_id": m["tmdb_id"], "imdb_id": m["imdb_id"],
+            "has_poster": bool(m["poster_url"]), "has_backdrop": bool(m["backdrop_url"]),
+            "owned": bool(m["has_file"]), "monitored": bool(m["monitored"]),
+            "file": (dict(f) if f else None),
+        }
+
     # ── paged/filtered/sorted library query (server-side, like music) ─────────
     def query_library(self, kind: str, *, search=None, letter=None, sort="title",
                       status="all", page=1, limit=75) -> dict:

tests/test_video_api.py

@@ -40,6 +40,39 @@ def test_blueprint_exposes_dashboard_route():
     assert "/api/video/enrichment/<service>/unmatched" in rules
     assert "/api/video/enrichment/config" in rules
     assert "/api/video/enrichment/<service>/test" in rules
+    assert "/api/video/detail/show/<int:show_id>" in rules
+    assert "/api/video/detail/movie/<int:movie_id>" in rules
+    assert any(r.startswith("/api/video/backdrop/") for r in rules)
+
+
+def test_show_detail_endpoint(tmp_path):
+    client, videoapi = _make_client(tmp_path)
+    try:
+        sid = videoapi._video_db.upsert_show_tree("plex", {
+            "server_id": "s1", "title": "Show", "seasons": [
+                {"season_number": 1, "episodes": [
+                    {"episode_number": 1, "title": "Pilot",
+                     "file": {"relative_path": "e1.mkv", "size_bytes": 5}}]}]})
+        resp = client.get("/api/video/detail/show/%d" % sid)
+        assert resp.status_code == 200
+        d = resp.get_json()
+        assert d["kind"] == "show" and d["episode_total"] == 1 and d["episode_owned"] == 1
+        assert d["seasons"][0]["episodes"][0]["title"] == "Pilot"
+        assert client.get("/api/video/detail/show/999999").status_code == 404
+    finally:
+        videoapi._video_db = None
+
+
+def test_movie_detail_endpoint(tmp_path):
+    client, videoapi = _make_client(tmp_path)
+    try:
+        mid = videoapi._video_db.upsert_movie("plex", {"server_id": "m1", "title": "Dune"})
+        resp = client.get("/api/video/detail/movie/%d" % mid)
+        assert resp.status_code == 200
+        assert resp.get_json()["title"] == "Dune"
+        assert client.get("/api/video/detail/movie/999999").status_code == 404
+    finally:
+        videoapi._video_db = None
 
 
 def test_dashboard_endpoint_returns_zeroed_json(tmp_path):

tests/test_video_database.py

@@ -241,6 +241,61 @@ def test_upsert_stores_provider_ids(db):
     assert erow["tvdb_id"] == 349232
 
 
+def test_show_detail_builds_season_episode_tree_with_rollups(db):
+    sid = db.upsert_show_tree("plex", {
+        "server_id": "s1", "title": "Show", "year": 2019, "overview": "A show",
+        "network": "HBO", "content_rating": "TV-MA", "status": "ended",
+        "poster_url": "/p.jpg", "seasons": [
+            {"season_number": 0, "episodes": [{"episode_number": 1, "title": "Special"}]},
+            {"season_number": 1, "title": "Season One", "episodes": [
+                {"episode_number": 1, "title": "Pilot", "air_date": "2019-01-01",
+                 "file": {"relative_path": "e1.mkv", "size_bytes": 5}},
+                {"episode_number": 2, "title": "Two", "air_date": "2019-01-08"}]}]})
+    # backdrop_url is filled by TMDB enrichment, not the scan — simulate that.
+    with db.connect() as c:
+        c.execute("UPDATE shows SET backdrop_url='/b.jpg' WHERE id=?", (sid,))
+        c.commit()
+    d = db.show_detail(sid)
+    assert d["title"] == "Show" and d["network"] == "HBO" and d["status"] == "ended"
+    assert d["has_poster"] and d["has_backdrop"]
+    assert (d["episode_total"], d["episode_owned"], d["season_count"]) == (3, 1, 2)
+    # Season 0 renders as "Specials"; season 1 keeps its title; ordered by number.
+    assert [s["season_number"] for s in d["seasons"]] == [0, 1]
+    assert d["seasons"][0]["title"] == "Specials"
+    s1 = d["seasons"][1]
+    assert s1["title"] == "Season One"
+    assert (s1["episode_total"], s1["episode_owned"]) == (2, 1)
+    assert s1["episodes"][0]["owned"] is True and s1["episodes"][1]["owned"] is False
+
+
+def test_show_detail_returns_none_for_missing(db):
+    assert db.show_detail(999999) is None
+
+
+def test_movie_detail_includes_owned_and_file(db):
+    mid = db.upsert_movie("plex", {
+        "server_id": "m1", "title": "Dune", "year": 2021, "overview": "Sand",
+        "tmdb_id": 438631, "poster_url": "/p.jpg",
+        "file": {"relative_path": "dune.mkv", "size_bytes": 99, "resolution": "2160p"}})
+    d = db.movie_detail(mid)
+    assert d["title"] == "Dune" and d["owned"] is True and d["tmdb_id"] == 438631
+    assert d["file"] and d["file"]["resolution"] == "2160p"
+    # A wishlist movie with no file reports owned False, file None.
+    mid2 = db.upsert_movie("plex", {"server_id": "m2", "title": "Wanted"})
+    d2 = db.movie_detail(mid2)
+    assert d2["owned"] is False and d2["file"] is None
+
+
+def test_get_art_ref_poster_vs_backdrop(db):
+    sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "poster_url": "/p.jpg"})
+    with db.connect() as c:           # backdrop is enrichment-filled, not scanned
+        c.execute("UPDATE shows SET backdrop_url='/b.jpg' WHERE id=?", (sid,))
+        c.commit()
+    assert db.get_art_ref("show", sid, "poster")["poster_url"] == "/p.jpg"
+    assert db.get_art_ref("show", sid, "backdrop")["poster_url"] == "/b.jpg"
+    assert db.get_art_ref("show", sid, "bogus") is None
+
+
 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).