E5 S2: TMDB enrichment engine (Plex catalog)
Matches each mirrored Plex movie/show to TMDB by its stored canonical guid, fetches once (append_to_response), and PERSISTS the enrichment onto the row — served from the DB, never a live TMDB call at serve time. Additive: the Plex-sourced imdb_rating/rating are untouched; tmdb_rating is the distinct TMDb community score (ratings scope = TMDb + Plex's IMDb/audience, no OMDb/RT). - config.py: tmdb_api_key + api_base/language/interval/batch defaults. - sysconfig SPECS: tmdb_api_key (secret, Fernet) + tmdb_language, auto-rendered in the admin Config page; read like the Plex token. - migration 0071 (ADD-only nullable, reversible): plex_items + plex_shows gain tmdb_id / tmdb_enriched_at (watermark) / tmdb_rating / tmdb_vote_count / budget / revenue / keywords / crew / studios / production_countries / recommendations / backdrops; plex_people gains tmdb_person_id (the S2 foundation for S6 person bios). No watermark reset needed. - app/plex/tmdb.py: TmdbClient (mirrors PlexClient; throttled; 404->None, transient->TmdbError), parse_tmdb / apply_tmdb_detail (one-parser rule), guid_to_tmdb_id (tmdb: direct, imdb:/tvdb: via /find), enrich_batch (movies then shows, transient error leaves the watermark NULL to retry, definitive no-match still stamps so it isn't re-fetched), _link_people (upserts tmdb_person_id by name), serve_block (recommendations resolved to library rows by guid), run_enrich (no-op without a key). - scheduler: tmdb_enrich nightly job (default 120 min, no YouTube quota), runtime-tunable + "run now" from the Scheduler dashboard. - detail endpoints emit `tmdb` (movies + shows) so it's curl-verifiable now; the FE render lands in S5. Note: _link_people inserts plex_people rows for TMDB cast/crew; unreferenced crew (composers etc., not in cast_names/directors) get pruned by the orphan sweep — acceptable, the person page is cast-driven and cast rows survive. Gate: ruff + py_compile green; backend test lane FULL suite 387 passed (incl. migration drift downgrade->re-upgrade + test_plex_tmdb 9 tests). Frontend unchanged since S1. Not merged (epic branch, stacks on S1).
This commit is contained in:
@@ -0,0 +1,360 @@
|
||||
"""TMDB enrichment for the mirrored Plex catalog (E5 S2).
|
||||
|
||||
A thin synchronous TMDB v3 client plus the parse/persist of the enrichment fields onto plex_items /
|
||||
plex_shows, matched by the stored canonical `guid` ("tmdb:12345", or imdb/tvdb resolved via /find).
|
||||
Server-side ONLY (the api_key from sysconfig). Everything fetched is PERSISTED and served from the DB
|
||||
row — there is never a live TMDB call at serve time (the user's explicit rule). Additive: the existing
|
||||
Plex-sourced `imdb_rating` / `rating` are left untouched; `tmdb_rating` is the distinct TMDb score.
|
||||
|
||||
One-parser rule (mirrors app/plex/detail.py): `parse_tmdb` is the pure parser, `apply_tmdb_detail` writes
|
||||
the row. The enrich batch + scheduler entry live here too so the whole TMDB integration is one module.
|
||||
"""
|
||||
import logging
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import httpx
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.dialects.postgresql import insert as pg_insert
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app import sysconfig
|
||||
from app.config import settings
|
||||
from app.models import PlexItem, PlexPerson, PlexShow
|
||||
|
||||
log = logging.getLogger("siftlode.tmdb")
|
||||
|
||||
# Politeness throttle: TMDB tolerates ~50 req/s; ~33/s leaves comfortable headroom for a shared host.
|
||||
_MIN_INTERVAL = 0.03
|
||||
|
||||
# The crew jobs worth surfacing on an info page (TMDB lists dozens per title). Order-preserving pick.
|
||||
_CREW_JOBS = (
|
||||
"Director",
|
||||
"Writer",
|
||||
"Screenplay",
|
||||
"Story",
|
||||
"Producer",
|
||||
"Executive Producer",
|
||||
"Editor",
|
||||
"Director of Photography",
|
||||
"Original Music Composer",
|
||||
)
|
||||
|
||||
# append_to_response bundles the sub-resources into ONE request per title (keywords + credits +
|
||||
# recommendations + images + external ids). TV uses aggregate_credits (per-person aggregated jobs).
|
||||
_MOVIE_APPEND = "keywords,credits,recommendations,images,external_ids"
|
||||
_TV_APPEND = "keywords,aggregate_credits,recommendations,images,external_ids"
|
||||
|
||||
|
||||
class TmdbError(Exception):
|
||||
"""Any transient failure talking to TMDB (network, non-404 HTTP, bad payload). The caller leaves
|
||||
the row's `tmdb_enriched_at` NULL so the next run retries."""
|
||||
|
||||
|
||||
class TmdbNotConfigured(TmdbError):
|
||||
"""No TMDB API key is set — the module can't do anything."""
|
||||
|
||||
|
||||
class TmdbClient:
|
||||
"""Bound to the configured api_key + language. Use as a context manager so the httpx client closes."""
|
||||
|
||||
def __init__(self, db: Session):
|
||||
self.key = sysconfig.get_str(db, "tmdb_api_key")
|
||||
if not self.key:
|
||||
raise TmdbNotConfigured("TMDB API key is not configured")
|
||||
self.base = settings.tmdb_api_base.rstrip("/")
|
||||
self.language = sysconfig.get_str(db, "tmdb_language") or settings.tmdb_language
|
||||
self._http = httpx.Client(timeout=30.0)
|
||||
self._last = 0.0
|
||||
|
||||
def __enter__(self) -> "TmdbClient":
|
||||
return self
|
||||
|
||||
def __exit__(self, *exc) -> None:
|
||||
self._http.close()
|
||||
|
||||
def close(self) -> None:
|
||||
self._http.close()
|
||||
|
||||
def _throttle(self) -> None:
|
||||
wait = self._last + _MIN_INTERVAL - time.monotonic()
|
||||
if wait > 0:
|
||||
time.sleep(wait)
|
||||
self._last = time.monotonic()
|
||||
|
||||
def _get(self, path: str, params: dict | None = None) -> dict | None:
|
||||
"""GET a TMDB path. Returns the JSON dict, None on a 404 (a definitive "no such id"), and raises
|
||||
TmdbError on any transient failure (network / other HTTP status / non-JSON)."""
|
||||
self._throttle()
|
||||
p = {"api_key": self.key, "language": self.language, **(params or {})}
|
||||
try:
|
||||
r = self._http.get(f"{self.base}{path}", params=p)
|
||||
except httpx.HTTPError as e:
|
||||
raise TmdbError(str(e)) from e
|
||||
if r.status_code == 404:
|
||||
return None
|
||||
try:
|
||||
r.raise_for_status()
|
||||
except httpx.HTTPStatusError as e:
|
||||
raise TmdbError(str(e)) from e
|
||||
try:
|
||||
return r.json()
|
||||
except ValueError as e:
|
||||
raise TmdbError("TMDB returned a non-JSON response") from e
|
||||
|
||||
def fetch(self, tmdb_id: int, media_type: str) -> dict | None:
|
||||
"""Full enrichment payload for a movie or show (one request via append_to_response)."""
|
||||
if media_type == "tv":
|
||||
return self._get(f"/tv/{tmdb_id}", {"append_to_response": _TV_APPEND})
|
||||
return self._get(f"/movie/{tmdb_id}", {"append_to_response": _MOVIE_APPEND})
|
||||
|
||||
def find(self, external_id: str, source: str) -> dict | None:
|
||||
"""Resolve an external id (imdb_id / tvdb_id) to TMDB items via /find."""
|
||||
return self._get(f"/find/{external_id}", {"external_source": source})
|
||||
|
||||
|
||||
def guid_to_tmdb_id(guid: str | None, media_type: str, tmdb: TmdbClient) -> int | None:
|
||||
"""Map a stored canonical guid to a TMDB id. `tmdb:` is direct; `imdb:`/`tvdb:` go through /find
|
||||
(picking the movie_results or tv_results list per `media_type`). Returns None for an unmatchable
|
||||
guid — the caller then stamps the watermark so it isn't retried every run."""
|
||||
if not guid or ":" not in guid:
|
||||
return None
|
||||
provider, _, ext = guid.partition(":")
|
||||
if provider == "tmdb":
|
||||
try:
|
||||
return int(ext)
|
||||
except ValueError:
|
||||
return None
|
||||
source = {"imdb": "imdb_id", "tvdb": "tvdb_id"}.get(provider)
|
||||
if not source:
|
||||
return None
|
||||
res = tmdb.find(ext, source)
|
||||
if not res:
|
||||
return None
|
||||
items = res.get("tv_results" if media_type == "tv" else "movie_results") or []
|
||||
return items[0].get("id") if items else None
|
||||
|
||||
|
||||
def _dedup_crew(crew: list[dict]) -> list[dict]:
|
||||
seen: set[tuple] = set()
|
||||
out: list[dict] = []
|
||||
for c in crew:
|
||||
key = (c["name"], c["job"])
|
||||
if key in seen:
|
||||
continue
|
||||
seen.add(key)
|
||||
out.append(c)
|
||||
return out
|
||||
|
||||
|
||||
def parse_tmdb(payload: dict, media_type: str) -> dict:
|
||||
"""Pure parser: pull the enrichment fields out of a TMDB movie/tv payload (append_to_response form).
|
||||
Best-effort — any missing sub-resource is just omitted. Returns the row fields plus a `people` list
|
||||
([{name, tmdb_id}] over cast+crew) for the plex_people id-linking. TMDB image fields are stored as
|
||||
bare paths ("/abc.jpg"); the client builds the image.tmdb.org URL."""
|
||||
is_tv = media_type == "tv"
|
||||
votes = payload.get("vote_count") or 0
|
||||
|
||||
kw_container = payload.get("keywords") or {}
|
||||
kw_list = kw_container.get("results") if is_tv else kw_container.get("keywords")
|
||||
keywords = [k.get("name") for k in (kw_list or []) if k.get("name")][:20]
|
||||
|
||||
credits = payload.get("aggregate_credits" if is_tv else "credits") or {}
|
||||
crew: list[dict] = []
|
||||
for c in credits.get("crew") or []:
|
||||
# Movie crew entries carry `job`; TV aggregate_credits carry a `jobs` array per person.
|
||||
job = (c.get("jobs") or [{}])[0].get("job") if is_tv else c.get("job")
|
||||
if job in _CREW_JOBS and c.get("name"):
|
||||
crew.append(
|
||||
{"name": c["name"], "job": job, "tmdb_id": c.get("id"), "profile_path": c.get("profile_path")}
|
||||
)
|
||||
crew = _dedup_crew(crew)[:12]
|
||||
|
||||
# name -> tmdb person id, for the plex_people foundation (S6 fetches bios off these ids).
|
||||
people: list[dict] = []
|
||||
for c in (credits.get("cast") or [])[:30]:
|
||||
if c.get("name") and c.get("id"):
|
||||
people.append({"name": c["name"], "tmdb_id": c["id"]})
|
||||
for c in crew:
|
||||
if c.get("tmdb_id"):
|
||||
people.append({"name": c["name"], "tmdb_id": c["tmdb_id"]})
|
||||
|
||||
recs: list[dict] = []
|
||||
for r in ((payload.get("recommendations") or {}).get("results") or [])[:20]:
|
||||
recs.append(
|
||||
{
|
||||
"tmdb_id": r.get("id"),
|
||||
"title": r.get("title") or r.get("name"),
|
||||
"poster_path": r.get("poster_path"),
|
||||
"media_type": r.get("media_type") or media_type,
|
||||
}
|
||||
)
|
||||
|
||||
backdrops = [
|
||||
b.get("file_path")
|
||||
for b in ((payload.get("images") or {}).get("backdrops") or [])
|
||||
if b.get("file_path")
|
||||
][:10]
|
||||
|
||||
return {
|
||||
"tmdb_id": payload.get("id"),
|
||||
# vote_average is meaningless with no votes (TMDB returns 0.0) → treat as unknown.
|
||||
"tmdb_rating": payload.get("vote_average") if votes else None,
|
||||
"tmdb_vote_count": votes or None,
|
||||
# 0 is TMDB's "unknown" for money fields; TV has neither.
|
||||
"budget": (payload.get("budget") or None) if not is_tv else None,
|
||||
"revenue": (payload.get("revenue") or None) if not is_tv else None,
|
||||
"keywords": keywords or None,
|
||||
"crew": crew or None,
|
||||
"studios": [c.get("name") for c in (payload.get("production_companies") or []) if c.get("name")][:8]
|
||||
or None,
|
||||
"production_countries": [
|
||||
c.get("name") for c in (payload.get("production_countries") or []) if c.get("name")
|
||||
]
|
||||
or None,
|
||||
"recommendations": recs or None,
|
||||
"backdrops": backdrops or None,
|
||||
"people": people,
|
||||
}
|
||||
|
||||
|
||||
def apply_tmdb_detail(row, parsed: dict) -> None:
|
||||
"""Persist the parsed TMDB fields onto a movie/show row. Does NOT touch full_cast / imdb_rating /
|
||||
rating (Plex-owned). The caller stamps `tmdb_enriched_at` and links people separately."""
|
||||
row.tmdb_id = parsed["tmdb_id"]
|
||||
row.tmdb_rating = parsed["tmdb_rating"]
|
||||
row.tmdb_vote_count = parsed["tmdb_vote_count"]
|
||||
row.budget = parsed["budget"]
|
||||
row.revenue = parsed["revenue"]
|
||||
row.keywords = parsed["keywords"]
|
||||
row.crew = parsed["crew"]
|
||||
row.studios = parsed["studios"]
|
||||
row.production_countries = parsed["production_countries"]
|
||||
row.recommendations = parsed["recommendations"]
|
||||
row.backdrops = parsed["backdrops"]
|
||||
|
||||
|
||||
def _link_people(db: Session, links: dict[str, int]) -> None:
|
||||
"""Upsert plex_people by name, stamping `tmdb_person_id` — keeping any id already set (first wins),
|
||||
filling it where NULL, inserting a bare row where the person isn't cached yet. Deduped by name so a
|
||||
single statement can't hit the conflict target twice (the same rule as people.enrich_row_people)."""
|
||||
values = [{"name": nm, "tmdb_person_id": pid} for nm, pid in links.items() if nm and pid]
|
||||
if not values:
|
||||
return
|
||||
stmt = pg_insert(PlexPerson).values(values)
|
||||
db.execute(
|
||||
stmt.on_conflict_do_update(
|
||||
index_elements=[PlexPerson.name],
|
||||
set_={"tmdb_person_id": func.coalesce(PlexPerson.tmdb_person_id, stmt.excluded.tmdb_person_id)},
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
def enrich_batch(db: Session, tmdb: TmdbClient, limit: int) -> int:
|
||||
"""Enrich up to `limit` not-yet-TMDB-enriched titles (movies first, shows fill the remainder —
|
||||
mirrors _enrich_people_batch). A transient TmdbError leaves the watermark NULL for retry; a
|
||||
definitive no-match (unresolvable guid / TMDB 404) still stamps the watermark so it isn't re-fetched
|
||||
every run. One commit at the end. Returns the number processed (matched OR definitively unmatched)."""
|
||||
movies = (
|
||||
db.execute(
|
||||
select(PlexItem)
|
||||
.where(
|
||||
PlexItem.kind == "movie",
|
||||
PlexItem.guid.is_not(None),
|
||||
PlexItem.tmdb_enriched_at.is_(None),
|
||||
)
|
||||
.order_by(PlexItem.id)
|
||||
.limit(limit)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
shows: list[PlexShow] = []
|
||||
if len(movies) < limit:
|
||||
shows = (
|
||||
db.execute(
|
||||
select(PlexShow)
|
||||
.where(PlexShow.guid.is_not(None), PlexShow.tmdb_enriched_at.is_(None))
|
||||
.order_by(PlexShow.id)
|
||||
.limit(limit - len(movies))
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
|
||||
processed = 0
|
||||
links: dict[str, int] = {}
|
||||
for row, media_type in [(m, "movie") for m in movies] + [(s, "tv") for s in shows]:
|
||||
try:
|
||||
tmdb_id = guid_to_tmdb_id(row.guid, media_type, tmdb)
|
||||
payload = tmdb.fetch(tmdb_id, media_type) if tmdb_id else None
|
||||
except TmdbError as e:
|
||||
log.warning("tmdb enrich %s failed (will retry): %s", row.rating_key, e)
|
||||
continue # transient — leave tmdb_enriched_at NULL for the next run
|
||||
if payload:
|
||||
parsed = parse_tmdb(payload, media_type)
|
||||
apply_tmdb_detail(row, parsed)
|
||||
for p in parsed["people"]:
|
||||
links.setdefault(p["name"], p["tmdb_id"])
|
||||
# Stamp even on a definitive no-match so an unresolvable title isn't re-fetched every run.
|
||||
row.tmdb_enriched_at = datetime.now(timezone.utc)
|
||||
processed += 1
|
||||
|
||||
if links:
|
||||
_link_people(db, links)
|
||||
db.commit()
|
||||
return processed
|
||||
|
||||
|
||||
def run_enrich(db: Session) -> dict:
|
||||
"""Scheduler entry point. No-op (never an error) when the TMDB key isn't configured."""
|
||||
if not sysconfig.get_str(db, "tmdb_api_key"):
|
||||
return {"enriched": 0, "skipped": "tmdb_api_key not set"}
|
||||
with TmdbClient(db) as tmdb:
|
||||
n = enrich_batch(db, tmdb, settings.tmdb_enrich_batch)
|
||||
return {"enriched": n}
|
||||
|
||||
|
||||
def _resolve_recs(db: Session, recs: list[dict] | None) -> list[dict]:
|
||||
"""Tag each recommendation with whether it's in OUR library (matched by guid `tmdb:<id>`) and, if so,
|
||||
its rating_key — so the client can deep-link the ones we have and merely display the rest."""
|
||||
if not recs:
|
||||
return []
|
||||
guids = [f"tmdb:{r['tmdb_id']}" for r in recs if r.get("tmdb_id")]
|
||||
in_lib: dict[str, str] = {}
|
||||
if guids:
|
||||
for guid, rk in db.execute(
|
||||
select(PlexItem.guid, PlexItem.rating_key).where(
|
||||
PlexItem.kind == "movie", PlexItem.guid.in_(guids)
|
||||
)
|
||||
):
|
||||
in_lib[guid] = rk
|
||||
for guid, rk in db.execute(
|
||||
select(PlexShow.guid, PlexShow.rating_key).where(PlexShow.guid.in_(guids))
|
||||
):
|
||||
in_lib.setdefault(guid, rk)
|
||||
out: list[dict] = []
|
||||
for r in recs:
|
||||
rk = in_lib.get(f"tmdb:{r.get('tmdb_id')}")
|
||||
out.append({**r, "in_library": rk is not None, "rating_key": rk})
|
||||
return out
|
||||
|
||||
|
||||
def serve_block(db: Session, row) -> dict | None:
|
||||
"""The TMDB extras for a movie/show detail response, served from the row (no live TMDB call).
|
||||
None until the row has been TMDB-enriched (the client then shows nothing extra)."""
|
||||
if row.tmdb_enriched_at is None:
|
||||
return None
|
||||
return {
|
||||
"tmdb_id": row.tmdb_id,
|
||||
"rating": row.tmdb_rating,
|
||||
"vote_count": row.tmdb_vote_count,
|
||||
"budget": row.budget,
|
||||
"revenue": row.revenue,
|
||||
"keywords": row.keywords or [],
|
||||
"crew": row.crew or [],
|
||||
"studios": row.studios or [],
|
||||
"production_countries": row.production_countries or [],
|
||||
"recommendations": _resolve_recs(db, row.recommendations),
|
||||
"backdrops": row.backdrops or [],
|
||||
}
|
||||
Reference in New Issue
Block a user