Video enrichment: backfill-worker framework + 4 new workers (fanart.tv, OpenSubtitles, Return YouTube Dislike, SponsorBlock)

Adds a VideoBackfillWorker base that enriches already-identified items BY id
(vs the matcher workers), with the exact same lifecycle + get_stats() shape so
the engine registry, /api/video/enrichment routes, and Manage-Workers modal
drive them identically.

Workers:
- fanart.tv (free key): gap-fills logo/clearart/banner/backdrop/poster art
- OpenSubtitles (free key): records subtitle-language availability per title
- Return YouTube Dislike (no key): like/dislike estimates on cached videos
- SponsorBlock (no key): crowd segments per video

DB: youtube_video_stats + youtube_video_segments tables; fanart/subs columns;
backfill_next/mark/breakdown + youtube_enrich_* helpers; likes/dislikes merged
into get_channel_videos. Seam tests in tests/test_video_backfill.py.

core/video/enrichment/backfill.py

@@ -0,0 +1,493 @@
+"""Video BACKFILL workers — enrich an already-identified item BY id.
+
+The matcher workers (worker.py) find an external id for a title. These workers
+take items we can already address (a tmdb/imdb/tvdb id, or a YouTube video id)
+and fetch SUPPLEMENTARY data for them: artwork (fanart.tv), subtitle
+availability (OpenSubtitles), and the no-key YouTube extras — Return YouTube
+Dislike (like/dislike estimates) and SponsorBlock (crowd segments).
+
+Each worker presents the EXACT same lifecycle + get_stats() shape as
+VideoEnrichmentWorker, so the engine registry, the /api/video/enrichment routes,
+and the Manage-Workers modal (cards / animations / pause-resume) drive them
+identically — a new worker is just another entry in engine.workers.
+
+Isolated: imports only video.db helpers + requests; no music code.
+"""
+
+from __future__ import annotations
+
+import json
+import threading
+import time
+
+from utils.logging_config import get_logger
+
+logger = get_logger("video_enrichment.backfill")
+
+_UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+       "(KHTML, like Gecko) Chrome/124.0 Safari/537.36")
+
+
+# ── tiny HTTP helper with the status semantics the workers rely on ────────────
+class _RateLimited(Exception):
+    def __init__(self, retry_after=60):
+        self.retry_after = retry_after
+
+
+class _Unauthorized(Exception):
+    pass
+
+
+def _http_get_json(url, params=None, headers=None, timeout=12):
+    """GET → parsed JSON (list or dict), or None for a 404 / unparseable body.
+    Raises _Unauthorized (401/403), _RateLimited (429), or the underlying error
+    so the worker can record 'error' vs back off vs mark 'not_found'."""
+    import requests
+    h = {"User-Agent": _UA}
+    if headers:
+        h.update(headers)
+    r = requests.get(url, params=params, headers=h, timeout=timeout)
+    if r.status_code == 404:
+        return None
+    if r.status_code in (401, 403):
+        raise _Unauthorized()
+    if r.status_code == 429:
+        try:
+            ra = int(r.headers.get("Retry-After") or 60)
+        except (TypeError, ValueError):
+            ra = 60
+        raise _RateLimited(ra)
+    r.raise_for_status()
+    try:
+        return r.json()
+    except Exception:
+        return None
+
+
+# ── base worker (lifecycle + loop + status; mirrors VideoEnrichmentWorker) ────
+class VideoBackfillWorker:
+    is_ratings = False
+
+    def __init__(self, db, service, display_name, interval=1.0):
+        self.db = db
+        self.service = service
+        self.display_name = display_name
+        self.interval = interval
+        self.running = False
+        self.paused = False
+        self.should_stop = False
+        self._thread = None
+        self._stop = threading.Event()
+        self.current_item = None
+        self.stats = {"matched": 0, "not_found": 0, "errors": 0}
+        self.note = None
+        self._cooldown_until = 0.0
+
+    # ── lifecycle ─────────────────────────────────────────────────────────────
+    def start(self):
+        if self.running:
+            return
+        self.should_stop = False
+        self._stop.clear()
+        self.running = True
+        self._thread = threading.Thread(target=self._run, daemon=True)
+        self._thread.start()
+
+    def stop(self):
+        self.should_stop = True
+        self._stop.set()
+        if self._thread:
+            self._thread.join(timeout=1.0)
+        self.running = False
+
+    def pause(self, persist=True):
+        self.paused = True
+        if persist:
+            self._persist_paused()
+
+    def resume(self, persist=True):
+        self.paused = False
+        if persist:
+            self._persist_paused()
+
+    def _persist_paused(self):
+        try:
+            self.db.set_setting(self.service + "_paused", "1" if self.paused else "0")
+        except Exception:
+            logger.exception("video backfill: could not persist pause for %s", self.service)
+
+    def restore_paused(self):
+        try:
+            self.paused = str(self.db.get_setting(self.service + "_paused") or "") == "1"
+        except Exception:
+            logger.exception("video backfill: could not restore pause for %s", self.service)
+
+    @property
+    def enabled(self):
+        try:
+            return self._enabled()
+        except Exception:
+            return False
+
+    # ── subclass hooks ────────────────────────────────────────────────────────
+    def _enabled(self) -> bool:
+        return True
+
+    def test(self):
+        return (True, self.display_name + " OK")
+
+    def next_item(self):
+        raise NotImplementedError
+
+    def fetch(self, item):
+        """Return data (truthy) on a hit, falsy for a genuine 'no data', or raise
+        on a call failure (network/rate-limit/auth)."""
+        raise NotImplementedError
+
+    def record_ok(self, item, data):
+        raise NotImplementedError
+
+    def record_empty(self, item):
+        raise NotImplementedError
+
+    def record_error(self, item):
+        raise NotImplementedError
+
+    def breakdown(self) -> dict:
+        raise NotImplementedError
+
+    # ── loop ──────────────────────────────────────────────────────────────────
+    def _run(self):
+        while not self.should_stop:
+            if self.paused or not self.enabled:
+                self._stop.wait(1.0)
+                continue
+            if self._cooldown_until > time.monotonic():
+                self.current_item = None
+                self._stop.wait(15.0)
+                continue
+            try:
+                did = self.process_one()
+            except Exception:
+                logger.exception("video backfill %s loop error", self.service)
+                self.stats["errors"] += 1
+                self._stop.wait(5.0)
+                continue
+            if did:
+                self._stop.wait(self.interval)
+            else:
+                self.current_item = None
+                self._stop.wait(10.0)
+
+    def process_one(self) -> bool:
+        item = self.next_item()
+        if not item:
+            return False
+        self.current_item = {"type": item.get("kind"), "name": item.get("name")}
+        try:
+            data = self.fetch(item)
+        except _RateLimited as e:
+            self._cooldown_until = time.monotonic() + max(15, e.retry_after)
+            self.note = self.display_name + " rate-limited — backing off"
+            logger.warning("%s rate-limited; idling %ss", self.display_name, e.retry_after)
+            return False
+        except _Unauthorized:
+            self.note = self.display_name + " rejected the API key"
+            self.pause(persist=False)        # transient: fixing the key rebuilds the engine
+            logger.warning("%s rejected the API key — pausing until fixed", self.display_name)
+            return False
+        except Exception:
+            logger.exception("video backfill %s fetch failed for %s", self.service, item.get("name"))
+            self.stats["errors"] += 1
+            self.record_error(item)
+            return True
+        self.note = None
+        if data:
+            self.record_ok(item, data)
+            self.stats["matched"] += 1
+            logger.info("Enriched %s '%s' via %s", item.get("kind"), item.get("name"), self.display_name)
+        else:
+            self.record_empty(item)
+            self.stats["not_found"] += 1
+        return True
+
+    # ── status (same shape the music/video enrichment API returns) ────────────
+    def get_stats(self) -> dict:
+        breakdown = self.breakdown()
+        pending = sum(b["pending"] + b.get("errors", 0)
+                      for b in breakdown.values() if not b.get("coverage_only"))
+        cooling = self._cooldown_until > time.monotonic()
+        running = self.running and not self.paused and self.enabled and not cooling
+        idle = running and pending == 0 and self.current_item is None
+        progress = {}
+        for kind, b in breakdown.items():
+            total = b["matched"] + b["not_found"] + b.get("errors", 0) + b["pending"]
+            done = b["matched"] + b["not_found"]
+            progress[kind] = {"matched": b["matched"], "total": total,
+                              "percent": round(done / total * 100) if total else 0}
+        return {
+            "enabled": self.enabled,
+            "running": running,
+            "paused": self.paused or cooling,
+            "idle": idle,
+            "current_item": self.current_item,
+            "note": self.note,
+            "cooldown": cooling,
+            "stats": {**self.stats, "pending": pending},
+            "progress": progress,
+            "breakdown": breakdown,
+        }
+
+
+# ── Return YouTube Dislike (no key) ───────────────────────────────────────────
+class RydWorker(VideoBackfillWorker):
+    URL = "https://returnyoutubedislikeapi.com/votes"
+
+    def __init__(self, db):
+        super().__init__(db, "ryd", "Return YouTube Dislike", interval=0.6)
+
+    def _enabled(self):
+        return str(self.db.get_setting("ryd_enabled") or "1") != "0"
+
+    def test(self):
+        try:
+            j = _http_get_json(self.URL, {"videoId": "dQw4w9WgXcQ"})
+            return (j is not None, "Return YouTube Dislike reachable" if j else "No response")
+        except Exception as e:
+            return (False, str(e))
+
+    def next_item(self):
+        return self.db.youtube_enrich_next("ryd_status")
+
+    def fetch(self, item):
+        j = _http_get_json(self.URL, {"videoId": item["youtube_id"]})
+        if not isinstance(j, dict) or j.get("dislikes") is None:
+            return None
+        return {"likes": j.get("likes"), "dislikes": j.get("dislikes")}
+
+    def record_ok(self, item, data):
+        self.db.apply_youtube_votes(item["youtube_id"], data.get("likes"), data.get("dislikes"), "ok")
+
+    def record_empty(self, item):
+        self.db.apply_youtube_votes(item["youtube_id"], None, None, "not_found")
+
+    def record_error(self, item):
+        self.db.apply_youtube_votes(item["youtube_id"], None, None, "error")
+
+    def breakdown(self):
+        return self.db.youtube_enrich_breakdown("ryd_status")
+
+
+# ── SponsorBlock (no key) ─────────────────────────────────────────────────────
+_SB_CATS = ["sponsor", "selfpromo", "interaction", "intro", "outro",
+            "preview", "music_offtopic", "filler", "poi_highlight"]
+
+
+class SponsorBlockWorker(VideoBackfillWorker):
+    URL = "https://sponsor.ajay.app/api/skipSegments"
+
+    def __init__(self, db):
+        super().__init__(db, "sponsorblock", "SponsorBlock", interval=0.6)
+
+    def _enabled(self):
+        return str(self.db.get_setting("sponsorblock_enabled") or "1") != "0"
+
+    def test(self):
+        try:
+            _http_get_json(self.URL, {"videoID": "dQw4w9WgXcQ", "categories": json.dumps(_SB_CATS)})
+            return (True, "SponsorBlock reachable")
+        except Exception as e:
+            return (False, str(e))
+
+    def next_item(self):
+        return self.db.youtube_enrich_next("sb_status")
+
+    def fetch(self, item):
+        data = _http_get_json(self.URL, {"videoID": item["youtube_id"],
+                                         "categories": json.dumps(_SB_CATS)})
+        if not isinstance(data, list) or not data:
+            return None
+        segs = []
+        for s in data:
+            seg = s.get("segment") or []
+            if len(seg) != 2 or not s.get("UUID") or not s.get("category"):
+                continue
+            segs.append({"category": s.get("category"), "start_sec": seg[0], "end_sec": seg[1],
+                         "votes": s.get("votes"), "uuid": s.get("UUID")})
+        return segs or None
+
+    def record_ok(self, item, data):
+        self.db.apply_youtube_segments(item["youtube_id"], data, "ok")
+
+    def record_empty(self, item):
+        self.db.apply_youtube_segments(item["youtube_id"], None, "not_found")
+
+    def record_error(self, item):
+        self.db.apply_youtube_segments(item["youtube_id"], None, "error")
+
+    def breakdown(self):
+        return self.db.youtube_enrich_breakdown("sb_status")
+
+
+# ── fanart.tv (free key) ──────────────────────────────────────────────────────
+def _fa_first(j, *keys):
+    """First artwork URL from the first present key, preferring English / textless
+    (lang '') and most-liked."""
+    for k in keys:
+        arr = j.get(k) or []
+        if not isinstance(arr, list) or not arr:
+            continue
+        best = sorted(arr, key=lambda a: (a.get("lang") not in ("en", ""),
+                                          -int(a.get("likes") or 0)))
+        url = best[0].get("url") if best else None
+        if url:
+            return url
+    return None
+
+
+class FanartWorker(VideoBackfillWorker):
+    BASE = "https://webservice.fanart.tv/v3"
+
+    def __init__(self, db):
+        super().__init__(db, "fanart", "fanart.tv", interval=1.0)
+
+    def _key(self):
+        return (self.db.get_setting("fanart_api_key") or "").strip()
+
+    def _enabled(self):
+        return bool(self._key())
+
+    def test(self):
+        key = self._key()
+        if not key:
+            return (False, "No fanart.tv API key")
+        try:
+            j = _http_get_json(self.BASE + "/movies/550", {"api_key": key})
+            return (j is not None, "fanart.tv key OK" if j else "No artwork returned")
+        except _Unauthorized:
+            return (False, "fanart.tv rejected the API key")
+        except Exception as e:
+            return (False, str(e))
+
+    def next_item(self):
+        return self.db.backfill_next("fanart")
+
+    def fetch(self, item):
+        key = self._key()
+        if not key:
+            return None
+        if item["kind"] == "movie":
+            ident = item.get("tmdb_id") or item.get("imdb_id")
+            if not ident:
+                return None
+            j = _http_get_json(self.BASE + "/movies/" + str(ident), {"api_key": key})
+            if not isinstance(j, dict):
+                return None
+            out = {"logo_url": _fa_first(j, "hdmovielogo", "clearlogo"),
+                   "clearart_url": _fa_first(j, "hdmovieclearart", "movieart"),
+                   "banner_url": _fa_first(j, "moviebanner"),
+                   "backdrop_url": _fa_first(j, "moviebackground"),
+                   "poster_url": _fa_first(j, "movieposter")}
+        else:
+            ident = item.get("tvdb_id")
+            if not ident:
+                return None
+            j = _http_get_json(self.BASE + "/tv/" + str(ident), {"api_key": key})
+            if not isinstance(j, dict):
+                return None
+            out = {"logo_url": _fa_first(j, "hdtvlogo", "clearlogo"),
+                   "clearart_url": _fa_first(j, "hdclearart", "clearart"),
+                   "banner_url": _fa_first(j, "tvbanner"),
+                   "backdrop_url": _fa_first(j, "showbackground"),
+                   "poster_url": _fa_first(j, "tvposter")}
+        out = {k: v for k, v in out.items() if v}
+        return out or None
+
+    def record_ok(self, item, data):
+        self.db.backfill_mark("fanart", item["kind"], item["id"], "ok", columns=data)
+
+    def record_empty(self, item):
+        self.db.backfill_mark("fanart", item["kind"], item["id"], "not_found")
+
+    def record_error(self, item):
+        self.db.backfill_mark("fanart", item["kind"], item["id"], "error")
+
+    def breakdown(self):
+        return self.db.backfill_breakdown("fanart")
+
+
+# ── OpenSubtitles (free key) — subtitle-language availability ─────────────────
+class OpenSubtitlesWorker(VideoBackfillWorker):
+    BASE = "https://api.opensubtitles.com/api/v1"
+
+    def __init__(self, db):
+        super().__init__(db, "opensubtitles", "OpenSubtitles", interval=1.5)
+
+    def _key(self):
+        return (self.db.get_setting("opensubtitles_api_key") or "").strip()
+
+    def _enabled(self):
+        return bool(self._key())
+
+    def _headers(self):
+        return {"Api-Key": self._key(), "Accept": "application/json",
+                "User-Agent": "SoulSync v1.0"}
+
+    def test(self):
+        if not self._key():
+            return (False, "No OpenSubtitles API key")
+        try:
+            j = _http_get_json(self.BASE + "/subtitles", {"tmdb_id": 550, "languages": "en"},
+                               headers=self._headers())
+            return (j is not None, "OpenSubtitles key OK" if j else "No response")
+        except _Unauthorized:
+            return (False, "OpenSubtitles rejected the API key")
+        except Exception as e:
+            return (False, str(e))
+
+    def next_item(self):
+        return self.db.backfill_next("opensubtitles")
+
+    def fetch(self, item):
+        if not self._key():
+            return None
+        params = {}
+        imdb = str(item.get("imdb_id") or "").lower()
+        if imdb.startswith("tt"):
+            imdb = imdb[2:]
+        if imdb:
+            params["imdb_id"] = imdb
+        elif item.get("tmdb_id"):
+            params["tmdb_id"] = item["tmdb_id"]
+        else:
+            return None
+        j = _http_get_json(self.BASE + "/subtitles", params, headers=self._headers())
+        if not isinstance(j, dict):
+            return None
+        langs = set()
+        for row in (j.get("data") or []):
+            lng = ((row.get("attributes") or {}).get("language") or "").lower()
+            if lng:
+                langs.add(lng)
+        if not langs:
+            return None
+        return {"subtitle_langs": json.dumps(sorted(langs))}
+
+    def record_ok(self, item, data):
+        self.db.backfill_mark("opensubtitles", item["kind"], item["id"], "ok", columns=data)
+
+    def record_empty(self, item):
+        self.db.backfill_mark("opensubtitles", item["kind"], item["id"], "not_found")
+
+    def record_error(self, item):
+        self.db.backfill_mark("opensubtitles", item["kind"], item["id"], "error")
+
+    def breakdown(self):
+        return self.db.backfill_breakdown("opensubtitles")
+
+
+def build_backfill_workers(db) -> dict:
+    """All backfill workers, keyed by service id, for the engine registry."""
+    return {w.service: w for w in (
+        RydWorker(db), SponsorBlockWorker(db), FanartWorker(db), OpenSubtitlesWorker(db),
+    )}

