From 278ba47519a6675a379519f3b4d61eb20eecc03f Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Wed, 17 Jun 2026 23:26:46 -0700 Subject: [PATCH] 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 | 493 ++++++++++++++++++++++++++++++ core/video/enrichment/engine.py | 7 + database/video_database.py | 238 ++++++++++++++- database/video_schema.sql | 28 ++ tests/test_video_api.py | 4 +- tests/test_video_backfill.py | 189 ++++++++++++ tests/test_video_enrichment.py | 8 +- 7 files changed, 962 insertions(+), 5 deletions(-) create mode 100644 core/video/enrichment/backfill.py create mode 100644 tests/test_video_backfill.py diff --git a/core/video/enrichment/backfill.py b/core/video/enrichment/backfill.py new file mode 100644 index 00000000..774385cf --- /dev/null +++ b/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), + )} diff --git a/core/video/enrichment/engine.py b/core/video/enrichment/engine.py index c315b76e..bb1d771a 100644 --- a/core/video/enrichment/engine.py +++ b/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 diff --git a/database/video_database.py b/database/video_database.py index dd4f1ad4..b81d2b15 100644 --- a/database/video_database.py +++ b/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() diff --git a/database/video_schema.sql b/database/video_schema.sql index 172c6e9c..3e9e06de 100644 --- a/database/video_schema.sql +++ b/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. diff --git a/tests/test_video_api.py b/tests/test_video_api.py index 3e9e1d38..5ff8736a 100644 --- a/tests/test_video_api.py +++ b/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() diff --git a/tests/test_video_backfill.py b/tests/test_video_backfill.py new file mode 100644 index 00000000..748cc720 --- /dev/null +++ b/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}" diff --git a/tests/test_video_enrichment.py b/tests/test_video_enrichment.py index 199e5dcc..1c27a13e 100644 --- a/tests/test_video_enrichment.py +++ b/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()