diff --git a/backend/.basedpyright/baseline.json b/backend/.basedpyright/baseline.json index 3bb34af2..8af825b4 100644 --- a/backend/.basedpyright/baseline.json +++ b/backend/.basedpyright/baseline.json @@ -708,108 +708,6 @@ } } ], - "./app/routes/plex/discover.py": [ - { - "code": "reportCallIssue", - "range": { - "startColumn": 4, - "endColumn": 63, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 18, - "endColumn": 48, - "lineCount": 1 - } - }, - { - "code": "reportCallIssue", - "range": { - "startColumn": 4, - "endColumn": 63, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 18, - "endColumn": 48, - "lineCount": 1 - } - } - ], - "./app/routes/plex/playback.py": [ - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 6, - "endColumn": 14, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 51, - "endColumn": 54, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 51, - "endColumn": 54, - "lineCount": 1 - } - } - ], - "./app/routes/plex/playlists.py": [ - { - "code": "reportArgumentType", - "range": { - "startColumn": 77, - "endColumn": 83, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 77, - "endColumn": 83, - "lineCount": 1 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 8, - "endColumn": 9, - "lineCount": 5 - } - }, - { - "code": "reportArgumentType", - "range": { - "startColumn": 8, - "endColumn": 9, - "lineCount": 5 - } - }, - { - "code": "reportAttributeAccessIssue", - "range": { - "startColumn": 10, - "endColumn": 18, - "lineCount": 1 - } - } - ], "./app/routes/scheduler.py": [ { "code": "reportOptionalMemberAccess", diff --git a/backend/app/discover/queries.py b/backend/app/discover/queries.py index 0ff0d40b..a3e1c96f 100644 --- a/backend/app/discover/queries.py +++ b/backend/app/discover/queries.py @@ -6,7 +6,7 @@ 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 +`plex.rollup` and the `plex.cards` serializers rather than `plex/discover.py`: that read-model 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. @@ -39,7 +39,7 @@ from app.plex.query import apply_show_state_filter, content_libs from app.userscope import owned_by # 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 +# (`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 @@ -221,7 +221,7 @@ def plex_in_progress( ) -> 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 + Mirrors `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: diff --git a/backend/app/discover/rails.py b/backend/app/discover/rails.py index 73e16c81..c4c0f7f5 100644 --- a/backend/app/discover/rails.py +++ b/backend/app/discover/rails.py @@ -76,7 +76,7 @@ def surprise( ) -> list[dict]: """The hero's draw pool: unseen things that fit the budget, from both sources under ONE seed. - Interleaved rather than concatenated — the same alternation `routes/plex/discover.py` uses for its + Interleaved rather than concatenated — the same alternation `plex/discover.py` uses for its movie/show split — so a short pool never comes back all-YouTube or all-Plex just because one side happened to be queried first. diff --git a/backend/app/plex/discover.py b/backend/app/plex/discover.py new file mode 100644 index 00000000..ac9ed64a --- /dev/null +++ b/backend/app/plex/discover.py @@ -0,0 +1,536 @@ +"""Plex Discover's reads: the rail-surface read-model. + +A consolidated payload for the dedicated Discover landing — a stack of horizontal rails: +Continue watching · Recently added · Browse-by-genre tiles · Because you watched X — plus a +separate, reshuffling Surprise-me sample. Serve-from-DB only: the recommendation rail reads the +recs the TMDB engine already persisted (`PlexItem.recommendations`), never a live TMDB call, and +every card comes from the same `movie_card`/`show_card` serializers the grid uses (so rails inherit +their `art` + rating/summary for free). The rails carry EN metadata (like the other non-grid +card paths); the HU overlay coalesces only in the main grid. + +The sidebar metadata filters (genre / rating / year / collection / …) narrow EVERY rail too, so the +Discover surface behaves like the Browse grid — the same `LibraryFilters` / `meta_filter_conds` the +grid uses, applied to each rail's movie + show query (a rail may legitimately come back empty, which +the client hides). Genre-language matches the grid; duration is movie-only. +""" + +from datetime import datetime + +from sqlalchemy import and_, func, or_, select +from sqlalchemy.orm import Session + +from app.collation import sorted_words +from app.discover.queries import RAIL_LIMIT +from app.models import User +from app.plex import rollup +from app.plex.cards import SHOW_NEW, display_title, leaf_card, movie_card, show_card +from app.plex.models import PlexItem, PlexShow, PlexState +from app.plex.query import ( + LibraryFilters, + apply_meta_filters, + apply_show_state_filter, + content_libs, + genres_col, + meta_filter_conds, + top_art_keys, +) +from app.plex.tmdb import resolve_recs +from app.userscope import owned_by + +GENRE_TILES = 12 +SURPRISE_PER_KIND = 3 # Surprise-me: at most this many movies + this many shows (6 total) + + +def _scope_libs(db: Session, scope: str) -> tuple[list[int], list[int]]: + """The movie/show library ids visible for this scope (empty list = out of scope).""" + movie_ids, show_ids = content_libs(db) + return ( + movie_ids if scope in ("movie", "both") else [], + show_ids if scope in ("show", "both") else [], + ) + + +def _movie_meta(q, p: dict, lang: str, f: LibraryFilters): + """Apply the shared metadata filters + movie-only duration to a movie query (same as the grid).""" + q = apply_meta_filters(q, PlexItem, p, lang) + if f.duration_min is not None: + q = q.where(PlexItem.duration_s >= f.duration_min) + if f.duration_max is not None: + q = q.where(PlexItem.duration_s <= f.duration_max) + return q + + +def _movie_cards(db: Session, uid: int, items: list[PlexItem], lang: str) -> list[dict]: + sts = rollup.item_state_map(db, uid, [it.id for it in items]) + return [movie_card(it, sts.get(it.id), lang=lang) for it in items] + + +def _show_cards(db: Session, uid: int, shows: list[PlexShow], lang: str) -> list[dict]: + statuses = rollup.state_map(db, uid, [s.id for s in shows]) + return [show_card(s, statuses.get(s.id, SHOW_NEW), lang=lang) for s in shows] + + +def _recently_added( + db: Session, + uid: int, + movie_ids: list[int], + show_ids: list[int], + p: dict, + lang: str, + f: LibraryFilters, +) -> list[dict]: + """Newest movies + shows, interleaved by `added_at` desc into one rail.""" + movies = ( + db.execute( + _movie_meta( + select(PlexItem).where( + PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids) + ), + p, + lang, + f, + ) + .order_by(PlexItem.added_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ) + .scalars() + .all() + if movie_ids + else [] + ) + shows = ( + db.execute( + apply_meta_filters( + select(PlexShow).where(PlexShow.library_id.in_(show_ids)), PlexShow, p, lang + ) + .order_by(PlexShow.added_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ) + .scalars() + .all() + if show_ids + else [] + ) + # Merge the two type-homogeneous lists by added_at, then serialize each side in one batch so the + # per-card state lookups stay batched (no N+1). Same `or NO_TS` key as the two rails below — a + # review found this one still on the older `(is not None, value)` form, i.e. one rule in two + # spellings, with the two comments explaining it differently. + tagged = [(m.added_at, "movie", m) for m in movies] + [(s.added_at, "show", s) for s in shows] + tagged.sort(key=lambda t: t[0] or rollup.NO_TS, reverse=True) + top = tagged[:RAIL_LIMIT] + # Key the serialized cards back by rating_key (globally unique in Plex — a movie and a show never + # share one), so the merge doesn't depend on object identity staying alive. + top_movies = [o for _, kind, o in top if kind == "movie"] + top_shows = [o for _, kind, o in top if kind == "show"] + mcards = { + m.rating_key: c + for m, c in zip(top_movies, _movie_cards(db, uid, top_movies, lang), strict=True) + } + scards = { + s.rating_key: c + for s, c in zip(top_shows, _show_cards(db, uid, top_shows, lang), strict=True) + } + return [mcards[o.rating_key] if kind == "movie" else scards[o.rating_key] for _, kind, o in top] + + +def _recently_watched( + db: Session, + uid: int, + movie_ids: list[int], + show_ids: list[int], + p: dict, + lang: str, + f: LibraryFilters, +) -> list[dict]: + """Recently watched rail: the titles the user last FULLY watched, newest first. A movie is its own + card; a show contributes the specific EPISODE watched (carrying its show title). Movies honour the + sidebar filters (like the other rails); episodes carry no filterable metadata, so they don't.""" + rows: list[tuple[datetime | None, dict]] = [] # (watched_at, card) + if movie_ids: + mq = ( + select(PlexItem, PlexState) + .join( + PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid)) + ) + .where( + PlexItem.kind == "movie", + PlexItem.library_id.in_(movie_ids), + PlexState.status == "watched", + ) + ) + for it, st in db.execute( + _movie_meta(mq, p, lang, f) + .order_by(PlexState.watched_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ).all(): + rows.append((st.watched_at, leaf_card(it, st, lang=lang))) + if show_ids: + eq = ( + select(PlexItem, PlexState, PlexShow) + .join( + PlexState, and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid)) + ) + .join(PlexShow, PlexShow.id == PlexItem.show_id) + .where( + PlexItem.kind == "episode", + PlexItem.library_id.in_(show_ids), + PlexState.status == "watched", + ) + .order_by(PlexState.watched_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ) + for ep, st, sh in db.execute(eq).all(): + card = leaf_card( + ep, st, show_title=display_title(sh, lang), show_rk=sh.rating_key, lang=lang + ) + rows.append((st.watched_at, card)) + rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True) + return [card for _, card in rows[:RAIL_LIMIT]] + + +def _continue_watching( + db: Session, + uid: int, + movie_ids: list[int], + show_ids: list[int], + p: dict, + lang: str, + f: LibraryFilters, +) -> list[dict]: + """Resume rail: in-progress movies + partially-watched shows, most-recently-progressed first. + A movie's resume time is its own `progress_updated_at`; a show's is the max across its episodes.""" + rows: list[tuple[datetime | None, dict]] = [] # (resumed_at, card) + + if movie_ids: + mq = ( + select(PlexItem, PlexState) + .join( + PlexState, + and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid)), + ) + .where( + PlexItem.kind == "movie", + PlexItem.library_id.in_(movie_ids), + PlexState.position_seconds > 0, + # A NULL status is still "started" (mirrors apply_movie_state_filter's in_progress) — + # `notin_` alone would drop it, since `NULL NOT IN (...)` is NULL, not TRUE. + or_( + PlexState.status.is_(None), + PlexState.status.notin_(["watched", "hidden"]), + ), + ) + ) + mrows = db.execute( + _movie_meta(mq, p, lang, f) + .order_by(PlexState.progress_updated_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ).all() + for it, st in mrows: + rows.append((st.progress_updated_at, movie_card(it, st, lang=lang))) + + if show_ids: + # Reuse the shared per-show episode rollup (total/watched/inprog) rather than re-inlining it, plus + # its `resumed_at` = max(progress_updated_at); the in-progress predicate comes from the same shared + # `apply_show_state_filter` the grid uses, so the rail can't drift from the grid's definition. + prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery() + sq = apply_show_state_filter( + select(PlexShow, prog.c.resumed_at, prog.c.total, prog.c.watched, prog.c.inprog).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_ids)) + srows = db.execute( + apply_meta_filters(sq, PlexShow, p, lang) + .order_by(prog.c.resumed_at.desc().nullslast()) + .limit(RAIL_LIMIT) + ).all() + for sh, resumed_at, total, watched, inprog in srows: + state = rollup.show_state(total, watched, inprog) + rows.append((resumed_at, show_card(sh, state, lang=lang))) + + rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True) + return [card for _, card in rows[:RAIL_LIMIT]] + + +def _genre_tiles( + db: Session, movie_ids: list[int], show_ids: list[int], p: dict, lang: str +) -> list[dict]: + """The most-common genres across the scope, as {value, count} tiles → tap to filter the grid. + The genre vocabulary is language-specific (coalesce(genres_hu, genres) in HU), matching the grid. + The active metadata filters narrow the counts too, so the tiles agree with the filtered rails.""" + counts: dict[str, int] = {} + for model, ids in ((PlexItem, movie_ids), (PlexShow, show_ids)): + if not ids: + continue + gcol = genres_col(model, lang) + base = select(func.jsonb_array_elements_text(gcol).label("g")).where( + model.library_id.in_(ids), + func.jsonb_typeof(gcol) == "array", + *meta_filter_conds(model, p, lang), + ) + if model is PlexItem: + base = base.where(model.kind == "movie") + expanded = base.subquery() + for g, c in db.execute(select(expanded.c.g, func.count()).group_by(expanded.c.g)).all(): + counts[g] = counts.get(g, 0) + c + alpha = {w: i for i, w in enumerate(sorted_words(db, list(counts), lang))} + top = sorted(counts.items(), key=lambda kv: (-kv[1], alpha[kv[0]]))[:GENRE_TILES] + # A 2×2 poster collage per tile: the top movies in that genre (color hash is the client fallback). + art = top_art_keys( + db, + PlexItem, + movie_ids, + genres_col(PlexItem, lang), + [g for g, _ in top], + extra_conds=(PlexItem.kind == "movie", *meta_filter_conds(PlexItem, p, lang)), + ) + return [{"value": g, "count": c, "art": art.get(g, [])} for g, c in top] + + +def _because_you_watched( + db: Session, + uid: int, + movie_ids: list[int], + show_ids: list[int], + p: dict, + lang: str, + f: LibraryFilters, +) -> dict | None: + """Recommendations anchored on the user's most-recently-watched, TMDB-enriched movie. Only the + recs that resolve to OUR library (by guid) become cards — the rest aren't actionable here. Returns + None (rail hidden) when there's no enriched watch history or none of its recs are in the library. + The metadata filters narrow the resolved recs (not the anchor pick) so the rail matches the grid.""" + if not movie_ids: + return None + anchor = ( + db.execute( + select(PlexItem) + .join( + PlexState, + and_(PlexState.item_id == PlexItem.id, owned_by(PlexState.user_id, uid)), + ) + .where( + PlexItem.kind == "movie", + PlexItem.library_id.in_(movie_ids), + PlexState.status == "watched", + PlexItem.recommendations.isnot(None), + ) + .order_by(PlexState.watched_at.desc().nullslast()) + .limit(1) + ) + .scalars() + .first() + ) + if anchor is None: + return None + resolved = resolve_recs(db, anchor.recommendations, lang=lang) + # Preserve TMDB's recommendation order, keep only in-library hits, and drop the anchor itself. + rks = [ + r["rating_key"] + for r in resolved + if r.get("in_library") and r.get("rating_key") != anchor.rating_key + ] + if not rks: + return None + movies = { + m.rating_key: m + for m in db.execute( + _movie_meta( + select(PlexItem).where(PlexItem.kind == "movie", PlexItem.rating_key.in_(rks)), + p, + lang, + f, + ) + ) + .scalars() + .all() + } + shows = { + s.rating_key: s + for s in db.execute( + apply_meta_filters( + select(PlexShow).where(PlexShow.rating_key.in_(rks)), PlexShow, p, lang + ) + ) + .scalars() + .all() + } + ordered_movies = [movies[rk] for rk in rks if rk in movies] + ordered_shows = [shows[rk] for rk in rks if rk in shows] + # rating_key is globally unique in Plex, so it keys the serialized cards back safely (no id()). + mcards = { + m.rating_key: c + for m, c in zip(ordered_movies, _movie_cards(db, uid, ordered_movies, lang), strict=True) + } + scards = { + s.rating_key: c + for s, c in zip(ordered_shows, _show_cards(db, uid, ordered_shows, lang), strict=True) + } + items: list[dict] = [] + for rk in rks: + if rk in mcards: + items.append(mcards[rk]) + elif rk in scards: + items.append(scards[rk]) + if not items: + return None + return { + "anchor": {"id": anchor.rating_key, "title": display_title(anchor, lang)}, + "items": items, + } + + +def discover_page(db: Session, user: User, scope: str, lang: str, f: LibraryFilters) -> dict: + """The Discover landing payload — every rail in one pass (serve-from-DB), narrowed by the active + sidebar filters. Empty rails come back as empty lists (the client hides them); `because_you_watched` + is null when there's no anchor. `lang` is already normalised.""" + p = f.as_dict() + movie_ids, show_ids = _scope_libs(db, scope) + return { + "scope": scope, + "continue_watching": _continue_watching(db, user.id, movie_ids, show_ids, p, lang, f), + "recently_watched": _recently_watched(db, user.id, movie_ids, show_ids, p, lang, f), + "recently_added": _recently_added(db, user.id, movie_ids, show_ids, p, lang, f), + "genres": _genre_tiles(db, movie_ids, show_ids, p, lang), + "because_you_watched": _because_you_watched(db, user.id, movie_ids, show_ids, p, lang, f), + } + + +def _draw(db: Session, stmt, key_col, per: int, exclude: list[str]) -> list: + """`per` random rows of `stmt`, the previous draw's (`exclude`) left out while the pool allows: a + reshuffle over a small filtered pool kept dealing the same few titles. What the fresh rows leave + short is topped up from the excluded ones, so a pool smaller than a draw still fills the rail.""" + fresh = ( + db.execute(stmt.where(key_col.not_in(exclude)).order_by(func.random()).limit(per)) + .scalars() + .all() + if exclude + else [] + ) + if len(fresh) == per: + return list(fresh) + taken = [key_col.not_in([r.rating_key for r in fresh])] if fresh else [] + rest = db.execute(stmt.where(*taken).order_by(func.random()).limit(per - len(fresh))).scalars() + return [*fresh, *rest.all()] + + +def _kept( + db: Session, uid: int, movie_ids: list[int], show_ids: list[int], keep: list[str], lang: str +) -> list[dict]: + """The `keep` titles' cards in their given order, as they stand now: a title watched since the draw + stays in it, marked watched, and one gone from the scope's libraries drops out.""" + movies = ( + db.execute( + select(PlexItem).where( + PlexItem.kind == "movie", + PlexItem.library_id.in_(movie_ids), + PlexItem.rating_key.in_(keep), + ) + ) + .scalars() + .all() + if movie_ids + else [] + ) + shows = ( + db.execute( + select(PlexShow).where(PlexShow.library_id.in_(show_ids), PlexShow.rating_key.in_(keep)) + ) + .scalars() + .all() + if show_ids + else [] + ) + cards = { + c["id"]: c + for c in [ + *_movie_cards(db, uid, list(movies), lang), + *_show_cards(db, uid, list(shows), lang), + ] + } + return [cards[k] for k in keep if k in cards] + + +def surprise( + db: Session, + user: User, + scope: str, + lang: str, + limit: int, + exclude: list[str], + keep: list[str], + f: LibraryFilters, +) -> dict: + """A fresh random sample for the reshuffling Surprise-me rail (a random draw each call), narrowed by + the active filters. Only titles the user has NOT seen — a watched movie / a fully-watched show is + excluded (the point is to surface something to watch). Capped at SURPRISE_PER_KIND movies + + SURPRISE_PER_KIND shows, drawn separately then interleaved so a lopsided library doesn't crowd out + one type.""" + p = f.as_dict() + movie_ids, show_ids = _scope_libs(db, scope) + if keep: + return {"scope": scope, "items": _kept(db, user.id, movie_ids, show_ids, keep, lang)} + per = min(limit, SURPRISE_PER_KIND) + movies = [] + if movie_ids: + # Exclude movies this user has watched. + watched_movie = ( + select(PlexState.id) + .where( + PlexState.item_id == PlexItem.id, + owned_by(PlexState.user_id, user), + PlexState.status == "watched", + ) + .exists() + ) + movies = _draw( + db, + _movie_meta( + select(PlexItem).where( + PlexItem.kind == "movie", + PlexItem.library_id.in_(movie_ids), + ~watched_movie, + ), + p, + lang, + f, + ), + PlexItem.rating_key, + per, + exclude, + ) + shows = [] + if show_ids: + # Exclude fully-watched shows (LEFT JOIN so a never-started show — no rollup row — still counts + # as unseen). `total IS NULL / == 0 / watched < total` == "not fully watched". + prog = rollup.agg_subquery(db, user.id).subquery() + not_fully_watched = or_( + prog.c.total.is_(None), prog.c.total == 0, prog.c.watched < prog.c.total + ) + shows = _draw( + db, + apply_meta_filters( + select(PlexShow) + .outerjoin(prog, prog.c.show_id == PlexShow.id) + .where(PlexShow.library_id.in_(show_ids), not_fully_watched), + PlexShow, + p, + lang, + ), + PlexShow.rating_key, + per, + exclude, + ) + mcards = _movie_cards(db, user.id, movies, lang) + scards = _show_cards(db, user.id, shows, lang) + # Interleave movie/show cards, then trim to `limit`. + items: list[dict] = [] + for i in range(max(len(mcards), len(scards))): + if i < len(mcards): + items.append(mcards[i]) + if i < len(scards): + items.append(scards[i]) + return {"scope": scope, "items": items[:limit]} diff --git a/backend/app/plex/queries.py b/backend/app/plex/queries.py index b8d911f0..9d1f7cb3 100644 --- a/backend/app/plex/queries.py +++ b/backend/app/plex/queries.py @@ -17,13 +17,16 @@ from app.plex.models import ( PlexCollection, PlexItem, PlexLibrary, + PlexLink, PlexPerson, + PlexPlaylist, + PlexPlaylistItem, PlexSeason, PlexShow, PlexState, ) from app.plex.query import content_libs, runtime_by_key, title_col, top_art_keys -from app.userscope import owned_by +from app.userscope import owned_by, owned_row from app.utils import like_escape # --- one row --------------------------------------------------------------------------------- @@ -47,6 +50,18 @@ def collection(db: Session, rating_key: str) -> PlexCollection | None: return by_rating_key(db, PlexCollection, rating_key) +def season(db: Session, rating_key: str) -> PlexSeason | None: + return by_rating_key(db, PlexSeason, rating_key) + + +def season_episodes(db: Session, se: PlexSeason) -> list[PlexItem]: + return list( + db.execute( + select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode") + ).scalars() + ) + + def show_by_id(db: Session, show_id: int) -> PlexShow | None: return db.get(PlexShow, show_id) @@ -83,10 +98,10 @@ def person(db: Session, name: str) -> PlexPerson | None: # --- a user's watch state -------------------------------------------------------------------- -def state(db: Session, user: User | int, it: PlexItem) -> PlexState | None: +def state(db: Session, user: User | int, item_id: int) -> PlexState | None: return ( db.execute( - select(PlexState).where(owned_by(PlexState.user_id, user), PlexState.item_id == it.id) + select(PlexState).where(owned_by(PlexState.user_id, user), PlexState.item_id == item_id) ) .scalars() .first() @@ -380,3 +395,127 @@ def collection_tiles( art_s = top_art_keys(db, PlexShow, show_ids, PlexShow.collection_keys, keys, per=2) runtime = runtime_by_key(db, PlexItem, movie_ids, PlexItem.collection_keys, keys) return art_m, art_s, runtime + + +# --- the admin page -------------------------------------------------------------------------- + + +def sample_file_path(db: Session) -> str | None: + """Any mirrored item's Plex-side file path — a sample to test the mount and path map against.""" + return db.scalar(select(PlexItem.file_path).where(PlexItem.file_path.isnot(None))) + + +def watch_link(db: Session, user: User | int) -> PlexLink | None: + """The user's Plex watch-sync link, if they have one.""" + return db.execute(select(PlexLink).where(owned_by(PlexLink.user_id, user))).scalars().first() + + +# --- a user's playlists ---------------------------------------------------------------------- + + +def own_playlist( + db: Session, user: User, playlist_id: int, *, lock: bool = False +) -> PlexPlaylist | None: + """The user's playlist with this id; someone else's is the same answer as a missing one. `lock` + takes it FOR UPDATE: every add, remove and reorder does, so two of them never read the same max + position, and the list row is always locked before its items.""" + return owned_row(db, PlexPlaylist, playlist_id, user, lock=lock) + + +def playlists_of(db: Session, user: User) -> list[PlexPlaylist]: + """The user's playlists, newest-touched first.""" + return list( + db.execute( + select(PlexPlaylist) + .where(owned_by(PlexPlaylist.user_id, user)) + .order_by(PlexPlaylist.updated_at.desc()) + ).scalars() + ) + + +def playlist_count(db: Session, playlist_id: int) -> int: + return ( + db.scalar( + select(func.count(PlexPlaylistItem.id)).where( + PlexPlaylistItem.playlist_id == playlist_id + ) + ) + or 0 + ) + + +def _playlist_rows(playlist_id: int): + return ( + select(PlexPlaylistItem, PlexItem) + .join(PlexItem, PlexItem.id == PlexPlaylistItem.item_id) + .where(PlexPlaylistItem.playlist_id == playlist_id) + .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) + ) + + +def playlist_cover(db: Session, playlist_id: int) -> str | None: + """The rating key of a playlist's first item — its cover.""" + first = db.execute(_playlist_rows(playlist_id)).first() + return first[1].rating_key if first else None + + +def playlist_items(db: Session, playlist_id: int) -> list[PlexItem]: + """A playlist's items in play order.""" + return [it for _row, it in db.execute(_playlist_rows(playlist_id)).all()] + + +def playlist_holds(db: Session, playlist_id: int, item_ids: set[int]) -> int: + """How many of these items the playlist holds.""" + if not item_ids: + return 0 + return ( + db.scalar( + select(func.count(PlexPlaylistItem.id)).where( + PlexPlaylistItem.playlist_id == playlist_id, + PlexPlaylistItem.item_id.in_(item_ids), + ) + ) + or 0 + ) + + +def item_id(db: Session, rating_key: str) -> int | None: + """The id of the mirrored item with this key. Unlike `item`, a NUL in the key reaches the query, + which the app answers as a 422 (`main._nul_in_text`), as for every other compared string.""" + return db.scalar(select(PlexItem.id).where(PlexItem.rating_key == str(rating_key))) + + +def item_ids(db: Session, rating_keys: list[str]) -> set[int]: + """The ids of the mirrored items with these keys.""" + if not rating_keys: + return set() + return set( + db.execute(select(PlexItem.id).where(PlexItem.rating_key.in_(rating_keys))).scalars() + ) + + +def items_by_keys(db: Session, rating_keys: list) -> list[PlexItem]: + """The mirrored items with these keys, in the given order; an unknown key and a repeat drop out.""" + order = [str(k) for k in rating_keys if str(k)] + if not order: + return [] + by_rk = { + it.rating_key: it + for it in db.execute(select(PlexItem).where(PlexItem.rating_key.in_(order))).scalars() + } + seen: set[str] = set() + out: list[PlexItem] = [] + for rk in order: + it = by_rk.get(rk) + if it is not None and rk not in seen: + seen.add(rk) + out.append(it) + return out + + +def shows_by_ids(db: Session, show_ids: set[int]) -> dict[int, PlexShow]: + if not show_ids: + return {} + return { + sh.id: sh for sh in db.execute(select(PlexShow).where(PlexShow.id.in_(show_ids))).scalars() + } diff --git a/backend/app/plex/rollup.py b/backend/app/plex/rollup.py index 69cb9762..78144fac 100644 --- a/backend/app/plex/rollup.py +++ b/backend/app/plex/rollup.py @@ -8,6 +8,7 @@ from datetime import datetime, timezone from sqlalchemy import and_, case, func, or_, select from sqlalchemy.orm import Session, aliased +from app.userscope import owned_by from app.plex.models import PlexItem, PlexState from app.plex.cards import ShowState @@ -92,7 +93,7 @@ def agg_subquery(db: Session, user_id: int, include_resumed: bool = False): ) return ( select(*cols) - .outerjoin(ps, and_(ps.item_id == PlexItem.id, ps.user_id == user_id)) + .outerjoin(ps, and_(ps.item_id == PlexItem.id, owned_by(ps.user_id, user_id))) .where(PlexItem.kind == "episode", PlexItem.show_id.isnot(None)) .group_by(PlexItem.show_id) ) @@ -114,6 +115,8 @@ def item_state_map(db: Session, user_id: int, item_ids: list[int]) -> dict[int, return { s.item_id: s for s in db.execute( - select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id.in_(item_ids)) + select(PlexState).where( + owned_by(PlexState.user_id, user_id), PlexState.item_id.in_(item_ids) + ) ).scalars() } diff --git a/backend/app/plex/service.py b/backend/app/plex/service.py index 70e5154f..0ca9ae70 100644 --- a/backend/app/plex/service.py +++ b/backend/app/plex/service.py @@ -8,6 +8,8 @@ after a write-back, its local cleanup after a delete. from datetime import datetime, timezone +from sqlalchemy import CursorResult, delete, func, select, update +from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app import logs @@ -16,7 +18,20 @@ from app.plex import detail as plex_detail from app.plex import people as plex_people from app.plex import tmdb as plex_tmdb from app.plex.lookups import by_rating_key -from app.plex.models import PlexCollection, PlexItem, PlexLibrary, PlexPerson, PlexShow +from app.models import User +from app.plex import queries +from app.plex.models import ( + PlexCollection, + PlexItem, + PlexLibrary, + PlexLink, + PlexPerson, + PlexPlaylist, + PlexPlaylistItem, + PlexShow, + PlexState, +) +from app.userscope import owned_by log = logs.get_logger(__name__) @@ -79,3 +94,295 @@ def enrich_person_bg(person_id: int) -> None: log.warning("background person bio enrich failed for id=%s", person_id, exc_info=True) finally: db.close() + + +def set_watch_link(db: Session, user: User, enabled: bool) -> PlexLink: + """Turn the user's two-way watch sync on or off, creating their link on first use. The link uses + the server admin token (`uses_admin`), so it needs no Plex login of its own.""" + link = queries.watch_link(db, user) + if link is None: + link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc)) + db.add(link) + link.sync_enabled = enabled + db.flush() + return link + + +# --- a user's playlists ---------------------------------------------------------------------- + + +def create_playlist(db: Session, user: User, title: str, seed: PlexItem | None) -> PlexPlaylist: + """A new playlist of the user's, holding `seed` when one is given.""" + pl = PlexPlaylist(user_id=user.id, title=title) + db.add(pl) + db.flush() + if seed is not None: + db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=seed.id, position=0)) + db.flush() + return pl + + +def rename_playlist(pl: PlexPlaylist, title: str) -> None: + pl.title = title + + +def delete_playlist(db: Session, pl: PlexPlaylist) -> None: + db.delete(pl) + db.flush() + + +def _touch(pl: PlexPlaylist) -> None: + pl.updated_at = datetime.now(timezone.utc) + + +def _last_position(db: Session, playlist_id: int) -> int: + return db.execute( + select(func.coalesce(func.max(PlexPlaylistItem.position), -1)).where( + PlexPlaylistItem.playlist_id == playlist_id + ) + ).scalar_one() + + +def add_item(db: Session, pl: PlexPlaylist, it: PlexItem) -> None: + """Append `it` to the (locked) playlist; an item already in it stays where it is.""" + exists = db.execute( + select(PlexPlaylistItem.id).where( + PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id == it.id + ) + ).first() + if exists is None: + pos = _last_position(db, pl.id) + 1 + db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=pos)) + _touch(pl) + db.flush() + + +def remove_item(db: Session, pl: PlexPlaylist, it: PlexItem) -> None: + db.execute( + delete(PlexPlaylistItem).where( + PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id == it.id + ) + ) + _touch(pl) + db.flush() + + +def add_items(db: Session, pl: PlexPlaylist, items: list[PlexItem]) -> int: + """Append many items in their given order (a whole season or show); those already in the list + are skipped. Says how many were added.""" + existing = set( + db.execute( + select(PlexPlaylistItem.item_id).where(PlexPlaylistItem.playlist_id == pl.id) + ).scalars() + ) + pos = _last_position(db, pl.id) + added = 0 + for it in items: + if it.id in existing: + continue + pos += 1 + db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=pos)) + existing.add(it.id) + added += 1 + if added: + _touch(pl) + db.flush() + return added + + +def insert_items_at(db: Session, pl: PlexPlaylist, placed: list[tuple[int, PlexItem]]) -> int: + """Insert new rows at their indices (ascending, so each index is where the row stood when the + rows before it were back), renumber the whole list 0..n-1, and say how many were added. An item + already in the list does not move: the add stays idempotent.""" + rows = list( + db.execute( + select(PlexPlaylistItem) + .where(PlexPlaylistItem.playlist_id == pl.id) + .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) + ).scalars() + ) + present = {r.item_id for r in rows} + added = 0 + for at, it in sorted(placed, key=lambda p: p[0]): + if it.id in present: + continue + row = PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=0) + db.add(row) + rows.insert(min(at, len(rows)), row) + present.add(it.id) + added += 1 + for i, r in enumerate(rows): + r.position = i + if added: + _touch(pl) + db.flush() + return added + + +def remove_items(db: Session, pl: PlexPlaylist, item_ids: list[int]) -> int: + """Remove these items from the playlist; says how many were there.""" + if not item_ids: + return 0 + result = db.execute( + delete(PlexPlaylistItem).where( + PlexPlaylistItem.playlist_id == pl.id, PlexPlaylistItem.item_id.in_(item_ids) + ), + execution_options={"synchronize_session": False}, + ) + removed = result.rowcount if isinstance(result, CursorResult) else 0 + if removed: + _touch(pl) + db.flush() + return removed + + +def reorder_playlist(db: Session, pl: PlexPlaylist, rating_keys: list[str]) -> None: + """Set the play order from an explicit list of item keys (0-based positions).""" + id_by_rk = { + it.rating_key: it.id + for it in db.execute( + select(PlexItem).where(PlexItem.rating_key.in_(rating_keys or [""])) + ).scalars() + } + rows = { + r.item_id: r + for r in db.execute( + select(PlexPlaylistItem).where(PlexPlaylistItem.playlist_id == pl.id) + ).scalars() + } + pos = 0 + seen: set[int] = set() + for rk in rating_keys: + row = rows.get(id_by_rk.get(rk, -1)) + if row is not None and row.item_id not in seen: + row.position = pos + pos += 1 + seen.add(row.item_id) + # Rows the payload omitted (added in another tab meanwhile) keep their relative order after + # the given ones, so no two rows share a position (the YouTube reorder does the same). + for row in sorted( + (r for iid, r in rows.items() if iid not in seen), key=lambda r: (r.position, r.id) + ): + row.position = pos + pos += 1 + _touch(pl) + db.flush() + + +# --- a user's watch state -------------------------------------------------------------------- + + +def mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool: + """Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end + checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a + read-then-write let both see "not watched" and both scrobble. The transition is decided by the + write itself: an insert that did not conflict, or an update conditional on the old status — + under READ COMMITTED the second writer re-reads the committed row and matches nothing.""" + values = { + "status": "watched", + "watched_at": now, + "position_seconds": 0, + "progress_updated_at": now, + "synced_to_plex": False, + } + inserted = db.execute( + pg_insert(PlexState) + .values(user_id=user_id, item_id=item_id, **values) + .on_conflict_do_nothing(constraint="uq_plex_user_item") + .returning(PlexState.id) + ).first() + if inserted is not None: + return True + mine = owned_by(PlexState.user_id, user_id) & (PlexState.item_id == item_id) + result = db.execute(update(PlexState).where(mine, PlexState.status != "watched").values(values)) + if isinstance(result, CursorResult) and result.rowcount: + return True + # Already watched: a re-finish still restamps the row, as it always did — just no scrobble. + db.execute(update(PlexState).where(mine).values(values)) + return False + + +def clear_progress(db: Session, st: PlexState | None, now: datetime) -> None: + """A checkpoint too early to keep: a state that held nothing else goes, any other forgets its + position.""" + if st is not None and st.status == "new": + db.delete(st) + elif st is not None: + st.position_seconds = 0 + st.progress_updated_at = now + db.flush() + + +def checkpoint( + db: Session, + user: User, + item_id: int, + st: PlexState | None, + position: int, + now: datetime, + *, + dirty: bool, +) -> PlexState: + """Record a resume position. `dirty` marks it not yet mirrored to Plex — set before a push, so + the reconcile re-pushes it if the push fails and the push can settle it once Plex confirms.""" + if st is None: + st = PlexState(user_id=user.id, item_id=item_id) + db.add(st) + st.position_seconds = position + st.progress_updated_at = now + if dirty: + st.synced_to_plex = False + db.flush() + return st + + +def forget_state(db: Session, st: PlexState) -> None: + db.delete(st) + db.flush() + + +def set_status( + db: Session, user: User, item_id: int, st: PlexState | None, status: str +) -> PlexState: + """Set a watched or hidden status; watched also clears the resume point.""" + if st is None: + st = PlexState(user_id=user.id, item_id=item_id) + db.add(st) + st.status = status + if status == "watched": + st.watched_at = datetime.now(timezone.utc) + st.position_seconds = 0 + st.synced_to_plex = False + db.flush() + return st + + +def bulk_state( + db: Session, user_id: int, eps: list[PlexItem], watched: bool +) -> list[tuple[int, str, str]]: + """Mark every episode in `eps` watched (or unwatched) for a user and say what changed, as the + (item id, rating key, action) pushes a linked Plex account needs. Unwatching keeps the user's + private "hidden" marks.""" + now = datetime.now(timezone.utc) + existing = queries.states(db, user_id, [e.id for e in eps]) + to_push: list[tuple[int, str, str]] = [] + for e in eps: + st = existing.get(e.id) + prev = st.status if st is not None else "new" + if watched: + if st is None: + st = PlexState(user_id=user_id, item_id=e.id) + db.add(st) + st.status = "watched" + st.watched_at = now + st.position_seconds = 0 + st.synced_to_plex = False + if prev != "watched": + to_push.append((e.id, e.rating_key, "watched")) + else: # → new: drop any state (but keep the user's private "hidden" marks) + if st is not None and prev != "hidden": + db.delete(st) + if prev != "new": + to_push.append((e.id, e.rating_key, "unwatched")) + db.flush() + return to_push diff --git a/backend/app/plex/tmdb.py b/backend/app/plex/tmdb.py index 71e8075c..ea17bd5d 100644 --- a/backend/app/plex/tmdb.py +++ b/backend/app/plex/tmdb.py @@ -619,9 +619,9 @@ def enrich_person(db: Session, person: PlexPerson) -> bool: if not person.tmdb_person_id or not sysconfig.get_str(db, "tmdb_api_key"): return False try: - # Called from a background task (off the request path — see people._enrich_person_bg), so it no - # longer blocks a serve; still cap the outbound TMDB call with a short timeout so a hanging TMDB - # doesn't tie up a threadpool worker for the default 30s. + # Called from a background task (off the request path — see `plex.service.enrich_person_bg`), + # so it no longer blocks a serve; still cap the outbound TMDB call with a short timeout so a + # hanging TMDB doesn't tie up a threadpool worker for the default 30s. with TmdbClient(db, timeout=8.0) as tmdb: _fetch_person_into(tmdb, person) except TmdbError as exc: diff --git a/backend/app/plex/watch_sync.py b/backend/app/plex/watch_sync.py index f76dfa7c..5f6487ef 100644 --- a/backend/app/plex/watch_sync.py +++ b/backend/app/plex/watch_sync.py @@ -29,6 +29,7 @@ from sqlalchemy.orm import Session from app import logs, sysconfig from app.db import SessionLocal +from app.userscope import owned_by from app.plex.models import PlexItem, PlexLibrary, PlexLink, PlexState from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.plex.sync import ( @@ -196,7 +197,7 @@ def link_for_push(db: Session, user_id: int) -> PlexLink | None: """The active owner watch-push link for this user, or None when push is off / not the owner. Push is gated on `initial_import_done` so we never push a half-known local state over Plex's authoritative one before the one-time "Plex is master" import has reconciled them.""" - link = db.execute(select(PlexLink).where(PlexLink.user_id == user_id)).scalars().first() + link = db.execute(select(PlexLink).where(owned_by(PlexLink.user_id, user_id))).scalars().first() if link and link.uses_admin and link.sync_enabled and link.initial_import_done: return link return None @@ -253,7 +254,9 @@ def push_state_to_plex( # re-edited since we captured `expect_ts` (leave it dirty for its own push / the reconcile). st = ( db.execute( - select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + select(PlexState).where( + owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id + ) ) .scalars() .first() @@ -318,7 +321,7 @@ def push_bulk_state_to_plex(user_id: int, changes: list[tuple[int, str, str]]) - if synced_ids: db.execute( update(PlexState) - .where(PlexState.user_id == user_id, PlexState.item_id.in_(synced_ids)) + .where(owned_by(PlexState.user_id, user_id), PlexState.item_id.in_(synced_ids)) .values(synced_to_plex=True) ) db.commit() @@ -393,7 +396,9 @@ def _pull_apply( full reconcile's job) — this only pulls in watched/resume.""" st = ( db.execute( - select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) + select(PlexState).where( + owned_by(PlexState.user_id, user_id), PlexState.item_id == item_id + ) ) .scalars() .first() @@ -449,7 +454,7 @@ def _repush_dirty(db: Session, plex: PlexClient, link: PlexLink) -> int: dirty = ( db.execute( select(PlexState).where( - PlexState.user_id == link.user_id, + owned_by(PlexState.user_id, link.user_id), PlexState.synced_to_plex.is_(False), PlexState.status != "hidden", ) @@ -505,7 +510,7 @@ def _reconcile_inprogress_movies( and aren't affected). Note: like the rest of this job, `metadata()` reads the TOKEN account (the owner) — a HTPC playing under a managed Plex user (accountID>1) is out of scope here as elsewhere.""" conds = [ - PlexState.user_id == link.user_id, + owned_by(PlexState.user_id, link.user_id), PlexState.status.notin_(("watched", "hidden")), PlexState.position_seconds > _PROGRESS_MIN_S, PlexItem.kind == "movie", @@ -627,7 +632,7 @@ def _clear_unwatched(db: Session, user_id: int, plex_states: dict, settled: set[ select(PlexState) .join(PlexItem, PlexItem.id == PlexState.item_id) .join(PlexLibrary, PlexLibrary.id == PlexItem.library_id) - .where(PlexState.user_id == user_id, PlexLibrary.plex_key.in_(settled)) + .where(owned_by(PlexState.user_id, user_id), PlexLibrary.plex_key.in_(settled)) ) for st in rows.scalars().all(): if st.item_id in plex_states or st.status == "hidden" or not st.synced_to_plex: diff --git a/backend/app/routes/plex/admin.py b/backend/app/routes/plex/admin.py index 8387f496..cd8b27fc 100644 --- a/backend/app/routes/plex/admin.py +++ b/backend/app/routes/plex/admin.py @@ -1,24 +1,26 @@ """Plex admin: connection test, catalog sync, and owner watch-link config.""" -from datetime import datetime, timezone +from typing import TYPE_CHECKING from fastapi import APIRouter, Depends from pydantic import BaseModel -from sqlalchemy import select from sqlalchemy.orm import Session from app import logs, sysconfig from app.auth import admin_user, current_user from app.db import get_db from app.models import User -from app.plex.models import PlexItem, PlexLink from app.plex import paths as plex_paths +from app.plex import queries, service from app.plex import sync as plex_sync from app.plex import watch_sync as plex_watch from app.plex.client import PlexClient, PlexError, PlexNotConfigured from app.utils import iso from app.refusal import refused +if TYPE_CHECKING: + from app.plex.models import PlexLink + log = logs.get_logger(__name__) router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -66,7 +68,7 @@ def _sample_plex_path(db: Session, plex: PlexClient, sections: list[dict]) -> st """A real Plex-side media file path to validate the mount + path-map against. Prefers the already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled section straight from Plex.""" - fp = db.scalar(select(PlexItem.file_path).where(PlexItem.file_path.isnot(None))) + fp = queries.sample_file_path(db) if fp: return fp wanted = plex_sync._enabled_section_keys(db) @@ -103,7 +105,7 @@ def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) - return {"ok": True, **plex_sync.sync(db)} -def _link_dict(link: PlexLink | None) -> dict: +def _link_dict(link: "PlexLink | None") -> dict: return { "linked": link is not None, "sync_enabled": bool(link and link.sync_enabled), @@ -116,8 +118,7 @@ def _link_dict(link: PlexLink | None) -> dict: @router.get("/watch/link") def watch_link_status(user: User = Depends(current_user), db: Session = Depends(get_db)) -> dict: """This user's Plex watch-sync link status (drives the Settings toggle).""" - link = db.execute(select(PlexLink).where(PlexLink.user_id == user.id)).scalars().first() - return _link_dict(link) + return _link_dict(queries.watch_link(db, user)) class WatchLinkIn(BaseModel): @@ -136,11 +137,7 @@ def watch_link_set( if not sysconfig.get_bool(db, "plex_enabled"): raise refused(409, "plex_disabled") enabled = body.enabled - link = db.execute(select(PlexLink).where(PlexLink.user_id == user.id)).scalars().first() - if link is None: - link = PlexLink(user_id=user.id, uses_admin=True, linked_at=datetime.now(timezone.utc)) - db.add(link) - link.sync_enabled = enabled + link = service.set_watch_link(db, user, enabled) db.commit() result: dict = {} @@ -153,7 +150,7 @@ def watch_link_set( def watch_import_now(user: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict: """Admin: (re-)run the "Plex is master" import for this account — Plex re-wins on the intersection; Siftlode-only states are preserved. Idempotent.""" - link = db.execute(select(PlexLink).where(PlexLink.user_id == user.id)).scalars().first() + link = queries.watch_link(db, user) if link is None or not link.sync_enabled: raise refused(409, "watch_sync_off") result = plex_watch.import_owner_watch_state(db, link) diff --git a/backend/app/routes/plex/detail.py b/backend/app/routes/plex/detail.py index 5aa1b0e0..10eaf028 100644 --- a/backend/app/routes/plex/detail.py +++ b/backend/app/routes/plex/detail.py @@ -309,7 +309,7 @@ def item_detail( prev/next, and the per-user resume position + watch status.""" lang = _norm_lang(lang) it = item_or_404(db, rating_key) - st = queries.state(db, user, it) + st = queries.state(db, user, it.id) # Markers + audio/subtitle streams are file-specific, so they stay LIVE; the rich detail (cast/IMDb/ # tagline/genres) is served from the DB row for movies (persisted below / by the backfill) and only diff --git a/backend/app/routes/plex/discover.py b/backend/app/routes/plex/discover.py index 1fd57675..2e1aa3f8 100644 --- a/backend/app/routes/plex/discover.py +++ b/backend/app/routes/plex/discover.py @@ -1,392 +1,18 @@ -"""Plex Discover: the rail-surface read-model. - -A consolidated payload for the dedicated Discover landing — a stack of horizontal rails: -Continue watching · Recently added · Browse-by-genre tiles · Because you watched X — plus a -separate, reshuffling Surprise-me sample. Serve-from-DB only: the recommendation rail reads the -recs the TMDB engine already persisted (`PlexItem.recommendations`), never a live TMDB call, and -every card comes from the same `movie_card`/`show_card` serializers the grid uses (so rails inherit -their `art` + rating/summary for free). The rails carry EN metadata (like the other non-grid -card paths); the HU overlay coalesces only in the main grid. - -The sidebar metadata filters (genre / rating / year / collection / …) narrow EVERY rail too, so the -Discover surface behaves like the Browse grid — the same `LibraryFilters` / `meta_filter_conds` the -grid uses, applied to each rail's movie + show query (a rail may legitimately come back empty, which -the client hides). Genre-language matches the grid; duration is movie-only. -""" +"""Plex Discover: the rail-surface landing and its Surprise-me draw (`plex.discover`).""" from fastapi import APIRouter, Depends, Query -from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session -from app.collation import sorted_words from app.auth import current_user from app.db import get_db from app.models import User -from app.plex.models import PlexItem, PlexShow, PlexState -from app.plex import rollup -from app.plex.cards import ( - SHOW_NEW, - display_title, - leaf_card, - movie_card, - show_card, -) -from app.plex.query import ( - LibraryFilters, - apply_meta_filters, - apply_show_state_filter, - content_libs, - genres_col, - meta_filter_conds, - top_art_keys, -) -from app.discover.queries import RAIL_LIMIT -from app.plex.tmdb import resolve_recs +from app.plex import discover as plex_discover +from app.plex.discover import SURPRISE_PER_KIND +from app.plex.query import LibraryFilters from app.routes.plex._shared import _norm_lang router = APIRouter(prefix="/api/plex", tags=["plex"]) -GENRE_TILES = 12 -SURPRISE_PER_KIND = 3 # Surprise-me: at most this many movies + this many shows (6 total) - - -def _scope_libs(db: Session, scope: str) -> tuple[list[int], list[int]]: - """The movie/show library ids visible for this scope (empty list = out of scope).""" - movie_ids, show_ids = content_libs(db) - return ( - movie_ids if scope in ("movie", "both") else [], - show_ids if scope in ("show", "both") else [], - ) - - -def _movie_meta(q, p: dict, lang: str, f: LibraryFilters): - """Apply the shared metadata filters + movie-only duration to a movie query (same as the grid).""" - q = apply_meta_filters(q, PlexItem, p, lang) - if f.duration_min is not None: - q = q.where(PlexItem.duration_s >= f.duration_min) - if f.duration_max is not None: - q = q.where(PlexItem.duration_s <= f.duration_max) - return q - - -def _movie_cards(db: Session, uid: int, items: list[PlexItem], lang: str) -> list[dict]: - sts = rollup.item_state_map(db, uid, [it.id for it in items]) - return [movie_card(it, sts.get(it.id), lang=lang) for it in items] - - -def _show_cards(db: Session, uid: int, shows: list[PlexShow], lang: str) -> list[dict]: - statuses = rollup.state_map(db, uid, [s.id for s in shows]) - return [show_card(s, statuses.get(s.id, SHOW_NEW), lang=lang) for s in shows] - - -def _recently_added( - db: Session, - uid: int, - movie_ids: list[int], - show_ids: list[int], - p: dict, - lang: str, - f: LibraryFilters, -) -> list[dict]: - """Newest movies + shows, interleaved by `added_at` desc into one rail.""" - movies = ( - db.execute( - _movie_meta( - select(PlexItem).where( - PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids) - ), - p, - lang, - f, - ) - .order_by(PlexItem.added_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ) - .scalars() - .all() - if movie_ids - else [] - ) - shows = ( - db.execute( - apply_meta_filters( - select(PlexShow).where(PlexShow.library_id.in_(show_ids)), PlexShow, p, lang - ) - .order_by(PlexShow.added_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ) - .scalars() - .all() - if show_ids - else [] - ) - # Merge the two type-homogeneous lists by added_at, then serialize each side in one batch so the - # per-card state lookups stay batched (no N+1). Same `or NO_TS` key as the two rails below — a - # review found this one still on the older `(is not None, value)` form, i.e. one rule in two - # spellings, with the two comments explaining it differently. - tagged = [(m.added_at, "movie", m) for m in movies] + [(s.added_at, "show", s) for s in shows] - tagged.sort(key=lambda t: t[0] or rollup.NO_TS, reverse=True) - top = tagged[:RAIL_LIMIT] - # Key the serialized cards back by rating_key (globally unique in Plex — a movie and a show never - # share one), so the merge doesn't depend on object identity staying alive. - top_movies = [o for _, kind, o in top if kind == "movie"] - top_shows = [o for _, kind, o in top if kind == "show"] - mcards = { - m.rating_key: c - for m, c in zip(top_movies, _movie_cards(db, uid, top_movies, lang), strict=True) - } - scards = { - s.rating_key: c - for s, c in zip(top_shows, _show_cards(db, uid, top_shows, lang), strict=True) - } - return [mcards[o.rating_key] if kind == "movie" else scards[o.rating_key] for _, kind, o in top] - - -def _recently_watched( - db: Session, - uid: int, - movie_ids: list[int], - show_ids: list[int], - p: dict, - lang: str, - f: LibraryFilters, -) -> list[dict]: - """Recently watched rail: the titles the user last FULLY watched, newest first. A movie is its own - card; a show contributes the specific EPISODE watched (carrying its show title). Movies honour the - sidebar filters (like the other rails); episodes carry no filterable metadata, so they don't.""" - rows: list[tuple[object, dict]] = [] # (watched_at, card) - if movie_ids: - 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_ids), - PlexState.status == "watched", - ) - ) - for it, st in db.execute( - _movie_meta(mq, p, lang, f) - .order_by(PlexState.watched_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ).all(): - rows.append((st.watched_at, leaf_card(it, st, lang=lang))) - if show_ids: - eq = ( - select(PlexItem, PlexState, PlexShow) - .join(PlexState, and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid)) - .join(PlexShow, PlexShow.id == PlexItem.show_id) - .where( - PlexItem.kind == "episode", - PlexItem.library_id.in_(show_ids), - PlexState.status == "watched", - ) - .order_by(PlexState.watched_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ) - for ep, st, sh in db.execute(eq).all(): - card = leaf_card( - ep, st, show_title=display_title(sh, lang), show_rk=sh.rating_key, lang=lang - ) - rows.append((st.watched_at, card)) - rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True) - return [card for _, card in rows[:RAIL_LIMIT]] - - -def _continue_watching( - db: Session, - uid: int, - movie_ids: list[int], - show_ids: list[int], - p: dict, - lang: str, - f: LibraryFilters, -) -> list[dict]: - """Resume rail: in-progress movies + partially-watched shows, most-recently-progressed first. - A movie's resume time is its own `progress_updated_at`; a show's is the max across its episodes.""" - rows: list[tuple[object, dict]] = [] # (resumed_at, card) - - if movie_ids: - 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_ids), - PlexState.position_seconds > 0, - # A NULL status is still "started" (mirrors apply_movie_state_filter's in_progress) — - # `notin_` alone would drop it, since `NULL NOT IN (...)` is NULL, not TRUE. - or_( - PlexState.status.is_(None), - PlexState.status.notin_(["watched", "hidden"]), - ), - ) - ) - mrows = db.execute( - _movie_meta(mq, p, lang, f) - .order_by(PlexState.progress_updated_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ).all() - for it, st in mrows: - rows.append((st.progress_updated_at, movie_card(it, st, lang=lang))) - - if show_ids: - # Reuse the shared per-show episode rollup (total/watched/inprog) rather than re-inlining it, plus - # its `resumed_at` = max(progress_updated_at); the in-progress predicate comes from the same shared - # `apply_show_state_filter` the grid uses, so the rail can't drift from the grid's definition. - prog = rollup.agg_subquery(db, uid, include_resumed=True).subquery() - sq = apply_show_state_filter( - select(PlexShow, prog.c.resumed_at, prog.c.total, prog.c.watched, prog.c.inprog).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_ids)) - srows = db.execute( - apply_meta_filters(sq, PlexShow, p, lang) - .order_by(prog.c.resumed_at.desc().nullslast()) - .limit(RAIL_LIMIT) - ).all() - for sh, resumed_at, total, watched, inprog in srows: - state = rollup.show_state(total, watched, inprog) - rows.append((resumed_at, show_card(sh, state, lang=lang))) - - rows.sort(key=lambda r: r[0] or rollup.NO_TS, reverse=True) - return [card for _, card in rows[:RAIL_LIMIT]] - - -def _genre_tiles( - db: Session, movie_ids: list[int], show_ids: list[int], p: dict, lang: str -) -> list[dict]: - """The most-common genres across the scope, as {value, count} tiles → tap to filter the grid. - The genre vocabulary is language-specific (coalesce(genres_hu, genres) in HU), matching the grid. - The active metadata filters narrow the counts too, so the tiles agree with the filtered rails.""" - counts: dict[str, int] = {} - for model, ids in ((PlexItem, movie_ids), (PlexShow, show_ids)): - if not ids: - continue - gcol = genres_col(model, lang) - base = select(func.jsonb_array_elements_text(gcol).label("g")).where( - model.library_id.in_(ids), - func.jsonb_typeof(gcol) == "array", - *meta_filter_conds(model, p, lang), - ) - if model is PlexItem: - base = base.where(model.kind == "movie") - expanded = base.subquery() - for g, c in db.execute(select(expanded.c.g, func.count()).group_by(expanded.c.g)).all(): - counts[g] = counts.get(g, 0) + c - alpha = {w: i for i, w in enumerate(sorted_words(db, list(counts), lang))} - top = sorted(counts.items(), key=lambda kv: (-kv[1], alpha[kv[0]]))[:GENRE_TILES] - # A 2×2 poster collage per tile: the top movies in that genre (color hash is the client fallback). - art = top_art_keys( - db, - PlexItem, - movie_ids, - genres_col(PlexItem, lang), - [g for g, _ in top], - extra_conds=(PlexItem.kind == "movie", *meta_filter_conds(PlexItem, p, lang)), - ) - return [{"value": g, "count": c, "art": art.get(g, [])} for g, c in top] - - -def _because_you_watched( - db: Session, - uid: int, - movie_ids: list[int], - show_ids: list[int], - p: dict, - lang: str, - f: LibraryFilters, -) -> dict | None: - """Recommendations anchored on the user's most-recently-watched, TMDB-enriched movie. Only the - recs that resolve to OUR library (by guid) become cards — the rest aren't actionable here. Returns - None (rail hidden) when there's no enriched watch history or none of its recs are in the library. - The metadata filters narrow the resolved recs (not the anchor pick) so the rail matches the grid.""" - if not movie_ids: - return None - anchor = ( - db.execute( - select(PlexItem) - .join( - PlexState, - and_(PlexState.item_id == PlexItem.id, PlexState.user_id == uid), - ) - .where( - PlexItem.kind == "movie", - PlexItem.library_id.in_(movie_ids), - PlexState.status == "watched", - PlexItem.recommendations.isnot(None), - ) - .order_by(PlexState.watched_at.desc().nullslast()) - .limit(1) - ) - .scalars() - .first() - ) - if anchor is None: - return None - resolved = resolve_recs(db, anchor.recommendations, lang=lang) - # Preserve TMDB's recommendation order, keep only in-library hits, and drop the anchor itself. - rks = [ - r["rating_key"] - for r in resolved - if r.get("in_library") and r.get("rating_key") != anchor.rating_key - ] - if not rks: - return None - movies = { - m.rating_key: m - for m in db.execute( - _movie_meta( - select(PlexItem).where(PlexItem.kind == "movie", PlexItem.rating_key.in_(rks)), - p, - lang, - f, - ) - ) - .scalars() - .all() - } - shows = { - s.rating_key: s - for s in db.execute( - apply_meta_filters( - select(PlexShow).where(PlexShow.rating_key.in_(rks)), PlexShow, p, lang - ) - ) - .scalars() - .all() - } - ordered_movies = [movies[rk] for rk in rks if rk in movies] - ordered_shows = [shows[rk] for rk in rks if rk in shows] - # rating_key is globally unique in Plex, so it keys the serialized cards back safely (no id()). - mcards = { - m.rating_key: c - for m, c in zip(ordered_movies, _movie_cards(db, uid, ordered_movies, lang), strict=True) - } - scards = { - s.rating_key: c - for s, c in zip(ordered_shows, _show_cards(db, uid, ordered_shows, lang), strict=True) - } - items: list[dict] = [] - for rk in rks: - if rk in mcards: - items.append(mcards[rk]) - elif rk in scards: - items.append(scards[rk]) - if not items: - return None - return { - "anchor": {"id": anchor.rating_key, "title": display_title(anchor, lang)}, - "items": items, - } - @router.get("/discover") def discover( @@ -399,72 +25,7 @@ def discover( """The Discover landing payload — every rail in one pass (serve-from-DB), narrowed by the active sidebar filters. Empty rails come back as empty lists (the client hides them); `because_you_watched` is null when there's no anchor.""" - lang = _norm_lang(lang) - p = f.as_dict() - movie_ids, show_ids = _scope_libs(db, scope) - return { - "scope": scope, - "continue_watching": _continue_watching(db, user.id, movie_ids, show_ids, p, lang, f), - "recently_watched": _recently_watched(db, user.id, movie_ids, show_ids, p, lang, f), - "recently_added": _recently_added(db, user.id, movie_ids, show_ids, p, lang, f), - "genres": _genre_tiles(db, movie_ids, show_ids, p, lang), - "because_you_watched": _because_you_watched(db, user.id, movie_ids, show_ids, p, lang, f), - } - - -def _draw(db: Session, stmt, key_col, per: int, exclude: list[str]) -> list: - """`per` random rows of `stmt`, the previous draw's (`exclude`) left out while the pool allows: a - reshuffle over a small filtered pool kept dealing the same few titles. What the fresh rows leave - short is topped up from the excluded ones, so a pool smaller than a draw still fills the rail.""" - fresh = ( - db.execute(stmt.where(key_col.not_in(exclude)).order_by(func.random()).limit(per)) - .scalars() - .all() - if exclude - else [] - ) - if len(fresh) == per: - return list(fresh) - taken = [key_col.not_in([r.rating_key for r in fresh])] if fresh else [] - rest = db.execute(stmt.where(*taken).order_by(func.random()).limit(per - len(fresh))).scalars() - return [*fresh, *rest.all()] - - -def _kept( - db: Session, uid: int, movie_ids: list[int], show_ids: list[int], keep: list[str], lang: str -) -> list[dict]: - """The `keep` titles' cards in their given order, as they stand now: a title watched since the draw - stays in it, marked watched, and one gone from the scope's libraries drops out.""" - movies = ( - db.execute( - select(PlexItem).where( - PlexItem.kind == "movie", - PlexItem.library_id.in_(movie_ids), - PlexItem.rating_key.in_(keep), - ) - ) - .scalars() - .all() - if movie_ids - else [] - ) - shows = ( - db.execute( - select(PlexShow).where(PlexShow.library_id.in_(show_ids), PlexShow.rating_key.in_(keep)) - ) - .scalars() - .all() - if show_ids - else [] - ) - cards = { - c["id"]: c - for c in [ - *_movie_cards(db, uid, list(movies), lang), - *_show_cards(db, uid, list(shows), lang), - ] - } - return [cards[k] for k in keep if k in cards] + return plex_discover.discover_page(db, user, scope, _norm_lang(lang), f) @router.get("/surprise") @@ -486,69 +47,4 @@ def surprise( excluded (the point is to surface something to watch). Capped at SURPRISE_PER_KIND movies + SURPRISE_PER_KIND shows, drawn separately then interleaved so a lopsided library doesn't crowd out one type.""" - lang = _norm_lang(lang) - p = f.as_dict() - movie_ids, show_ids = _scope_libs(db, scope) - if keep: - return {"scope": scope, "items": _kept(db, user.id, movie_ids, show_ids, keep, lang)} - per = min(limit, SURPRISE_PER_KIND) - movies = [] - if movie_ids: - # Exclude movies this user has watched. - watched_movie = ( - select(PlexState.id) - .where( - PlexState.item_id == PlexItem.id, - PlexState.user_id == user.id, - PlexState.status == "watched", - ) - .exists() - ) - movies = _draw( - db, - _movie_meta( - select(PlexItem).where( - PlexItem.kind == "movie", - PlexItem.library_id.in_(movie_ids), - ~watched_movie, - ), - p, - lang, - f, - ), - PlexItem.rating_key, - per, - exclude, - ) - shows = [] - if show_ids: - # Exclude fully-watched shows (LEFT JOIN so a never-started show — no rollup row — still counts - # as unseen). `total IS NULL / == 0 / watched < total` == "not fully watched". - prog = rollup.agg_subquery(db, user.id).subquery() - not_fully_watched = or_( - prog.c.total.is_(None), prog.c.total == 0, prog.c.watched < prog.c.total - ) - shows = _draw( - db, - apply_meta_filters( - select(PlexShow) - .outerjoin(prog, prog.c.show_id == PlexShow.id) - .where(PlexShow.library_id.in_(show_ids), not_fully_watched), - PlexShow, - p, - lang, - ), - PlexShow.rating_key, - per, - exclude, - ) - mcards = _movie_cards(db, user.id, movies, lang) - scards = _show_cards(db, user.id, shows, lang) - # Interleave movie/show cards, then trim to `limit`. - items: list[dict] = [] - for i in range(max(len(mcards), len(scards))): - if i < len(mcards): - items.append(mcards[i]) - if i < len(scards): - items.append(scards[i]) - return {"scope": scope, "items": items[:limit]} + return plex_discover.surprise(db, user, scope, _norm_lang(lang), limit, exclude, keep, f) diff --git a/backend/app/routes/plex/playback.py b/backend/app/routes/plex/playback.py index 457859e5..d8cea012 100644 --- a/backend/app/routes/plex/playback.py +++ b/backend/app/routes/plex/playback.py @@ -1,6 +1,7 @@ """Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes.""" import threading +from typing import TYPE_CHECKING from datetime import datetime, timedelta, timezone import httpx @@ -16,17 +17,13 @@ from fastapi import ( ) from fastapi.responses import FileResponse from pydantic import BaseModel -from sqlalchemy import select, update -from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Session from app.auth import current_user, resolved_user_id from app.db import SessionLocal, get_db from app.models import User -from app.plex.models import PlexItem, PlexSeason, PlexShow, PlexState -from app.plex import imgcache, subtitles +from app.plex import imgcache, queries, service, subtitles from app.plex import paths as plex_paths -from app.plex.lookups import by_rating_key from app.plex import people as plex_people from app.plex import stream as plex_stream from app.plex import watch_sync as plex_watch @@ -35,6 +32,9 @@ from app.routes.plex._shared import item_or_404 from app.refusal import refused from app.routes._common import ProgressIn, not_found +if TYPE_CHECKING: + from app.plex.models import PlexItem + router = APIRouter(prefix="/api/plex", tags=["plex"]) @@ -275,14 +275,7 @@ def _push_watch( return # Capture the row's freshest timestamp NOW so the background push can tell if a newer local edit # lands before it settles synced_to_plex (which would otherwise strand that newer value). - st = ( - db.execute( - select(PlexState).where(PlexState.user_id == user_id, PlexState.item_id == item_id) - ) - .scalars() - .first() - ) - marker = plex_watch._state_marker(st) + marker = plex_watch._state_marker(queries.state(db, user_id, item_id)) if action == "resume" and intent is not None: background.add_task( _push_resume_guarded, user_id, item_id, rating_key, time_ms, duration_ms, marker, intent @@ -300,37 +293,6 @@ def _push_watch( ) -def _mark_finished(db: Session, user_id: int, item_id: int, now: datetime) -> bool: - """Mark the item watched (position cleared) and say whether THIS call made it so. Two near-end - checkpoints arrive together routinely (the 10 s tick and the debounced `seeked` flush), and a - read-then-write let both see "not watched" and both scrobble. The transition is decided by the - write itself: an insert that did not conflict, or an update conditional on the old status — - under READ COMMITTED the second writer re-reads the committed row and matches nothing.""" - values = { - "status": "watched", - "watched_at": now, - "position_seconds": 0, - "progress_updated_at": now, - "synced_to_plex": False, - } - inserted = db.execute( - pg_insert(PlexState) - .values(user_id=user_id, item_id=item_id, **values) - .on_conflict_do_nothing(constraint="uq_plex_user_item") - .returning(PlexState.id) - ).first() - if inserted is not None: - return True - mine = (PlexState.user_id == user_id) & (PlexState.item_id == item_id) - if db.execute( - update(PlexState).where(mine, PlexState.status != "watched").values(values) - ).rowcount: - return True - # Already watched: a re-finish still restamps the row, as it always did — just no scrobble. - db.execute(update(PlexState).where(mine).values(values)) - return False - - class ItemProgressIn(ProgressIn): """The shared checkpoint (its bounds included), plus whether the player is leaving the item.""" @@ -366,7 +328,7 @@ def item_progress( now = datetime.now(timezone.utc) near_end = duration > 0 and position > duration - _FINISH_MARGIN_S if near_end: - flipped = _mark_finished(db, user.id, it.id, now) + flipped = service.mark_finished(db, user.id, it.id, now) # The finish supersedes every resume push still in flight for this item: stamping it as the # latest push makes `_resume_push_superseded` drop them. Popping the entry instead (to keep # the map small) removed the very guard, and an older resume push could then land at Plex @@ -380,33 +342,17 @@ def item_progress( _push_watch(background, db, user.id, it.id, rating_key, "watched") return {"ok": True, "status": "watched", "position_seconds": 0} - st = ( - db.execute( - select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) - ) - .scalars() - .first() - ) + st = queries.state(db, user, it.id) if position < _PROGRESS_MIN_S: - if st is not None and st.status == "new": - db.delete(st) - elif st is not None: - st.position_seconds = 0 - st.progress_updated_at = now + service.clear_progress(db, st, now) db.commit() return {"ok": True, "position_seconds": 0} - if st is None: - st = PlexState(user_id=user.id, item_id=it.id) - db.add(st) - st.position_seconds = position - st.progress_updated_at = now push_now = _should_push_resume(user.id, it.id, now, final) - if push_now: - # Mark dirty before the (background) push so the reconcile knows to re-push if it fails, and so - # `push_state_to_plex` can settle synced_to_plex once Plex confirms. - st.synced_to_plex = False + # Marked dirty before the (background) push so the reconcile knows to re-push if it fails, and so + # `push_state_to_plex` can settle synced_to_plex once Plex confirms. + service.checkpoint(db, user, it.id, st, position, now, dirty=push_now) db.commit() if push_now: _push_watch( @@ -442,13 +388,7 @@ def item_state( status = body.status if status not in ("new", "watched", "hidden"): raise refused(400, "invalid_status") - st = ( - db.execute( - select(PlexState).where(PlexState.user_id == user.id, PlexState.item_id == it.id) - ) - .scalars() - .first() - ) + st = queries.state(db, user, it.id) prev_status = st.status if st is not None else "new" if status == "new": # Does Plex hold anything for this item that the un-mark has to clear? Two things qualify: a @@ -462,19 +402,12 @@ def item_state( st.status == "watched" or (st.position_seconds or 0) > 0 ) if st is not None: - db.delete(st) + service.forget_state(db, st) db.commit() if mirrored_on_plex: _push_watch(background, db, user.id, it.id, rating_key, "unwatched") return {"ok": True, "status": "new"} - if st is None: - st = PlexState(user_id=user.id, item_id=it.id) - db.add(st) - st.status = status - if status == "watched": - st.watched_at = datetime.now(timezone.utc) - st.position_seconds = 0 - st.synced_to_plex = False + service.set_status(db, user, it.id, st, status) db.commit() if status == "watched" and prev_status != "watched": _push_watch(background, db, user.id, it.id, rating_key, "watched") @@ -485,44 +418,15 @@ def _bulk_state( background: BackgroundTasks, db: Session, user_id: int, - eps: list[PlexItem], + eps: "list[PlexItem]", watched: bool, ) -> int: """Mark every episode in `eps` watched (or unwatched) for a user, mirroring each change to a - linked Plex account in the background (best-effort). Returns how many rows actually changed.""" - now = datetime.now(timezone.utc) - existing = { - s.item_id: s - for s in db.execute( - select(PlexState).where( - PlexState.user_id == user_id, - PlexState.item_id.in_([e.id for e in eps] or [0]), - ) - ).scalars() - } + linked Plex account in the background (best-effort). Returns how many rows actually changed. + The handler commits before the response — which is when the background push runs, so the push + always reads the committed marks.""" pushable = plex_watch.link_for_push(db, user_id) is not None - to_push: list[tuple[int, str, str]] = [] - for e in eps: - st = existing.get(e.id) - prev = st.status if st is not None else "new" - if watched: - if st is None: - st = PlexState(user_id=user_id, item_id=e.id) - db.add(st) - st.status = "watched" - st.watched_at = now - st.position_seconds = 0 - st.synced_to_plex = False - if prev != "watched": - to_push.append((e.id, e.rating_key, "watched")) - else: # → new: drop any state (but keep the user's private "hidden" marks) - if st is not None and prev != "hidden": - db.delete(st) - if prev != "new": - to_push.append((e.id, e.rating_key, "unwatched")) - # Flush, not commit: the handler commits, and it does so before the response — which is when - # the background push below runs, so the push always reads the committed marks. - db.flush() + to_push = service.bulk_state(db, user_id, eps, watched) if pushable and to_push: # One coalesced background task for the whole batch (reuses a single session + Plex client), # not one task per episode — a 200-ep show mark is 1 task, not 200. @@ -543,15 +447,11 @@ def show_state( db: Session = Depends(get_db), ) -> dict: """Mark a whole show watched/unwatched (all its episodes) for the current user + push to Plex.""" - sh = by_rating_key(db, PlexShow, rating_key) + sh = queries.show(db, rating_key) if sh is None: raise not_found("show") watched = body.watched - eps = ( - db.execute(select(PlexItem).where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")) - .scalars() - .all() - ) + eps = queries.episodes(db, sh) changed = _bulk_state(background, db, user.id, eps, watched) db.commit() return {"ok": True, "changed": changed, "watched": watched} @@ -566,15 +466,11 @@ def season_state( db: Session = Depends(get_db), ) -> dict: """Mark a whole season watched/unwatched (all its episodes) for the current user + push to Plex.""" - se = by_rating_key(db, PlexSeason, rating_key) + se = queries.season(db, rating_key) if se is None: raise not_found("season") watched = body.watched - eps = ( - db.execute(select(PlexItem).where(PlexItem.season_id == se.id, PlexItem.kind == "episode")) - .scalars() - .all() - ) + eps = queries.season_episodes(db, se) changed = _bulk_state(background, db, user.id, eps, watched) db.commit() return {"ok": True, "changed": changed, "watched": watched} diff --git a/backend/app/routes/plex/playlists.py b/backend/app/routes/plex/playlists.py index 14b1879f..bb6ddfc5 100644 --- a/backend/app/routes/plex/playlists.py +++ b/backend/app/routes/plex/playlists.py @@ -1,27 +1,26 @@ """Plex playlists: the current user's personal, orderable playlists of playable items.""" -from datetime import datetime, timezone +from typing import TYPE_CHECKING, Annotated from fastapi import APIRouter, Depends -from typing import Annotated - from pydantic import BaseModel, Field, model_validator -from sqlalchemy import delete, func, select from sqlalchemy.orm import Session from app.auth import current_user from app.db import get_db from app.models import User -from app.plex.models import PlexItem, PlexPlaylist, PlexPlaylistItem, PlexShow, PlexState +from app.plex import queries, service from app.plex.cards import display_title, leaf_card, norm_lang -from app.plex.lookups import by_rating_key -from app.routes._common import owned_or_404, not_found from app.refusal import refused +from app.routes._common import not_found + +if TYPE_CHECKING: + from app.plex.models import PlexPlaylist router = APIRouter(prefix="/api/plex", tags=["plex"]) -def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: +def _playlist_card(pl: "PlexPlaylist", count: int, thumb_rk: str | None) -> dict: return { "id": pl.id, "title": pl.title, @@ -30,13 +29,15 @@ def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict: } -def _playlist_item_query(pid: int): - return ( - select(PlexPlaylistItem, PlexItem) - .join(PlexItem, PlexItem.id == PlexPlaylistItem.item_id) - .where(PlexPlaylistItem.playlist_id == pid) - .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) - ) +def _card_of(db: Session, pl: "PlexPlaylist") -> dict: + return _playlist_card(pl, queries.playlist_count(db, pl.id), queries.playlist_cover(db, pl.id)) + + +def _own_playlist(db: Session, user: User, pid: int, *, lock: bool = False) -> "PlexPlaylist": + pl = queries.own_playlist(db, user, pid, lock=lock) + if pl is None: + raise not_found("playlist") + return pl @router.get("/playlists") @@ -50,49 +51,17 @@ def playlists( `contains`=an item rating_key adds `has_item` per playlist (for the single 'add to playlist' dialog). `contains_group`=a comma-separated list of rating_keys adds `group_in` per playlist (how many of the group it already holds) + a top-level `group_size`, for the 'add a whole season/show' dialog.""" - contains_id = None - if contains: - contains_id = db.scalar(select(PlexItem.id).where(PlexItem.rating_key == str(contains))) + contains_id = queries.item_id(db, contains) if contains else None group_ids: set[int] | None = None if contains_group is not None: - rks = [s for s in (contains_group or "").split(",") if s] - group_ids = set( - db.execute(select(PlexItem.id).where(PlexItem.rating_key.in_(rks or [""]))).scalars() - ) + group_ids = queries.item_ids(db, [s for s in (contains_group or "").split(",") if s]) out = [] - for pl in db.execute( - select(PlexPlaylist) - .where(PlexPlaylist.user_id == user.id) - .order_by(PlexPlaylist.updated_at.desc()) - ).scalars(): - count = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where(PlexPlaylistItem.playlist_id == pl.id) - ) - or 0 - ) - first = db.execute(_playlist_item_query(pl.id)).first() - card = _playlist_card(pl, count, first[1].rating_key if first else None) + for pl in queries.playlists_of(db, user): + card = _card_of(db, pl) if contains_id is not None: - card["has_item"] = ( - db.execute( - select(PlexPlaylistItem.id).where( - PlexPlaylistItem.playlist_id == pl.id, - PlexPlaylistItem.item_id == contains_id, - ) - ).first() - is not None - ) + card["has_item"] = queries.playlist_holds(db, pl.id, {contains_id}) > 0 if group_ids is not None: - card["group_in"] = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where( - PlexPlaylistItem.playlist_id == pl.id, - PlexPlaylistItem.item_id.in_(group_ids or [0]), - ) - ) - or 0 - ) + card["group_in"] = queries.playlist_holds(db, pl.id, group_ids) out.append(card) resp: dict = {"playlists": out} if group_ids is not None: @@ -114,19 +83,10 @@ def create_playlist( title = body.title.strip() if not title: raise refused(400, "title_required") - pl = PlexPlaylist(user_id=user.id, title=title) - db.add(pl) - db.flush() - seed = body.item_rating_key - count = 0 - thumb_rk = None - if seed: - it = by_rating_key(db, PlexItem, seed) - if it is not None: - db.add(PlexPlaylistItem(playlist_id=pl.id, item_id=it.id, position=0)) - count, thumb_rk = 1, it.rating_key + seed = queries.item(db, body.item_rating_key) if body.item_rating_key else None + pl = service.create_playlist(db, user, title, seed) db.commit() - return _playlist_card(pl, count, thumb_rk) + return _playlist_card(pl, 1 if seed else 0, seed.rating_key if seed else None) @router.get("/playlists/{pid}") @@ -137,26 +97,10 @@ def playlist_detail( db: Session = Depends(get_db), ) -> dict: lang = norm_lang(lang) - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") - rows = db.execute(_playlist_item_query(pid)).all() - items = [it for (_pi, it) in rows] - states = { - s.item_id: s - for s in db.execute( - select(PlexState).where( - PlexState.user_id == user.id, - PlexState.item_id.in_([it.id for it in items] or [0]), - ) - ).scalars() - } - shows = { - sh.id: sh - for sh in db.execute( - select(PlexShow).where( - PlexShow.id.in_([it.show_id for it in items if it.show_id] or [0]) - ) - ).scalars() - } + pl = _own_playlist(db, user, pid) + items = queries.playlist_items(db, pl.id) + states = queries.states(db, user, [it.id for it in items]) + shows = queries.shows_by_ids(db, {it.show_id for it in items if it.show_id}) cards = [ leaf_card( it, @@ -181,28 +125,21 @@ def rename_playlist( user: User = Depends(current_user), db: Session = Depends(get_db), ) -> dict: - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") + pl = _own_playlist(db, user, pid) title = body.title.strip() if not title: raise refused(400, "title_required") - pl.title = title + service.rename_playlist(pl, title) db.commit() - count = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where(PlexPlaylistItem.playlist_id == pl.id) - ) - or 0 - ) - first = db.execute(_playlist_item_query(pl.id)).first() - return _playlist_card(pl, count, first[1].rating_key if first else None) + return _playlist_card(pl, queries.playlist_count(db, pl.id), queries.playlist_cover(db, pl.id)) @router.delete("/playlists/{pid}") def delete_playlist( pid: int, user: User = Depends(current_user), db: Session = Depends(get_db) ) -> dict: - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist") - db.delete(pl) + pl = _own_playlist(db, user, pid) + service.delete_playlist(db, pl) db.commit() return {"deleted": pid} @@ -219,27 +156,11 @@ def playlist_add_item( db: Session = Depends(get_db), ) -> dict: # Locked (like every add / reorder here): two adds otherwise read the same max position. - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) - it = by_rating_key(db, PlexItem, body.item_rating_key) + pl = _own_playlist(db, user, pid, lock=True) + it = queries.item(db, body.item_rating_key) if it is None: raise not_found("item") - exists = ( - db.execute( - select(PlexPlaylistItem).where( - PlexPlaylistItem.playlist_id == pid, PlexPlaylistItem.item_id == it.id - ) - ) - .scalars() - .first() - ) - if exists is None: - maxpos = db.scalar( - select(func.coalesce(func.max(PlexPlaylistItem.position), -1)).where( - PlexPlaylistItem.playlist_id == pid - ) - ) - db.add(PlexPlaylistItem(playlist_id=pid, item_id=it.id, position=int(maxpos) + 1)) - pl.updated_at = datetime.now(timezone.utc) + service.add_item(db, pl, it) db.commit() return {"ok": True} @@ -253,15 +174,10 @@ def playlist_remove_item( ) -> dict: # Locked like the adds and the reorder: the list row first, then its items — a remove that # locked an item row before a reorder locked the list deadlocked against it. - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) - it = by_rating_key(db, PlexItem, item_rating_key) + pl = _own_playlist(db, user, pid, lock=True) + it = queries.item(db, item_rating_key) if it is not None: - db.execute( - delete(PlexPlaylistItem).where( - PlexPlaylistItem.playlist_id == pid, PlexPlaylistItem.item_id == it.id - ) - ) - pl.updated_at = datetime.now(timezone.utc) + service.remove_item(db, pl, it) db.commit() return {"ok": True} @@ -270,25 +186,6 @@ class PlexItemKeysIn(BaseModel): item_rating_keys: list[str] = [] -def _items_by_rks(db: Session, rks: list) -> list[PlexItem]: - """Resolve rating_keys to PlexItems, preserving the input order (drops unknown keys).""" - order = [str(k) for k in rks if str(k)] - if not order: - return [] - by_rk = { - it.rating_key: it - for it in db.execute(select(PlexItem).where(PlexItem.rating_key.in_(order))).scalars() - } - seen: set[str] = set() - out: list[PlexItem] = [] - for rk in order: - it = by_rk.get(rk) - if it is not None and rk not in seen: - seen.add(rk) - out.append(it) - return out - - class PlexItemAddBulkIn(PlexItemKeysIn): # Where each key goes back (the list page's "Undo" of a removal): an index over the # ORDERED rows, one per key. The two-call shape (append, then send the whole order) leaves the rows @@ -303,32 +200,6 @@ class PlexItemAddBulkIn(PlexItemKeysIn): return self -def _insert_at(db: Session, pid: int, placed: list[tuple[int, PlexItem]]) -> int: - """Insert new rows at their indices (ascending, so each index is where the row stood when the - rows before it were back), renumber the whole list 0..n-1, and say how many were added. An item - already in the list does not move: the add stays idempotent.""" - rows = list( - db.execute( - select(PlexPlaylistItem) - .where(PlexPlaylistItem.playlist_id == pid) - .order_by(PlexPlaylistItem.position, PlexPlaylistItem.id) - ).scalars() - ) - present = {r.item_id for r in rows} - added = 0 - for at, it in sorted(placed, key=lambda p: p[0]): - if it.id in present: - continue - row = PlexPlaylistItem(playlist_id=pid, item_id=it.id, position=0) - db.add(row) - rows.insert(min(at, len(rows)), row) - present.add(it.id) - added += 1 - for i, r in enumerate(rows): - r.position = i - return added - - @router.post("/playlists/{pid}/items/bulk") def playlist_add_bulk( pid: int, @@ -339,56 +210,20 @@ def playlist_add_bulk( """Append many items at once (a whole season/show). Already-present items are skipped; the input order is preserved for the newly-added ones. With `positions`, each key is put back at its index instead of appended. Returns how many were added + the new total.""" - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) + pl = _own_playlist(db, user, pid, lock=True) + items = queries.items_by_keys(db, body.item_rating_keys) if body.positions is not None: - by_rk = {it.rating_key: it for it in _items_by_rks(db, body.item_rating_keys)} + by_rk = {it.rating_key: it for it in items} placed = [ (at, by_rk[str(rk)]) for rk, at in zip(body.item_rating_keys, body.positions, strict=True) if str(rk) in by_rk ] - added = _insert_at(db, pid, placed) - if added: - pl.updated_at = datetime.now(timezone.utc) - db.commit() - total = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where(PlexPlaylistItem.playlist_id == pid) - ) - or 0 - ) - return {"ok": True, "added": added, "item_count": total} - items = _items_by_rks(db, body.item_rating_keys) - existing = set( - db.execute( - select(PlexPlaylistItem.item_id).where(PlexPlaylistItem.playlist_id == pid) - ).scalars() - ) - pos = int( - db.scalar( - select(func.coalesce(func.max(PlexPlaylistItem.position), -1)).where( - PlexPlaylistItem.playlist_id == pid - ) - ) - ) - added = 0 - for it in items: - if it.id in existing: - continue - pos += 1 - db.add(PlexPlaylistItem(playlist_id=pid, item_id=it.id, position=pos)) - existing.add(it.id) - added += 1 - if added: - pl.updated_at = datetime.now(timezone.utc) + added = service.insert_items_at(db, pl, placed) + else: + added = service.add_items(db, pl, items) db.commit() - total = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where(PlexPlaylistItem.playlist_id == pid) - ) - or 0 - ) - return {"ok": True, "added": added, "item_count": total} + return {"ok": True, "added": added, "item_count": queries.playlist_count(db, pid)} @router.post("/playlists/{pid}/items/remove-bulk") @@ -400,26 +235,11 @@ def playlist_remove_bulk( ) -> dict: """Remove many items at once (a whole season/show, or a multi-select). Returns how many were removed + the new total.""" - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) - ids = [it.id for it in _items_by_rks(db, body.item_rating_keys)] - removed = 0 - if ids: - removed = db.execute( - delete(PlexPlaylistItem).where( - PlexPlaylistItem.playlist_id == pid, PlexPlaylistItem.item_id.in_(ids) - ), - execution_options={"synchronize_session": False}, - ).rowcount - if removed: - pl.updated_at = datetime.now(timezone.utc) + pl = _own_playlist(db, user, pid, lock=True) + ids = [it.id for it in queries.items_by_keys(db, body.item_rating_keys)] + removed = service.remove_items(db, pl, ids) db.commit() - total = ( - db.scalar( - select(func.count(PlexPlaylistItem.id)).where(PlexPlaylistItem.playlist_id == pid) - ) - or 0 - ) - return {"ok": True, "removed": removed, "item_count": total} + return {"ok": True, "removed": removed, "item_count": queries.playlist_count(db, pid)} @router.put("/playlists/{pid}/order") @@ -430,35 +250,7 @@ def reorder_playlist( db: Session = Depends(get_db), ) -> dict: """Set the playlist order from an explicit list of item rating_keys (0-based positions).""" - pl = owned_or_404(db, PlexPlaylist, pid, user, what="playlist", lock=True) - order = body.item_rating_keys - id_by_rk = { - it.rating_key: it.id - for it in db.execute( - select(PlexItem).where(PlexItem.rating_key.in_(order or [""])) - ).scalars() - } - rows = { - r.item_id: r - for r in db.execute( - select(PlexPlaylistItem).where(PlexPlaylistItem.playlist_id == pid) - ).scalars() - } - pos = 0 - seen: set[int] = set() - for rk in order: - row = rows.get(id_by_rk.get(rk, -1)) - if row is not None and row.item_id not in seen: - row.position = pos - pos += 1 - seen.add(row.item_id) - # Rows the payload omitted (added in another tab meanwhile) keep their relative order after - # the given ones, so no two rows share a position (the YouTube reorder does the same). - for row in sorted( - (r for iid, r in rows.items() if iid not in seen), key=lambda r: (r.position, r.id) - ): - row.position = pos - pos += 1 - pl.updated_at = datetime.now(timezone.utc) + pl = _own_playlist(db, user, pid, lock=True) + service.reorder_playlist(db, pl, body.item_rating_keys) db.commit() return {"ok": True} diff --git a/backend/tests/test_plex_finish.py b/backend/tests/test_plex_finish.py index fccee9e6..fea93d65 100644 --- a/backend/tests/test_plex_finish.py +++ b/backend/tests/test_plex_finish.py @@ -14,6 +14,7 @@ from sqlalchemy import select from app.db import SessionLocal from app.plex.models import PlexItem, PlexLibrary, PlexState +from app.plex import service as plex_service from app.routes.plex import playback @@ -36,10 +37,12 @@ def test_two_racing_finishes_flip_the_row_once(db, make_user): first, second = SessionLocal(), SessionLocal() try: - assert playback._mark_finished(first, user.id, item.id, now) is True # holds the row lock + assert ( + plex_service.mark_finished(first, user.id, item.id, now) is True + ) # holds the row lock out: list[bool] = [] racer = threading.Thread( - target=lambda: out.append(playback._mark_finished(second, user.id, item.id, now)) + target=lambda: out.append(plex_service.mark_finished(second, user.id, item.id, now)) ) racer.start() racer.join(0.5) @@ -57,9 +60,9 @@ def test_a_first_ever_finish_inserts_and_counts_once(db, make_user): user = make_user() item = _item(db) now = datetime.now(timezone.utc) - assert playback._mark_finished(db, user.id, item.id, now) is True + assert plex_service.mark_finished(db, user.id, item.id, now) is True db.commit() - assert playback._mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False + assert plex_service.mark_finished(db, user.id, item.id, now + timedelta(seconds=10)) is False db.commit() st = db.execute(select(PlexState).where(PlexState.item_id == item.id)).scalars().one() assert (st.status, st.position_seconds) == ("watched", 0) diff --git a/backend/tests/test_plex_service.py b/backend/tests/test_plex_service.py index df28d037..a15f6fd8 100644 --- a/backend/tests/test_plex_service.py +++ b/backend/tests/test_plex_service.py @@ -4,7 +4,8 @@ The routes are thin callers now, and their own tests stay the contract a client what each read and write does on its own: that a user's watch state is theirs alone (another user's state is the same answer as none), that a write leaves the commit to its caller, and the rule each one carries (an explicit empty key list asks for nothing, a kept collection is editable and -holds its seed). DB-backed (real Postgres). +holds its seed, a finish scrobbles once, an unwatch keeps a hidden mark, a playlist add is +idempotent). DB-backed (real Postgres). """ from datetime import datetime, timezone @@ -68,8 +69,8 @@ def test_a_users_watch_state_is_theirs_alone(db, make_user): lib = _lib(db) it = _movie(db, lib, "m1") _state(db, owner, it, status="watched") - assert queries.state(db, owner, it).status == "watched" - assert queries.state(db, stranger, it) is None + assert queries.state(db, owner, it.id).status == "watched" + assert queries.state(db, stranger, it.id) is None assert queries.states(db, stranger, [it.id]) == {} assert set(queries.states(db, owner, [it.id])) == {it.id} assert queries.states(db, owner, []) == {} @@ -262,3 +263,161 @@ def test_marking_a_collection_editable_is_left_to_the_caller(db): db.commit() db.expire_all() assert db.scalar(select(PlexCollection.editable)) is True + + +# --- playlists ------------------------------------------------------------------------------- + + +def _items(db, lib, *rks): + return [_movie(db, lib, rk) for rk in rks] + + +def test_a_playlist_is_its_owners_alone(db, make_user): + owner, stranger = make_user(), make_user() + pl = service.create_playlist(db, owner, "Mine", None) + db.commit() + assert queries.own_playlist(db, stranger, pl.id) is None + assert queries.own_playlist(db, owner, pl.id).title == "Mine" + assert queries.playlists_of(db, stranger) == [] + assert [p.id for p in queries.playlists_of(db, owner)] == [pl.id] + + +def test_playlist_adds_append_once_and_removes_stay_in_their_list(db, make_user): + user = make_user() + lib = _lib(db) + a, b, c = _items(db, lib, "a", "b", "c") + pl = service.create_playlist(db, user, "L", a) + other = service.create_playlist(db, user, "Other", b) + service.add_item(db, pl, b) + service.add_item(db, pl, a) # already in it: stays first + assert service.add_items(db, pl, [c, b]) == 1 + db.commit() + assert [it.rating_key for it in queries.playlist_items(db, pl.id)] == ["a", "b", "c"] + assert queries.playlist_holds(db, pl.id, {a.id, c.id}) == 2 + assert service.remove_items(db, pl, [b.id, 999]) == 1 + service.remove_item(db, pl, c) + db.commit() + assert [it.rating_key for it in queries.playlist_items(db, pl.id)] == ["a"] + assert [it.rating_key for it in queries.playlist_items(db, other.id)] == ["b"] + assert queries.playlist_cover(db, other.id) == "b" + + +def test_an_undo_puts_items_back_at_their_places_and_renumbers(db, make_user): + user = make_user() + lib = _lib(db) + a, b, c, d = _items(db, lib, "a", "b", "c", "d") + pl = service.create_playlist(db, user, "L", a) + service.add_items(db, pl, [c]) + assert service.insert_items_at(db, pl, [(3, d), (1, b), (0, a)]) == 2 # a is there already + db.commit() + assert [it.rating_key for it in queries.playlist_items(db, pl.id)] == ["a", "b", "c", "d"] + + +def test_a_reorder_keeps_rows_it_was_not_told_about_after_the_rest(db, make_user): + user = make_user() + lib = _lib(db) + a, b, c = _items(db, lib, "a", "b", "c") + pl = service.create_playlist(db, user, "L", a) + service.add_items(db, pl, [b, c]) + service.reorder_playlist(db, pl, ["c", "b", "nope"]) # `a` (position 0) not named + db.commit() + assert [it.rating_key for it in queries.playlist_items(db, pl.id)] == ["c", "b", "a"] + assert queries.items_by_keys(db, ["b", "x", "b", "a"]) == [b, a] + + +# --- watch state ----------------------------------------------------------------------------- + + +def test_a_finish_flips_once_and_leaves_another_users_state_alone(db, make_user): + me, other = make_user(), make_user() + lib = _lib(db) + it = _movie(db, lib, "m1") + _state(db, other, it, status="new", position_seconds=50) + assert service.mark_finished(db, me.id, it.id, NOW) is True + db.commit() + assert service.mark_finished(db, me.id, it.id, NOW) is False + db.commit() + db.expire_all() + assert ( + queries.state(db, other.id, it.id).status, + queries.state(db, other.id, it.id).position_seconds, + ) == ("new", 50) + + +def test_an_early_checkpoint_drops_a_bare_state_and_zeroes_a_marked_one(db, make_user): + user = make_user() + lib = _lib(db) + bare, hid = _movie(db, lib, "bare"), _movie(db, lib, "hid") + st = service.checkpoint(db, user, bare.id, None, 120, NOW, dirty=True) + assert (st.position_seconds, st.synced_to_plex) == (120, False) + _state(db, user, hid, status="hidden", position_seconds=40) + service.clear_progress(db, queries.state(db, user, bare.id), NOW) + service.clear_progress(db, queries.state(db, user, hid.id), NOW) + db.commit() + assert queries.state(db, user, bare.id) is None + assert ( + queries.state(db, user, hid.id).status, + queries.state(db, user, hid.id).position_seconds, + ) == ("hidden", 0) + + +def test_unwatching_a_season_keeps_hidden_marks_and_says_what_to_push(db, make_user): + me, other = make_user(), make_user() + lib = _lib(db) + seen, hid, fresh = _items(db, lib, "seen", "hid", "fresh") + _state(db, me, seen, status="watched") + _state(db, me, hid, status="hidden") + _state(db, other, seen, status="watched") + assert service.bulk_state(db, me.id, [seen, hid, fresh], watched=False) == [ + (seen.id, "seen", "unwatched") + ] + db.commit() + assert queries.state(db, me, hid.id).status == "hidden" + assert queries.state(db, other, seen.id).status == "watched" + pushes = service.bulk_state(db, me.id, [seen, hid], watched=True) + db.commit() + assert pushes == [(seen.id, "seen", "watched"), (hid.id, "hid", "watched")] + + +def test_a_watch_link_is_made_once_and_is_its_owners(db, make_user): + me, other = make_user(), make_user() + first = service.set_watch_link(db, me, True) + again = service.set_watch_link(db, me, False) + db.commit() + assert first.id == again.id and again.sync_enabled is False and again.uses_admin is True + assert queries.watch_link(db, other) is None + + +# --- discover -------------------------------------------------------------------------------- + + +def test_the_discover_rails_follow_the_callers_state(db, make_user): + from app.plex import discover + + me, other = make_user(), make_user() + lib = _lib(db) + going, done = _movie(db, lib, "going"), _movie(db, lib, "done") + _state(db, other, going, status="new", position_seconds=300, progress_updated_at=NOW) + _state(db, other, done, status="watched", watched_at=NOW) + mine = discover.discover_page(db, me, "both", "en", LibraryFilters()) + theirs = discover.discover_page(db, other, "both", "en", LibraryFilters()) + assert ([c["id"] for c in mine["continue_watching"]], mine["recently_watched"]) == ([], []) + assert [c["id"] for c in theirs["continue_watching"]] == ["going"] + assert [c["id"] for c in theirs["recently_watched"]] == ["done"] + drawn = discover.surprise(db, me, "movie", "en", 6, [], [], LibraryFilters()) + assert {c["id"] for c in drawn["items"]} == {"going", "done"} # their watch is not mine + drawn = discover.surprise(db, other, "movie", "en", 6, [], [], LibraryFilters()) + assert {c["id"] for c in drawn["items"]} == {"going"} + + +def test_only_the_owners_own_link_pushes_their_watch_state(db, make_user): + """A push is scheduled per user: another account's enabled link must not make this user's + marks reach Plex.""" + from app.plex import watch_sync + + owner, other = make_user(), make_user() + link = service.set_watch_link(db, owner, True) + link.initial_import_done = True + db.commit() + assert watch_sync.link_for_push(db, owner.id).id == link.id + assert watch_sync.link_for_push(db, other.id) is None diff --git a/backend/tests/test_unit_fences.py b/backend/tests/test_unit_fences.py index 37f4755c..e51eb292 100644 --- a/backend/tests/test_unit_fences.py +++ b/backend/tests/test_unit_fences.py @@ -109,12 +109,7 @@ UNIT_ROUTES = { # What each HTTP file still does, by file: the debt a unit's sprint pays off. Exact counts. A core # route's count is only the unit tables it imports; it is paid by the sprint of the unit it reaches. -ALLOWED = { - "plex/admin.py": 12, - "plex/discover.py": 44, - "plex/playback.py": 42, - "plex/playlists.py": 80, -} +ALLOWED: dict[str, int] = {} # The session methods that read or write data. `commit` and `rollback` stay with the handler. SESSION_DATA = { @@ -342,11 +337,19 @@ UNIT_PACKAGES = { "sync/subscriptions.py", "sync/videos.py", ), + # The Plex mirror: its sync, the watch-state sync with a linked account, TMDB enrichment, the + # pages' reads and writes. + "plex": ("plex/",), } # The units fenced so far; the rest join with their sprint. -FENCED = {"capture", "reader", "catalog"} +FENCED = {"capture", "reader", "catalog", "plex"} # Whose tables a unit's own code may READ. Core code reads a fenced unit's tables through its queries. -READS = {"capture": {"catalog", "plex"}, "reader": {"catalog", "plex"}, "catalog": set()} +READS = { + "capture": {"catalog", "plex"}, + "reader": {"catalog", "plex"}, + "catalog": set(), + "plex": set(), +} # The columns that say whose a row is. OWNER_COLUMNS = {"user_id", "to_user_id", "from_user_id", "created_by"} # Where a fenced unit's code may commit: the jobs that own their session (the worker and what it @@ -399,6 +402,26 @@ UNIT_COMMITS = { "sync/videos.py:run_shorts_classification", "sync/videos.py:ensure_videos", }, + # The Plex jobs commit per step too: a library sync per section and per enrich/probe batch, the + # TMDB enrichment per batch, the watch sync per pull and push. The pushes and a person's bio + # fetch run as background tasks after the response, on a session of their own; the owner's + # first import runs after the handler committed the link it reads. + "plex": { + "plex/people.py:prune_orphan_people", + "plex/sync.py:sync", + "plex/sync.py:_probe_media_batch", + "plex/sync.py:_enrich_people_batch", + "plex/tmdb.py:enrich_batch", + "plex/tmdb.py:backfill_country_codes", + "plex/tmdb.py:enrich_person", + "plex/tmdb.py:enrich_people_batch", + "plex/watch_sync.py:import_owner_watch_state", + "plex/watch_sync.py:push_state_to_plex", + "plex/watch_sync.py:push_bulk_state_to_plex", + "plex/watch_sync.py:_resolve_account_id", + "plex/watch_sync.py:run_plex_watch_sync", + "plex/watch_sync.py:run_plex_watch_reconcile", + }, } SQL_WRITES = {"insert", "update", "delete", "pg_insert"} @@ -533,7 +556,8 @@ PACKAGE_CASES = { ("downloads/x.py", "from app.models import Video\nupdate(Video)"): 1, ("downloads/x.py", "from app.downloads.models import DownloadJob\nDownloadJob()"): 0, ("search.py", "from app.downloads.models import DownloadJob"): 1, # core reads capture - ("search.py", "from app.models import PlexItem\nPlexItem()"): 0, # neither side fenced yet + ("search.py", "from app.plex.models import PlexItem\nPlexItem()"): 2, # core reads, writes plex + ("plex/x.py", "from app.catalog.models import Video"): 1, # no plex→catalog edge ("search.py", "from app.models import Video\nVideo()"): 2, # core reads and writes catalog ("sync/videos.py", "from app.models import Playlist"): 1, # no catalog→reader edge ("downloads/x.py", "q = DownloadJob.user_id == 1"): 1, diff --git a/docs/gates.md b/docs/gates.md index d6990113..2810eaf2 100644 --- a/docs/gates.md +++ b/docs/gates.md @@ -113,7 +113,7 @@ project checks needs a row. | CVE gate | HIGH/CRITICAL with a fix available only | `--ignore-unfixed` + `--severity HIGH,CRITICAL`: a finding with no fix cannot be acted on by an upgrade; it shows in the scan output. A fixable one fails the publish, and the only way past it is `.trivyignore.yaml`. | 0 open (E32 S1) | | npm audit | `--omit=dev --audit-level=high` | What ships to the browser, at high or above. The dev toolchain's advisories were burned down by hand in E32 S1 (18 → 0). | 0 | | knip | `entry` — one file added | Not a narrowing: knip's Vite plugin reads entries from index.html only, so the shared link's page (`watch.html` → `src/watchMain.tsx`) is named here, or the lane reports that whole entry unused. It adds files to the lane, never takes any out. | 1 entry | -| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) → 178 in 4 (S4a: Plex browse, detail, people, collections 0) | +| unit fence | the database work the HTTP layer still does, pinned | `backend/tests/test_unit_fences.py`: a unit's route file may not build SQL, call the session's data methods, set a model's attributes or import a unit's tables; a core route may not import a unit's tables. Switched on unit by unit: `ALLOWED` pins each file's count exactly, so a rise fails and a fall fails until the pin is lowered — the list may only shrink. End state: empty, and the pin mechanism goes. A unit in `FENCED` is held below its HTTP layer too: its code reads another unit's tables only along `READS`, writes them never, filters owners through `userscope`, and commits only where `UNIT_COMMITS` names a job's own session. | 826 in 23 files (E42 S0) → 667 in 19 (S1: Capture 0, fenced) → 525 in 14 (S2: Reader 0, fenced) → 354 in 9 (S3: Catalog 0, fenced; core routes 0) → 178 in 4 (S4a: Plex browse, detail, people, collections 0) → 0 (S4b: Plex 0, fenced — every unit fenced) | | React Compiler | the functions it skips, pinned | `components/compilerCoverage.contract.test.ts` — the list may only shrink. End state: only VirtualGrid (TanStack Virtual, an IncompatibleLibrary) remains. | 93 (E32 S1) → 29 (E32 S3) → 1, VirtualGrid (E32 S4) | ## Accepted findings