video enrichment: cascade episode backfill from the TMDB show worker

Episodes ride along with their show instead of being a separate (tens-of-thousands)
queue: when the TMDB worker matches a show, it now backfills every season's
episodes — still / overview / rating — via /tv/<id>/season/<n> (one call per
season, gap-only so server data is never clobbered). Also backfills season
overviews.

The worker manager 'knows about it': the TMDB breakdown gains an Episodes
coverage entry (matched = has art, rest = pending), shown as its own card; the
Episodes view lists episodes still missing art. It's coverage-only, kept out of
the worker's idle/pending calc so it never blocks 'Complete'.

Seam tests: client season parse, worker cascade fills episodes, gap-only backfill
+ season overview, breakdown coverage (tmdb only), missing-art list, idle calc
ignores episode coverage.

core/video/enrichment/clients.py

@@ -115,6 +115,29 @@ class TMDBClient:
             logger.exception("TMDB details fetch failed for %s", title or tmdb_id)
         return {"id": tmdb_id, "metadata": {k: v for k, v in meta.items() if v}}
 
+    def season_episodes(self, tv_id, season_number):
+        """Episode-level data for one season (still/overview/rating) — the show
+        worker cascades over a show's seasons to backfill episodes the media
+        server lacked. Returns {'overview', 'episodes': [...]} or None."""
+        if not self.api_key or tv_id is None or season_number is None:
+            return None
+        import requests
+        r = requests.get(self.BASE + "/tv/" + str(tv_id) + "/season/" + str(season_number),
+                         params={"api_key": self.api_key}, timeout=15)
+        r.raise_for_status()
+        data = r.json() or {}
+        out = []
+        for e in (data.get("episodes") or []):
+            en = e.get("episode_number")
+            if en is None:
+                continue
+            ep = {"episode_number": en, "overview": e.get("overview"),
+                  "rating": e.get("vote_average") or None}
+            if e.get("still_path"):
+                ep["still_url"] = self.IMG + e["still_path"]
+            out.append(ep)
+        return {"overview": data.get("overview"), "episodes": out}
+
 
 class TVDBClient:
     BASE = "https://api4.thetvdb.com/v4"

core/video/enrichment/worker.py

@@ -123,18 +123,41 @@ class VideoEnrichmentWorker:
             logger.info("Matched %s '%s' -> %s ID: %s%s", item["kind"], item["title"],
                         self.display_name, result["id"],
                         " (by server id)" if item.get("known_id") else "")
+            # Cascade: a matched show backfills its episodes' art/overview/rating
+            # from the same provider (one call per season), so episodes ride along
+            # with their show instead of being a separate (huge) queue.
+            if item["kind"] == "show" and hasattr(self.client, "season_episodes"):
+                self._cascade_episodes(item["id"], result["id"])
         else:
             self.db.enrichment_apply(self.service, item["kind"], item["id"], matched=False)
             self.stats["not_found"] += 1
             logger.info("No %s match for %s '%s'", self.display_name, item["kind"], item["title"])
         return True
 
+    def _cascade_episodes(self, show_id, tv_id) -> None:
+        """Backfill a show's episodes from the provider (one call per season).
+        Best-effort: a season failure never aborts the show's enrichment."""
+        try:
+            seasons = self.db.show_season_numbers(show_id)
+        except Exception:
+            logger.exception("episode backfill: season list failed for show %s", show_id)
+            return
+        for snum in seasons:
+            try:
+                data = self.client.season_episodes(tv_id, snum)
+                if data and data.get("episodes"):
+                    self.db.backfill_episodes(show_id, snum, data["episodes"], data.get("overview"))
+            except Exception:
+                logger.exception("episode backfill failed: show %s season %s", show_id, snum)
+
     # ── status (same shape the music enrichment API returns) ──────────────────
     def get_stats(self) -> dict:
         breakdown = self.db.enrichment_breakdown(self.service)
         # Errored items are outstanding (retried later), so they count as pending
