diff --git a/core/video/discovery_gaps.py b/core/video/discovery_gaps.py new file mode 100644 index 00000000..9376c8c7 --- /dev/null +++ b/core/video/discovery_gaps.py @@ -0,0 +1,101 @@ +"""The "what am I missing?" gap engine for video discovery. + +Two pure diffs that power the discover rails: + +* **collection_gaps** — given the TMDB items of a franchise the user has *started* + (owns >=1 of) and the set of tmdb ids they already own, return the franchise + entries they're missing ("Complete the Matrix Collection — you're missing 2"). +* **filmography_gaps** — given a person's credits (combined_credits) and the owned + set, return the titles of theirs the user doesn't have ("More from Christopher + Nolan — 3 you don't own"). + +Both are pure (no I/O, no TMDB, no DB) — the discover API wires them to real data +(``owned tmdb ids`` from the library, collection items from the TMDB client, person +credits from the enrichment engine), so the diff + ranking logic is unit-testable in +isolation. Items are plain dicts carrying at least ``tmdb_id``; ``kind`` / +``popularity`` / ``vote_count`` are used for filtering + ranking when present. +""" + +from __future__ import annotations + +from typing import Any, Dict, Iterable, List, Set + + +def _owned_set(owned_tmdb_ids: Iterable) -> Set[int]: + out: Set[int] = set() + for x in owned_tmdb_ids or []: + try: + out.add(int(x)) + except (TypeError, ValueError): + continue + return out + + +def collection_gaps(owned_tmdb_ids: Iterable, collection_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + """Franchise entries the user is missing. + + ``collection_items`` is the full ordered film list of a TMDB collection (each a + dict with ``tmdb_id``). Returns the entries whose tmdb id isn't in + ``owned_tmdb_ids``, in the collection's original order (chronological / as TMDB + returns it), deduped. Entries without a usable tmdb id are skipped. + """ + owned = _owned_set(owned_tmdb_ids) + seen: Set[int] = set() + out: List[Dict[str, Any]] = [] + for it in collection_items or []: + if not isinstance(it, dict): + continue + tid = it.get("tmdb_id") + try: + tid = int(tid) + except (TypeError, ValueError): + continue + if tid in owned or tid in seen: + continue + seen.add(tid) + out.append(it) + return out + + +def filmography_gaps( + owned_tmdb_ids: Iterable, + credits: List[Dict[str, Any]], + *, + kinds: Iterable[str] = ("movie",), + min_vote_count: int = 0, + limit: int = 0, +) -> List[Dict[str, Any]]: + """Titles from a person's credits the user doesn't own, ranked by popularity. + + ``credits`` is a person's combined credits (each a dict with ``tmdb_id`` and + ``kind``). Keeps only the requested ``kinds``, drops owned + duplicate ids, and + (when ``min_vote_count`` > 0) filters out obscure entries lacking enough votes — + so the rail surfaces real films, not every uncredited cameo. Sorted by + ``popularity`` (then ``vote_count``) descending; ``limit`` caps the result (0 = all). + """ + owned = _owned_set(owned_tmdb_ids) + wanted = {str(k) for k in kinds} + seen: Set[int] = set() + out: List[Dict[str, Any]] = [] + for c in credits or []: + if not isinstance(c, dict): + continue + if c.get("kind") is not None and str(c.get("kind")) not in wanted: + continue + tid = c.get("tmdb_id") + try: + tid = int(tid) + except (TypeError, ValueError): + continue + if tid in owned or tid in seen: + continue + if min_vote_count and (c.get("vote_count") or 0) < min_vote_count: + continue + seen.add(tid) + out.append(c) + + out.sort(key=lambda x: ((x.get("popularity") or 0), (x.get("vote_count") or 0)), reverse=True) + return out[:limit] if limit and limit > 0 else out + + +__all__ = ["collection_gaps", "filmography_gaps"] diff --git a/tests/video/test_discovery_gaps.py b/tests/video/test_discovery_gaps.py new file mode 100644 index 00000000..9bf7f769 --- /dev/null +++ b/tests/video/test_discovery_gaps.py @@ -0,0 +1,71 @@ +"""Video discovery gap engine — the "what am I missing?" diffs. + +Pins: collection gaps keep franchise order + drop owned; filmography gaps dedup, +filter by kind + vote_count, rank by popularity, and respect the owned set. +""" + +from __future__ import annotations + +from core.video.discovery_gaps import collection_gaps, filmography_gaps + + +def _m(tid, **kw): + d = {"tmdb_id": tid, "kind": "movie"} + d.update(kw) + return d + + +# ── collection gaps ────────────────────────────────────────────────────────── + +def test_collection_returns_unowned_in_order(): + coll = [_m(1, title="A"), _m(2, title="B"), _m(3, title="C")] + out = collection_gaps({2}, coll) + assert [x["title"] for x in out] == ["A", "C"] + + +def test_collection_all_owned_is_empty(): + coll = [_m(1), _m(2)] + assert collection_gaps({1, 2}, coll) == [] + + +def test_collection_dedups_and_skips_bad_ids(): + coll = [_m(1, title="A"), _m(1, title="dup"), _m(None, title="noid"), {"title": "nokey"}] + out = collection_gaps(set(), coll) + assert [x["title"] for x in out] == ["A"] + + +def test_collection_owned_ids_coerced(): + # owned ids may arrive as strings from the DB + assert collection_gaps(["2"], [_m(1), _m(2)]) == [_m(1)] + + +# ── filmography gaps ───────────────────────────────────────────────────────── + +def test_filmography_unowned_ranked_by_popularity(): + creds = [_m(1, popularity=5), _m(2, popularity=50), _m(3, popularity=20)] + out = filmography_gaps({99}, creds) + assert [x["tmdb_id"] for x in out] == [2, 3, 1] + + +def test_filmography_drops_owned_and_dupes(): + creds = [_m(1, popularity=9), _m(1, popularity=9), _m(2, popularity=1)] + out = filmography_gaps({2}, creds) + assert [x["tmdb_id"] for x in out] == [1] + + +def test_filmography_filters_by_kind(): + creds = [_m(1, popularity=9), {"tmdb_id": 2, "kind": "show", "popularity": 99}] + assert [x["tmdb_id"] for x in filmography_gaps(set(), creds, kinds=("movie",))] == [1] + assert [x["tmdb_id"] for x in filmography_gaps(set(), creds, kinds=("movie", "show"))] == [2, 1] + + +def test_filmography_min_vote_count_filters_obscure(): + creds = [_m(1, popularity=9, vote_count=2), _m(2, popularity=8, vote_count=500)] + out = filmography_gaps(set(), creds, min_vote_count=50) + assert [x["tmdb_id"] for x in out] == [2] + + +def test_filmography_limit(): + creds = [_m(i, popularity=i) for i in range(1, 6)] + out = filmography_gaps(set(), creds, limit=2) + assert [x["tmdb_id"] for x in out] == [5, 4]