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

"""Plex browse: libraries, facets, collections, playlists, show/item detail, people."""

import threading
import time
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import (
    Integer,
    String,
    and_,
    case,
    cast,
    delete,
    func,
    literal,
    null,
    or_,
    select,
    update,
)
from sqlalchemy.orm import Session, aliased

from app import sysconfig
from app.auth import admin_user, current_user
from app.db import get_db
from app.models import (
    PlexCollection,
    PlexItem,
    PlexLibrary,
    PlexPerson,
    PlexPlaylist,
    PlexPlaylistItem,
    PlexSeason,
    PlexShow,
    PlexState,
    User,
)
from app.plex import detail as plex_detail
from app.plex import people as plex_people
from app.plex import rollup, subtitles
from app.plex import sync as plex_sync
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key
from app.feedquery import to_tsquery_str
from app.routes._common import owned_or_404
from app.routes.plex._shared import item_or_404

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

_TS_CONFIG = "public.unaccent_simple"

# Field-scoped search: the scope-pill in the header search box narrows a full-text query to ONE
# facet of the row. `all` uses the GIN-indexed weighted `search_vector` untouched (byte-identical to
# the pre-scope behaviour); a specific field keeps that GIN prefilter AND re-checks a per-column
# tsvector, so the scoped hits are always a strict subset of Everywhere. The per-field expressions
# mirror the `search_vector` Computed definition (same `_TS_CONFIG`, same `left(summary, 1000)`
# truncation) so scoping never surfaces a match the vector itself wouldn't. `episode` is handled
# separately in the query body (it targets episode rows, not the movie/show grid).
_FIELD_COLUMNS = ("title", "people", "summary", "episode")


def _field_recheck(model, field: str, tsq):
    """A per-field `to_tsvector(...) @@ tsq` clause for a non-`all`, non-`episode` scope, or None."""
    if field == "title":
        expr = func.to_tsvector(_TS_CONFIG, func.coalesce(model.title, ""))
    elif field == "people":
        expr = func.to_tsvector(_TS_CONFIG, func.coalesce(model.people_text, ""))
    elif field == "summary":
        expr = func.to_tsvector(
            _TS_CONFIG, func.left(func.coalesce(model.summary, ""), 1000)
        )
    else:
        return None
    return expr.op("@@")(tsq)


