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.
111 lines
4.6 KiB
Python
111 lines
4.6 KiB
Python
"""What a user's Plex history leans towards, as a weight for a surprise draw (`app.draw`).
|
||
|
||
Two signals, both already in the database (no TMDB call):
|
||
- **genres** of the titles the user watched, the recent ones weighing more (a half-life of
|
||
HALF_LIFE_DAYS), each scored 0..1 against the strongest;
|
||
- the **TMDB recommendations** stored on those titles (`recommendations`), each scored by the most
|
||
recent title that recommends it.
|
||
|
||
A title counts once however many of its episodes were watched — a binge of one show would otherwise
|
||
outvote every film — and only the RECENT_TITLES latest titles count. A candidate's weight is
|
||
1 + GENRE_WEIGHT × (the mean score of its genres) + REC_WEIGHT × (its recommendation score), so a
|
||
title with no affinity keeps weight 1 and still comes up.
|
||
"""
|
||
|
||
import json
|
||
from dataclasses import dataclass, field
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import Float, cast, func, literal, literal_column, select
|
||
from sqlalchemy.dialects.postgresql import JSONB
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy.sql.elements import ColumnElement
|
||
|
||
from app import draw
|
||
from app.plex.models import PlexItem, PlexShow, PlexState
|
||
from app.userscope import owned_by
|
||
|
||
HALF_LIFE_DAYS = 180
|
||
RECENT_TITLES = 200
|
||
GENRE_WEIGHT = 2.0
|
||
REC_WEIGHT = 3.0
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Affinity:
|
||
genres: dict[str, float] = field(default_factory=dict)
|
||
movie_recs: dict[str, float] = field(default_factory=dict)
|
||
show_recs: dict[str, float] = field(default_factory=dict)
|
||
|
||
def __bool__(self) -> bool:
|
||
return bool(self.genres or self.movie_recs or self.show_recs)
|
||
|
||
def weight(self, model: type[PlexItem] | type[PlexShow]) -> ColumnElement[float]:
|
||
"""The draw weight of a `model` row (a movie or a show)."""
|
||
scores = cast(literal(json.dumps(self.genres)), JSONB)
|
||
recs = self.show_recs if model is PlexShow else self.movie_recs
|
||
genre = func.jsonb_array_elements_text(model.genres).table_valued("value").alias("g")
|
||
genre_aff = (
|
||
select(
|
||
func.coalesce(
|
||
func.avg(func.coalesce(cast(scores.op("->>")(genre.c.value), Float), 0.0)),
|
||
0.0,
|
||
)
|
||
)
|
||
.select_from(genre)
|
||
.scalar_subquery()
|
||
)
|
||
rec_aff = func.coalesce(
|
||
cast(cast(literal(json.dumps(recs)), JSONB).op("->>")(model.guid), Float), 0.0
|
||
)
|
||
return literal_column("1.0") + GENRE_WEIGHT * genre_aff + REC_WEIGHT * rec_aff
|
||
|
||
|
||
def for_user(db: Session, uid: int) -> Affinity:
|
||
"""The user's affinity from their latest watched titles; empty (`not aff`) with no history."""
|
||
# One key per title: a show for an episode, the item for a film. concat() of a NULL show_id is
|
||
# just "s", which nullif turns back into NULL so the film's key takes over.
|
||
title = func.coalesce(
|
||
func.nullif(func.concat("s", PlexItem.show_id), "s"), func.concat("m", PlexItem.id)
|
||
).label("title")
|
||
latest = (
|
||
select(title, PlexState.watched_at, PlexItem.id.label("item_id"), PlexItem.show_id)
|
||
.join(PlexItem, PlexItem.id == PlexState.item_id)
|
||
.where(owned_by(PlexState.user_id, uid), PlexState.status == "watched")
|
||
.distinct(title)
|
||
.order_by(title, PlexState.watched_at.desc().nullslast())
|
||
.subquery()
|
||
)
|
||
rows = db.execute(
|
||
select(
|
||
latest.c.watched_at,
|
||
PlexItem.genres,
|
||
PlexItem.recommendations,
|
||
PlexShow.genres,
|
||
PlexShow.recommendations,
|
||
)
|
||
.select_from(latest)
|
||
.join(PlexItem, PlexItem.id == latest.c.item_id)
|
||
.outerjoin(PlexShow, PlexShow.id == latest.c.show_id)
|
||
.order_by(latest.c.watched_at.desc().nullslast())
|
||
.limit(RECENT_TITLES)
|
||
).all()
|
||
if not rows:
|
||
return Affinity()
|
||
|
||
now = datetime.now(timezone.utc)
|
||
genres: dict[str, float] = {}
|
||
movie_recs: dict[str, float] = {}
|
||
show_recs: dict[str, float] = {}
|
||
for watched_at, item_genres, item_recs, show_genres, show_recs_json in rows:
|
||
r = draw.recency(watched_at, now, HALF_LIFE_DAYS)
|
||
for g in (show_genres if show_genres is not None else item_genres) or []:
|
||
genres[g] = genres.get(g, 0.0) + r
|
||
for rec in (show_recs_json if show_recs_json is not None else item_recs) or []:
|
||
if not rec.get("tmdb_id"):
|
||
continue
|
||
bucket = show_recs if rec.get("media_type") == "tv" else movie_recs
|
||
guid = f"tmdb:{rec['tmdb_id']}"
|
||
bucket[guid] = max(bucket.get(guid, 0.0), r)
|
||
return Affinity(draw.scaled(genres), draw.scaled(movie_recs), draw.scaled(show_recs))
|