core/video/enrichment/engine.py

@@ -26,6 +26,13 @@ class VideoEnrichmentEngine:
             service: VideoEnrichmentWorker(db, service, client, display_name=_DISPLAY.get(service))
             for service, client in clients.items()
         }
+        # Backfill workers (artwork / subtitles / no-key YouTube extras). Same
+        # lifecycle + get_stats() shape, so the registry/API/UI drive them too.
+        try:
+            from .backfill import build_backfill_workers
+            self.workers.update(build_backfill_workers(db))
+        except Exception:
+            logger.exception("video enrichment: backfill workers unavailable")
         # OMDb ratings (IMDb/RT/Metacritic) — not a matcher, so not a worker;
         # backfilled on the lazy detail refresh.
         self.ratings_client = ratings_client

database/video_database.py

@@ -71,6 +71,28 @@ _ENRICH_META_COLS = {
               "imdb_id", "tmdb_id", "tvdb_id"},
 }
 
+# Backfill-worker plumbing (parallels _ENRICH, but for services that enrich an
+# already-identified library item BY id rather than matching a title). Maps a
+# service + kind to (table, status_col, attempted_col, where_has_required_id).
+_BACKFILL = {
+    "fanart": {
+        "movie": ("movies", "fanart_status", "fanart_attempted",
+                  "(tmdb_id IS NOT NULL OR imdb_id IS NOT NULL)"),
+        "show": ("shows", "fanart_status", "fanart_attempted", "tvdb_id IS NOT NULL"),
+    },
+    "opensubtitles": {
+        "movie": ("movies", "subs_status", "subs_attempted",
+                  "(imdb_id IS NOT NULL OR tmdb_id IS NOT NULL)"),
+        "show": ("shows", "subs_status", "subs_attempted",
+                 "(imdb_id IS NOT NULL OR tmdb_id IS NOT NULL)"),
+    },
+}
+# Columns each backfill service may gap-fill (whitelist; never clobbers server data).
+_BACKFILL_COLS = {
+    "fanart": {"logo_url", "backdrop_url", "poster_url", "clearart_url", "banner_url"},
+    "opensubtitles": {"subtitle_langs"},
+}
+
 # Columns ensured on existing DBs (ALTER TABLE ADD COLUMN; idempotent).
 _COLUMN_MIGRATIONS = [
     ("movies", "tmdb_match_status", "TEXT"),
@@ -115,6 +137,16 @@ _COLUMN_MIGRATIONS = [
     # per-video duration + approximate view count on the remembered catalog
     ("youtube_channel_videos", "duration", "TEXT"),
     ("youtube_channel_videos", "view_count", "INTEGER"),
+    # fanart.tv artwork backfill (gap-fill only; logo/backdrop/poster live already)
+    ("movies", "clearart_url", "TEXT"), ("movies", "banner_url", "TEXT"),
+    ("movies", "fanart_status", "TEXT"), ("movies", "fanart_attempted", "TEXT"),
+    ("shows", "clearart_url", "TEXT"), ("shows", "banner_url", "TEXT"),
+    ("shows", "fanart_status", "TEXT"), ("shows", "fanart_attempted", "TEXT"),
+    # OpenSubtitles availability backfill (which languages exist for a title)
+    ("movies", "subtitle_langs", "TEXT"),            # JSON array of language codes
+    ("movies", "subs_status", "TEXT"), ("movies", "subs_attempted", "TEXT"),
+    ("shows", "subtitle_langs", "TEXT"),
+    ("shows", "subs_status", "TEXT"), ("shows", "subs_attempted", "TEXT"),
 ]
 
 
@@ -335,6 +367,12 @@ class VideoDatabase:
     def enrichment_breakdown(self, service: str) -> dict:
         if service == "omdb":
             return self._ratings_breakdown()
+        if service == "ryd":
+            return self.youtube_enrich_breakdown("ryd_status")
+        if service == "sponsorblock":
+            return self.youtube_enrich_breakdown("sb_status")
+        if service in _BACKFILL:
+            return self.backfill_breakdown(service)
         kinds = _ENRICH.get(service, {})
         out = {}
         conn = self._get_connection()
@@ -360,6 +398,166 @@ class VideoDatabase:
         finally:
             conn.close()
 
+    # ── backfill-worker plumbing (artwork / subtitles, by id) ─────────────────
+    def backfill_next(self, service: str) -> dict | None:
+        """Next library item needing a backfill service: a row that already has the
+        id the service needs and no status yet. Returns
+        {kind, id, title, tmdb_id, imdb_id[, tvdb_id]} or None."""
+        kinds = _BACKFILL.get(service)
+        if not kinds:
+            return None
+        conn = self._get_connection()
+        try:
+            for kind, (tbl, sc, _ac, has_id) in kinds.items():
+                cols = "id, title, tmdb_id, imdb_id" + (", tvdb_id" if tbl == "shows" else "")
+                row = conn.execute(
+                    f"SELECT {cols} FROM {tbl} WHERE {sc} IS NULL AND {has_id} "
+                    f"ORDER BY id LIMIT 1").fetchone()
+                if row:
+                    d = dict(row)
+                    d["kind"] = kind
+                    return d
+            return None
+        finally:
+            conn.close()
+
+    def backfill_mark(self, service: str, kind: str, item_id: int, status: str,
+                      columns: dict | None = None) -> None:
+        """Record a backfill result (status + attempted) and gap-fill whitelisted
+        columns (COALESCE — never clobbers). status: 'ok'|'not_found'|'error'."""
+        spec = _BACKFILL.get(service, {}).get(kind)
+        if not spec:
+            return
+        tbl, sc, ac, _has = spec
+        allowed = _BACKFILL_COLS.get(service, set())
+        sets = [f"{sc}=?", f"{ac}=CURRENT_TIMESTAMP"]
+        params: list = [status]
+        for col, val in (columns or {}).items():
+            if val is None or col not in allowed:
+                continue
+            sets.append(f"{col}=COALESCE(NULLIF({col}, ''), ?)")
+            params.append(val)
+        params.append(item_id)
+        conn = self._get_connection()
+        try:
+            conn.execute(f"UPDATE {tbl} SET {', '.join(sets)} WHERE id=?", params)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def backfill_breakdown(self, service: str) -> dict:
+        kinds = _BACKFILL.get(service, {})
+        out = {}
+        conn = self._get_connection()
+        try:
+            for kind, (tbl, sc, _ac, has_id) in kinds.items():
+                base = f"FROM {tbl} WHERE {has_id}"
+                out[kind] = {
+                    "matched": conn.execute(f"SELECT COUNT(*) {base} AND {sc}='ok'").fetchone()[0],
+                    "not_found": conn.execute(f"SELECT COUNT(*) {base} AND {sc}='not_found'").fetchone()[0],
+                    "errors": conn.execute(f"SELECT COUNT(*) {base} AND {sc}='error'").fetchone()[0],
+                    "pending": conn.execute(f"SELECT COUNT(*) {base} AND {sc} IS NULL").fetchone()[0],
+                }
+            return out
+        finally:
+            conn.close()
+
+    # ── per-video YouTube enrichment (no-key: RYD votes + SponsorBlock) ────────
+    def youtube_enrich_next(self, status_col: str) -> dict | None:
+        """Next cached YouTube video missing a per-video enrichment. status_col is
+        'ryd_status' or 'sb_status'. Distinct by youtube_id (a video shared across
+        playlists is enriched once). Returns {kind:'video', id, name, youtube_id}."""
+        if status_col not in ("ryd_status", "sb_status"):
+            return None
+        conn = self._get_connection()
+        try:
+            row = conn.execute(
+                "SELECT cv.youtube_id AS youtube_id, MIN(cv.title) AS title "
+                "FROM youtube_channel_videos cv "
+                "LEFT JOIN youtube_video_stats s ON s.youtube_id = cv.youtube_id "
+                f"WHERE s.youtube_id IS NULL OR s.{status_col} IS NULL "
+                "GROUP BY cv.youtube_id LIMIT 1").fetchone()
+            if not row:
+                return None
+            return {"kind": "video", "id": row["youtube_id"],
+                    "name": row["title"], "youtube_id": row["youtube_id"]}
+        finally:
+            conn.close()
+
+    def apply_youtube_votes(self, youtube_id, like_count, dislike_count, status: str) -> None:
+        yid = str(youtube_id or "").strip()
+        if not yid:
+            return
+        conn = self._get_connection()
+        try:
+            conn.execute(
+                "INSERT INTO youtube_video_stats "
+                "(youtube_id, like_count, dislike_count, ryd_status, ryd_attempted) "
+                "VALUES (?,?,?,?,CURRENT_TIMESTAMP) ON CONFLICT(youtube_id) DO UPDATE SET "
+                "like_count=COALESCE(excluded.like_count, like_count), "
+                "dislike_count=COALESCE(excluded.dislike_count, dislike_count), "
+                "ryd_status=excluded.ryd_status, ryd_attempted=CURRENT_TIMESTAMP",
+                (yid, like_count, dislike_count, status))
+            conn.commit()
+        finally:
+            conn.close()
+
+    def apply_youtube_segments(self, youtube_id, segments, status: str) -> None:
+        yid = str(youtube_id or "").strip()
+        if not yid:
+            return
+        conn = self._get_connection()
+        try:
+            conn.execute(
+                "INSERT INTO youtube_video_stats (youtube_id, sb_status, sb_attempted) "
+                "VALUES (?,?,CURRENT_TIMESTAMP) ON CONFLICT(youtube_id) DO UPDATE SET "
+                "sb_status=excluded.sb_status, sb_attempted=CURRENT_TIMESTAMP", (yid, status))
+            if segments:
+                conn.execute("DELETE FROM youtube_video_segments WHERE youtube_id=?", (yid,))
+                rows = [(yid, s.get("category"), s.get("start_sec"), s.get("end_sec"),
+                         s.get("votes"), s.get("uuid"))
+                        for s in segments if s.get("uuid") and s.get("category")]
+                if rows:
+                    conn.executemany(
+                        "INSERT OR IGNORE INTO youtube_video_segments "
+                        "(youtube_id, category, start_sec, end_sec, votes, uuid) "
+                        "VALUES (?,?,?,?,?,?)", rows)
+            conn.commit()
+        finally:
+            conn.close()
+
+    def youtube_enrich_breakdown(self, status_col: str) -> dict:
+        if status_col not in ("ryd_status", "sb_status"):
+            return {}
+        conn = self._get_connection()
+        try:
+            total = conn.execute(
+                "SELECT COUNT(DISTINCT youtube_id) FROM youtube_channel_videos").fetchone()[0]
+
+            def c(st):
+                return conn.execute(
+                    f"SELECT COUNT(*) FROM youtube_video_stats WHERE {status_col}=?", (st,)).fetchone()[0]
+
+            matched, nf, err = c("ok"), c("not_found"), c("error")
+            return {"video": {"matched": matched, "not_found": nf, "errors": err,
+                              "pending": max(0, total - matched - nf - err)}}
+        finally:
+            conn.close()
+
+    def youtube_video_segments(self, youtube_id) -> list:
+        """SponsorBlock segments for a video (detail UI / skip logic)."""
+        yid = str(youtube_id or "").strip()
+        if not yid:
+            return []
+        conn = self._get_connection()
+        try:
+            rows = conn.execute(
+                "SELECT category, start_sec, end_sec, votes FROM youtube_video_segments "
+                "WHERE youtube_id=? ORDER BY start_sec", (yid,)).fetchall()
+            return [dict(r) for r in rows]
+        finally:
+            conn.close()
+
     def show_match_info(self, show_id: int) -> dict | None:
         """Title/year/tmdb_id for one show — for on-demand (lazy) art refresh."""
         conn = self._get_connection()
@@ -746,6 +944,39 @@ class VideoDatabase:
                 return cur.rowcount
             finally:
                 conn.close()
+        if service in ("ryd", "sponsorblock"):
+            col = "ryd_status" if service == "ryd" else "sb_status"
+            att = "ryd_attempted" if service == "ryd" else "sb_attempted"
+            conn = self._get_connection()
+            try:
+                if scope == "item" and item_id is not None:
+                    cur = conn.execute(
+                        f"UPDATE youtube_video_stats SET {col}=NULL, {att}=NULL WHERE youtube_id=?",
+                        (str(item_id),))
+                else:
+                    cur = conn.execute(
+                        f"UPDATE youtube_video_stats SET {col}=NULL, {att}=NULL "
+                        f"WHERE {col} IN ('not_found','error')")
+                conn.commit()
+                return cur.rowcount
+            finally:
+                conn.close()
+        if service in _BACKFILL:
+            spec = _BACKFILL[service].get(kind)
+            if not spec:
+                return 0
+            tbl, sc, ac, _has = spec
+            conn = self._get_connection()
+            try:
+                if scope == "item" and item_id is not None:
+                    cur = conn.execute(f"UPDATE {tbl} SET {sc}=NULL, {ac}=NULL WHERE id=?", (item_id,))
+                else:
+                    cur = conn.execute(
+                        f"UPDATE {tbl} SET {sc}=NULL, {ac}=NULL WHERE {sc} IN ('not_found','error')")
+                conn.commit()
+                return cur.rowcount
+            finally:
+                conn.close()
         spec = _ENRICH.get(service, {}).get(kind)
         if not spec:
             return 0
@@ -2200,14 +2431,17 @@ class VideoDatabase:
         conn = self._get_connection()
         try:
             rows = conn.execute(
-                "SELECT v.youtube_id, v.title, v.thumbnail_url, v.duration, v.view_count, d.published_at "
+                "SELECT v.youtube_id, v.title, v.thumbnail_url, v.duration, v.view_count, "
+                "d.published_at, s.like_count, s.dislike_count "
                 "FROM youtube_channel_videos v "
                 "LEFT JOIN youtube_video_dates d ON d.youtube_id = v.youtube_id "
+                "LEFT JOIN youtube_video_stats s ON s.youtube_id = v.youtube_id "
                 "WHERE v.channel_id=? "
                 "ORDER BY (d.published_at IS NULL), d.published_at DESC, v.rowid",
                 (cid,)).fetchall()
             return [{"youtube_id": r["youtube_id"], "title": r["title"], "thumbnail_url": r["thumbnail_url"],
-                     "duration": r["duration"], "view_count": r["view_count"], "published_at": r["published_at"]}
+                     "duration": r["duration"], "view_count": r["view_count"], "published_at": r["published_at"],
+                     "like_count": r["like_count"], "dislike_count": r["dislike_count"]}
                     for r in rows]
         finally:
             conn.close()