-        # work — the worker isn't "Complete" while any remain.
-        pending = sum(b["pending"] + b.get("errors", 0) for b in breakdown.values())
+        # work — the worker isn't "Complete" while any remain. Episode art is a
+        # coverage-only cascade (no queue), so it's excluded from idle/pending.
+        pending = sum(b["pending"] + b.get("errors", 0)
+                      for b in breakdown.values() if not b.get("coverage_only"))
         running = self.running and not self.paused and self.enabled
         idle = running and pending == 0 and self.current_item is None
         progress = {}

database/video_database.py

@@ -277,12 +277,67 @@ class VideoDatabase:
                     "errors": conn.execute(f"SELECT COUNT(*) FROM {tbl} WHERE {sc}='error'").fetchone()[0],
                     "pending": conn.execute(f"SELECT COUNT(*) FROM {tbl} WHERE {sc} IS NULL").fetchone()[0],
                 }
+            # TMDB also cascades episode art (still) backfill from the show worker,
+            # so the manager sees episode coverage. Not a queue (matched = has a
+            # still; the rest are "pending" art) — kept out of the idle/pending
+            # calc by the worker so it never blocks "Complete".
+            if service == "tmdb":
+                total = conn.execute("SELECT COUNT(*) FROM episodes").fetchone()[0]
+                with_still = conn.execute(
+                    "SELECT COUNT(*) FROM episodes WHERE still_url IS NOT NULL AND still_url<>''").fetchone()[0]
+                out["episode"] = {"matched": with_still, "not_found": 0, "errors": 0,
+                                  "pending": total - with_still, "coverage_only": True}
             return out
         finally:
             conn.close()
 
+    def show_season_numbers(self, show_id: int) -> list:
+        conn = self._get_connection()
+        try:
+            return [r["season_number"] for r in conn.execute(
+                "SELECT season_number FROM seasons WHERE show_id=? ORDER BY season_number",
+                (show_id,)).fetchall()]
+        finally:
+            conn.close()
+
+    def backfill_episodes(self, show_id: int, season_number: int, episodes: list,
+                          season_overview: str | None = None) -> int:
+        """Gap-only backfill of episode still/overview/rating (and the season's
+        overview) from enrichment — never clobbers what the server provided.
+        Returns the number of episode rows touched."""
+        conn = self._get_connection()
+        touched = 0
+        try:
+            for e in (episodes or []):
+                en = e.get("episode_number")
+                if en is None:
+                    continue
+                sets, params = [], []
+                for col in ("still_url", "overview", "rating"):
+                    val = e.get(col)
+                    if val is None:
+                        continue
+                    sets.append(f"{col}=COALESCE(NULLIF({col}, ''), ?)")
+                    params.append(val)
+                if not sets:
+                    continue
+                params += [show_id, season_number, en]
+                cur = conn.execute(
+                    f"UPDATE episodes SET {', '.join(sets)} "
+                    "WHERE show_id=? AND season_number=? AND episode_number=?", params)
+                touched += cur.rowcount
+            if season_overview:
+                conn.execute("UPDATE seasons SET overview=COALESCE(NULLIF(overview, ''), ?) "
+                             "WHERE show_id=? AND season_number=?", (season_overview, show_id, season_number))
+            conn.commit()
+            return touched
+        finally:
+            conn.close()
+
     def enrichment_unmatched(self, service: str, kind: str, status: str = "not_found",
                              search=None, limit: int = 50, offset: int = 0) -> dict:
+        if kind == "episode" and service == "tmdb":
+            return self._episodes_missing_art(search, limit, offset)
         spec = _ENRICH.get(service, {}).get(kind)
         if not spec:
             return {"items": [], "total": 0}
