siftlode/backend/app/routes/plex/discover.py

"""Plex Discover: the rail-surface read-model (E5 S4).

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 S2 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
S1's `art` + S3's 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.
"""

from fastapi import APIRouter, Depends, Query
from sqlalchemy import and_, case, func, or_, select
from sqlalchemy.orm import Session, aliased

from app.auth import current_user
from app.db import get_db
from app.models import PlexItem, PlexShow, PlexState, User
from app.plex import rollup
from app.plex.cards import movie_card, show_card
from app.plex.query import genres_col
from app.plex.tmdb import resolve_recs
from app.routes.plex._shared import _content_libs, _norm_lang

router = APIRouter(prefix="/api/plex", tags=["plex"])

RAIL_LIMIT = 20
GENRE_TILES = 12


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_states(db: Session, uid: int, items: list[PlexItem]) -> dict[int, PlexState]:
    """One batched fetch of the user's watch-state for a set of movies (avoids a per-card query)."""
    ids = [it.id for it in items]
    if not ids:
        return {}
    rows = db.execute(
        select(PlexState).where(PlexState.user_id == uid, PlexState.item_id.in_(ids))
    ).scalars().all()
    return {r.item_id: r for r in rows}


def _movie_cards(db: Session, uid: int, items: list[PlexItem]) -> list[dict]:
    sts = _movie_states(db, uid, items)
    return [movie_card(it, sts.get(it.id)) for it in items]


def _show_cards(db: Session, uid: int, shows: list[PlexShow]) -> list[dict]:
    statuses = rollup.state_map(db, uid, [s.id for s in shows])
    return [show_card(s, statuses.get(s.id, "new")) for s in shows]


def _recently_added(
    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
) -> list[dict]:
    """Newest movies + shows, interleaved by `added_at` desc into one rail."""
    movies = (
        db.execute(
            select(PlexItem)
            .where(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids))
            .order_by(PlexItem.added_at.desc().nullslast())
            .limit(RAIL_LIMIT)
        ).scalars().all()
        if movie_ids
        else []
    )
    shows = (
        db.execute(
            select(PlexShow)
            .where(PlexShow.library_id.in_(show_ids))
            .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). The `(is not None, value)` key sorts a missing
    # timestamp last under reverse.
    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] is not None, t[0]), 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))}
    scards = {s.rating_key: c for s, c in zip(top_shows, _show_cards(db, uid, top_shows))}
    return [
        mcards[o.rating_key] if kind == "movie" else scards[o.rating_key]
        for _, kind, o in top
    ]


def _continue_watching(
    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
) -> 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:
        mrows = db.execute(
            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"]),
                ),
            )
            .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)))

    if show_ids:
        ps = aliased(PlexState)
        prog = (
            select(
                PlexItem.show_id.label("show_id"),
                func.max(ps.progress_updated_at).label("resumed_at"),
                func.count(PlexItem.id).label("total"),
                func.count(case((ps.status == "watched", 1))).label("watched"),
                func.count(
                    case(
                        (
                            and_(
                                ps.position_seconds > 0,
                                or_(ps.status.is_(None), ps.status != "watched"),
                            ),
                            1,
                        )
                    )
                ).label("inprog"),
            )
            .outerjoin(ps, and_(ps.item_id == PlexItem.id, ps.user_id == uid))
            .where(PlexItem.kind == "episode", PlexItem.show_id.isnot(None))
            .group_by(PlexItem.show_id)
            .subquery()
        )
        # in-progress == started but not fully watched (mirrors apply_show_state_filter's `in_progress`).
        srows = db.execute(
            select(PlexShow, prog.c.resumed_at)
            .join(prog, prog.c.show_id == PlexShow.id)
            .where(
                PlexShow.library_id.in_(show_ids),
                or_(prog.c.watched > 0, prog.c.inprog > 0),
                prog.c.watched < prog.c.total,
            )
            .order_by(prog.c.resumed_at.desc().nullslast())
            .limit(RAIL_LIMIT)
        ).all()
        for sh, resumed_at in srows:
            rows.append((resumed_at, show_card(sh, "in_progress")))

    rows.sort(key=lambda r: (r[0] is not None, r[0]), reverse=True)
    return [card for _, card in rows[:RAIL_LIMIT]]


def _genre_tiles(
    db: Session, movie_ids: list[int], show_ids: list[int], 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."""
    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"
        )
        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
    top = sorted(counts.items(), key=lambda kv: (-kv[1], kv[0]))[:GENRE_TILES]
    return [{"value": g, "count": c} for g, c in top]


def _because_you_watched(
    db: Session, uid: int, movie_ids: list[int], show_ids: list[int]
) -> 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."""
    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)
    # 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(
            select(PlexItem).where(
                PlexItem.kind == "movie", PlexItem.rating_key.in_(rks)
            )
        ).scalars().all()
    }
    shows = {
        s.rating_key: s
        for s in db.execute(
            select(PlexShow).where(PlexShow.rating_key.in_(rks))
        ).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))}
    scards = {s.rating_key: c for s, c in zip(ordered_shows, _show_cards(db, uid, ordered_shows))}
    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": anchor.title}, "items": items}


@router.get("/discover")
def discover(
    scope: str = "both",
    lang: str = "en",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """The Discover landing payload — every rail in one pass (serve-from-DB). 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)
    movie_ids, show_ids = _scope_libs(db, scope)
    return {
        "scope": scope,
        "continue_watching": _continue_watching(db, user.id, movie_ids, show_ids),
        "recently_added": _recently_added(db, user.id, movie_ids, show_ids),
        "genres": _genre_tiles(db, movie_ids, show_ids, lang),
        "because_you_watched": _because_you_watched(db, user.id, movie_ids, show_ids),
    }


@router.get("/surprise")
def surprise(
    scope: str = "both",
    limit: int = Query(default=RAIL_LIMIT, ge=1, le=40),
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """A fresh random sample for the reshuffling Surprise-me rail (a random draw each call). Movies and
    shows are drawn separately then interleaved so a lopsided library doesn't crowd out one type."""
    movie_ids, show_ids = _scope_libs(db, scope)
    movies = (
        db.execute(
            select(PlexItem)
            .where(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_ids))
            .order_by(func.random())
            .limit(limit)
        ).scalars().all()
        if movie_ids
        else []
    )
    shows = (
        db.execute(
            select(PlexShow)
            .where(PlexShow.library_id.in_(show_ids))
            .order_by(func.random())
            .limit(limit)
        ).scalars().all()
        if show_ids
        else []
    )
    mcards = _movie_cards(db, user.id, movies)
    scards = _show_cards(db, user.id, shows)
    # 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]}