database/video_schema.sql

@@ -310,6 +310,34 @@ CREATE TABLE IF NOT EXISTS youtube_channel_meta (
     cached_at        TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
 );
 
+-- Per-video supplementary stats from the no-key YouTube enrichers (keyed by
+-- youtube_id, NOT by channel, so a video shared across playlists is enriched
+-- once). Merged onto the cached catalog on read.
+--   ryd_*  Return YouTube Dislike  -> like/dislike estimates
+--   sb_*   SponsorBlock            -> crowd-sourced segments (in youtube_video_segments)
+-- status columns: NULL = pending, 'ok' | 'not_found' | 'error'.
+CREATE TABLE IF NOT EXISTS youtube_video_stats (
+    youtube_id    TEXT PRIMARY KEY,
+    like_count    INTEGER,
+    dislike_count INTEGER,
+    ryd_status    TEXT,
+    ryd_attempted TEXT,
+    sb_status     TEXT,
+    sb_attempted  TEXT
+);
+
+-- SponsorBlock crowd segments (sponsor/intro/outro/selfpromo/…) for a video.
+CREATE TABLE IF NOT EXISTS youtube_video_segments (
+    youtube_id TEXT NOT NULL,
+    category   TEXT NOT NULL,           -- sponsor | intro | outro | selfpromo | interaction | music_offtopic | preview | filler | poi_highlight | chapter
+    start_sec  REAL NOT NULL,
+    end_sec    REAL NOT NULL,
+    votes      INTEGER,
+    uuid       TEXT NOT NULL,
+    PRIMARY KEY (youtube_id, uuid)
+);
+CREATE INDEX IF NOT EXISTS idx_yvseg_video ON youtube_video_segments(youtube_id);
+
 -- ── Owned media files (the Library = content that has a file) ────────────────
 -- Exactly one owner FK is set (no polymorphic id). 1 row per physical file;
 -- usually 1:1 with its content, but the table allows history/extras.

