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

import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, 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.cards import episode_card, leaf_card, movie_card, show_card
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.plex.lookups import by_rating_key
from app.plex.query import (
    FIELD_COLUMNS,
    TS_CONFIG,
    LibraryFilters,
    apply_meta_filters,
    field_recheck,
    genres_col,
    like_escape,
    meta_filter_conds,
    split_csv,
    title_col,
    vector_col,
)
from app.routes._common import owned_or_404
from app.routes.plex._shared import item_or_404
from app.utils import to_tsquery_str

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


@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 _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 _norm_lang(lang: str | None) -> str:
    """The overlay has an EN base + an optional HU twin; anything other than `hu` serves EN."""
    return "hu" if lang == "hu" else "en"


@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",
    lang: str = "en",
    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)
    lang = _norm_lang(lang)
    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"),
                title_col(PlexItem, lang).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(vector_col(PlexItem, lang), 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, lang)
        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(vector_col(PlexItem, lang).op("@@")(tsq))
        if recheck is not None:
            mq = mq.where(recheck(PlexItem, field, tsq, lang))
        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"),
                title_col(PlexShow, lang).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(vector_col(PlexShow, lang), 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, lang)
        if tsq is not None:
            sq = sq.where(vector_col(PlexShow, lang).op("@@")(tsq))
        if recheck is not None:
            sq = sq.where(recheck(PlexShow, field, tsq, lang))
        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",
    lang: str = "en",
    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
    lang = _norm_lang(lang)
    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, lang))
        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.
        # In HU the vocabulary comes from the active-language genre column (coalesce(genres_hu, genres)).
        gcol = genres_col(model, lang)
        expanded = (
            wsq(
                select(func.jsonb_array_elements_text(gcol).label("g")),
                model,
                ids,
                genre_pp,
                kind_movie,
                True,
            )
            .where(func.jsonb_typeof(gcol) == "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 split_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 _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,
    lang: str = "en",
    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."""
    hu = _norm_lang(lang) == "hu"
    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 = _relatedshow_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_hu or sh.title) if hu else sh.title,
            "summary": (sh.summary_hu or sh.summary) if hu else 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_hu or sh.genres if hu else 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 _relatedshow_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 = split_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,
    lang: str = "en",
    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."""
    lang = _norm_lang(lang)
    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()

    hu = lang == "hu"
    # 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_hu or it.genres if hu else 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_hu or it.title) if hu else it.title,
        "summary": (it.summary_hu or it.summary) if hu else 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"),
        # Audience score (Plex audienceRating) served alongside the IMDb score — the UI shows the
        # IMDb badge when imdb_rating is present, else this. Episodes carry it on the row too.
        "rating": it.rating,
        "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_hu or show.title if hu else 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),
    }