diff --git a/core/video/enrichment/__init__.py b/core/video/enrichment/__init__.py new file mode 100644 index 00000000..962e0f91 --- /dev/null +++ b/core/video/enrichment/__init__.py @@ -0,0 +1,7 @@ +"""SoulSync — isolated VIDEO enrichment subsystem. + +Mirrors the music enrichment-worker pattern (per-source match status, a worker +loop, a registry) but operates entirely on video.db and never imports the music +enrichment code. The shared Manage-Workers modal / worker-orbs visuals are made +kind-aware separately; this package is the backend half. +""" diff --git a/core/video/enrichment/clients.py b/core/video/enrichment/clients.py new file mode 100644 index 00000000..4bbf3bec --- /dev/null +++ b/core/video/enrichment/clients.py @@ -0,0 +1,111 @@ +"""TMDB / TVDB match clients for the video enrichment workers. + +Thin adapters: ``.enabled`` (an API key is configured) and ``.match(kind, title, +year) -> {"id", "metadata"} | None``. These talk to real TMDB/TVDB APIs and are +validated against the live services; the worker LOGIC is unit-tested with a fake +client. Keys come from video_settings. +""" + +from __future__ import annotations + +from utils.logging_config import get_logger + +logger = get_logger("video_enrichment.clients") + + +def _int(val): + try: + return int(val) + except (TypeError, ValueError): + return None + + +class TMDBClient: + BASE = "https://api.themoviedb.org/3" + IMG = "https://image.tmdb.org/t/p/original" + + def __init__(self, api_key): + self.api_key = api_key or None + + @property + def enabled(self): + return bool(self.api_key) + + def match(self, kind, title, year): + if not self.api_key or not title: + return None + import requests + path = "/search/movie" if kind == "movie" else "/search/tv" + params = {"api_key": self.api_key, "query": title} + if year: + params["year" if kind == "movie" else "first_air_date_year"] = year + resp = requests.get(self.BASE + path, params=params, timeout=15) + results = (resp.json() or {}).get("results") or [] + if not results: + return None + top = results[0] + tmdb_id = top.get("id") + meta = {"overview": top.get("overview")} + try: + detail_path = "/movie/" if kind == "movie" else "/tv/" + dr = requests.get(self.BASE + detail_path + str(tmdb_id), + params={"api_key": self.api_key, "append_to_response": "external_ids"}, + timeout=15).json() or {} + meta["overview"] = dr.get("overview") or meta["overview"] + if dr.get("backdrop_path"): + meta["backdrop_url"] = self.IMG + dr["backdrop_path"] + ext = dr.get("external_ids") or {} + meta["imdb_id"] = ext.get("imdb_id") or dr.get("imdb_id") + if kind == "movie": + meta["release_date"] = dr.get("release_date") + else: + meta["status"] = dr.get("status") + meta["tvdb_id"] = _int(ext.get("tvdb_id")) + except Exception: + logger.exception("TMDB details fetch failed for %s", title) + return {"id": tmdb_id, "metadata": {k: v for k, v in meta.items() if v}} + + +class TVDBClient: + BASE = "https://api4.thetvdb.com/v4" + + def __init__(self, api_key): + self.api_key = api_key or None + self._token = None + + @property + def enabled(self): + return bool(self.api_key) + + def _auth(self): + if self._token: + return self._token + import requests + r = requests.post(self.BASE + "/login", json={"apikey": self.api_key}, timeout=15).json() or {} + self._token = (r.get("data") or {}).get("token") + return self._token + + def match(self, kind, title, year): + if kind != "show" or not self.api_key or not title: + return None + import requests + token = self._auth() + if not token: + return None + r = requests.get(self.BASE + "/search", headers={"Authorization": "Bearer " + token}, + params={"query": title, "type": "series"}, timeout=15).json() or {} + results = r.get("data") or [] + if not results: + return None + top = results[0] + tvdb_id = _int(top.get("tvdb_id") or top.get("id")) + meta = {"overview": top.get("overview")} + return {"id": tvdb_id, "metadata": {k: v for k, v in meta.items() if v}} + + +def build_clients(db) -> dict: + """Construct the source clients from the saved API keys (in video_settings).""" + return { + "tmdb": TMDBClient(db.get_setting("tmdb_api_key")), + "tvdb": TVDBClient(db.get_setting("tvdb_api_key")), + } diff --git a/core/video/enrichment/engine.py b/core/video/enrichment/engine.py new file mode 100644 index 00000000..9f03f9e9 --- /dev/null +++ b/core/video/enrichment/engine.py @@ -0,0 +1,60 @@ +"""Video enrichment engine — owns the per-source workers (registry). + +Parallels music's enrichment registry but is isolated to the video side. Built +lazily as a process-wide singleton; starts the workers (each idles until its API +key is configured). Imports only video.db + this package. +""" + +from __future__ import annotations + +import threading + +from utils.logging_config import get_logger + +from .worker import VideoEnrichmentWorker + +logger = get_logger("video_enrichment.engine") + +_DISPLAY = {"tmdb": "TMDB", "tvdb": "TVDB"} + + +class VideoEnrichmentEngine: + def __init__(self, db, clients: dict): + self.db = db + self.workers = { + service: VideoEnrichmentWorker(db, service, client, display_name=_DISPLAY.get(service)) + for service, client in clients.items() + } + + def start_all(self): + for w in self.workers.values(): + w.start() + + def stop_all(self): + for w in self.workers.values(): + w.stop() + + def worker(self, service): + return self.workers.get(service) + + def services(self) -> list: + return [{"id": s, "display_name": w.display_name} for s, w in self.workers.items()] + + +_engine = None +_lock = threading.Lock() + + +def get_video_enrichment_engine(): + """Process-wide engine, created (and started) on first use.""" + global _engine + if _engine is None: + with _lock: + if _engine is None: + from database.video_database import VideoDatabase + from .clients import build_clients + db = VideoDatabase() + eng = VideoEnrichmentEngine(db, build_clients(db)) + eng.start_all() + _engine = eng + return _engine diff --git a/core/video/enrichment/worker.py b/core/video/enrichment/worker.py new file mode 100644 index 00000000..406a195a --- /dev/null +++ b/core/video/enrichment/worker.py @@ -0,0 +1,125 @@ +"""Video enrichment worker — one per source (TMDB, TVDB). + +Mirrors the music worker: a daemon loop that pulls the next item needing +enrichment from video.db, asks its CLIENT to match it, and records the result. +The client is injected (a thin TMDB/TVDB adapter), so the worker's loop/queue/ +status logic is fully testable with a fake client. Isolated: imports only +video.db helpers; no music code. +""" + +from __future__ import annotations + +import threading + +from utils.logging_config import get_logger + +logger = get_logger("video_enrichment.worker") + + +class VideoEnrichmentWorker: + def __init__(self, db, service, client, display_name=None, interval=2.0, retry_days=30): + self.db = db + self.service = service + self.client = client + self.display_name = display_name or service.upper() + self.interval = interval + self.retry_days = retry_days + + 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} + + # ── 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): + self.paused = True + + def resume(self): + self.paused = False + + @property + def enabled(self): + return bool(getattr(self.client, "enabled", False)) + + # ── loop ────────────────────────────────────────────────────────────────── + def _run(self): + while not self.should_stop: + if self.paused or not self.enabled: + self._stop.wait(1.0) + continue + try: + did = self.process_one() + except Exception: + logger.exception("video enrichment %s loop error", self.service) + self.stats["errors"] += 1 + self._stop.wait(5.0) + continue + if did: + self._stop.wait(self.interval) # rate-limit between items + else: + self.current_item = None + self._stop.wait(10.0) # nothing to do — back off + + def process_one(self) -> bool: + """Process a single item. Returns True if one was processed.""" + item = self.db.enrichment_next(self.service, self.retry_days) + if not item: + return False + self.current_item = {"type": item["kind"], "name": item["title"]} + try: + result = self.client.match(item["kind"], item["title"], item.get("year")) + except Exception: + logger.exception("video enrichment %s match failed for %s", self.service, item["title"]) + self.stats["errors"] += 1 + self.db.enrichment_apply(self.service, item["kind"], item["id"], matched=False) + return True + if result and result.get("id"): + self.db.enrichment_apply(self.service, item["kind"], item["id"], matched=True, + external_id=result["id"], metadata=result.get("metadata")) + self.stats["matched"] += 1 + else: + self.db.enrichment_apply(self.service, item["kind"], item["id"], matched=False) + self.stats["not_found"] += 1 + return True + + # ── status (same shape the music enrichment API returns) ────────────────── + def get_stats(self) -> dict: + breakdown = self.db.enrichment_breakdown(self.service) + pending = sum(b["pending"] for b in breakdown.values()) + running = self.running and not self.paused and self.enabled + 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["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, + "idle": idle, + "current_item": self.current_item, + "stats": {**self.stats, "pending": pending}, + "progress": progress, + "breakdown": breakdown, + } diff --git a/tests/test_video_enrichment.py b/tests/test_video_enrichment.py new file mode 100644 index 00000000..46575266 --- /dev/null +++ b/tests/test_video_enrichment.py @@ -0,0 +1,105 @@ +"""Seam tests for the video enrichment workers (experimental branch). + +The worker's loop/queue/status logic is driven by a FAKE client so it's tested +without hitting TMDB/TVDB. Also guards that the enrichment package imports +nothing from the music side. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from database.video_database import VideoDatabase +from core.video.enrichment.worker import VideoEnrichmentWorker +from core.video.enrichment.engine import VideoEnrichmentEngine + + +class FakeClient: + enabled = True + + def __init__(self, result): + self._result = result + self.calls = [] + + def match(self, kind, title, year): + self.calls.append((kind, title, year)) + return self._result + + +@pytest.fixture() +def db(tmp_path): + return VideoDatabase(database_path=str(tmp_path / "video_library.db")) + + +def test_worker_process_one_matches(db): + db.upsert_movie("plex", {"server_id": "m1", "title": "Dune", "year": 2021}) + client = FakeClient({"id": 438631, "metadata": {"overview": "O", "backdrop_url": "/b.jpg"}}) + w = VideoEnrichmentWorker(db, "tmdb", client) + assert w.process_one() is True + assert client.calls == [("movie", "Dune", 2021)] + with db.connect() as c: + row = c.execute("SELECT tmdb_id, tmdb_match_status, overview FROM movies").fetchone() + assert (row["tmdb_id"], row["tmdb_match_status"], row["overview"]) == (438631, "matched", "O") + assert w.stats["matched"] == 1 + + +def test_worker_process_one_not_found(db): + db.upsert_movie("plex", {"server_id": "m1", "title": "X"}) + w = VideoEnrichmentWorker(db, "tmdb", FakeClient(None)) + assert w.process_one() is True + with db.connect() as c: + assert c.execute("SELECT tmdb_match_status FROM movies").fetchone()["tmdb_match_status"] == "not_found" + assert w.stats["not_found"] == 1 + + +def test_worker_process_one_no_items_returns_false(db): + assert VideoEnrichmentWorker(db, "tmdb", FakeClient(None)).process_one() is False + + +def test_worker_match_exception_marks_not_found_and_counts_error(db): + db.upsert_movie("plex", {"server_id": "m1", "title": "X"}) + + class Boom: + enabled = True + def match(self, *a): raise RuntimeError("api down") + + w = VideoEnrichmentWorker(db, "tmdb", Boom()) + assert w.process_one() is True # doesn't crash the loop + assert w.stats["errors"] == 1 + with db.connect() as c: + assert c.execute("SELECT tmdb_match_status FROM movies").fetchone()["tmdb_match_status"] == "not_found" + + +def test_worker_get_stats_shape(db): + db.upsert_movie("plex", {"server_id": "m1", "title": "X"}) + s = VideoEnrichmentWorker(db, "tmdb", FakeClient(None)).get_stats() + assert {"enabled", "running", "paused", "idle", "current_item", "stats", + "progress", "breakdown"} <= set(s) + assert s["enabled"] is True + assert s["stats"]["pending"] == 1 + + +def test_worker_disabled_without_key(db): + class NoKey: + enabled = False + + w = VideoEnrichmentWorker(db, "tmdb", NoKey()) + assert w.enabled is False + assert w.get_stats()["enabled"] is False + + +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"} + assert eng.worker("tmdb") is not None and eng.worker("nope") is None + + +def test_enrichment_package_imports_nothing_from_music(): + base = Path(__file__).resolve().parent.parent / "core" / "video" / "enrichment" + for py in base.glob("*.py"): + for line in py.read_text(encoding="utf-8").splitlines(): + s = line.strip() + if s.startswith("import ") or s.startswith("from "): + assert "music" not in s.lower(), f"{py.name}: music import leaked: {s!r}"