peter
e6b76e156a
The comfort shelf resolved each pinned channel with three queries (the row, its unwatched count, its newest video) and each playlist with two. channels_summary gets the count and the newest row per channel as window functions over one filtered set, playlists_summary reads every pinned list in one query, and the rows come in one IN() each — so the shelf's cost no longer grows with its length. A test counts the statements. RAIL_LIMIT had two definitions held in step by a comment; the Plex Discover route now imports the domain's. And rollup's "continue the latest-started episode" was false: it takes the last in-progress one in broadcast order, which is also what Home's next_episode picks.
362 lines
17 KiB
Python
362 lines
17 KiB
Python
"""The row fetchers behind the Home surface — one place where "what is startable right now" is
|
|
expressed as SQL, for both sources.
|
|
|
|
Two rules shape this module:
|
|
|
|
**Reuse the query layer, not the route.** The YouTube half calls `feedquery.filtered_query` /
|
|
`sort_keys`, so the Home surface inherits the feed's own definitions of subscribed, blocked,
|
|
explore-visible and in-progress for free, and cannot drift from them. The Plex half calls
|
|
`plex.rollup` and the `plex.cards` serializers rather than `routes/plex/discover.py`: that route
|
|
narrows EVERY rail by the Plex sidebar's `LibraryFilters`, which is right for a surface that HAS a
|
|
sidebar and wrong for a landing page that does not.
|
|
|
|
**One seed drives both halves.** The feed orders a shuffle by `md5(id || seed)` so a keyset cursor
|
|
stays valid across pages; the Plex surprise rail uses `ORDER BY random()`, which is fine for a single
|
|
un-paged draw but cannot be reproduced. A cross-source reroll needs both halves to move together —
|
|
otherwise "Mást kérek" re-draws the Plex picks while the YouTube ones stand still — so the Plex side
|
|
here takes the feed's seeded-md5 idiom instead of `random()`.
|
|
"""
|
|
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import Session, aliased
|
|
|
|
from app.discover.budget import band
|
|
from app.feedquery import (
|
|
feed_columns,
|
|
filtered_query,
|
|
not_blocked_cond,
|
|
order_by,
|
|
serialize,
|
|
sort_keys,
|
|
)
|
|
from app.models import (
|
|
Channel,
|
|
PlaylistItem,
|
|
PlexItem,
|
|
PlexShow,
|
|
PlexState,
|
|
User,
|
|
Video,
|
|
VideoState,
|
|
)
|
|
from app.plex import rollup
|
|
from app.plex.cards import leaf_card, movie_card, show_card
|
|
from app.plex.query import apply_show_state_filter, content_libs
|
|
|
|
# How many rows a single rail may carry — Home's rails and the Plex module's Discover rails alike
|
|
# (`routes/plex/discover.py` imports it): the two surfaces should feel like the same app, and a rail
|
|
# is a rail. One definition, because a comment was all that held two copies in step.
|
|
RAIL_LIMIT = 20
|
|
|
|
# The feed parameters a Home rail runs with. Spelled out rather than borrowed from the route's
|
|
# `_feed_params` defaults, because those are an HTTP contract that may change for the feed's own
|
|
# reasons; the landing page's scope should not move with them silently.
|
|
_FEED_BASE: dict = {
|
|
"tags": [],
|
|
"tag_modes": {},
|
|
"untagged": False,
|
|
"channel_ids": [],
|
|
"q": None,
|
|
"min_duration": None,
|
|
"max_duration": None,
|
|
"max_age_days": None,
|
|
"published_after": None,
|
|
"published_before": None,
|
|
"show_normal": True,
|
|
"include_shorts": False,
|
|
"include_live": False,
|
|
"show": "unwatched",
|
|
"scope": "my",
|
|
"library_source": "organic",
|
|
}
|
|
|
|
|
|
def _seeded(col, seed: int):
|
|
"""The feed's deterministic shuffle expression, reused for Plex so one seed moves both halves."""
|
|
return func.md5(func.concat(col, str(seed)))
|
|
|
|
|
|
# ── YouTube ───────────────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def yt_in_progress(db: Session, user: User, limit: int = RAIL_LIMIT) -> list[tuple[object, dict]]:
|
|
"""Started-but-not-finished videos, most recently progressed first, as (progressed_at, row).
|
|
|
|
The predicate is `feedquery`'s `show="in_progress"` verbatim (a stored resume position, status not
|
|
watched/hidden) — but the query is built here rather than through `filtered_query` because this
|
|
rail wants an ordering by `progress_updated_at`, and that column lives on the state alias the feed
|
|
query keeps private. Driving from `video_states` is also the faster plan: it is a per-user table of
|
|
a few hundred rows, against a catalog of many thousands.
|
|
|
|
Building it here means the feed's OTHER predicates have to be chosen deliberately rather than
|
|
inherited. `not_blocked_cond` is shared, because "never show me this channel" outranks "you
|
|
started it" — blocking leaves the state row untouched, so without this the video the user just
|
|
banned heads the resume rail. The feed's SCOPE clauses (subscribed / via_explore accessibility)
|
|
are deliberately NOT applied, for the reason `yt_video_row` gives: something you have already
|
|
started must stay resumable even after you unsubscribe or an exploration expires."""
|
|
st = aliased(VideoState)
|
|
q = (
|
|
select(*feed_columns(user, st), st.progress_updated_at.label("progressed_at"))
|
|
.join(Channel, Channel.id == Video.channel_id)
|
|
.join(st, and_(st.video_id == Video.id, st.user_id == user.id))
|
|
.where(
|
|
st.position_seconds > 0,
|
|
st.status.notin_(("watched", "hidden")),
|
|
Video.unavailable_since.is_(None),
|
|
not_blocked_cond(user),
|
|
)
|
|
.order_by(st.progress_updated_at.desc().nullslast())
|
|
.limit(limit)
|
|
)
|
|
return [(r.progressed_at, serialize(r)) for r in db.execute(q).all()]
|
|
|
|
|
|
def yt_surprise(db: Session, user: User, budget: str, seed: int, limit: int) -> list[dict]:
|
|
"""Unwatched videos from the user's own feed scope, in seeded-shuffle order, inside the budget.
|
|
|
|
Shorts are opened up only for the shortest band: "I have ten minutes" is the one answer a Short is
|
|
actually a good response to, and the feed's own default keeps them out otherwise."""
|
|
lo, hi = band(budget)
|
|
params = dict(_FEED_BASE, min_duration=lo, max_duration=hi, include_shorts=budget == "short")
|
|
query, rank_expr = filtered_query(db, user, **params)
|
|
keys = sort_keys("shuffle", seed, rank_expr)
|
|
# Over-fetch, then drop anything already started. The feed's `unwatched` scope keeps watched and
|
|
# hidden out but deliberately keeps in-progress rows in (they belong in the feed). Here they must
|
|
# go: the resume tier has already had its chance at them, and it may have passed precisely because
|
|
# the REMAINING time did not fit the budget — offering the same item back as a surprise, whose
|
|
# runtime does fit, would label a half-finished thing "Meglepetés" and contradict the question.
|
|
rows = db.execute(query.order_by(*order_by(keys)).limit(limit * 4)).all()
|
|
out = [serialize(r) for r in rows]
|
|
return [r for r in out if not r["position_seconds"]][:limit]
|
|
|
|
|
|
def yt_video_row(db: Session, user: User, video_id: str) -> dict | None:
|
|
"""One video as a feed row, with this user's watch state.
|
|
|
|
Built directly rather than through `filtered_query` on purpose: a PINNED video must keep working
|
|
even when it falls outside the feed's scope — the user unsubscribed from the channel, or blocked
|
|
it, or the video only ever arrived through a search. The shelf is the surface you fall back to;
|
|
it silently losing an entry because a filter moved is exactly the failure it must not have."""
|
|
st = aliased(VideoState)
|
|
row = db.execute(
|
|
select(*feed_columns(user, st))
|
|
.join(Channel, Channel.id == Video.channel_id)
|
|
.outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
|
|
.where(Video.id == video_id)
|
|
).first()
|
|
return serialize(row) if row is not None else None
|
|
|
|
|
|
def channels_summary(
|
|
db: Session, user: User, channel_ids: list[str]
|
|
) -> dict[str, tuple[int, dict | None]]:
|
|
"""For each pinned channel: (unwatched count, its newest unwatched video) — the tile's only honest
|
|
promise, since a channel has no runtime of its own, and what "put on that channel" resolves to.
|
|
|
|
ONE query for the whole shelf, not two per channel: the count and the first row per channel are
|
|
window functions over the same filtered set. `library_source="all"`: a pinned channel means the
|
|
whole channel, including videos that only ever arrived through a search — the feed's "organic"
|
|
default is a feed-scoping choice, not a statement about what belongs to a channel."""
|
|
if not channel_ids:
|
|
return {}
|
|
params = dict(_FEED_BASE, channel_ids=list(channel_ids), scope="all", library_source="all")
|
|
query, rank_expr = filtered_query(db, user, **params)
|
|
newest = order_by(sort_keys("newest", 0, rank_expr))
|
|
per_channel = {"partition_by": Video.channel_id}
|
|
sub = query.add_columns(
|
|
func.row_number().over(**per_channel, order_by=newest).label("_rn"),
|
|
func.count().over(**per_channel).label("_total"),
|
|
).subquery()
|
|
rows = db.execute(select(sub).where(sub.c._rn == 1)).all()
|
|
return {r.channel_id: (int(r._total), serialize(r)) for r in rows}
|
|
|
|
|
|
def playlists_summary(
|
|
db: Session, user: User, playlist_ids: list[int]
|
|
) -> dict[int, tuple[int, dict | None]]:
|
|
"""For each pinned playlist: (item count, the row to play next) — its first not-yet-watched item,
|
|
or its first item once the whole list is watched. The same row doubles as the tile's artwork, so
|
|
a shelf of playlists is scannable instead of a row of blank boxes. One query for the whole shelf."""
|
|
if not playlist_ids:
|
|
return {}
|
|
st = aliased(VideoState)
|
|
rows = db.execute(
|
|
select(*feed_columns(user, st), PlaylistItem.playlist_id.label("_playlist"))
|
|
.join(PlaylistItem, PlaylistItem.video_id == Video.id)
|
|
.join(Channel, Channel.id == Video.channel_id)
|
|
.outerjoin(st, and_(st.video_id == Video.id, st.user_id == user.id))
|
|
.where(PlaylistItem.playlist_id.in_(playlist_ids))
|
|
.order_by(PlaylistItem.playlist_id, PlaylistItem.position.asc())
|
|
).all()
|
|
grouped: dict[int, list] = {}
|
|
for r in rows:
|
|
grouped.setdefault(r._playlist, []).append(r)
|
|
out: dict[int, tuple[int, dict | None]] = {}
|
|
for pid, items in grouped.items():
|
|
nxt = next((r for r in items if (r.status or "new") not in ("watched", "hidden")), items[0])
|
|
out[pid] = (len(items), serialize(nxt))
|
|
return out
|
|
|
|
|
|
# ── Plex ──────────────────────────────────────────────────────────────────────────────────────────
|
|
|
|
|
|
def _unseen_movie_conds(uid: int):
|
|
"""A movie the user has neither finished nor hidden. A LEFT-JOIN-IS-NULL would also drop rows that
|
|
merely carry a resume position, which are exactly the ones Continue watching wants back."""
|
|
seen = (
|
|
select(1)
|
|
.select_from(PlexState)
|
|
.where(
|
|
PlexState.item_id == PlexItem.id,
|
|
PlexState.user_id == uid,
|
|
or_(
|
|
PlexState.status.in_(("watched", "hidden")),
|
|
# Started counts as "not a surprise" too — see the note in `yt_surprise`.
|
|
PlexState.position_seconds > 0,
|
|
),
|
|
)
|
|
.exists()
|
|
)
|
|
return ~seen
|
|
|
|
|
|
def plex_in_progress(db: Session, uid: int, limit: int = RAIL_LIMIT) -> list[tuple[object, dict]]:
|
|
"""In-progress movies + partially-watched shows as (progressed_at, card).
|
|
|
|
Mirrors `routes/plex/discover.py::_continue_watching` — same `PlexState` predicate for movies (a
|
|
NULL status still counts as started, so `notin_` alone would drop it), and the same shared
|
|
`rollup.agg_subquery` + `apply_show_state_filter` for shows, so the two rails cannot disagree on
|
|
what "in progress" means. What it does NOT inherit is that route's `LibraryFilters` narrowing:
|
|
this surface has no Plex sidebar to read a filter from."""
|
|
movie_libs, show_libs = content_libs(db)
|
|
out: list[tuple[object, dict]] = []
|
|
|
|
if movie_libs:
|
|
mq = (
|
|
select(PlexItem, PlexState)
|
|
.join(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
|
|
.where(
|
|
PlexItem.kind == "movie",
|
|
PlexItem.library_id.in_(movie_libs),
|
|
PlexState.position_seconds > 0,
|
|
or_(
|
|
PlexState.status.is_(None),
|
|
PlexState.status.notin_(("watched", "hidden")),
|
|
),
|
|
)
|
|
.order_by(PlexState.progress_updated_at.desc().nullslast())
|
|
.limit(limit)
|
|
)
|
|
for it, st in db.execute(mq).all():
|
|
out.append((st.progress_updated_at, movie_card(it, st)))
|
|
|
|
if show_libs:
|
|
prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery()
|
|
sq = apply_show_state_filter(
|
|
select(PlexShow, prog.c.resumed_at).join(prog, prog.c.show_id == PlexShow.id),
|
|
prog.c.total,
|
|
prog.c.watched,
|
|
prog.c.inprog,
|
|
"in_progress",
|
|
).where(PlexShow.library_id.in_(show_libs))
|
|
for sh, resumed_at in db.execute(
|
|
sq.order_by(prog.c.resumed_at.desc().nullslast()).limit(limit)
|
|
).all():
|
|
out.append((resumed_at, show_card(sh, "in_progress")))
|
|
|
|
return out
|
|
|
|
|
|
def plex_surprise(db: Session, uid: int, budget: str, seed: int, limit: int) -> list[dict]:
|
|
"""Unseen playable movies inside the budget, in seeded-shuffle order.
|
|
|
|
Movies only. A show is not a thing you can start playing, and resolving every candidate show to
|
|
its next episode just to draw three cards would be a query per row for a rail whose whole value is
|
|
that it answers instantly — shows reach this surface through Continue watching and the comfort
|
|
shelf, which is where "put the next episode on" actually belongs.
|
|
|
|
No `playable` filter, matching the Plex surprise rail: `PlexItem.playable` is a playback CLASS
|
|
("direct" / "remux" / "transcode"), not a yes/no, and NULL there means "not probed yet" rather than
|
|
"will not play". Excluding un-probed items would quietly hide a fresh library from the one surface
|
|
that is supposed to have an answer. The class travels on the card, so the client still shows it."""
|
|
movie_libs, _ = content_libs(db)
|
|
if not movie_libs:
|
|
return []
|
|
lo, hi = band(budget)
|
|
q = (
|
|
select(PlexItem)
|
|
.where(
|
|
PlexItem.kind == "movie",
|
|
PlexItem.library_id.in_(movie_libs),
|
|
_unseen_movie_conds(uid),
|
|
)
|
|
.order_by(_seeded(PlexItem.rating_key, seed))
|
|
.limit(limit)
|
|
)
|
|
if lo is not None:
|
|
q = q.where(PlexItem.duration_s >= lo)
|
|
if hi is not None:
|
|
q = q.where(PlexItem.duration_s <= hi)
|
|
items = db.execute(q).scalars().all()
|
|
states = rollup.item_state_map(db, uid, [it.id for it in items])
|
|
return [movie_card(it, states.get(it.id)) for it in items]
|
|
|
|
|
|
# Watch states that take an episode out of the running for "what would play next".
|
|
_DONE = ("watched", "hidden")
|
|
|
|
|
|
def next_episode(db: Session, uid: int, show_id: int) -> dict | None:
|
|
"""What "put that show on" means: the episode in progress, else the first unwatched one, else the
|
|
first — over the same `(season, episode)` broadcast ordering `detail.py::_episode_nav` uses.
|
|
|
|
Three ordered `LIMIT 1` probes rather than one scan picked apart in Python. The Python version
|
|
loaded EVERY episode of the show to choose one of them — 236 rows for Friends, 227 for JAG — which
|
|
the comfort shelf could just about afford (a handful of hand-picked pins) and the resume rail
|
|
could not, since it resolves a show per rail row on the landing page.
|
|
|
|
Returned as a `leaf_card` so it carries the show title the episode row itself does not."""
|
|
asc = (
|
|
PlexItem.season_number.asc().nullsfirst(),
|
|
PlexItem.episode_number.asc().nullsfirst(),
|
|
)
|
|
base = (
|
|
select(PlexItem, PlexState)
|
|
.outerjoin(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid))
|
|
.where(PlexItem.show_id == show_id, PlexItem.kind == "episode")
|
|
.limit(1)
|
|
)
|
|
# In progress — the LAST one in broadcast order, which is `plex.rollup.rollup`'s `inprog[-1]`,
|
|
# i.e. what the show page's own Resume button plays. Two episodes of one show are in progress at
|
|
# once more often than it sounds (`watch_sync._reconcile_inprogress_movies` documents that a
|
|
# scrub to the end leaves a stale `status='new', position>0` row until the daily reconcile), and
|
|
# taking opposite ends of that set is how one show came to have two different "continue" answers:
|
|
# the Home tile ranked it by its LATEST touch and then offered the EARLIEST started episode.
|
|
# `position_seconds > 0` already requires the state row to exist, so no null guard.
|
|
pick = db.execute(
|
|
base.where(PlexState.position_seconds > 0, PlexState.status.notin_(_DONE)).order_by(
|
|
PlexItem.season_number.desc().nullslast(),
|
|
PlexItem.episode_number.desc().nullslast(),
|
|
)
|
|
).first()
|
|
# Unwatched — INCLUDING an episode with no state row at all, which is the common case and the one
|
|
# a bare `status NOT IN (...)` silently drops (NULL is not false, it is unknown, so the row fails
|
|
# the filter).
|
|
pick = (
|
|
pick
|
|
or db.execute(
|
|
base.where(or_(PlexState.id.is_(None), PlexState.status.notin_(_DONE))).order_by(*asc)
|
|
).first()
|
|
)
|
|
# Fully watched show: start it again from the top.
|
|
pick = pick or db.execute(base.order_by(*asc)).first()
|
|
if pick is None:
|
|
return None
|
|
|
|
ep, st = pick
|
|
show = db.get(PlexShow, show_id)
|
|
return leaf_card(
|
|
ep, st, show_title=show.title if show else None, show_rk=show.rating_key if show else None
|
|
)
|