From 093e14bd5dc261dca83c1eee44f343df5ffff4eb Mon Sep 17 00:00:00 2001 From: BoulderBadgeDad Date: Sun, 14 Jun 2026 11:18:46 -0700 Subject: [PATCH] video enrichment 1a: DB layer (match-status cols + migration + helpers) Foundation for the video enrichment workers, mirroring music's per-source columns/queries on video.db. - Schema: tmdb_match_status/tmdb_last_attempted on movies; tmdb_+tvdb_ on shows. Idempotent ALTER-TABLE migration adds them to existing DBs on init. - VideoDatabase helpers (service+kind -> columns map): enrichment_next (pending first, then not_found past retry window), enrichment_apply (sets id/status/last_attempted + whitelisted metadata, backfill-safe), enrichment_breakdown, enrichment_unmatched (paged), and enrichment_retry. Same shape as music's enrichment API so the shared modal can drive it. 30 DB tests green. --- database/video_database.py | 160 +++++++++++++++++++++++++++++++++++ database/video_schema.sql | 6 ++ tests/test_video_database.py | 55 ++++++++++++ 3 files changed, 221 insertions(+) diff --git a/database/video_database.py b/database/video_database.py index 20723ed7..abe47caf 100644 --- a/database/video_database.py +++ b/database/video_database.py @@ -20,6 +20,7 @@ import os import re import sqlite3 import threading +from datetime import datetime, timedelta, timezone from pathlib import Path from utils.logging_config import get_logger @@ -46,6 +47,38 @@ def _sort_title(title) -> str: return _ARTICLE_RE.sub("", (title or "").strip()).lower() +# Enrichment plumbing (parallels music's per-source columns). Maps a service + +# content kind to (table, id_col, match_status_col, last_attempted_col). +_ENRICH = { + "tmdb": { + "movie": ("movies", "tmdb_id", "tmdb_match_status", "tmdb_last_attempted"), + "show": ("shows", "tmdb_id", "tmdb_match_status", "tmdb_last_attempted"), + }, + "tvdb": { + "show": ("shows", "tvdb_id", "tvdb_match_status", "tvdb_last_attempted"), + }, +} + +# Whitelist of metadata columns enrichment may write per table (guards against +# arbitrary keys; backfill semantics applied by the caller). +_ENRICH_META_COLS = { + "movies": {"overview", "backdrop_url", "release_date", "status", "content_rating", + "runtime_minutes", "studio", "imdb_id", "tmdb_id"}, + "shows": {"overview", "backdrop_url", "status", "network", "content_rating", + "imdb_id", "tmdb_id", "tvdb_id"}, +} + +# Columns ensured on existing DBs (ALTER TABLE ADD COLUMN; idempotent). +_COLUMN_MIGRATIONS = [ + ("movies", "tmdb_match_status", "TEXT"), + ("movies", "tmdb_last_attempted", "TEXT"), + ("shows", "tmdb_match_status", "TEXT"), + ("shows", "tmdb_last_attempted", "TEXT"), + ("shows", "tvdb_match_status", "TEXT"), + ("shows", "tvdb_last_attempted", "TEXT"), +] + + class VideoDatabase: """Connection + schema manager for the isolated video library DB.""" @@ -90,6 +123,7 @@ class VideoDatabase: conn = self._get_connection() try: conn.executescript(schema) + self._ensure_columns(conn) conn.execute(f"PRAGMA user_version = {int(SCHEMA_VERSION)}") conn.commit() logger.info( @@ -103,6 +137,132 @@ class VideoDatabase: finally: conn.close() + @staticmethod + def _ensure_columns(conn) -> None: + """Add any new columns to an existing DB (idempotent ALTER TABLE).""" + for table, col, coltype in _COLUMN_MIGRATIONS: + cols = {r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()} + if col not in cols: + conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {coltype}") + + # ── enrichment plumbing (per-source match status, like music) ───────────── + def enrichment_next(self, service: str, retry_days: int = 30) -> dict | None: + """Next item that needs enrichment for a service: pending (never tried) + first, then a not_found item older than retry_days. Returns + {kind, id, title, year} or None.""" + kinds = _ENRICH.get(service) + if not kinds: + return None + cutoff = (datetime.now(timezone.utc) - timedelta(days=retry_days)).strftime("%Y-%m-%d %H:%M:%S") + conn = self._get_connection() + try: + for kind, (tbl, _idc, sc, _ac) in kinds.items(): + row = conn.execute( + f"SELECT id, title, year FROM {tbl} WHERE {sc} IS NULL ORDER BY id LIMIT 1").fetchone() + if row: + return {"kind": kind, "id": row["id"], "title": row["title"], "year": row["year"]} + for kind, (tbl, _idc, sc, ac) in kinds.items(): + row = conn.execute( + f"SELECT id, title, year FROM {tbl} WHERE {sc}='not_found' " + f"AND ({ac} IS NULL OR {ac} < ?) ORDER BY {ac} LIMIT 1", (cutoff,)).fetchone() + if row: + return {"kind": kind, "id": row["id"], "title": row["title"], "year": row["year"]} + return None + finally: + conn.close() + + def enrichment_apply(self, service: str, kind: str, item_id: int, matched: bool, + external_id=None, metadata: dict | None = None) -> None: + """Record a match result: set match_status + last_attempted, the external + id (when matched), and any whitelisted metadata columns.""" + spec = _ENRICH.get(service, {}).get(kind) + if not spec: + return + tbl, idc, sc, ac = spec + sets = [f"{sc}=?", f"{ac}=CURRENT_TIMESTAMP"] + params = ["matched" if matched else "not_found"] + if matched and external_id is not None: + sets.append(f"{idc}=?") + params.append(external_id) + allowed = _ENRICH_META_COLS.get(tbl, set()) + for col, val in (metadata or {}).items(): + if val is not None and col in allowed: + sets.append(f"{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 enrichment_breakdown(self, service: str) -> dict: + kinds = _ENRICH.get(service, {}) + out = {} + conn = self._get_connection() + try: + for kind, (tbl, _idc, sc, _ac) in kinds.items(): + out[kind] = { + "matched": conn.execute(f"SELECT COUNT(*) FROM {tbl} WHERE {sc}='matched'").fetchone()[0], + "not_found": conn.execute(f"SELECT COUNT(*) FROM {tbl} WHERE {sc}='not_found'").fetchone()[0], + "pending": conn.execute(f"SELECT COUNT(*) FROM {tbl} WHERE {sc} IS NULL").fetchone()[0], + } + return out + finally: + conn.close() + + def enrichment_unmatched(self, service: str, kind: str, status: str = "not_found", + search=None, limit: int = 50, offset: int = 0) -> dict: + spec = _ENRICH.get(service, {}).get(kind) + if not spec: + return {"items": [], "total": 0} + tbl, _idc, sc, ac = spec + where, params = [], [] + if status == "pending": + where.append(f"{sc} IS NULL") + elif status == "unmatched": + where.append(f"({sc} IS NULL OR {sc}='not_found')") + else: + where.append(f"{sc}='not_found'") + if search: + where.append("title LIKE ? COLLATE NOCASE") + params.append("%" + search + "%") + where_sql = " WHERE " + " AND ".join(where) + conn = self._get_connection() + try: + total = conn.execute(f"SELECT COUNT(*) FROM {tbl}{where_sql}", params).fetchone()[0] + rows = conn.execute( + f"SELECT id, title, year, {ac} AS last_attempted, " + f"(poster_url IS NOT NULL AND poster_url<>'') AS has_poster " + f"FROM {tbl}{where_sql} ORDER BY COALESCE(sort_title, title) COLLATE NOCASE " + f"LIMIT ? OFFSET ?", params + [limit, offset]).fetchall() + items = [] + for r in rows: + d = dict(r) + d["has_poster"] = bool(d.get("has_poster")) + items.append(d) + return {"items": items, "total": total} + finally: + conn.close() + + def enrichment_retry(self, service: str, kind: str, scope: str = "failed", item_id=None) -> int: + """Re-queue items by resetting status/last_attempted to NULL.""" + spec = _ENRICH.get(service, {}).get(kind) + if not spec: + return 0 + tbl, _idc, sc, ac = 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}='not_found'") + conn.commit() + return cur.rowcount + finally: + conn.close() + @property def schema_version(self) -> int: conn = self._get_connection() diff --git a/database/video_schema.sql b/database/video_schema.sql index 5533cd87..df38c446 100644 --- a/database/video_schema.sql +++ b/database/video_schema.sql @@ -65,6 +65,8 @@ CREATE TABLE IF NOT EXISTS movies ( server_id TEXT, -- media server native id (Plex ratingKey / Jellyfin Item Id) tmdb_id INTEGER, -- not unique: same film can sit in >1 library imdb_id TEXT, + tmdb_match_status TEXT, -- enrichment: NULL=pending | matched | not_found | error + tmdb_last_attempted TEXT, title TEXT NOT NULL, sort_title TEXT, year INTEGER, @@ -100,6 +102,10 @@ CREATE TABLE IF NOT EXISTS shows ( tvdb_id INTEGER, -- not unique (same series can sit in >1 library) tmdb_id INTEGER, imdb_id TEXT, + tmdb_match_status TEXT, -- enrichment match state per source + tmdb_last_attempted TEXT, + tvdb_match_status TEXT, + tvdb_last_attempted TEXT, title TEXT NOT NULL, sort_title TEXT, year INTEGER, diff --git a/tests/test_video_database.py b/tests/test_video_database.py index f2eb9f96..5fef251d 100644 --- a/tests/test_video_database.py +++ b/tests/test_video_database.py @@ -322,6 +322,61 @@ def test_query_library_shows_status_and_counts(db): assert (owned["episode_count"], owned["owned_count"]) == (1, 1) +# ── enrichment plumbing ─────────────────────────────────────────────────────── + +def test_enrichment_columns_present(db): + with db.connect() as c: + mcols = {r[1] for r in c.execute("PRAGMA table_info(movies)").fetchall()} + scols = {r[1] for r in c.execute("PRAGMA table_info(shows)").fetchall()} + assert {"tmdb_match_status", "tmdb_last_attempted"} <= mcols + assert {"tmdb_match_status", "tvdb_match_status", "tvdb_last_attempted"} <= scols + + +def test_ensure_columns_is_idempotent(db): + # Running the migration again on an already-migrated DB must not error. + with db.connect() as c: + db._ensure_columns(c) + c.commit() + assert db.enrichment_breakdown("tmdb")["movie"]["pending"] == 0 + + +def test_enrichment_next_pending_then_none_when_fresh(db): + db.upsert_movie("plex", {"server_id": "m1", "title": "A"}) + db.upsert_movie("plex", {"server_id": "m2", "title": "B", "tmdb_id": 5}) + nxt = db.enrichment_next("tmdb") + assert nxt and nxt["kind"] == "movie" + db.enrichment_apply("tmdb", "movie", nxt["id"], matched=True, external_id=1) + nxt2 = db.enrichment_next("tmdb") + assert nxt2 and nxt2["id"] != nxt["id"] + db.enrichment_apply("tmdb", "movie", nxt2["id"], matched=False) + # both attempted; not_found is fresh (<30d) so nothing is due + assert db.enrichment_next("tmdb") is None + + +def test_enrichment_apply_matched_sets_id_status_and_metadata(db): + mid = db.upsert_movie("plex", {"server_id": "m1", "title": "A"}) + db.enrichment_apply("tmdb", "movie", mid, matched=True, external_id=27205, + metadata={"overview": "O", "backdrop_url": "/b.jpg", "imdb_id": "tt1", + "bogus_col": "x"}) + with db.connect() as c: + row = c.execute("SELECT tmdb_id, tmdb_match_status, overview, backdrop_url, imdb_id " + "FROM movies WHERE id=?", (mid,)).fetchone() + assert (row["tmdb_id"], row["tmdb_match_status"]) == (27205, "matched") + assert (row["overview"], row["backdrop_url"], row["imdb_id"]) == ("O", "/b.jpg", "tt1") + + +def test_enrichment_breakdown_unmatched_retry(db): + a = db.upsert_movie("plex", {"server_id": "m1", "title": "A"}) + b = db.upsert_movie("plex", {"server_id": "m2", "title": "B"}) + db.enrichment_apply("tmdb", "movie", a, matched=True, external_id=1) + db.enrichment_apply("tmdb", "movie", b, matched=False) + assert db.enrichment_breakdown("tmdb")["movie"] == {"matched": 1, "not_found": 1, "pending": 0} + un = db.enrichment_unmatched("tmdb", "movie", status="not_found") + assert [i["title"] for i in un["items"]] == ["B"] and un["total"] == 1 + assert db.enrichment_retry("tmdb", "movie", scope="failed") == 1 + assert db.enrichment_breakdown("tmdb")["movie"]["pending"] == 1 + + # ── isolation: the video DB imports nothing from music ─────────────────────── def test_video_db_module_imports_nothing_from_music():