@@ -315,6 +370,31 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    def _episodes_missing_art(self, search, limit, offset) -> dict:
+        """Episodes still lacking a still image (for the manager's Episodes view).
+        Read-only: episode art is backfilled as a cascade, not a retry queue."""
+        where = ["(e.still_url IS NULL OR e.still_url='')"]
+        params: list = []
+        if search:
+            where.append("(e.title LIKE ? COLLATE NOCASE OR sh.title LIKE ? COLLATE NOCASE)")
+            params += ["%" + search + "%", "%" + search + "%"]
+        where_sql = " WHERE " + " AND ".join(where)
+        conn = self._get_connection()
+        try:
+            total = conn.execute(
+                f"SELECT COUNT(*) FROM episodes e JOIN shows sh ON sh.id=e.show_id{where_sql}",
+                params).fetchone()[0]
+            rows = conn.execute(
+                "SELECT e.id, sh.title || ' · S' || e.season_number || 'E' || e.episode_number "
+                "|| COALESCE(' · ' || e.title, '') AS title, e.air_date AS year, "
+                "0 AS has_poster, NULL AS last_attempted "
+                f"FROM episodes e JOIN shows sh ON sh.id=e.show_id{where_sql} "
+                "ORDER BY sh.sort_title, e.season_number, e.episode_number LIMIT ? OFFSET ?",
+                params + [limit, offset]).fetchall()
+            return {"items": [dict(r) | {"has_poster": False} for r in rows], "total": total}
+        finally:
+            conn.close()
+
     def enrichment_retry(self, service: str, kind: str, scope: str = "failed", item_id=None) -> int:
         """Re-queue items by resetting status/last_attempted to NULL."""
         spec = _ENRICH.get(service, {}).get(kind)

tests/test_video_database.py

@@ -558,6 +558,43 @@ def test_enrichment_backfills_season_posters_only_when_missing(db):
     assert db.show_detail(sid)["seasons"][0]["has_poster"] is True
 
 
+def test_backfill_episodes_gap_only_and_season_overview(db):
+    sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
+        {"season_number": 1, "episodes": [
+            {"episode_number": 1, "overview": "server overview"},   # has overview already
+            {"episode_number": 2}]}]})                              # bare
+    n = db.backfill_episodes(sid, 1, [
+        {"episode_number": 1, "still_url": "/e1.jpg", "overview": "tmdb overview", "rating": 8.0},
+        {"episode_number": 2, "still_url": "/e2.jpg", "overview": "tmdb e2", "rating": 7.0},
+    ], season_overview="Season one")
+    assert n == 2
+    eps = {e["episode_number"]: e for e in db.show_detail(sid)["seasons"][0]["episodes"]}
+    assert eps[1]["has_still"] is True and eps[1]["overview"] == "server overview"  # overview kept
+    assert eps[2]["has_still"] is True and eps[2]["overview"] == "tmdb e2"          # gap filled
+    with db.connect() as c:
+        assert c.execute("SELECT overview FROM seasons WHERE show_id=? AND season_number=1",
+                         (sid,)).fetchone()["overview"] == "Season one"
+
+
+def test_breakdown_reports_episode_art_coverage_for_tmdb(db):
+    db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
+        {"season_number": 1, "episodes": [{"episode_number": 1, "still_url": "/e1.jpg"},
+                                          {"episode_number": 2}]}]})
+    bd = db.enrichment_breakdown("tmdb")
+    assert bd["episode"]["matched"] == 1 and bd["episode"]["pending"] == 1
+    assert bd["episode"].get("coverage_only") is True
+    # TVDB doesn't cascade episodes → no episode coverage entry.
+    assert "episode" not in db.enrichment_breakdown("tvdb")
+
+
+def test_unmatched_lists_episodes_missing_art(db):
+    db.upsert_show_tree("plex", {"server_id": "s1", "title": "Show", "seasons": [
+        {"season_number": 1, "episodes": [{"episode_number": 1, "title": "Has", "still_url": "/e1.jpg"},
+                                          {"episode_number": 2, "title": "Missing"}]}]})
+    res = db.enrichment_unmatched("tmdb", "episode", status="unmatched")
+    assert res["total"] == 1 and "Missing" in res["items"][0]["title"]
+
+
 def test_error_status_is_distinct_and_retryable_in_ui(db):
     a = db.upsert_movie("plex", {"server_id": "m1", "title": "A"})
     db.enrichment_apply("tmdb", "movie", a, matched=False, error=True)

