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 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 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.routes._common import owned_or_404
from app.routes.admin import admin_user
from app.routes.feed import to_tsquery_str
from app.routes.plex._shared import item_or_404

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

_TS_CONFIG = "public.unaccent_simple"


@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)."""
    libs = db.execute(
        select(PlexLibrary).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,
    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()
    libs = db.execute(
        select(PlexLibrary).where(PlexLibrary.enabled.is_(True))
    ).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"]
    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

    selects = []
    if want_movies:
        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))
        selects.append(mq)
    if want_shows:
        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))
        selects.append(sq)

    if not selects:
        return {
            "scope": scope,
            "total": 0,
            "offset": offset,
            "limit": limit,
            "items": [],
            "episodes": [],
        }
    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()

    items = []
    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,
                }
            )

    episodes = []
    if tsq is not None and want_shows:
        ep_st = aliased(PlexState)
        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),
                PlexItem.search_vector.op("@@")(tsq),
            )
            .order_by(func.ts_rank(PlexItem.search_vector, 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,
    }
    libs = db.execute(
        select(PlexLibrary).where(PlexLibrary.enabled.is_(True))
    ).scalars().all()
    movie_ids = (
        [lb.id for lb in libs if lb.kind == "movie"]
        if scope in ("movie", "both")
        else []
    )
    show_ids = (
        [lb.id for lb in libs if lb.kind == "show"] 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))
            ).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 (best-effort live metadata; same shape as the movie/episode info page). The two live
    # Plex calls are cached per-show + fetched in parallel by _show_enrichment; the per-user shaping
    # (_related_show_cards adds this user's watch-state) stays out of the shared cache.
    rich: dict = {}
    related: list[dict] = []
    try:
        meta, related_raw = _show_enrichment(db, sh.rating_key)
        rich = _rich_meta(meta)
        related = _related_show_cards(db, user.id, related_raw, exclude=sh.id)
    except Exception:
        # Best-effort enrichment only — the local episode list is already assembled, so any live-Plex
        # failure (not configured, HTTP/JSON shape, _rich_meta parse error) must degrade, not 500.
        pass

    # Lazy enrichment: persist the show's full cast so its cast members become filterable (same as the
    # movie info page — the sync only stored the listing's top ~3), and stamp people_enriched_at so the
    # backfill skips it. Best-effort.
    if rich.get("cast"):
        plex_people.enrich_row_people(db, sh, rich["cast"])
        sh.people_enriched_at = datetime.now(timezone.utc)
        try:
            db.commit()
        except Exception:
            db.rollback()

    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": rich.get("imdb_rating"),
            "imdb_id": rich.get("imdb_id"),
            "imdb_url": rich.get("imdb_url"),
            "cast": rich.get("cast", []),
            # 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 _rich_meta(meta: dict) -> dict:
    """Pull the "info page" extras out of a full Plex metadata dict: an IMDb score + link, content
    rating, genres, director(s), studio, tagline, and cast (name/role/photo). Best-effort — any
    missing field is just omitted."""
    # Rating: prefer the explicit IMDb entry in the Rating array, else the generic audienceRating.
    imdb_rating = None
    for r in meta.get("Rating") or []:
        if str(r.get("image") or "").startswith("imdb://"):
            try:
                imdb_rating = round(float(r.get("value")), 1)
            except (TypeError, ValueError):
                pass
            break
    if imdb_rating is None:
        try:
            imdb_rating = round(
                float(meta.get("audienceRating") or meta.get("rating")), 1
            )
        except (TypeError, ValueError):
            imdb_rating = None
    # IMDb id (tt…) from the Guid list → a clickable link.
    imdb_id = None
    for g in meta.get("Guid") or []:
        gid = str(g.get("id") or "")
        if gid.startswith("imdb://"):
            imdb_id = gid.split("imdb://", 1)[1]
            break
    return {
        "imdb_rating": imdb_rating,
        "imdb_id": imdb_id,
        "imdb_url": f"https://www.imdb.com/title/{imdb_id}/" if imdb_id else None,
        "content_rating": meta.get("contentRating"),
        "genres": [g.get("tag") for g in (meta.get("Genre") or []) if g.get("tag")][:8],
        "directors": [
            d.get("tag") for d in (meta.get("Director") or []) if d.get("tag")
        ][:4],
        "studio": meta.get("studio"),
        "tagline": meta.get("tagline"),
        "cast": plex_people.cast_from_meta(meta),
    }


@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": []}
    libs = db.execute(
        select(PlexLibrary).where(PlexLibrary.enabled.is_(True))
    ).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"]
    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}


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

    rich: 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 {}
        rich = _rich_meta(meta)
        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  # metadata extras are best-effort; any live-Plex failure must degrade, not 500 — playback works without them

    # Lazy enrichment: the rich fetch above carries the FULL cast (the sync only stored the listing's
    # top ~3), so persist it onto this movie so every cast member shown here becomes filterable, and
    # stamp people_enriched_at so the backfill skips it. Movie only — episodes aren't in the grid; a
    # show's cast is enriched from /show. Best-effort.
    if it.kind == "movie" and rich.get("cast"):
        plex_people.enrich_row_people(db, it, rich["cast"])
        it.people_enriched_at = datetime.now(timezone.utc)
        try:
            db.commit()
        except Exception:
            db.rollback()

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