video enrichment 1b: worker engine + TMDB/TVDB clients

Isolated enrichment subsystem mirroring the music worker pattern, on video.db.
- VideoEnrichmentWorker: daemon loop pulling enrichment_next -> client.match ->
  enrichment_apply; pause/resume/stop; get_stats in the same shape the music
  enrichment API returns (enabled/running/paused/idle/current_item/stats/
  progress/breakdown). Client injected -> loop fully unit-tested with a fake.
- TMDB/TVDB clients (thin adapters, keys from video_settings): .enabled +
  match(kind,title,year) -> {id, metadata}; validated live, worker logic tested.
- VideoEnrichmentEngine registry + lazy get_video_enrichment_engine() singleton.
8 tests (match/not_found/error-resilience/stats/disabled/engine/isolation).

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.
+"""

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")),
+    }

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

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,
+        }

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}"