tests/test_video_api.py

@@ -293,7 +293,9 @@ def test_enrichment_endpoints(tmp_path):
     client = app.test_client()
     try:
         svc = client.get("/api/video/enrichment/services").get_json()
-        assert {s["id"] for s in svc["services"]} == {"tmdb", "tvdb"}
+        ids = {s["id"] for s in svc["services"]}
+        assert {"tmdb", "tvdb"} <= ids                    # matcher workers
+        assert {"ryd", "sponsorblock", "fanart", "opensubtitles"} <= ids  # backfill workers
 
         mid = db.upsert_movie("plex", {"server_id": "m1", "title": "X"})
         st = client.get("/api/video/enrichment/tmdb/status").get_json()

tests/test_video_backfill.py

@@ -0,0 +1,189 @@
+"""Seam tests for the video BACKFILL workers (artwork / subtitles / no-key
+YouTube extras). The network fetch is stubbed so the loop/queue/record/status
+logic is tested without hitting fanart.tv / OpenSubtitles / RYD / SponsorBlock.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from database.video_database import VideoDatabase
+from core.video.enrichment.backfill import (
+    RydWorker, SponsorBlockWorker, FanartWorker, OpenSubtitlesWorker,
+    VideoBackfillWorker, _RateLimited, _Unauthorized, build_backfill_workers,
+)
+
+
+@pytest.fixture()
+def db(tmp_path):
+    return VideoDatabase(database_path=str(tmp_path / "video_library.db"))
+
+
+# ── DB seams ──────────────────────────────────────────────────────────────────
+def test_youtube_enrich_queue_apply_breakdown(db):
+    db.cache_channel_videos("UC1", [{"youtube_id": "a", "title": "A"},
+                                    {"youtube_id": "b", "title": "B"}])
+    assert db.youtube_enrich_breakdown("ryd_status")["video"]["pending"] == 2
+    nxt = db.youtube_enrich_next("ryd_status")
+    assert nxt["youtube_id"] in ("a", "b") and nxt["kind"] == "video"
+
+    db.apply_youtube_votes("a", 100, 5, "ok")
+    db.apply_youtube_votes("b", None, None, "not_found")
+    bd = db.youtube_enrich_breakdown("ryd_status")["video"]
+    assert bd == {"matched": 1, "not_found": 1, "errors": 0, "pending": 0}
+    # likes/dislikes merge onto the cached catalog on read
+    vids = {v["youtube_id"]: v for v in db.get_channel_videos("UC1")}
+    assert vids["a"]["like_count"] == 100 and vids["a"]["dislike_count"] == 5
+
+
+def test_youtube_enrich_shared_video_counted_once(db):
+    # Same video cached under two channels/playlists → one enrichment unit.
+    db.cache_channel_videos("UC1", [{"youtube_id": "x", "title": "X"}])
+    db.cache_channel_videos("PLfoo", [{"youtube_id": "x", "title": "X"}])
+    assert db.youtube_enrich_breakdown("sb_status")["video"]["pending"] == 1
+
+
+def test_youtube_segments_stored_and_read(db):
+    db.cache_channel_videos("UC1", [{"youtube_id": "v", "title": "V"}])
+    segs = [{"category": "sponsor", "start_sec": 10.0, "end_sec": 20.0, "votes": 3, "uuid": "u1"}]
+    db.apply_youtube_segments("v", segs, "ok")
+    got = db.youtube_video_segments("v")
+    assert got == [{"category": "sponsor", "start_sec": 10.0, "end_sec": 20.0, "votes": 3}]
+
+
+def test_backfill_next_mark_breakdown(db):
+    mid = db.upsert_movie("plex", {"server_id": "m1", "title": "Fight Club", "year": 1999})
+    with db.connect() as c:
+        c.execute("UPDATE movies SET tmdb_id=550, imdb_id='tt0137523' WHERE id=?", (mid,))
+        c.commit()
+    nxt = db.backfill_next("fanart")
+    assert nxt["kind"] == "movie" and nxt["tmdb_id"] == 550
+
+    db.backfill_mark("fanart", "movie", mid, "ok",
+                     columns={"logo_url": "http://x/l.png", "bogus_col": "nope"})
+    with db.connect() as c:
+        r = c.execute("SELECT logo_url, fanart_status FROM movies WHERE id=?", (mid,)).fetchone()
+    assert r["logo_url"] == "http://x/l.png" and r["fanart_status"] == "ok"   # whitelist drops bogus_col
+    assert db.backfill_breakdown("fanart")["movie"]["matched"] == 1
+
+
+def test_backfill_mark_never_clobbers(db):
+    mid = db.upsert_movie("plex", {"server_id": "m1", "title": "M"})
+    with db.connect() as c:
+        c.execute("UPDATE movies SET tmdb_id=1, logo_url='SERVER_LOGO' WHERE id=?", (mid,))
+        c.commit()
+    db.backfill_mark("fanart", "movie", mid, "ok", columns={"logo_url": "FANART_LOGO"})
+    with db.connect() as c:
+        r = c.execute("SELECT logo_url FROM movies WHERE id=?", (mid,)).fetchone()
+    assert r["logo_url"] == "SERVER_LOGO"   # gap-fill only
+
+
+def test_backfill_show_requires_tvdb_id(db):
+    with db.connect() as c:
+        c.execute("INSERT INTO shows (title, year) VALUES ('S', 2008)")
+        sid = c.execute("SELECT id FROM shows WHERE title='S'").fetchone()["id"]
+        c.commit()
+    # No tvdb_id → fanart has nothing to key on, so it's not queued.
+    assert db.backfill_next("fanart") is None
+    with db.connect() as c:
+        c.execute("UPDATE shows SET tvdb_id=81189 WHERE id=?", (sid,))
+        c.commit()
+    assert db.backfill_next("fanart")["kind"] == "show"
+
+
+# ── worker loop seams (stubbed fetch) ─────────────────────────────────────────
+def _seed_video(db):
+    db.cache_channel_videos("UC1", [{"youtube_id": "v", "title": "V"}])
+
+
+def test_ryd_worker_records_each_outcome(db):
+    _seed_video(db)
+    w = RydWorker(db)
+    w.fetch = lambda item: {"likes": 9, "dislikes": 2}
+    assert w.process_one() is True
+    assert db.youtube_enrich_breakdown("ryd_status")["video"]["matched"] == 1
+    assert w.stats["matched"] == 1
+
+
+def test_ryd_worker_empty_marks_not_found(db):
+    _seed_video(db)
+    w = RydWorker(db)
+    w.fetch = lambda item: None
+    w.process_one()
+    assert db.youtube_enrich_breakdown("ryd_status")["video"]["not_found"] == 1
+
+
+def test_worker_call_error_records_error_not_notfound(db):
+    _seed_video(db)
+    w = SponsorBlockWorker(db)
+
+    def boom(item):
+        raise RuntimeError("network")
+    w.fetch = boom
+    w.process_one()
+    bd = db.youtube_enrich_breakdown("sb_status")["video"]
+    assert bd["errors"] == 1 and bd["not_found"] == 0
+    assert w.stats["errors"] == 1
+
+
+def test_rate_limit_sets_cooldown_and_pauses_pending(db):
+    _seed_video(db)
+    w = SponsorBlockWorker(db)
+
+    def limited(item):
+        raise _RateLimited(30)
+    w.fetch = limited
+    assert w.process_one() is False           # not consumed; will retry
+    assert w._cooldown_until > 0
+    assert w.get_stats()["cooldown"] is True
+
+
+def test_unauthorized_pauses_worker(db):
+    mid = db.upsert_movie("plex", {"server_id": "m1", "title": "M"})
+    with db.connect() as c:
+        c.execute("UPDATE movies SET tmdb_id=1 WHERE id=?", (mid,))
+        c.commit()
+    db.set_setting("fanart_api_key", "KEY")
+    w = FanartWorker(db)
+
+    def denied(item):
+        raise _Unauthorized()
+    w.fetch = denied
+    w.process_one()
+    assert w.paused is True and "key" in (w.note or "").lower()
+
+
+def test_key_gated_workers_disabled_without_key(db):
+    assert FanartWorker(db).enabled is False
+    assert OpenSubtitlesWorker(db).enabled is False
+    db.set_setting("fanart_api_key", "KEY")
+    assert FanartWorker(db).enabled is True
+
+
+def test_nokey_workers_enabled_by_default_and_toggle(db):
+    assert RydWorker(db).enabled is True
+    db.set_setting("ryd_enabled", "0")
+    assert RydWorker(db).enabled is False
+
+
+def test_get_stats_shape_matches_matcher_worker(db):
+    _seed_video(db)
+    stats = RydWorker(db).get_stats()
+    assert set(stats) == {"enabled", "running", "paused", "idle", "current_item",
+                          "note", "cooldown", "stats", "progress", "breakdown"}
+    assert set(stats["stats"]) == {"matched", "not_found", "errors", "pending"}
+
+
+def test_build_backfill_workers_set(db):
+    assert set(build_backfill_workers(db)) == {"ryd", "sponsorblock", "fanart", "opensubtitles"}
+
+
+def test_backfill_module_imports_nothing_from_music():
+    path = Path(__file__).resolve().parent.parent / "core" / "video" / "enrichment" / "backfill.py"
+    for line in path.read_text(encoding="utf-8").splitlines():
+        s = line.strip()
+        if s.startswith("import ") or s.startswith("from "):
+            assert "music" not in s.lower(), f"music import leaked: {s!r}"

tests/test_video_enrichment.py

@@ -918,7 +918,11 @@ def test_worker_disabled_without_key(db):
 
 def test_engine_builds_and_lists_workers(db):
     eng = VideoEnrichmentEngine(db, {"tmdb": FakeClient(None), "tvdb": FakeClient(None)})
-    assert {s["id"] for s in eng.services()} == {"tmdb", "tvdb"}
+    ids = {s["id"] for s in eng.services()}
+    assert {"tmdb", "tvdb"} <= ids                         # the matcher workers
+    # The backfill workers (artwork / subtitles / no-key YouTube extras) are always
+    # registered alongside, so the manager/API can drive them too.
+    assert {"ryd", "sponsorblock", "fanart", "opensubtitles"} <= ids
     assert eng.worker("tmdb") is not None and eng.worker("nope") is None
 
 
@@ -953,7 +957,7 @@ def test_scan_pause_resumes_only_what_it_paused(db):
     assert eng.worker("tvdb").paused and not eng.worker("tmdb").paused
 
     paused = eng.pause_for_scan()
-    assert paused == {"tmdb"}                        # only the running one
+    assert "tmdb" in paused and "tvdb" not in paused  # only running ones (tvdb was user-paused)
     assert eng.worker("tmdb").paused and eng.worker("tvdb").paused
 
     eng.resume_after_scan()