peter
bb91d0a22d
The YouTube channel affinity and the Plex one each had the recency decay, the undated-watch rule and the scale to the strongest; app.draw now holds them. Tests hold that another user's history leans neither draw, the rail's blind slot, and that a show leans on show recommendations.
336 lines
13 KiB
Python
336 lines
13 KiB
Python
"""Surprise me leans on what the user watches, and stays a surprise.
|
|
|
|
The Home hero's surprise tier and the Plex Discover rail were uniform draws over the unseen. They are
|
|
now weighted draws (`app.draw`): a title's weight is 1 + its affinity — for Plex the genres of the
|
|
films and shows the user watched (recent ones weigh more) and the TMDB recommendations stored on them;
|
|
for YouTube the channels the user watches. One slot of three stays a blind draw, and with no watch
|
|
history the order is exactly the old one. DB lane.
|
|
"""
|
|
|
|
import hashlib
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from sqlalchemy import select, text
|
|
|
|
from app import draw
|
|
from app.catalog.models import Channel, Subscription, Video
|
|
from app.discover import queries
|
|
from app.plex import affinity, discover as plex_discover
|
|
from app.plex.models import PlexItem, PlexLibrary, PlexShow, PlexState
|
|
from app.plex.query import LibraryFilters
|
|
from app.reader.models import VideoState
|
|
|
|
NOW = datetime.now(timezone.utc)
|
|
SEEDS = range(90)
|
|
|
|
|
|
# ── the draw itself ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_every_third_slot_is_blind_and_nothing_repeats():
|
|
weighted = ["a", "b", "c", "d", "e", "f"]
|
|
blind = ["f", "e", "d", "c", "b", "a"]
|
|
# offset 0: slots 0 and 3 are blind
|
|
assert draw.blend(weighted, blind, 0, 6) == ["f", "a", "b", "e", "c", "d"]
|
|
# offset 1: slot 2 and 5
|
|
assert draw.blend(weighted, blind, 1, 6) == ["a", "b", "f", "c", "d", "e"]
|
|
assert draw.blend(weighted, blind, 2, 3) == ["a", "f", "b"]
|
|
|
|
|
|
def test_a_short_side_is_topped_up_from_the_other():
|
|
assert draw.blend(["a"], ["a", "b", "c"], 0, 3) == ["a", "b", "c"]
|
|
assert draw.blend(["a", "b", "c"], [], 0, 3) == ["a", "b", "c"]
|
|
|
|
|
|
# ── YouTube: the channels the user watches ───────────────────────────────────────────────────────
|
|
|
|
|
|
def _md5_order(ids, seed):
|
|
return sorted(ids, key=lambda i: hashlib.md5(f"{i}{seed}".encode()).hexdigest())
|
|
|
|
|
|
def _seed_yt(db, make_user, *, watched_in_fav: int):
|
|
user = make_user()
|
|
fav, other = Channel(id="UCfav", title="Fav"), Channel(id="UCother", title="Other")
|
|
db.add_all([fav, other])
|
|
db.flush()
|
|
db.add_all(
|
|
[
|
|
Subscription(user_id=user.id, channel_id="UCfav"),
|
|
Subscription(user_id=user.id, channel_id="UCother"),
|
|
]
|
|
)
|
|
|
|
def video(vid, ch, days):
|
|
return Video(
|
|
id=vid,
|
|
channel_id=ch,
|
|
title=vid,
|
|
duration_seconds=20 * 60,
|
|
published_at=NOW - timedelta(days=days),
|
|
)
|
|
|
|
# One unseen video per channel, plus the user's watched history on the favourite.
|
|
db.add_all([video("fav_new", "UCfav", 1), video("other_new", "UCother", 1)])
|
|
for i in range(watched_in_fav):
|
|
db.add(video(f"fav_seen{i}", "UCfav", 10 + i))
|
|
db.flush()
|
|
for i in range(watched_in_fav):
|
|
db.add(
|
|
VideoState(
|
|
user_id=user.id,
|
|
video_id=f"fav_seen{i}",
|
|
status="watched",
|
|
watched_at=NOW - timedelta(days=i),
|
|
)
|
|
)
|
|
db.commit()
|
|
return user
|
|
|
|
|
|
def _first(rows):
|
|
return rows[0]["id"] if rows else None
|
|
|
|
|
|
def test_a_watched_channel_wins_most_draws_but_not_all(db, make_user):
|
|
user = _seed_yt(db, make_user, watched_in_fav=12)
|
|
leads = [_first(queries.yt_surprise(db, user, "any", s, 2)) for s in SEEDS]
|
|
fav = leads.count("fav_new") / len(leads)
|
|
# weight 1 + 3 against 1: 0.8 in a weighted slot, 0.5 in the blind one (a third of the seeds)
|
|
assert 0.6 < fav < 0.9, fav
|
|
|
|
|
|
def test_the_blind_slot_deals_the_plain_shuffle_however_strong_the_lean(db, make_user):
|
|
user = _seed_yt(db, make_user, watched_in_fav=12)
|
|
for s in range(0, 30, 3): # (0 + seed) % 3 == 0: the first slot is the blind one
|
|
got = _first(queries.yt_surprise(db, user, "any", s, 2))
|
|
assert got == _md5_order(["fav_new", "other_new"], s)[0]
|
|
|
|
|
|
def test_with_no_history_the_youtube_order_is_the_old_shuffle(db, make_user):
|
|
user = _seed_yt(db, make_user, watched_in_fav=0)
|
|
for s in range(12):
|
|
got = [r["id"] for r in queries.yt_surprise(db, user, "any", s, 2)]
|
|
assert got == _md5_order(["fav_new", "other_new"], s)
|
|
|
|
|
|
def test_a_watched_video_is_never_drawn(db, make_user):
|
|
user = _seed_yt(db, make_user, watched_in_fav=12)
|
|
drawn = {r["id"] for s in SEEDS for r in queries.yt_surprise(db, user, "any", s, 5)}
|
|
assert drawn == {"fav_new", "other_new"}
|
|
|
|
|
|
# ── Plex: genres and recommendations of what the user watched ────────────────────────────────────
|
|
|
|
|
|
def _seed_plex(db, make_user, *, history: bool):
|
|
user = make_user()
|
|
lib = PlexLibrary(plex_key="77", title="Movies", kind="movie", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
tmdb_ids = iter(range(5000, 6000))
|
|
|
|
def movie(key, genres, recs=None):
|
|
m = PlexItem(
|
|
rating_key=key,
|
|
library_id=lib.id,
|
|
kind="movie",
|
|
title=key,
|
|
duration_s=100 * 60,
|
|
genres=genres,
|
|
guid=f"tmdb:{next(tmdb_ids)}",
|
|
recommendations=recs,
|
|
)
|
|
db.add(m)
|
|
return m
|
|
|
|
horror = movie("horror_new", ["Horror"])
|
|
comedy = movie("comedy_new", ["Comedy"])
|
|
tipped = movie("tipped_new", ["Western"])
|
|
db.flush()
|
|
tip = [{"tmdb_id": int(tipped.guid.split(":")[1]), "media_type": "movie", "title": "t"}]
|
|
seen = [movie(f"horror_seen{i}", ["Horror"], tip if i == 0 else None) for i in range(3)]
|
|
db.flush()
|
|
if history:
|
|
for i, m in enumerate(seen):
|
|
db.add(
|
|
PlexState(
|
|
user_id=user.id,
|
|
item_id=m.id,
|
|
status="watched",
|
|
watched_at=NOW - timedelta(days=i),
|
|
)
|
|
)
|
|
db.commit()
|
|
from app import sysconfig
|
|
|
|
sysconfig.set_value(db, "plex_enabled", True)
|
|
db.commit()
|
|
return user, horror, comedy, tipped
|
|
|
|
|
|
def test_the_genre_and_the_recommendation_of_what_was_watched_win_most_draws(db, make_user):
|
|
user, *_ = _seed_plex(db, make_user, history=True)
|
|
leads = [_first(queries.plex_surprise(db, user.id, "any", s, 3, lang="en")) for s in SEEDS]
|
|
# Weights: Horror 1 + 2 (its genre), the recommended Western 1 + 3, the Comedy nobody chose 1. In
|
|
# a weighted slot the Comedy leads 1 time in 8, in the blind third 1 in 3: about 0.19 against
|
|
# 0.81 for the other two together (a uniform draw: 1 in 3 each). It still leads now and then —
|
|
# a surprise, not a ranking.
|
|
comedy = leads.count("comedy_new")
|
|
assert comedy > 0
|
|
assert leads.count("horror_new") > 1.5 * comedy and leads.count("tipped_new") > 1.5 * comedy
|
|
assert leads.count("horror_new") + leads.count("tipped_new") > 3 * comedy
|
|
|
|
|
|
def test_with_no_history_the_plex_order_is_the_old_shuffle(db, make_user):
|
|
user, *_ = _seed_plex(db, make_user, history=False)
|
|
keys = ["horror_new", "comedy_new", "tipped_new"] + [f"horror_seen{i}" for i in range(3)]
|
|
for s in range(12):
|
|
got = [c["id"] for c in queries.plex_surprise(db, user.id, "any", s, 6, lang="en")]
|
|
assert got == _md5_order(keys, s)
|
|
|
|
|
|
def test_a_watched_film_is_never_drawn(db, make_user):
|
|
user, *_ = _seed_plex(db, make_user, history=True)
|
|
drawn = {
|
|
c["id"] for s in SEEDS for c in queries.plex_surprise(db, user.id, "any", s, 6, lang="en")
|
|
}
|
|
assert drawn == {"horror_new", "comedy_new", "tipped_new"}
|
|
|
|
|
|
def test_the_plex_blind_slot_deals_the_plain_shuffle(db, make_user):
|
|
user, *_ = _seed_plex(db, make_user, history=True)
|
|
unseen = ["horror_new", "comedy_new", "tipped_new"]
|
|
for s in range(0, 30, 3):
|
|
got = _first(queries.plex_surprise(db, user.id, "any", s, 3, lang="en"))
|
|
assert got == _md5_order(unseen, s)[0]
|
|
|
|
|
|
def test_a_binged_show_counts_once_and_recent_watches_weigh_more(db, make_user):
|
|
user = make_user()
|
|
lib = PlexLibrary(plex_key="78", title="Shows", kind="show", enabled=True)
|
|
mlib = PlexLibrary(plex_key="79", title="Movies", kind="movie", enabled=True)
|
|
db.add_all([lib, mlib])
|
|
db.flush()
|
|
show = PlexShow(rating_key="sitcom", library_id=lib.id, title="Sitcom", genres=["Comedy"])
|
|
db.add(show)
|
|
db.flush()
|
|
eps = [
|
|
PlexItem(
|
|
rating_key=f"ep{i}",
|
|
library_id=lib.id,
|
|
kind="episode",
|
|
title=f"e{i}",
|
|
show_id=show.id,
|
|
season_number=1,
|
|
episode_number=i,
|
|
)
|
|
for i in range(10)
|
|
]
|
|
films = [
|
|
PlexItem(
|
|
rating_key=f"h{i}", library_id=mlib.id, kind="movie", title=f"h{i}", genres=["Horror"]
|
|
)
|
|
for i in range(2)
|
|
]
|
|
old = PlexItem(
|
|
rating_key="w0", library_id=mlib.id, kind="movie", title="w0", genres=["Western"]
|
|
)
|
|
db.add_all([*eps, *films, old])
|
|
db.flush()
|
|
for it in [*eps, *films]:
|
|
db.add(PlexState(user_id=user.id, item_id=it.id, status="watched", watched_at=NOW))
|
|
db.add(
|
|
PlexState(
|
|
user_id=user.id, item_id=old.id, status="watched", watched_at=NOW - timedelta(days=720)
|
|
)
|
|
)
|
|
db.commit()
|
|
|
|
genres = affinity.for_user(db, user.id).genres
|
|
# Ten episodes of one show are one title: two horror films outweigh it.
|
|
assert genres["Horror"] == 1.0
|
|
assert genres["Comedy"] == 0.5
|
|
# Four half-lives ago weighs a sixteenth of today.
|
|
assert abs(genres["Western"] - 0.5 / 16) < 0.01
|
|
|
|
|
|
def test_the_discover_rail_leans_the_same_way(db, make_user):
|
|
# The rail draws with random(), not a seed; setseed makes this session's draws repeatable.
|
|
user, *_ = _seed_plex(db, make_user, history=True)
|
|
db.execute(text("SELECT setseed(0.42)"))
|
|
leads = [
|
|
plex_discover.surprise(db, user, "movie", "en", 3, [], [], LibraryFilters())["items"][0][
|
|
"id"
|
|
]
|
|
for _ in SEEDS
|
|
]
|
|
assert leads.count("comedy_new") > 0
|
|
assert leads.count("horror_new") + leads.count("tipped_new") > 3 * leads.count("comedy_new")
|
|
|
|
|
|
# ── whose history: only the user's own ───────────────────────────────────────────────────────────
|
|
|
|
|
|
def test_another_users_youtube_history_does_not_lean_my_draw(db, make_user):
|
|
watcher = _seed_yt(db, make_user, watched_in_fav=12)
|
|
me = make_user()
|
|
db.add_all(
|
|
[
|
|
Subscription(user_id=me.id, channel_id="UCfav"),
|
|
Subscription(user_id=me.id, channel_id="UCother"),
|
|
]
|
|
)
|
|
db.commit()
|
|
assert queries._channel_affinity(db, watcher)
|
|
assert queries._channel_affinity(db, me) == {}
|
|
|
|
|
|
def test_another_users_plex_history_does_not_lean_my_draw(db, make_user):
|
|
_seed_plex(db, make_user, history=True)
|
|
me = make_user()
|
|
assert not affinity.for_user(db, me.id)
|
|
keys = ["horror_new", "comedy_new", "tipped_new"] + [f"horror_seen{i}" for i in range(3)]
|
|
for s in range(12):
|
|
got = [c["id"] for c in queries.plex_surprise(db, me.id, "any", s, 6, lang="en")]
|
|
assert got == _md5_order(keys, s)
|
|
|
|
|
|
# ── the rail: its blind slot, and a show's own recommendations ───────────────────────────────────
|
|
|
|
|
|
def test_the_rails_third_slot_is_the_blind_draw(db, make_user, monkeypatch):
|
|
# The rail draws with random(), so the blend is watched rather than its odds: every leaning draw
|
|
# goes through it, with the blind order beside the weighted one and the third slot blind.
|
|
user, *_ = _seed_plex(db, make_user, history=True)
|
|
calls = []
|
|
real = draw.blend
|
|
|
|
def spy(weighted, blind, offset, n, key):
|
|
calls.append((offset, n, sorted(map(key, weighted)), sorted(map(key, blind))))
|
|
return real(weighted, blind, offset, n, key=key)
|
|
|
|
monkeypatch.setattr(draw, "blend", spy)
|
|
got = plex_discover.surprise(db, user, "movie", "en", 3, [], [], LibraryFilters())["items"]
|
|
assert calls and all(offset == 1 and n == 3 for offset, n, *_ in calls)
|
|
pool = sorted(["horror_new", "comedy_new", "tipped_new"])
|
|
assert calls[0][2] == pool and calls[0][3] == pool
|
|
assert len(got) == 3
|
|
|
|
|
|
def test_a_show_leans_on_the_shows_recommended_not_the_films(db, make_user):
|
|
lib = PlexLibrary(plex_key="80", title="Shows", kind="show", enabled=True)
|
|
db.add(lib)
|
|
db.flush()
|
|
tipped = PlexShow(rating_key="tv1", library_id=lib.id, title="t", guid="tmdb:1", genres=[])
|
|
db.add(tipped)
|
|
db.commit()
|
|
shows = affinity.Affinity(show_recs={"tmdb:1": 1.0})
|
|
films = affinity.Affinity(movie_recs={"tmdb:1": 1.0})
|
|
|
|
def weight(aff):
|
|
return db.execute(select(aff.weight(PlexShow)).where(PlexShow.id == tipped.id)).scalar_one()
|
|
|
|
assert weight(shows) == 1 + affinity.REC_WEIGHT
|
|
assert weight(films) == 1
|