@router.get("/libraries")
def list_libraries(
    user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """The mirrored libraries the user can browse (drives the Plex scope selector). HU overlay
    libraries are excluded — they're not browsable, they only feed the EN twin's Hungarian metadata."""
    libs = db.execute(
        select(PlexLibrary)
        .where(PlexLibrary.overlay_of.is_(None))
        .order_by(PlexLibrary.kind.desc(), PlexLibrary.title)
    ).scalars().all()
    out = []
    for lib in libs:
        if lib.kind == "movie":
            count = db.scalar(
                select(func.count(PlexItem.id)).where(
                    PlexItem.library_id == lib.id, PlexItem.kind == "movie"
                )
            )
        else:
            count = db.scalar(
                select(func.count(PlexShow.id)).where(PlexShow.library_id == lib.id)
            )
        out.append(
            {
                "key": lib.plex_key,
                "title": lib.title,
                "kind": lib.kind,
                "count": count or 0,
            }
        )
    return {"enabled": sysconfig.get_bool(db, "plex_enabled"), "libraries": out}


def _movie_card(it: PlexItem, st: PlexState | None) -> dict:
    return {
        "id": it.rating_key,
        "type": "movie",
        "title": it.title,
        "year": it.year,
        "duration_seconds": it.duration_s,
        "thumb": f"/api/plex/image/{it.rating_key}",
        "playable": it.playable,
        "status": st.status if st else "new",
        "position_seconds": st.position_seconds if st else 0,
    }


def _show_card(sh: PlexShow, status: str = "new") -> dict:
    return {
        "id": sh.rating_key,
        "type": "show",
        "title": sh.title,
        "year": sh.year,
        "thumb": f"/api/plex/image/{sh.rating_key}",
        "season_count": sh.child_count,
        # Aggregate watch-state across the show's episodes (new | in_progress | watched) → grid badge.
        "status": status,
    }


def _lib_key(db: Session, library_id: int | None) -> str | None:
    """The Plex section key for a library id — the info/show page passes it to the admin collection
    editor (in the unified cross-library view there's no single library prop otherwise)."""
    if library_id is None:
        return None
    lib = db.get(PlexLibrary, library_id)
    return lib.plex_key if lib else None


def _episode_card(e: PlexItem, st: PlexState | None) -> dict:
    return {
        "id": e.rating_key,
        "type": "episode",
        "title": e.title,
        "summary": e.summary,
        "season_number": e.season_number,
        "episode_number": e.episode_number,
        "duration_seconds": e.duration_s,
        "thumb": f"/api/plex/image/{e.rating_key}",
        "playable": e.playable,
        "status": st.status if st else "new",
        "position_seconds": st.position_seconds if st else 0,
    }


def _leaf_card(
    it: PlexItem,
    st: PlexState | None,
    show_title: str | None = None,
    show_rk: str | None = None,
) -> dict:
    """A playable-leaf card (movie or episode). Used by playlists, which can mix both kinds.
    `show_rk` (the show's rating_key) lets the client group a playlist's episodes by show/season."""
    if it.kind == "episode":
        card = _episode_card(it, st)
        card["show_title"] = show_title
        card["show_id"] = show_rk
        return card
    return _movie_card(it, st)


def _csv(v: str | None) -> list[str]:
    return [s.strip() for s in v.split(",") if s.strip()] if v else []


_ADDED_WINDOWS = {"24h": 1, "1w": 7, "1m": 30, "6m": 182, "1y": 365}


def _added_cutoff(within: str | None) -> datetime | None:
    days = _ADDED_WINDOWS.get(within or "")
    return datetime.now(timezone.utc) - timedelta(days=days) if days else None


def _meta_filter_conds(model, p: dict) -> list:
    """The shared metadata-filter conditions (identical column names on plex_items + plex_shows) for
    `p` (the request's filter params). Duration is movie-only, handled by the caller. Returned as a
    list so faceted search can apply all-but-one group. DRY for browse, unified, and /facets."""
    conds: list = []
    gsel = _csv(p.get("genres"))
    if gsel:
        gc = [model.genres.contains([g]) for g in gsel]
        conds.append(and_(*gc) if p.get("genre_mode") == "all" else or_(*gc))
    crs = _csv(p.get("content_ratings"))
    if crs:
        conds.append(model.content_rating.in_(crs))
    if p.get("year_min") is not None:
        conds.append(model.year >= p["year_min"])
    if p.get("year_max") is not None:
        conds.append(model.year <= p["year_max"])
    if p.get("rating_min") is not None:
        conds.append(model.rating >= p["rating_min"])
    cutoff = _added_cutoff(p.get("added_within"))
    if cutoff is not None:
        conds.append(model.added_at >= cutoff)
    for d in _csv(p.get("directors")):  # AND
        conds.append(model.directors.contains([d]))
    alist = _csv(p.get("actors"))
    if alist:  # AND (all) or OR (any) — default OR, mirroring genre_mode
        ac = [model.cast_names.contains([a]) for a in alist]
        conds.append(and_(*ac) if p.get("actor_mode") == "all" else or_(*ac))
    stds = _csv(p.get("studios"))
    if stds:  # OR
        conds.append(model.studio.in_(stds))
    if p.get("collection"):
        conds.append(model.collection_keys.contains([p["collection"]]))
    return conds


def _apply_meta_filters(q, model, p: dict):
    conds = _meta_filter_conds(model, p)
    return q.where(*conds) if conds else q


@dataclass
class LibraryFilters:
    """The shared sidebar filter query-params for /library + /facets, declared once and injected via
    ``Depends()``. `as_dict()` yields the `p` dict the query builders read; `duration_*` is included
    for /facets' wsq but ignored by `_meta_filter_conds` (both callers apply duration separately)."""

    genres: str | None = None
    genre_mode: str = "any"
    content_ratings: str | None = None
    year_min: int | None = None
    year_max: int | None = None
    rating_min: float | None = None
    duration_min: int | None = None
    duration_max: int | None = None
    added_within: str | None = None
    directors: str | None = None
    actors: str | None = None
    actor_mode: str = (
        "any"  # how multiple actors combine: any (OR, default) | all (AND)
    )
    studios: str | None = None
    collection: str | None = None

    def as_dict(self) -> dict:
        return {
            "genres": self.genres,
            "genre_mode": self.genre_mode,
            "content_ratings": self.content_ratings,
            "year_min": self.year_min,
            "year_max": self.year_max,
            "rating_min": self.rating_min,
            "duration_min": self.duration_min,
            "duration_max": self.duration_max,
            "added_within": self.added_within,
            "directors": self.directors,
            "actors": self.actors,
            "actor_mode": self.actor_mode,
            "studios": self.studios,
            "collection": self.collection,
        }


@router.get("/library")
def unified_library(
    scope: str = "both",
    q: str | None = None,
    field: str = "all",
    sort: str = "added",
    sort_dir: str = "desc",
    show: str = "all",
    offset: int = 0,
    limit: int = Query(default=40, ge=1, le=100),
    f: LibraryFilters = Depends(),
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Unified cross-library browse: movies + shows in ONE mixed, paginated feed, scoped to
    movie|show|both, with the shared filters/search/watch-state (a show's state is the aggregate of
    its episodes). On a search that also matches episodes (and shows are in scope), the matching
    episodes come back in a separate `episodes` list (grouped result — the 'Episodes' section)."""
    offset = max(0, offset)
    p = f.as_dict()
    movie_lib_ids, show_lib_ids = _content_libs(db)
    want_movies = scope in ("movie", "both") and bool(movie_lib_ids)
    want_shows = scope in ("show", "both") and bool(show_lib_ids)

    ts = to_tsquery_str(q) if q else None
    tsq = func.to_tsquery(_TS_CONFIG, ts) if ts else None
    if field not in _FIELD_COLUMNS and field != "all":
        field = "all"
    # `episode` scope targets episode rows only — the movie/show grid stays empty; the recheck for
    # the specific title/people/summary fields is a strict narrowing on top of the GIN prefilter.
    grid_field = field != "episode"
    recheck = (
        _field_recheck if tsq is not None and field not in ("all", "episode") else None
    )

    selects = []
    if want_movies and grid_field:
        st = aliased(PlexState)
        m_status = case(
            (st.status == "watched", "watched"),
            (
                and_(
                    st.position_seconds > 0,
                    or_(st.status.is_(None), st.status != "watched"),
                ),
                "in_progress",
            ),
            else_="new",
        )
        mq = (
            select(
                PlexItem.rating_key.label("rk"),
                literal("movie").label("kind"),
                PlexItem.title.label("title"),
                PlexItem.year.label("year"),
                PlexItem.rating.label("rating"),
                PlexItem.added_at.label("added_at"),
                PlexItem.originally_available_at.label("rel"),
                PlexItem.duration_s.label("duration_s"),
                PlexItem.playable.label("playable"),
                cast(null(), Integer).label("season_count"),
                func.coalesce(st.position_seconds, 0).label("position_seconds"),
                m_status.label("status"),
                (
                    func.ts_rank(PlexItem.search_vector, tsq)
                    if tsq is not None
                    else literal(0.0)
                ).label("rank"),
            )
            .outerjoin(st, and_(st.item_id == PlexItem.id, st.user_id == user.id))
            .where(PlexItem.kind == "movie", PlexItem.library_id.in_(movie_lib_ids))
        )
        if show == "watched":
            mq = mq.where(st.status == "watched")
        elif show == "in_progress":
            mq = mq.where(
                st.position_seconds > 0,
                or_(st.status.is_(None), st.status != "watched"),
            )
        elif show == "unwatched":
            mq = mq.where(
                or_(st.status.is_(None), st.status.notin_(["watched", "hidden"]))
            )
        else:
            mq = mq.where(or_(st.status.is_(None), st.status != "hidden"))
        mq = _apply_meta_filters(mq, PlexItem, p)
        if f.duration_min is not None:
            mq = mq.where(PlexItem.duration_s >= f.duration_min)
        if f.duration_max is not None:
            mq = mq.where(PlexItem.duration_s <= f.duration_max)
        if tsq is not None:
            mq = mq.where(PlexItem.search_vector.op("@@")(tsq))
        if recheck is not None:
            mq = mq.where(recheck(PlexItem, field, tsq))
        selects.append(mq)
    if want_shows and grid_field:
        agg = rollup.agg_subquery(db, user.id).subquery()
        tot = func.coalesce(agg.c.total, 0)
        wat = func.coalesce(agg.c.watched, 0)
        inp = func.coalesce(agg.c.inprog, 0)
        s_status = case(
            (and_(tot > 0, wat >= tot), "watched"),
            (or_(wat > 0, inp > 0), "in_progress"),
            else_="new",
        )
        sq = (
            select(
                PlexShow.rating_key.label("rk"),
                literal("show").label("kind"),
                PlexShow.title.label("title"),
                PlexShow.year.label("year"),
                PlexShow.rating.label("rating"),
                PlexShow.added_at.label("added_at"),
                PlexShow.originally_available_at.label("rel"),
                cast(null(), Integer).label("duration_s"),
                cast(null(), String).label("playable"),
                PlexShow.child_count.label("season_count"),
                literal(0).label("position_seconds"),
                s_status.label("status"),
                (
                    func.ts_rank(PlexShow.search_vector, tsq)
                    if tsq is not None
                    else literal(0.0)
                ).label("rank"),
            )
            .outerjoin(agg, agg.c.show_id == PlexShow.id)
            .where(PlexShow.library_id.in_(show_lib_ids))
        )
        if show == "watched":
            sq = sq.where(tot > 0, wat >= tot)
        elif show == "in_progress":
            sq = sq.where(or_(wat > 0, inp > 0), wat < tot)
        elif show == "unwatched":
            sq = sq.where(wat == 0, inp == 0)
        sq = _apply_meta_filters(sq, PlexShow, p)
        if tsq is not None:
            sq = sq.where(PlexShow.search_vector.op("@@")(tsq))
        if recheck is not None:
            sq = sq.where(recheck(PlexShow, field, tsq))
        selects.append(sq)

    items = []
    total = 0
    if selects:
        union = (
            selects[0] if len(selects) == 1 else selects[0].union_all(*selects[1:])
        )
        u = union.subquery()
        total = db.scalar(select(func.count()).select_from(u)) or 0

        if tsq is not None:
            order = [u.c.rank.desc()]
        else:
            col = {
                "title": func.lower(u.c.title),
                "year": u.c.year,
                "rating": u.c.rating,
                "duration": u.c.duration_s,
                "release": u.c.rel,
            }.get(sort, u.c.added_at)
            order = [(col.asc() if sort_dir == "asc" else col.desc()).nullslast()]
        order.append(u.c.rk.desc())
        rows = db.execute(
            select(u).order_by(*order).offset(offset).limit(limit)
        ).all()

        for r in rows:
            if r.kind == "movie":
                items.append(
                    {
                        "id": r.rk,
                        "type": "movie",
                        "title": r.title,
                        "year": r.year,
                        "duration_seconds": r.duration_s,
                        "thumb": f"/api/plex/image/{r.rk}",
                        "playable": r.playable,
                        "status": r.status,
                        "position_seconds": r.position_seconds,
                    }
                )
            else:
                items.append(
                    {
                        "id": r.rk,
                        "type": "show",
                        "title": r.title,
                        "year": r.year,
                        "thumb": f"/api/plex/image/{r.rk}",
                        "season_count": r.season_count,
                        "status": r.status,
                    }
                )

    # Episode "grouped result" section: shown for Everywhere (full-vector match, today's behaviour)
    # and for the Episode scope (title-only match, empty item grid). The title/people/summary scopes
    # target the movie/show grid, so they suppress episodes.
    episodes = []
    if tsq is not None and want_shows and field in ("all", "episode"):
        ep_st = aliased(PlexState)
        ep_vec = (
            func.to_tsvector(_TS_CONFIG, func.coalesce(PlexItem.title, ""))
            if field == "episode"
            else PlexItem.search_vector
        )
        eps = db.execute(
            select(PlexItem, ep_st, PlexShow.title.label("show_title"))
            .outerjoin(
                ep_st, and_(ep_st.item_id == PlexItem.id, ep_st.user_id == user.id)
            )
            .outerjoin(PlexShow, PlexShow.id == PlexItem.show_id)
            .where(
                PlexItem.kind == "episode",
                PlexItem.library_id.in_(show_lib_ids),
                ep_vec.op("@@")(tsq),
            )
            .order_by(func.ts_rank(ep_vec, tsq).desc())
            .limit(24)
        ).all()
        for e, s, show_title in eps:
            card = _episode_card(e, s)
            card["show_title"] = show_title
            episodes.append(card)

    return {
        "scope": scope,
        "total": total,
        "offset": offset,
        "limit": limit,
        "items": items,
        "episodes": episodes,
    }


@router.get("/facets")
def facets(
    scope: str = "both",
    show: str = "all",
    f: LibraryFilters = Depends(),
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Available filter values for the unified library (scope movie|show|both), ADAPTED to the active
    filters (faceted search): each group's values/bounds are computed with all the OTHER active
    filters applied but NOT its own — so selecting a filter narrows what the other groups offer, while
    a multi-select group still lets you add more. Merged across the scope's libraries; duration is
    movie-only. (Search text is not factored in — facets track the sidebar filters.)"""
    empty = {
        "genres": [],
        "content_ratings": [],
        "year_min": None,
        "year_max": None,
        "rating_max": None,
        "duration_min": None,
        "duration_max": None,
    }
    all_movie_ids, all_show_ids = _content_libs(db)
    movie_ids = all_movie_ids if scope in ("movie", "both") else []
    show_ids = all_show_ids if scope in ("show", "both") else []
    if not movie_ids and not show_ids:
        return empty
    p = f.as_dict()

    uid = user.id

    def wsq(query, model, ids, pp, kind_movie, with_dur):
        """Apply the (self-excluded) metadata filters + library scope + the per-user watch-state to a
        facet aggregation query, so the facets reflect exactly the visible result set."""
        query = query.where(model.library_id.in_(ids), *_meta_filter_conds(model, pp))
        if kind_movie:
            query = query.where(model.kind == "movie")
            if with_dur and pp.get("duration_min") is not None:
                query = query.where(model.duration_s >= pp["duration_min"])
            if with_dur and pp.get("duration_max") is not None:
                query = query.where(model.duration_s <= pp["duration_max"])
            # Always join watch-state and mirror the grid's status branches (unified_library) exactly —
            # hidden movies never appear in the grid, so they must not inflate facet counts/bounds either.
            st = aliased(PlexState)
            query = query.outerjoin(st, and_(st.item_id == model.id, st.user_id == uid))
            if show == "watched":
                query = query.where(st.status == "watched")
            elif show == "in_progress":
                query = query.where(
                    st.position_seconds > 0,
                    or_(st.status.is_(None), st.status != "watched"),
                )
            elif show == "unwatched":
                query = query.where(
                    or_(st.status.is_(None), st.status.notin_(["watched", "hidden"]))
                )
            else:
                query = query.where(or_(st.status.is_(None), st.status != "hidden"))
        else:
            if show in ("watched", "in_progress", "unwatched"):
                agg = rollup.agg_subquery(db, uid).subquery()
                query = query.outerjoin(agg, agg.c.show_id == model.id)
                tot = func.coalesce(agg.c.total, 0)
                wat = func.coalesce(agg.c.watched, 0)
                inp = func.coalesce(agg.c.inprog, 0)
                if show == "watched":
                    query = query.where(tot > 0, wat >= tot)
                elif show == "in_progress":
                    query = query.where(or_(wat > 0, inp > 0), wat < tot)
                else:
                    query = query.where(wat == 0, inp == 0)
        return query

    genre_counts: dict[str, int] = {}
    cr_counts: dict[str, int] = {}
    years: list[int] = []
    ratings: list[float] = []
    durs: list[int] = []

    def per_table(model, ids, kind_movie: bool) -> None:
        if not ids:
            return
        # Genres — in "any"/OR mode, exclude the genre filter itself so each offered genre is a valid
        # OR-addition that only widens (keep adding). In "all"/AND mode, KEEP the filter applied so we
        # offer only genres that CO-OCCUR on the current result set — a non-co-occurring genre would AND
        # the result down to zero, i.e. a dead-end chip. (The already-selected genres still appear, since
        # every matching title carries them, so they stay togglable.)
        genre_pp = p if p.get("genre_mode") == "all" else {**p, "genres": None}
        # Only rows whose `genres` is a JSON array can be unnested. A genre-less item stores JSON
        # `null` (SQLAlchemy's JSONB persists a Python None as JSON null, not SQL NULL), and
        # jsonb_array_elements_text() raises on a scalar ("cannot extract elements from a scalar") —
        # so a single metadata-poor library (e.g. one bare Shows library with no genres) 500'd the
        # whole facets endpoint. Mirror the exact guard people.py already applies to the same columns.
        expanded = (
            wsq(
                select(func.jsonb_array_elements_text(model.genres).label("g")),
                model,
                ids,
                genre_pp,
                kind_movie,
                True,
            )
            .where(func.jsonb_typeof(model.genres) == "array")
            .subquery()
        )
        for g, c in db.execute(
            select(expanded.c.g, func.count()).group_by(expanded.c.g)
        ).all():
            genre_counts[g] = genre_counts.get(g, 0) + c
        # Content ratings — exclude its own.
        crq = (
            wsq(
                select(model.content_rating, func.count()),
                model,
                ids,
                {**p, "content_ratings": None},
                kind_movie,
                True,
            )
            .where(model.content_rating.isnot(None))
            .group_by(model.content_rating)
        )
        for cr, c in db.execute(crq).all():
            cr_counts[cr] = cr_counts.get(cr, 0) + c
        # Year bounds — exclude the year filter.
        by = db.execute(
            wsq(
                select(func.min(model.year), func.max(model.year)),
                model,
                ids,
                {**p, "year_min": None, "year_max": None},
                kind_movie,
                True,
            )
        ).first()
        if by[0] is not None:
            years.append(by[0])
        if by[1] is not None:
            years.append(by[1])
        # Rating max — exclude the rating filter.
        br = db.scalar(
            wsq(
                select(func.max(model.rating)),
                model,
                ids,
                {**p, "rating_min": None},
                kind_movie,
                True,
            )
        )
        if br is not None:
            ratings.append(float(br))
        # Duration bounds (movies) — exclude the duration filter.
        if kind_movie:
            bd = db.execute(
                wsq(
                    select(func.min(model.duration_s), func.max(model.duration_s)),
                    model,
                    ids,
                    p,
                    kind_movie,
                    False,
                )
            ).first()
            if bd[0] is not None:
                durs.append(bd[0])
            if bd[1] is not None:
                durs.append(bd[1])

    per_table(PlexItem, movie_ids, True)
    per_table(PlexShow, show_ids, False)
    # Always surface the actively-selected genres (even a zero-match AND combo, where they'd otherwise
    # be absent): the sidebar renders genres solely from this list and hides the whole genre section —
    # chips AND the Any/All toggle — when it's empty, which would trap the user with no per-chip way to
    # undo a zero-result selection. Count 0 is fine (genre chips don't display counts).
    for g in _csv(f.genres):
        genre_counts.setdefault(g, 0)
    return {
        "genres": [
            {"value": g, "count": c}
            for g, c in sorted(genre_counts.items(), key=lambda kv: kv[0])
        ],
        "content_ratings": [
            {"value": cr, "count": c}
            for cr, c in sorted(cr_counts.items(), key=lambda kv: -kv[1])
        ],
        "year_min": min(years) if years else None,
        "year_max": max(years) if years else None,
        "rating_max": max(ratings) if ratings else None,
        "duration_min": min(durs) if durs else None,
        "duration_max": max(durs) if durs else None,
    }


def _like_escape(s: str) -> str:
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _collection_editable(c: PlexCollection) -> bool:
    """Effective editability: the admin marked it editable AND it's a plain manual collection — never
    smart, never an external auto-list (IMDb/TMDb/…), which are managed outside Plex."""
    return bool(c.editable) and not c.smart and _collection_source(c) == "collection"


def _collection_card(c: PlexCollection) -> dict:
    return {
        "id": c.rating_key,
        "title": c.title,
        "summary": c.summary,
        "thumb": f"/api/plex/image/{c.rating_key}" if c.thumb_key else None,
        "child_count": c.child_count,
        "smart": c.smart,
        "source": _collection_source(c),
        "editable": c.editable,
        "can_edit": _collection_editable(
            c
        ),  # whether the edit controls should be offered (admin UI)
    }


@router.get("/collections")
def collections(
    library: str | None = None,
    q: str | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mirrored collections as cards (most-populated first), optionally name-filtered by `q`. With a
    `library` plex_key → that one library's collections (used by the collection editor). WITHOUT one →
    the UNION across all enabled libraries (the unified-scope sidebar picker, which has no single
    library selected)."""
    if library is not None:
        lib = db.execute(
            select(PlexLibrary).where(PlexLibrary.plex_key == str(library))
        ).scalars().first()
        if lib is None:
            return {"collections": []}
        query = select(PlexCollection).where(PlexCollection.library_id == lib.id)
    else:
        enabled_ids = list(
            db.execute(
                select(PlexLibrary.id).where(
                    PlexLibrary.enabled.is_(True), PlexLibrary.overlay_of.is_(None)
                )
            ).scalars()
        )
        if not enabled_ids:
            return {"collections": []}
        query = select(PlexCollection).where(
            PlexCollection.library_id.in_(enabled_ids)
        )
    term = (q or "").strip()
    if term:
        query = query.where(PlexCollection.title.ilike(f"%{_like_escape(term)}%"))
    rows = db.execute(
        query.order_by(
            PlexCollection.child_count.desc().nullslast(),
            func.lower(PlexCollection.title),
        )
    ).scalars().all()
    return {"collections": [_collection_card(c) for c in rows]}


def _editable_collection_or_403(db: Session, rating_key: str) -> PlexCollection:
    col = by_rating_key(db, PlexCollection, rating_key)
    if col is None:
        raise HTTPException(status_code=404, detail="Unknown collection")
    if not _collection_editable(col):
        raise HTTPException(status_code=403, detail="This collection is read-only")
    return col


class CreateCollectionIn(BaseModel):
    library: str = ""
    title: str = ""
    item_rating_key: str = ""


@router.post("/collections")
def create_collection(
    body: CreateCollectionIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Create a manual collection (seeded with one item — Plex needs at least one). Editable by default."""
    library = body.library
    title = body.title.strip()
    seed = body.item_rating_key
    if not title or not seed:
        raise HTTPException(
            status_code=400, detail="title and item_rating_key are required"
        )
    lib = db.execute(
        select(PlexLibrary).where(PlexLibrary.plex_key == library)
    ).scalars().first()
    if lib is None:
        raise HTTPException(status_code=404, detail="Unknown library")
    try:
        with PlexClient(db) as plex:
            new = plex.create_collection(lib.plex_key, title, seed)
            col = plex_sync.resync_collection(db, plex, lib, str(new.get("ratingKey")))
            col.editable = True
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return _collection_card(col)


@router.post("/collections/{rating_key}/items/{item_rating_key}")
def collection_add_item(
    rating_key: str,
    item_rating_key: str,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    try:
        with PlexClient(db) as plex:
            plex.add_collection_item(rating_key, str(item_rating_key))
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return _collection_card(col)


@router.delete("/collections/{rating_key}/items/{item_rating_key}")
def collection_remove_item(
    rating_key: str,
    item_rating_key: str,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    try:
        with PlexClient(db) as plex:
            plex.remove_collection_item(rating_key, str(item_rating_key))
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return _collection_card(col)


class CollectionRenameIn(BaseModel):
    title: str = ""


@router.patch("/collections/{rating_key}")
def collection_rename(
    rating_key: str,
    body: CollectionRenameIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=400, detail="title is required")
    lib = db.get(PlexLibrary, col.library_id)
    try:
        with PlexClient(db) as plex:
            plex.rename_collection(lib.plex_key, rating_key, title)
            col = plex_sync.resync_collection(db, plex, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return _collection_card(col)


@router.delete("/collections/{rating_key}")
def collection_delete(
    rating_key: str, _: User = Depends(admin_user), db: Session = Depends(get_db)
) -> dict:
    col = _editable_collection_or_403(db, rating_key)
    lib = db.get(PlexLibrary, col.library_id)
    try:
        with PlexClient(db) as plex:
            plex.delete_collection(rating_key)
        plex_sync.delete_collection_local(db, lib, rating_key)
        db.commit()
    except (PlexError, PlexNotConfigured) as e:
        db.rollback()
        raise HTTPException(status_code=422, detail=f"Plex write failed: {e}")
    return {"deleted": rating_key}


class CollectionEditableIn(BaseModel):
    editable: bool = False


@router.post("/collections/{rating_key}/editable")
def collection_set_editable(
    rating_key: str,
    body: CollectionEditableIn,
    _: User = Depends(admin_user),
    db: Session = Depends(get_db),
) -> dict:
    """Mark an existing Plex collection as 'mine to edit' (or unmark). Smart / auto-list collections
    (IMDb/TMDb/…) can never be made editable — they're regenerated outside Plex."""
    col = by_rating_key(db, PlexCollection, rating_key)
    if col is None:
        raise HTTPException(status_code=404, detail="Unknown collection")
    want = body.editable
    if want and (col.smart or _collection_source(col) != "collection"):
        raise HTTPException(
            status_code=400,
            detail="Smart / auto-list collections can't be made editable",
        )
    col.editable = want
    db.commit()
    return _collection_card(col)


def _own_playlist_or_404(db: Session, pid: int, user: User) -> PlexPlaylist:
    return owned_or_404(db, PlexPlaylist, pid, user, detail="Playlist not found")


def _playlist_card(pl: PlexPlaylist, count: int, thumb_rk: str | None) -> dict:
    return {
        "id": pl.id,
        "title": pl.title,
        "item_count": count,
        "thumb": f"/api/plex/image/{thumb_rk}" if thumb_rk else None,
    }


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)
    )


@router.get("/playlists")
def playlists(
    contains: str | None = None,
    contains_group: str | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """The current user's playlists (newest-touched first), each with an item count + a cover thumb.
    `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))
        )
    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()
        )
    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)
        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
            )
        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
            )
        out.append(card)
    resp: dict = {"playlists": out}
    if group_ids is not None:
        resp["group_size"] = len(group_ids)
    return resp


class CreatePlexPlaylistIn(BaseModel):
    title: str = ""
    item_rating_key: str = ""


@router.post("/playlists")
def create_playlist(
    body: CreatePlexPlaylistIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=400, detail="title is 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
    db.commit()
    return _playlist_card(pl, count, thumb_rk)


@router.get("/playlists/{pid}")
def playlist_detail(
    pid: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    pl = _own_playlist_or_404(db, pid, user)
    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()
    }
    cards = [
        _leaf_card(
            it,
            states.get(it.id),
            shows[it.show_id].title if it.show_id in shows else None,
            shows[it.show_id].rating_key if it.show_id in shows else None,
        )
        for it in items
    ]
    return {"id": pl.id, "title": pl.title, "items": cards}


class RenamePlexPlaylistIn(BaseModel):
    title: str = ""


@router.patch("/playlists/{pid}")
def rename_playlist(
    pid: int,
    body: RenamePlexPlaylistIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    pl = _own_playlist_or_404(db, pid, user)
    title = body.title.strip()
    if not title:
        raise HTTPException(status_code=400, detail="title is required")
    pl.title = 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)


@router.delete("/playlists/{pid}")
def delete_playlist(
    pid: int, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    pl = _own_playlist_or_404(db, pid, user)
    db.delete(pl)
    db.commit()
    return {"deleted": pid}


class PlexItemRefIn(BaseModel):
    item_rating_key: str = ""


@router.post("/playlists/{pid}/items")
def playlist_add_item(
    pid: int,
    body: PlexItemRefIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    pl = _own_playlist_or_404(db, pid, user)
    it = by_rating_key(db, PlexItem, body.item_rating_key)
    if it is None:
        raise HTTPException(status_code=404, detail="Unknown 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)
    db.commit()
    return {"ok": True}


@router.delete("/playlists/{pid}/items/{item_rating_key}")
def playlist_remove_item(
    pid: int,
    item_rating_key: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    pl = _own_playlist_or_404(db, pid, user)
    it = by_rating_key(db, PlexItem, 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)
    db.commit()
    return {"ok": True}


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


@router.post("/playlists/{pid}/items/bulk")
def playlist_add_bulk(
    pid: int,
    body: PlexItemKeysIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Append many items at once (a whole season/show). Already-present items are skipped; the input
    order is preserved for the newly-added ones. Returns how many were added + the new total."""
    pl = _own_playlist_or_404(db, pid, user)
    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)
    db.commit()
    total = (
        db.scalar(
            select(func.count(PlexPlaylistItem.id)).where(
                PlexPlaylistItem.playlist_id == pid
            )
        )
        or 0
    )
    return {"added": added, "item_count": total}


@router.post("/playlists/{pid}/items/remove-bulk")
def playlist_remove_bulk(
    pid: int,
    body: PlexItemKeysIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Remove many items at once (a whole season/show, or a multi-select). Returns how many were
    removed + the new total."""
    pl = _own_playlist_or_404(db, pid, user)
    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)
    db.commit()
    total = (
        db.scalar(
            select(func.count(PlexPlaylistItem.id)).where(
                PlexPlaylistItem.playlist_id == pid
            )
        )
        or 0
    )
    return {"removed": removed, "item_count": total}


@router.put("/playlists/{pid}/order")
def reorder_playlist(
    pid: int,
    body: PlexItemKeysIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Set the playlist order from an explicit list of item rating_keys (0-based positions)."""
    pl = _own_playlist_or_404(db, pid, user)
    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()
    }
    for pos, rk in enumerate(order):
        iid = id_by_rk.get(rk)
        if iid is not None:
            db.execute(
                update(PlexPlaylistItem)
                .where(
                    PlexPlaylistItem.playlist_id == pid, PlexPlaylistItem.item_id == iid
                )
                .values(position=pos)
            )
    pl.updated_at = datetime.now(timezone.utc)
    db.commit()
    return {"ok": True}


_SHOW_ENRICH_TTL_S = 300

_show_enrich_cache: dict[str, tuple[float, dict, list]] = {}

_show_enrich_lock = threading.Lock()


def _show_enrichment(db: Session, rating_key: str) -> tuple[dict, list]:
    """(meta, related_raw) for a show's live Plex enrichment — served from the per-show TTL cache when
    warm, otherwise fetched with the two Plex calls (metadata + related) run IN PARALLEL. Returns
    ({}, []) when Plex is unavailable/unconfigured (not cached, so it retries on the next open)."""
    now = time.monotonic()
    with _show_enrich_lock:
        hit = _show_enrich_cache.get(rating_key)
        if hit is not None and hit[0] > now:
            return hit[1], hit[2]
    try:
        with PlexClient(db) as plex:
            with ThreadPoolExecutor(max_workers=2) as ex:
                meta_f = ex.submit(plex.metadata, rating_key)
                related_f = ex.submit(
                    plex.related, rating_key
                )  # swallows its own errors → []
                meta = meta_f.result() or {}
                related_raw = related_f.result()
    except (PlexError, PlexNotConfigured):
        return {}, []
    # Don't cache an EMPTY related list: plex.related() swallows a transient failure as [] (same as a
    # genuinely no-related show), so caching it would serve an empty "related" section for the whole TTL
    # even after Plex recovers. Skip the cache in that case (retry next open, as before); shows that DO
    # have related items — the payload worth caching — still get cached.
    if related_raw:
        with _show_enrich_lock:
            _show_enrich_cache[rating_key] = (
                now + _SHOW_ENRICH_TTL_S,
                meta,
                related_raw,
            )
    return meta, related_raw


@router.get("/show/{rating_key}")
def show_detail(
    rating_key: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """A show's seasons + episodes (the drill-down view) with per-user watch state."""
    sh = by_rating_key(db, PlexShow, rating_key)
    if sh is None:
        raise HTTPException(status_code=404, detail="Unknown Plex show")
    seasons = db.execute(
        select(PlexSeason)
        .where(PlexSeason.show_id == sh.id)
        .order_by(PlexSeason.season_number.asc().nullsfirst())
    ).scalars().all()
    eps = db.execute(
        select(PlexItem)
        .where(PlexItem.show_id == sh.id, PlexItem.kind == "episode")
        .order_by(
            PlexItem.season_number.asc().nullsfirst(),
            PlexItem.episode_number.asc().nullsfirst(),
        )
    ).scalars().all()
    states = {
        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()
    }
    by_season: dict[int | None, list] = {}
    all_cards: list[
        dict
    ] = []  # every episode, in season/episode order → show-level rollup
    for e in eps:
        card = _episode_card(e, states.get(e.id))
        by_season.setdefault(e.season_id, []).append(card)
        all_cards.append(card)
    show_roll = rollup.rollup(all_cards)

    # Cast + IMDb + tagline are served from the DB row (persisted by the cast backfill, E5 S5b). The one
    # remaining live call is `related` — a Plex recommendation hub we don't mirror — which also carries
    # the metadata used to lazily fill a not-yet-enriched show. Any live failure just drops `related` and
    # the lazy fill; the row-served detail below still renders (outage-resilient).
    related: list[dict] = []
    try:
        meta, related_raw = _show_enrichment(db, sh.rating_key)
        related = _related_show_cards(db, user.id, related_raw, exclude=sh.id)
        if sh.people_enriched_at is None and meta:
            plex_people.enrich_row_people(db, sh, plex_people.cast_from_meta(meta))
            plex_detail.apply_rich_detail(sh, meta)  # imdb/tagline/full_cast for future views
            sh.people_enriched_at = datetime.now(timezone.utc)
            try:
                db.commit()
            except Exception:
                db.rollback()
    except Exception:
        # Best-effort only — the local episode list is already assembled, so any live-Plex failure
        # (not configured, HTTP/JSON shape) must degrade, not 500.
        pass

    imdb_id = sh.imdb_id
    return {
        "show": {
            "id": sh.rating_key,
            "title": sh.title,
            "summary": sh.summary,
            "year": sh.year,
            "thumb": f"/api/plex/image/{sh.rating_key}",
            "art": f"/api/plex/image/{sh.rating_key}?variant=art",
            "library": _lib_key(db, sh.library_id),
            "content_rating": sh.content_rating,
            "rating": sh.rating,
            "genres": sh.genres or [],
            "studio": sh.studio,
            "season_count": sh.child_count,
            "imdb_rating": sh.imdb_rating,
            "imdb_id": imdb_id,
            "imdb_url": f"https://www.imdb.com/title/{imdb_id}/" if imdb_id else None,
            "cast": sh.full_cast or [],
            # Show-level rollup for the hero action buttons.
            "status": show_roll["status"],
            "resume": show_roll["resume"],
            "first": show_roll["first"],
            "episode_count": show_roll["episode_count"],
            "collection_keys": sh.collection_keys or [],
        },
        "seasons": [_season_block(se, by_season.get(se.id, [])) for se in seasons],
        "related": related,
    }


def _season_block(se: PlexSeason, cards: list[dict]) -> dict:
    """A season's card for the show page (with its aggregate rollup) plus its episodes (used by the
    season subpage, read from the same cached show-detail payload)."""
    roll = rollup.rollup(cards)
    return {
        "id": se.rating_key,
        "season_number": se.season_number,
        "title": se.title
        or (f"Season {se.season_number}" if se.season_number else "Season"),
        "thumb": f"/api/plex/image/{se.rating_key}",
        "episode_count": roll["episode_count"],
        "status": roll["status"],
        "resume": roll["resume"],
        "first": roll["first"],
        "episodes": cards,
    }


def _related_show_cards(
    db: Session, user_id: int, related: list[dict], exclude: int
) -> list[dict]:
    """Map Plex 'related' metadata to show cards for the shows we actually mirror (so they're openable),
    excluding the current show. Order preserved; capped."""
    rks = [str(m.get("ratingKey")) for m in related if m.get("ratingKey")]
    if not rks:
        return []
    shows = {
        s.rating_key: s
        for s in db.execute(
            select(PlexShow).where(
                PlexShow.rating_key.in_(rks[:60]), PlexShow.id != exclude
            )
        ).scalars()
    }
    statuses = rollup.state_map(db, user_id, [s.id for s in shows.values()])
    out: list[dict] = []
    seen: set[str] = set()
    for rk in rks:
        sh = shows.get(rk)
        if sh is None or rk in seen:
            continue
        seen.add(rk)
        out.append(_show_card(sh, statuses.get(sh.id, "new")))
        if len(out) >= 20:
            break
    return out


def _content_libs(db: Session) -> tuple[list[int], list[int]]:
    """The BROWSABLE content libraries, partitioned into (movie_lib_ids, show_lib_ids) — the single
    source of the browse scope. Excludes HU overlay libraries (`overlay_of` set): those are never
    browsed on their own, they only lend their Hungarian metadata to the EN twin's rows (E5 S5)."""
    libs = db.execute(
        select(PlexLibrary).where(
            PlexLibrary.enabled.is_(True), PlexLibrary.overlay_of.is_(None)
        )
    ).scalars().all()
    movie_lib_ids = [lb.id for lb in libs if lb.kind == "movie"]
    show_lib_ids = [lb.id for lb in libs if lb.kind == "show"]
    return movie_lib_ids, show_lib_ids


@router.get("/people/cards")
def people_cards(
    names: str,
    scope: str = "both",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Cards for the actor filter-chips currently applied: each name + its cached headshot + how many
    movies/shows (in the given scope) feature them. Photos come from the plex_people cache (populated
    as items are rich-fetched — info-page view + the rich re-sync); counts are live against cast_names.
    A name with no cached photo yet still returns a card (the UI shows a placeholder)."""
    wanted = _csv(names)[:16]
    if not wanted:
        return {"people": []}
    movie_lib_ids, show_lib_ids = _content_libs(db)
    want_movies = scope in ("movie", "both") and bool(movie_lib_ids)
    want_shows = scope in ("show", "both") and bool(show_lib_ids)
    photos = {
        p.name: p.photo_thumb
        for p in db.execute(
            select(PlexPerson).where(PlexPerson.name.in_(wanted))
        ).scalars()
    }
    out = []
    for nm in wanted:
        count = 0
        if want_movies:
            count += (
                db.scalar(
                    select(func.count(PlexItem.id)).where(
                        PlexItem.kind == "movie",
                        PlexItem.library_id.in_(movie_lib_ids),
                        PlexItem.cast_names.contains([nm]),
                    )
                )
                or 0
            )
        if want_shows:
            count += (
                db.scalar(
                    select(func.count(PlexShow.id)).where(
                        PlexShow.library_id.in_(show_lib_ids),
                        PlexShow.cast_names.contains([nm]),
                    )
                )
                or 0
            )
        out.append({"name": nm, "photo": photos.get(nm), "count": count})
    return {"people": out}


# Cap each filmography section — a prolific person shouldn't return an unbounded list the client then
# mounts a card per row for (the grid paginates at PAGE; this page renders all at once). Ample for a
# personal library; the most-recent titles win via the year-desc order.
_PERSON_SECTION_MAX = 200


@router.get("/person/{name}")
def person_detail(
    name: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """One cast/crew member's filmography: their cached headshot + the movies and shows (in the enabled
    libraries) whose cast OR directors include this name — the inverse of the cast-name → library-filter
    link. The name string is the identity key (Plex cast carries no stable per-person id), so distinct
    people who share a name share a page. Each section is capped at _PERSON_SECTION_MAX. An unknown name
    simply returns empty sections."""
    photo = db.scalar(select(PlexPerson.photo_thumb).where(PlexPerson.name == name))
    movie_lib_ids, show_lib_ids = _content_libs(db)

    movies: list[dict] = []
    if movie_lib_ids:
        items = db.execute(
            select(PlexItem)
            .where(
                PlexItem.kind == "movie",
                PlexItem.library_id.in_(movie_lib_ids),
                or_(
                    PlexItem.cast_names.contains([name]),
                    PlexItem.directors.contains([name]),
                ),
            )
            .order_by(PlexItem.year.desc().nullslast(), func.lower(PlexItem.title))
            .limit(_PERSON_SECTION_MAX)
        ).scalars().all()
        item_ids = [it.id for it in items]
        states = {
            s.item_id: s
            for s in db.execute(
                select(PlexState).where(
                    PlexState.user_id == user.id,
                    PlexState.item_id.in_(item_ids or [0]),
                )
            ).scalars()
        }
        movies = [_movie_card(it, states.get(it.id)) for it in items]

    shows: list[dict] = []
    if show_lib_ids:
        show_rows = db.execute(
            select(PlexShow)
            .where(
                PlexShow.library_id.in_(show_lib_ids),
                or_(
                    PlexShow.cast_names.contains([name]),
                    PlexShow.directors.contains([name]),
                ),
            )
            .order_by(PlexShow.year.desc().nullslast(), func.lower(PlexShow.title))
            .limit(_PERSON_SECTION_MAX)
        ).scalars().all()
        statuses = rollup.state_map(db, user.id, [s.id for s in show_rows])
        shows = [_show_card(sh, statuses.get(sh.id, "new")) for sh in show_rows]

    return {"name": name, "photo": photo, "movies": movies, "shows": shows}


_STRIP_PROVIDERS = ("imdb", "tmdb", "tvdb", "trakt")


def _collection_source(col: PlexCollection) -> str:
    first = (col.title or "").strip().split(" ", 1)[0].lower()
    if first in _STRIP_PROVIDERS:
        return first
    if col.smart:
        return "smart"
    return "collection"


def _collection_strips(db: Session, it: PlexItem, user_id: int) -> list[dict]:
    """The collections this movie belongs to, each with its member titles (playable cards) — the
    'Avatar Collection' strip on the info page. Pure local lookup (no Plex call)."""
    keys = it.collection_keys or []
    if not keys or it.kind != "movie":
        return []
    cols = {
        c.rating_key: c
        for c in db.execute(
            select(PlexCollection).where(PlexCollection.rating_key.in_(keys))
        ).scalars()
    }
    # Smallest (most specific — e.g. a franchise) first; the big auto-lists (IMDb/TMDb Popular) last.
    ordered = sorted(
        (c for c in cols.values()), key=lambda c: (c.child_count or 0, c.title)
    )
    out = []
    for col in ordered:
        rk = col.rating_key
        members = db.execute(
            select(PlexItem)
            .where(
                PlexItem.library_id == it.library_id,
                PlexItem.kind == "movie",
                PlexItem.collection_keys.contains([rk]),
            )
            .order_by(PlexItem.year.asc().nullslast(), func.lower(PlexItem.title))
            .limit(60)
        ).scalars().all()
        ids = [m.id for m in members]
        states = {
            s.item_id: s
            for s in db.execute(
                select(PlexState).where(
                    PlexState.user_id == user_id, PlexState.item_id.in_(ids or [0])
                )
            ).scalars()
        }
        out.append(
            {
                "id": rk,
                "title": col.title,
                "source": _collection_source(col),
                "items": [_movie_card(m, states.get(m.id)) for m in members],
            }
        )
    return out


def _episode_nav(db: Session, it: PlexItem) -> tuple[str | None, str | None]:
    if it.kind != "episode" or it.show_id is None:
        return None, None
    keys = list(
        db.execute(
            select(PlexItem.rating_key)
            .where(PlexItem.show_id == it.show_id, PlexItem.kind == "episode")
            .order_by(
                PlexItem.season_number.asc().nullsfirst(),
                PlexItem.episode_number.asc().nullsfirst(),
            )
        ).scalars()
    )
    try:
        i = keys.index(it.rating_key)
    except ValueError:
        return None, None
    return (keys[i - 1] if i > 0 else None), (
        keys[i + 1] if i < len(keys) - 1 else None
    )


@router.get("/item/{rating_key}")
def item_detail(
    rating_key: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Everything the player needs: metadata, cast, intro/credit markers (from Plex), episode
    prev/next, and the per-user resume position + watch status."""
    it = item_or_404(db, rating_key)
    st = db.execute(
        select(PlexState).where(
            PlexState.user_id == user.id, PlexState.item_id == it.id
        )
    ).scalars().first()

    # 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
    # parsed from this live meta for episodes (never enriched — not in the grid, no HU overlay).
    meta: dict = {}
    markers: list[dict] = []
    audio_streams: list[dict] = []
    subtitle_streams: list[dict] = []
    try:
        with PlexClient(db) as plex:
            meta = plex.metadata(it.rating_key, markers=True) or {}
        for m in meta.get("Marker") or []:
            if m.get("type") in ("intro", "credits"):
                markers.append(
                    {
                        "type": m.get("type"),
                        "start_s": round((m.get("startTimeOffset") or 0) / 1000, 1),
                        "end_s": round((m.get("endTimeOffset") or 0) / 1000, 1),
                    }
                )
        # Audio + subtitle tracks (ordinals among their stream type — used to -map the selection).
        part = ((meta.get("Media") or [{}])[0].get("Part") or [{}])[0]
        for s in part.get("Stream") or []:
            if s.get("streamType") == 2:
                audio_streams.append(
                    {
                        "ord": len(audio_streams),
                        "label": s.get("displayTitle")
                        or s.get("language")
                        or f"Audio {len(audio_streams) + 1}",
                        "language": s.get("languageTag") or s.get("language"),
                        "default": bool(s.get("selected") or s.get("default")),
                    }
                )
            elif s.get("streamType") == 3:
                codec = (s.get("codec") or s.get("format") or "").lower()
                subtitle_streams.append(
                    {
                        "ord": len(subtitle_streams),
                        "label": s.get("displayTitle")
                        or s.get("language")
                        or f"Subtitle {len(subtitle_streams) + 1}",
                        "language": s.get("languageTag") or s.get("language"),
                        "codec": codec,
                        # Text subs (srt/ass/…) can be served as WebVTT tracks; image subs (PGS/VobSub)
                        # can't be turned into text, so the player hides them (burn-in is a later phase).
                        "text": codec not in subtitles.BITMAP_SUBS,
                    }
                )
    except Exception:
        pass  # live extras (markers/streams) are best-effort; any live-Plex failure must degrade, not 500 — playback works without them

    # Lazy enrichment (movies only): the live fetch above carries the FULL cast + rich detail (the sync
    # only stored the listing's top ~3), so fill a not-yet-enriched movie — its cast becomes filterable
    # and future views serve from the row. Episodes aren't enriched; a show's cast is filled from /show.
    if it.kind == "movie" and it.people_enriched_at is None and meta:
        cast = plex_people.cast_from_meta(meta)
        if cast:
            plex_people.enrich_row_people(db, it, cast)
        plex_detail.apply_rich_detail(it, meta)
        it.people_enriched_at = datetime.now(timezone.utc)
        try:
            db.commit()
        except Exception:
            db.rollback()

    # Movies serve rich detail from the DB row; episodes from the live meta just fetched.
    if it.kind == "movie":
        _imdb_id = it.imdb_id
        rich = {
            "cast": it.full_cast or [],
            "imdb_rating": it.imdb_rating,
            "imdb_id": _imdb_id,
            "imdb_url": f"https://www.imdb.com/title/{_imdb_id}/" if _imdb_id else None,
            "content_rating": it.content_rating,
            "genres": it.genres or [],
            "directors": it.directors or [],
            "studio": it.studio,
            "tagline": it.tagline,
        }
    else:
        rich = plex_detail.rich_detail(meta)

    prev_id, next_id = _episode_nav(db, it)
    show = db.get(PlexShow, it.show_id) if it.kind == "episode" and it.show_id else None
    return {
        "id": it.rating_key,
        "kind": it.kind,
        "title": it.title,
        "summary": it.summary,
        "year": it.year,
        "duration_seconds": it.duration_s,
        "playable": it.playable,
        "thumb": f"/api/plex/image/{it.rating_key}",
        "art": f"/api/plex/image/{it.rating_key}?variant=art",
        "library": _lib_key(db, it.library_id),
        "cast": rich.get("cast", []),
        "imdb_rating": rich.get("imdb_rating"),
        "imdb_id": rich.get("imdb_id"),
        "imdb_url": rich.get("imdb_url"),
        "content_rating": rich.get("content_rating"),
        "genres": rich.get("genres", []),
        "directors": rich.get("directors", []),
        "studio": rich.get("studio"),
        "tagline": rich.get("tagline"),
        "markers": markers,
        "audio_streams": audio_streams,
        "subtitle_streams": subtitle_streams,
        "status": st.status if st else "new",
        "position_seconds": st.position_seconds if st else 0,
        "show_title": show.title if show else None,
        "season_number": it.season_number,
        "episode_number": it.episode_number,
        "prev_id": prev_id,
        "next_id": next_id,
        "collections": _collection_strips(db, it, user.id),
    }