tests/test_video_enrichment.py

@@ -98,6 +98,51 @@ def test_tmdb_pulls_full_metadata(monkeypatch):
     assert m["genres"] == ["Sci-Fi", "Drama"] and m["status"] == "Released" and m["imdb_id"] == "tt1"
 
 
+def test_show_worker_cascades_episode_backfill(db):
+    # A matched show backfills its episodes' art via the client's season_episodes
+    # cascade (episodes ride along with their show — no separate queue).
+    sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
+        {"season_number": 1, "episodes": [{"episode_number": 1}, {"episode_number": 2}]}]})
+
+    class CascadeClient:
+        enabled = True
+        def match(self, kind, title, year, known_id=None):
+            return {"id": 1396, "metadata": {}}
+        def season_episodes(self, tv_id, snum):
+            assert tv_id == 1396
+            return {"overview": "S%d" % snum, "episodes": [
+                {"episode_number": 1, "still_url": "/e1.jpg", "overview": "O1", "rating": 8.0}]}
+
+    w = VideoEnrichmentWorker(db, "tmdb", CascadeClient())
+    assert w.process_one() is True
+    eps = {e["episode_number"]: e for e in db.show_detail(sid)["seasons"][0]["episodes"]}
+    assert eps[1]["has_still"] is True and eps[2]["has_still"] is False   # cascade filled E1
+
+
+def test_get_stats_excludes_episode_coverage_from_pending(db):
+    sid = db.upsert_show_tree("plex", {"server_id": "s1", "title": "S", "seasons": [
+        {"season_number": 1, "episodes": [{"episode_number": 1}]}]})   # episode has no still
+    db.enrichment_apply("tmdb", "show", sid, matched=True, external_id=1)   # show done
+    stats = VideoEnrichmentWorker(db, "tmdb", FakeClient(None)).get_stats()
+    assert "episode" in stats["breakdown"]            # manager sees episode coverage
+    assert stats["stats"]["pending"] == 0             # but it doesn't block "Complete"
+
+
+def test_tmdb_season_episodes_parses(monkeypatch):
+    class _Resp:
+        def __init__(self, b): self._b = b
+        def raise_for_status(self): pass
+        def json(self): return self._b
+    body = {"overview": "Season 1", "episodes": [
+        {"episode_number": 1, "still_path": "/a.jpg", "overview": "O", "vote_average": 8.1},
+        {"episode_number": 2, "still_path": None, "overview": "P", "vote_average": 0}]}
+    monkeypatch.setitem(sys.modules, "requests", types.SimpleNamespace(get=lambda u, **k: _Resp(body)))
+    res = TMDBClient("KEY").season_episodes(1396, 1)
+    assert res["overview"] == "Season 1" and len(res["episodes"]) == 2
+    assert res["episodes"][0]["still_url"] == "https://image.tmdb.org/t/p/original/a.jpg"
+    assert "still_url" not in res["episodes"][1]      # no still_path → omitted
+
+
 def test_tmdb_show_returns_season_posters(monkeypatch):
     class _Resp:
         def __init__(self, b): self._b = b

webui/static/video/video-enrichment-manager.js

@@ -31,8 +31,8 @@
         tmdb: 'https://www.themoviedb.org/assets/2/v4/logos/v2/blue_square_2-d537fb228cf3ded904ef09b136fe3fec72548ebc1fea3fbbd1ad9e36364db38b.svg',
         tvdb: 'https://www.svgrepo.com/show/443500/brand-tvdb.svg',
     };
-    var GLYPH = { movie: '🎬', show: '📺' };
-    var KIND_LABEL = { movie: 'Movies', show: 'Shows' };
+    var GLYPH = { movie: '🎬', show: '📺', episode: '🎞️' };
+    var KIND_LABEL = { movie: 'Movies', show: 'Shows', episode: 'Episodes' };
 
     var state = {
         open: false, selected: 'tmdb', statuses: {}, breakdown: null,