peter
533571f572
PlexLibrary, PlexShow, PlexSeason, PlexItem, PlexPerson, PlexCollection, PlexState, PlexLink, PlexPlaylist and PlexPlaylistItem leave app/models.py for app/plex/models.py; app.tables registers them. Every import is re-pointed; no schema change.
499 lines
19 KiB
Python
499 lines
19 KiB
Python
"""TMDB enrichment engine: the pure parser + persister, guid→tmdb-id resolution, and the
|
|
DB-lane batch/serve integration (enrich stamps the watermark + links people; serve_block resolves
|
|
recommendations to library rows; the whole thing no-ops without an API key).
|
|
|
|
The DB tests use the `db` fixture (real Postgres) — skipped when none is reachable."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import httpx
|
|
import pytest
|
|
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexPerson, PlexShow
|
|
from app.plex import tmdb
|
|
|
|
# A TMDB /movie/{id}?append_to_response=keywords,credits,recommendations,images,external_ids payload.
|
|
MOVIE_PAYLOAD = {
|
|
"id": 238,
|
|
"vote_average": 8.7,
|
|
"vote_count": 20000,
|
|
"budget": 6000000,
|
|
"revenue": 245066411,
|
|
"keywords": {"keywords": [{"id": 1, "name": "mafia"}, {"id": 2, "name": "crime family"}]},
|
|
"credits": {
|
|
"cast": [
|
|
{
|
|
"id": 3084,
|
|
"name": "Marlon Brando",
|
|
"character": "Vito",
|
|
"profile_path": "/a.jpg",
|
|
"order": 0,
|
|
},
|
|
{
|
|
"id": 1158,
|
|
"name": "Al Pacino",
|
|
"character": "Michael",
|
|
"profile_path": "/b.jpg",
|
|
"order": 1,
|
|
},
|
|
],
|
|
"crew": [
|
|
{
|
|
"id": 1776,
|
|
"name": "Francis Ford Coppola",
|
|
"job": "Director",
|
|
"profile_path": "/c.jpg",
|
|
},
|
|
{"id": 1776, "name": "Francis Ford Coppola", "job": "Writer", "profile_path": "/c.jpg"},
|
|
{"id": 9, "name": "Some Gaffer", "job": "Gaffer"}, # not a key job → dropped
|
|
],
|
|
},
|
|
"production_companies": [{"name": "Paramount"}, {"name": "Alfran"}],
|
|
"production_countries": [{"iso_3166_1": "US", "name": "United States of America"}],
|
|
"recommendations": {
|
|
"results": [
|
|
{
|
|
"id": 240,
|
|
"title": "The Godfather Part II",
|
|
"poster_path": "/p.jpg",
|
|
"media_type": "movie",
|
|
},
|
|
{"id": 999, "title": "Not In Library", "poster_path": "/q.jpg", "media_type": "movie"},
|
|
]
|
|
},
|
|
"images": {"backdrops": [{"file_path": "/bd1.jpg"}, {"file_path": "/bd2.jpg"}]},
|
|
"external_ids": {"imdb_id": "tt0068646"},
|
|
}
|
|
|
|
# A TMDB /tv payload: keywords under `results`, aggregate_credits with per-person `jobs`, no budget.
|
|
TV_PAYLOAD = {
|
|
"id": 1396,
|
|
"vote_average": 9.0,
|
|
"vote_count": 12000,
|
|
"created_by": [{"id": 66633, "name": "Vince Gilligan", "profile_path": "/vg.jpg"}],
|
|
"keywords": {"results": [{"name": "drugs"}, {"name": "new mexico"}]},
|
|
"aggregate_credits": {
|
|
"cast": [{"id": 17419, "name": "Bryan Cranston"}],
|
|
"crew": [
|
|
{"id": 66633, "name": "Vince Gilligan", "jobs": [{"job": "Executive Producer"}]},
|
|
{"id": 5, "name": "Someone", "jobs": [{"job": "Sound"}]}, # dropped
|
|
],
|
|
},
|
|
"production_companies": [{"name": "Sony Pictures Television"}],
|
|
"production_countries": [{"name": "United States of America"}],
|
|
"recommendations": {"results": []},
|
|
"images": {"backdrops": []},
|
|
}
|
|
|
|
|
|
class FakeTmdb:
|
|
"""Stand-in for TmdbClient in the parse/batch tests — no network."""
|
|
|
|
def __init__(self, payloads=None, find_map=None):
|
|
self.payloads = payloads or {} # (media_type, tmdb_id) -> payload
|
|
self.find_map = find_map or {} # external_id -> /find result
|
|
|
|
def fetch(self, tmdb_id, media_type):
|
|
return self.payloads.get((media_type, tmdb_id))
|
|
|
|
def find(self, external_id, source):
|
|
return self.find_map.get(external_id)
|
|
|
|
|
|
# ---- pure parse/persist --------------------------------------------------------------------------
|
|
|
|
|
|
def test_parse_movie_pulls_all_fields():
|
|
d = tmdb.parse_tmdb(MOVIE_PAYLOAD, "movie")
|
|
assert d["tmdb_id"] == 238
|
|
assert d["tmdb_rating"] == 8.7
|
|
assert d["tmdb_vote_count"] == 20000
|
|
assert d["budget"] == 6000000
|
|
assert d["revenue"] == 245066411
|
|
assert d["keywords"] == ["mafia", "crime family"]
|
|
assert d["crew"] == [
|
|
{
|
|
"name": "Francis Ford Coppola",
|
|
"job": "Director",
|
|
"tmdb_id": 1776,
|
|
"profile_path": "/c.jpg",
|
|
},
|
|
{
|
|
"name": "Francis Ford Coppola",
|
|
"job": "Writer",
|
|
"tmdb_id": 1776,
|
|
"profile_path": "/c.jpg",
|
|
},
|
|
] # the Gaffer is filtered; the two Coppola jobs survive (dedup is by name+job)
|
|
assert d["studios"] == ["Paramount", "Alfran"]
|
|
# The code beside the English name, so the client can name the country in the reader's language.
|
|
assert d["production_countries"] == [{"code": "US", "name": "United States of America"}]
|
|
assert [r["tmdb_id"] for r in d["recommendations"]] == [240, 999]
|
|
assert d["backdrops"] == ["/bd1.jpg", "/bd2.jpg"]
|
|
# cast + crew names → tmdb person ids for the plex_people foundation
|
|
links = {(p["name"], p["tmdb_id"]) for p in d["people"]}
|
|
assert ("Marlon Brando", 3084) in links
|
|
assert ("Al Pacino", 1158) in links
|
|
assert ("Francis Ford Coppola", 1776) in links
|
|
|
|
|
|
def test_parse_tv_uses_aggregate_credits_and_no_money():
|
|
d = tmdb.parse_tmdb(TV_PAYLOAD, "tv")
|
|
assert d["budget"] is None and d["revenue"] is None # TV has neither
|
|
assert d["keywords"] == ["drugs", "new mexico"] # from keywords.results
|
|
# The showrunner (created_by → Creator) leads; the aggregate-credits EP follows; the Sound person is
|
|
# filtered; TV jobs are read from jobs[0]. Crew is display-priority ordered (Creator before EP).
|
|
assert d["crew"] == [
|
|
{"name": "Vince Gilligan", "job": "Creator", "tmdb_id": 66633, "profile_path": "/vg.jpg"},
|
|
{
|
|
"name": "Vince Gilligan",
|
|
"job": "Executive Producer",
|
|
"tmdb_id": 66633,
|
|
"profile_path": None,
|
|
},
|
|
]
|
|
assert d["tmdb_rating"] == 9.0
|
|
assert ("Bryan Cranston", 17419) in {(p["name"], p["tmdb_id"]) for p in d["people"]}
|
|
|
|
|
|
def test_parse_zero_votes_and_money_are_unknown():
|
|
d = tmdb.parse_tmdb(
|
|
{"id": 1, "vote_average": 0.0, "vote_count": 0, "budget": 0, "revenue": 0}, "movie"
|
|
)
|
|
assert d["tmdb_rating"] is None # 0.0 with no votes is meaningless
|
|
assert d["tmdb_vote_count"] is None
|
|
assert d["budget"] is None and d["revenue"] is None
|
|
|
|
|
|
def test_apply_writes_row_fields_without_touching_plex_owned():
|
|
it = PlexItem(
|
|
rating_key="m1", kind="movie", title="The Godfather", imdb_rating=8.7, full_cast=[{"x": 1}]
|
|
)
|
|
tmdb.apply_tmdb_detail(it, tmdb.parse_tmdb(MOVIE_PAYLOAD, "movie"))
|
|
assert it.tmdb_id == 238
|
|
assert it.budget == 6000000
|
|
assert it.keywords == ["mafia", "crime family"]
|
|
assert it.imdb_rating == 8.7 and it.full_cast == [{"x": 1}] # Plex-owned, untouched
|
|
|
|
|
|
def test_guid_to_tmdb_id_direct_find_and_bad():
|
|
fake = FakeTmdb(find_map={"tt0068646": {"movie_results": [{"id": 238}], "tv_results": []}})
|
|
assert tmdb.guid_to_tmdb_id("tmdb:238", "movie", fake) == 238
|
|
assert tmdb.guid_to_tmdb_id("imdb:tt0068646", "movie", fake) == 238 # resolved via /find
|
|
assert tmdb.guid_to_tmdb_id("imdb:tt0000000", "movie", fake) is None # not in find_map
|
|
assert tmdb.guid_to_tmdb_id("tmdb:notanumber", "movie", fake) is None
|
|
assert tmdb.guid_to_tmdb_id(None, "movie", fake) is None
|
|
assert tmdb.guid_to_tmdb_id("local://abc", "movie", fake) is None
|
|
|
|
|
|
def test_guid_to_tmdb_id_tvdb_source_and_tv_media_type():
|
|
# A tvdb guid for a TV item goes through /find and must pick the tv_results list (not movie_results),
|
|
# and a tvdb source maps to the tvdb_id external_source. The media_type branch is the whole point.
|
|
fake = FakeTmdb(find_map={"81189": {"movie_results": [], "tv_results": [{"id": 1396}]}})
|
|
assert tmdb.guid_to_tmdb_id("tvdb:81189", "tv", fake) == 1396
|
|
# The SAME /find result, asked for a movie, reads the (empty) movie_results → None.
|
|
assert tmdb.guid_to_tmdb_id("tvdb:81189", "movie", fake) is None
|
|
# A direct tmdb guid for a tv item is just the int (no /find).
|
|
assert tmdb.guid_to_tmdb_id("tmdb:1396", "tv", fake) == 1396
|
|
# An unknown provider (only imdb/tvdb are /find sources) is unresolvable.
|
|
assert tmdb.guid_to_tmdb_id("plex:abc", "tv", fake) is None
|
|
|
|
|
|
# ---- DB-lane: batch enrich + serve --------------------------------------------------------------
|
|
|
|
|
|
def _lib(db, key="1", kind="movie"):
|
|
lib = PlexLibrary(plex_key=key, title="L", kind=kind, enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
return lib
|
|
|
|
|
|
def test_enrich_batch_persists_stamps_and_links_people(db):
|
|
lib = _lib(db)
|
|
mv = PlexItem(
|
|
rating_key="m1", library_id=lib.id, kind="movie", title="The Godfather", guid="tmdb:238"
|
|
)
|
|
unmatched = PlexItem(
|
|
rating_key="m2", library_id=lib.id, kind="movie", title="Nope", guid="tmdb:777"
|
|
)
|
|
db.add_all([mv, unmatched])
|
|
db.commit()
|
|
|
|
fake = FakeTmdb(
|
|
payloads={("movie", 238): MOVIE_PAYLOAD}
|
|
) # 777 has no payload → definitive no-match
|
|
n = tmdb.enrich_batch(db, fake, 50)
|
|
assert n == 2 # both processed (one matched, one definitively unmatched)
|
|
|
|
db.refresh(mv)
|
|
assert mv.tmdb_id == 238
|
|
assert mv.keywords == ["mafia", "crime family"]
|
|
assert mv.revenue == 245066411
|
|
assert mv.tmdb_enriched_at is not None
|
|
|
|
db.refresh(unmatched)
|
|
assert unmatched.tmdb_enriched_at is not None # stamped so it isn't re-fetched every run
|
|
assert unmatched.tmdb_id is None # but nothing stored
|
|
|
|
brando = (
|
|
db.execute(PlexPerson.__table__.select().where(PlexPerson.name == "Marlon Brando"))
|
|
.mappings()
|
|
.first()
|
|
)
|
|
assert brando is not None and brando["tmdb_person_id"] == 3084
|
|
|
|
|
|
def test_serve_block_none_until_enriched_and_resolves_recs(db):
|
|
lib = _lib(db)
|
|
# A recommendation target that IS in our library (guid tmdb:240) → deep-linkable.
|
|
sequel = PlexItem(
|
|
rating_key="m240", library_id=lib.id, kind="movie", title="Part II", guid="tmdb:240"
|
|
)
|
|
fresh = PlexItem(
|
|
rating_key="m1", library_id=lib.id, kind="movie", title="The Godfather", guid="tmdb:238"
|
|
)
|
|
db.add_all([sequel, fresh])
|
|
db.commit()
|
|
|
|
assert tmdb.serve_block(db, fresh, lang="en") is None # not enriched yet
|
|
|
|
tmdb.apply_tmdb_detail(fresh, tmdb.parse_tmdb(MOVIE_PAYLOAD, "movie"))
|
|
fresh.tmdb_enriched_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
|
|
block = tmdb.serve_block(db, fresh, lang="en")
|
|
assert block is not None
|
|
assert block["budget"] == 6000000
|
|
assert block["keywords"] == ["mafia", "crime family"]
|
|
recs = {r["tmdb_id"]: r for r in block["recommendations"]}
|
|
assert recs[240]["in_library"] is True and recs[240]["rating_key"] == "m240"
|
|
assert recs[999]["in_library"] is False and recs[999]["rating_key"] is None
|
|
|
|
|
|
def test_get_never_leaks_api_key_in_error_messages(monkeypatch):
|
|
# The api_key rides in the request URL query string, so an error message built from the httpx
|
|
# exception (which embeds that URL) would leak the secret to logs (enrich_batch logs it). Both the
|
|
# HTTP-status and the network error paths must report status/path only — never the key.
|
|
monkeypatch.setattr(
|
|
tmdb.sysconfig, "get_str", lambda d, k: "SECRETKEY" if k == "tmdb_api_key" else "en-US"
|
|
)
|
|
client = tmdb.TmdbClient(None)
|
|
req = httpx.Request(
|
|
"GET", "https://api.themoviedb.org/3/movie/1?api_key=SECRETKEY&language=en-US"
|
|
)
|
|
|
|
class _StatusHttp:
|
|
def get(self, url, params=None):
|
|
return httpx.Response(401, request=req) # non-404 → raise_for_status → HTTPStatusError
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
client._http = _StatusHttp()
|
|
with pytest.raises(tmdb.TmdbError) as ei:
|
|
client._get("/movie/1")
|
|
assert "SECRETKEY" not in str(ei.value) and "401" in str(ei.value)
|
|
|
|
class _NetHttp:
|
|
def get(self, url, params=None):
|
|
raise httpx.ConnectError("connect failed", request=req)
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
client._http = _NetHttp()
|
|
with pytest.raises(tmdb.TmdbError) as ei2:
|
|
client._get("/movie/1")
|
|
assert "SECRETKEY" not in str(ei2.value)
|
|
client.close()
|
|
|
|
|
|
def test_run_enrich_noops_without_key(db):
|
|
# No tmdb_api_key configured (default "") → clean no-op, never an error.
|
|
out = tmdb.run_enrich(db)
|
|
assert out["enriched"] == 0 and "skipped" in out
|
|
|
|
|
|
class _RaisingTmdb:
|
|
"""A client whose fetch raises a transient TmdbError (e.g. a 429), to exercise the retry path."""
|
|
|
|
def find(self, external_id, source):
|
|
return None
|
|
|
|
def fetch(self, tmdb_id, media_type):
|
|
raise tmdb.TmdbError("429 rate limited")
|
|
|
|
|
|
def test_enrich_batch_transient_error_leaves_watermark_null(db):
|
|
# A transient TmdbError mid-batch must leave the row's tmdb_enriched_at NULL so the next run retries
|
|
# it (unlike a definitive no-match, which stamps the watermark). The row is NOT counted as processed.
|
|
lib = _lib(db)
|
|
mv = PlexItem(
|
|
rating_key="m1", library_id=lib.id, kind="movie", title="Godfather", guid="tmdb:238"
|
|
)
|
|
db.add(mv)
|
|
db.commit()
|
|
|
|
n = tmdb.enrich_batch(db, _RaisingTmdb(), 50)
|
|
assert n == 0 # the transient failure is skipped, not processed
|
|
db.refresh(mv)
|
|
assert mv.tmdb_enriched_at is None # left NULL → retried next run
|
|
assert mv.tmdb_id is None
|
|
|
|
|
|
def test_enrich_batch_skips_shows_without_guid(db):
|
|
lib = _lib(db, key="2", kind="show")
|
|
sh = PlexShow(
|
|
rating_key="s1", library_id=lib.id, title="No Guid Show"
|
|
) # guid None → not selected
|
|
db.add(sh)
|
|
db.commit()
|
|
assert tmdb.enrich_batch(db, FakeTmdb(), 50) == 0
|
|
db.refresh(sh)
|
|
assert sh.tmdb_enriched_at is None
|
|
|
|
|
|
def test_resolve_library_guids_scopes_to_given_libraries(db):
|
|
# The scoped path (person page) must keep a credit that lives in a NON-scoped library (disabled or an
|
|
# HU-overlay twin) OUT of the result; the unscoped path (detail recs) matches across all libraries;
|
|
# and an explicit empty list skips that media side entirely (distinct from None = match all).
|
|
lib_on = _lib(db, key="on", kind="movie")
|
|
lib_off = _lib(db, key="off", kind="movie")
|
|
m_on = PlexItem(
|
|
rating_key="on1", library_id=lib_on.id, kind="movie", title="In Scope", guid="tmdb:100"
|
|
)
|
|
m_off = PlexItem(
|
|
rating_key="off1", library_id=lib_off.id, kind="movie", title="Out", guid="tmdb:200"
|
|
)
|
|
db.add_all([m_on, m_off])
|
|
db.commit()
|
|
entries = [{"tmdb_id": 100, "media_type": "movie"}, {"tmdb_id": 200, "media_type": "movie"}]
|
|
|
|
# Scoped to lib_on: the row in lib_off is excluded.
|
|
movies, shows = tmdb.resolve_library_guids(
|
|
db, entries, movie_lib_ids=[lib_on.id], show_lib_ids=[]
|
|
)
|
|
assert set(movies) == {"tmdb:100"}
|
|
assert shows == {}
|
|
|
|
# Unscoped (None defaults): matches across all libraries.
|
|
movies_all, _ = tmdb.resolve_library_guids(db, entries)
|
|
assert set(movies_all) == {"tmdb:100", "tmdb:200"}
|
|
|
|
# Empty list for a media side skips its bucket even with candidates present (vs None = match all).
|
|
_, shows_empty = tmdb.resolve_library_guids(
|
|
db, [{"tmdb_id": 100, "media_type": "tv"}], show_lib_ids=[]
|
|
)
|
|
assert shows_empty == {}
|
|
|
|
|
|
def test_resolve_recs_matches_by_media_type_not_bare_id(db):
|
|
# TMDB movie ids and tv ids are independent, so tmdb:500 can be BOTH a show (in our library) and a
|
|
# movie recommendation. The match must respect media_type — a bare-id match would deep-link the movie
|
|
# rec to the unrelated show (and vice-versa).
|
|
mlib = _lib(db, key="10", kind="movie")
|
|
slib = _lib(db, key="11", kind="show")
|
|
show500 = PlexShow(rating_key="s500", library_id=slib.id, title="Show 500", guid="tmdb:500")
|
|
movie900 = PlexItem(
|
|
rating_key="m900", library_id=mlib.id, kind="movie", title="Movie 900", guid="tmdb:900"
|
|
)
|
|
db.add_all([show500, movie900])
|
|
db.commit()
|
|
|
|
# A movie rec for id 500 must NOT match show s500; a tv rec for id 900 must NOT match movie m900.
|
|
miss = {
|
|
r["tmdb_id"]: r
|
|
for r in tmdb.resolve_recs(
|
|
db,
|
|
[
|
|
{"tmdb_id": 500, "title": "Movie 500", "media_type": "movie"},
|
|
{"tmdb_id": 900, "title": "Show 900", "media_type": "tv"},
|
|
],
|
|
lang="en",
|
|
)
|
|
}
|
|
assert miss[500]["in_library"] is False and miss[500]["rating_key"] is None
|
|
assert miss[900]["in_library"] is False and miss[900]["rating_key"] is None
|
|
|
|
# The right media_type DOES resolve to the right row.
|
|
hit = {
|
|
r["tmdb_id"]: r
|
|
for r in tmdb.resolve_recs(
|
|
db,
|
|
[
|
|
{"tmdb_id": 900, "media_type": "movie"},
|
|
{"tmdb_id": 500, "media_type": "tv"},
|
|
],
|
|
lang="en",
|
|
)
|
|
}
|
|
assert hit[900]["in_library"] is True and hit[900]["rating_key"] == "m900"
|
|
assert hit[500]["in_library"] is True and hit[500]["rating_key"] == "s500"
|
|
|
|
|
|
# ---- countries by ISO code -----------------------------------------------------------------------
|
|
|
|
|
|
class FakeCountries(FakeTmdb):
|
|
"""TMDB's /configuration/countries, counting the requests."""
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.calls = 0
|
|
|
|
def countries(self):
|
|
self.calls += 1
|
|
return [
|
|
{"iso_3166_1": "US", "english_name": "United States of America"},
|
|
{"iso_3166_1": "YU", "english_name": "Yugoslavia"},
|
|
]
|
|
|
|
|
|
def test_countries_stored_before_codes_get_them_from_tmdbs_list(db):
|
|
lib = _lib(db)
|
|
old = PlexItem(
|
|
rating_key="m1",
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title="Old",
|
|
production_countries=["Yugoslavia", "Atlantis"],
|
|
tmdb_enriched_at=datetime.now(timezone.utc),
|
|
)
|
|
show_lib = _lib(db, key="2", kind="show")
|
|
show = PlexShow(
|
|
rating_key="s1",
|
|
library_id=show_lib.id,
|
|
title="Show",
|
|
production_countries=["United States of America"],
|
|
)
|
|
new = PlexItem(
|
|
rating_key="m2",
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title="New",
|
|
production_countries=[{"code": "US", "name": "United States of America"}],
|
|
)
|
|
db.add_all([old, show, new])
|
|
db.commit()
|
|
|
|
# Served before the backfill: the bare name, with no code — never a crash on the old shape.
|
|
assert tmdb.serve_block(db, old, lang="hu")["production_countries"] == [
|
|
{"code": None, "name": "Yugoslavia"},
|
|
{"code": None, "name": "Atlantis"},
|
|
]
|
|
|
|
fake = FakeCountries()
|
|
assert tmdb.backfill_country_codes(db, fake) == 2
|
|
db.refresh(old)
|
|
db.refresh(show)
|
|
assert old.production_countries == [
|
|
{"code": "YU", "name": "Yugoslavia"},
|
|
{"code": None, "name": "Atlantis"}, # not in TMDB's list → no code, the name kept
|
|
]
|
|
assert show.production_countries == [{"code": "US", "name": "United States of America"}]
|
|
|
|
# Nothing left to code: no request at all.
|
|
assert tmdb.backfill_country_codes(db, fake) == 0
|
|
assert fake.calls == 1
|