"""Plex integration API.

Read endpoints (libraries/browse/show/image) are available to ANY authenticated user (demo +
pure-Siftlode included) — the module is an access layer to the Plex library WITHOUT requiring a
plex.tv account, and playback is a local file. Admin endpoints test the connection and run the
catalog sync.
"""
import hashlib
import logging
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import quote

import httpx
from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
from fastapi.responses import FileResponse
from sqlalchemy import and_, func, or_, text
from sqlalchemy.orm import Session, aliased

from app import sysconfig
from app.auth import current_user, resolved_user_id
from app.config import settings
from app.db import SessionLocal, get_db
from app.models import PlexCollection, PlexItem, PlexLibrary, PlexSeason, PlexShow, PlexState, User
from app.plex import paths as plex_paths
from app.plex import stream as plex_stream
from app.plex import sync as plex_sync
from app.plex.client import PlexClient, PlexError, PlexNotConfigured
from app.routes.admin import admin_user
from app.routes.feed import _to_tsquery_str

log = logging.getLogger("siftlode.plex")

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

_TS_CONFIG = "public.unaccent_simple"


# --- Admin: connectivity + sync ---------------------------------------------------------------


@router.post("/test")
def test_connection(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
    """Admin: verify the configured Plex server URL + token, returning the server name and the
    movie/show sections (for the config library-picker). Also probes ONE real media file through
    the local mount, so a missing bind-mount / wrong path-map is caught here rather than only
    surfacing as a dead player later."""
    try:
        with PlexClient(db) as plex:
            info = plex.server_info()
            sections = plex.sections()
            media_check = _media_check(db, plex, sections)
    except PlexNotConfigured as e:
        raise HTTPException(status_code=400, detail=str(e))
    except PlexError as e:
        raise HTTPException(status_code=422, detail=f"Plex connection failed: {e}")
    return {
        "ok": True,
        "server_name": info.get("friendlyName") or info.get("title1") or "Plex",
        "version": info.get("version"),
        "sections": [
            {"key": str(s.get("key")), "title": s.get("title"), "type": s.get("type"), "count": s.get("count")}
            for s in sections
            if s.get("type") in ("movie", "show")
        ],
        "media_check": media_check,
    }


def _sample_plex_path(db: Session, plex: PlexClient, sections: list[dict]) -> str | None:
    """A real Plex-side media file path to validate the mount + path-map against. Prefers the
    already-mirrored catalog (cheap); before the first sync, pulls one movie leaf from an enabled
    section straight from Plex."""
    row = db.query(PlexItem.file_path).filter(PlexItem.file_path.isnot(None)).first()
    if row and row[0]:
        return row[0]
    wanted = plex_sync._enabled_section_keys(db)
    for s in sections:
        if s.get("type") != "movie":
            continue
        key = str(s.get("key"))
        if wanted is not None and key not in wanted:
            continue
        items, _ = plex.section_items(key, 1, 0, 1)
        if items:
            fp = plex_sync._media_facts(items[0]).get("file_path")
            if fp:
                return fp
    return None


def _media_check(db: Session, plex: PlexClient, sections: list[dict]) -> dict:
    """Resolve a sample Plex file path to its local mount path and confirm it's readable, so the
    admin sees at Test time whether the physical media is actually reachable (bind-mount +
    path-map correct). ``checked=False`` when no sample is available (empty library)."""
    plex_path = _sample_plex_path(db, plex, sections)
    if not plex_path:
        return {"checked": False}
    local_path = plex_paths.map_path(db, plex_path)
    ok = plex_paths.local_media_path(db, plex_path) is not None
    return {"checked": True, "ok": ok, "plex_path": plex_path, "local_path": local_path}


@router.post("/sync")
def trigger_sync(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
    """Admin: mirror the enabled Plex sections into the local catalog. Synchronous (can take a
    while on a large library); a background/scheduled sync also runs periodically."""
    return plex_sync.sync(db)


# --- Read: libraries / browse / show / image --------------------------------------------------


def _enabled() -> None:
    """Guard read endpoints when the module is off (avoids leaking a stale mirror)."""
    # Read endpoints stay usable as long as a mirror exists; the toggle only gates sync + UI.
    return None


@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.query(PlexLibrary).order_by(PlexLibrary.kind.desc(), PlexLibrary.title).all()
    out = []
    for lib in libs:
        if lib.kind == "movie":
            count = db.query(func.count(PlexItem.id)).filter_by(library_id=lib.id, kind="movie").scalar()
        else:
            count = db.query(func.count(PlexShow.id)).filter_by(library_id=lib.id).scalar()
        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) -> 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,
    }


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


@router.get("/browse")
def browse(
    library: str,
    q: str | None = None,
    sort: str = "added",
    show: str = "all",
    offset: int = 0,
    limit: int = Query(default=40, ge=1, le=100),
    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,
    studios: str | None = None,
    collection: str | None = None,
    sort_dir: str = "desc",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """List a library's movies (leaves) or shows, with optional FTS search + sorting + a per-user
    watch-state filter (`show`: all|unwatched|in_progress|watched — movie libraries only). Movie
    libraries return playable movie cards; show libraries return show cards (drill in via /show)."""
    lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
    if lib is None:
        raise HTTPException(status_code=404, detail="Unknown Plex library")
    offset = max(0, offset)
    model = PlexItem if lib.kind == "movie" else PlexShow

    query = db.query(model)
    if lib.kind == "movie":
        query = query.filter(PlexItem.library_id == lib.id, PlexItem.kind == "movie")
        # Per-user watch-state filter (movies only; a show isn't a single watch unit).
        st = aliased(PlexState)
        query = query.outerjoin(st, and_(st.item_id == PlexItem.id, st.user_id == user.id))
        if show == "watched":
            query = query.filter(st.status == "watched")
        elif show == "in_progress":
            query = query.filter(st.position_seconds > 0, or_(st.status.is_(None), st.status != "watched"))
        elif show == "unwatched":
            query = query.filter(or_(st.status.is_(None), st.status.notin_(["watched", "hidden"])))
        else:  # all — hide only the explicitly hidden
            query = query.filter(or_(st.status.is_(None), st.status != "hidden"))
        # --- Metadata filters (movie libraries; @> containment is GIN-indexed) ---
        gsel = _csv(genres)
        if gsel:
            conds = [PlexItem.genres.contains([g]) for g in gsel]
            query = query.filter(and_(*conds) if genre_mode == "all" else or_(*conds))
        crs = _csv(content_ratings)
        if crs:
            query = query.filter(PlexItem.content_rating.in_(crs))
        if year_min is not None:
            query = query.filter(PlexItem.year >= year_min)
        if year_max is not None:
            query = query.filter(PlexItem.year <= year_max)
        if rating_min is not None:
            query = query.filter(PlexItem.rating >= rating_min)
        if duration_min is not None:
            query = query.filter(PlexItem.duration_s >= duration_min)
        if duration_max is not None:
            query = query.filter(PlexItem.duration_s <= duration_max)
        cutoff = _added_cutoff(added_within)
        if cutoff is not None:
            query = query.filter(PlexItem.added_at >= cutoff)
        for d in _csv(directors):  # AND — titles this whole set directed
            query = query.filter(PlexItem.directors.contains([d]))
        for a in _csv(actors):  # AND — titles featuring all these people
            query = query.filter(PlexItem.cast_names.contains([a]))
        stds = _csv(studios)
        if stds:  # OR — a title has one studio
            query = query.filter(PlexItem.studio.in_(stds))
    else:
        query = query.filter(PlexShow.library_id == lib.id)

    # Collection filter applies to both movies and shows (both carry collection_keys).
    if collection:
        query = query.filter(model.collection_keys.contains([collection]))

    ts = _to_tsquery_str(q) if q else None
    tsq = None
    if ts:
        tsq = func.to_tsquery(_TS_CONFIG, ts)
        query = query.filter(model.search_vector.op("@@")(tsq))

    total = query.count()

    if tsq is not None:
        order = [func.ts_rank(model.search_vector, tsq).desc()]
    else:
        if sort == "title":
            col = func.lower(model.title)
        elif sort == "year" and model is PlexItem:
            col = PlexItem.year
        elif sort == "rating" and model is PlexItem:
            col = PlexItem.rating
        elif sort == "duration" and model is PlexItem:
            col = PlexItem.duration_s
        elif sort == "release" and model is PlexItem:
            col = PlexItem.originally_available_at
        else:  # added
            col = model.added_at
        asc = sort_dir == "asc"
        order = [(col.asc() if asc else col.desc()).nullslast()]
    order.append(model.id.desc())

    rows = query.order_by(*order).offset(offset).limit(limit).all()

    if lib.kind == "movie":
        by_item = {r.id: r for r in rows}
        states = {
            s.item_id: s
            for s in db.query(PlexState).filter(
                PlexState.user_id == user.id, PlexState.item_id.in_(list(by_item.keys()) or [0])
            )
        }
        items = [_movie_card(r, states.get(r.id)) for r in rows]
    else:
        items = [_show_card(r) for r in rows]

    return {"kind": lib.kind, "total": total, "offset": offset, "limit": limit, "items": items}


@router.get("/facets")
def facets(
    library: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Available filter values for a movie library — genres + content ratings (with counts) and the
    year/rating/duration bounds — so the sidebar only offers what the library actually contains."""
    empty = {
        "genres": [],
        "content_ratings": [],
        "year_min": None,
        "year_max": None,
        "rating_max": None,
        "duration_min": None,
        "duration_max": None,
    }
    lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
    if lib is None or lib.kind != "movie":
        return empty
    base = "FROM plex_items WHERE library_id = :lib AND kind = 'movie'"
    genres = db.execute(
        text(
            f"SELECT g, count(*) AS c FROM (SELECT jsonb_array_elements_text(genres) AS g {base}) x "
            "GROUP BY g ORDER BY c DESC, g"
        ),
        {"lib": lib.id},
    ).all()
    crs = db.execute(
        text(f"SELECT content_rating, count(*) AS c {base} AND content_rating IS NOT NULL GROUP BY content_rating ORDER BY c DESC"),
        {"lib": lib.id},
    ).all()
    b = db.execute(
        text(f"SELECT min(year), max(year), max(rating), min(duration_s), max(duration_s) {base}"),
        {"lib": lib.id},
    ).first()
    return {
        "genres": [{"value": g, "count": c} for g, c in genres],
        "content_ratings": [{"value": cr, "count": c} for cr, c in crs],
        "year_min": b[0],
        "year_max": b[1],
        "rating_max": float(b[2]) if b[2] is not None else None,
        "duration_min": b[3],
        "duration_max": b[4],
    }


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


@router.get("/people")
def people(
    q: str,
    library: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Cast/crew whose name matches the search term — drives the virtual person cards shown above the
    grid. Returns up to a few matches (most films first) with a count + a photo."""
    term = (q or "").strip()
    lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
    if lib is None or lib.kind != "movie" or len(term) < 2:
        return {"people": []}
    esc = _like_escape(term)
    pfx, word = f"{esc}%", f"% {esc}%"  # name starts with the term, or any of its words does
    rows = db.execute(
        text(
            "SELECT name, kind, count(*) c FROM ("
            "  SELECT jsonb_array_elements_text(cast_names) AS name, 'actor' AS kind FROM plex_items"
            "    WHERE library_id = :lib AND kind = 'movie'"
            "  UNION ALL"
            "  SELECT jsonb_array_elements_text(directors) AS name, 'director' AS kind FROM plex_items"
            "    WHERE library_id = :lib AND kind = 'movie'"
            ") x WHERE name ILIKE :pfx OR name ILIKE :word "
            "GROUP BY name, kind ORDER BY c DESC, name LIMIT 5"
        ),
        {"lib": lib.id, "pfx": pfx, "word": word},
    ).all()
    return {
        "people": [
            {"name": name, "kind": kind, "count": c, "photo": _person_photo(db, lib.id, name, kind)}
            for name, kind, c in rows
        ]
    }


def _person_photo(db: Session, lib_id: int, name: str, kind: str) -> str | None:
    """A person's headshot from one representative film's live metadata (image bytes are disk-cached)."""
    col = PlexItem.directors if kind == "director" else PlexItem.cast_names
    rk = (
        db.query(PlexItem.rating_key)
        .filter(PlexItem.library_id == lib_id, col.contains([name]))
        .limit(1)
        .scalar()
    )
    if not rk:
        return None
    try:
        with PlexClient(db) as plex:
            meta = plex.metadata(rk) or {}
    except (PlexError, PlexNotConfigured):
        return None
    for r in meta.get("Director" if kind == "director" else "Role") or []:
        if r.get("tag") == name:
            thumb = str(r.get("thumb") or "")
            return (
                f"/api/plex/person-image?u={quote(thumb, safe='')}"
                if thumb.startswith(_PERSON_IMG_HOST)
                else None
            )
    return None


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,
        "editable": c.editable,
    }


@router.get("/collections")
def collections(
    library: str,
    q: str | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """The library's mirrored collections as cards (most-populated first). Optional name filter `q`."""
    lib = db.query(PlexLibrary).filter_by(plex_key=str(library)).first()
    if lib is None:
        return {"collections": []}
    query = db.query(PlexCollection).filter(PlexCollection.library_id == lib.id)
    term = (q or "").strip()
    if term:
        query = query.filter(PlexCollection.title.ilike(f"%{_like_escape(term)}%"))
    rows = query.order_by(
        PlexCollection.child_count.desc().nullslast(), func.lower(PlexCollection.title)
    ).all()
    return {"collections": [_collection_card(c) for c in rows]}


@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 = db.query(PlexShow).filter_by(rating_key=str(rating_key)).first()
    if sh is None:
        raise HTTPException(status_code=404, detail="Unknown Plex show")
    seasons = (
        db.query(PlexSeason).filter_by(show_id=sh.id).order_by(PlexSeason.season_number.asc().nullsfirst()).all()
    )
    eps = (
        db.query(PlexItem)
        .filter_by(show_id=sh.id, kind="episode")
        .order_by(PlexItem.season_number.asc().nullsfirst(), PlexItem.episode_number.asc().nullsfirst())
        .all()
    )
    states = {
        s.item_id: s
        for s in db.query(PlexState).filter(
            PlexState.user_id == user.id, PlexState.item_id.in_([e.id for e in eps] or [0])
        )
    }
    by_season: dict[int | None, list] = {}
    for e in eps:
        by_season.setdefault(e.season_id, []).append(_episode_card(e, states.get(e.id)))
    return {
        "show": {
            "id": sh.rating_key,
            "title": sh.title,
            "summary": sh.summary,
            "year": sh.year,
            "thumb": f"/api/plex/image/{sh.rating_key}",
            "art": f"/api/plex/image/{sh.rating_key}?variant=art",
        },
        "seasons": [
            {
                "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}",
                "episodes": by_season.get(se.id, []),
            }
            for se in seasons
        ],
    }


def _image_key(db: Session, rating_key: str, variant: str) -> str | None:
    """The stored Plex image path for a known rating_key (item/show/season). Only mirrored keys
    are proxyable — this is not an open image proxy."""
    it = db.query(PlexItem).filter_by(rating_key=rating_key).first()
    if it is not None:
        return it.art_key if variant == "art" else it.thumb_key
    sh = db.query(PlexShow).filter_by(rating_key=rating_key).first()
    if sh is not None:
        return sh.art_key if variant == "art" else sh.thumb_key
    se = db.query(PlexSeason).filter_by(rating_key=rating_key).first()
    if se is not None:
        return se.thumb_key
    col = db.query(PlexCollection).filter_by(rating_key=rating_key).first()
    if col is not None:
        return col.thumb_key
    return None


# --- Thin on-disk image cache -----------------------------------------------------------------
# Posters/art/cast-photos are proxied on demand; without a cache every scroll re-hits Plex over the
# LAN (janky). We cache the fetched bytes on local disk keyed by a stable id, so repeat views (and
# other users) serve instantly from disk. On-demand + bounded by the library size; no eviction (a
# few hundred MB at most for a big library — acceptable, matching the download disk trade-offs).
_IMG_CACHE = Path(settings.download_root) / ".plex-img-cache"
_EXT_CT = [("jpg", "image/jpeg"), ("png", "image/png"), ("webp", "image/webp")]
_CT_EXT = {ct: ext for ext, ct in _EXT_CT}


def _img_cache_read(key: str) -> tuple[bytes, str] | None:
    for ext, ct in _EXT_CT:
        p = _IMG_CACHE / f"{key}.{ext}"
        try:
            if p.exists() and p.stat().st_size > 0:
                return p.read_bytes(), ct
        except OSError:
            pass
    return None


def _img_cache_write(key: str, data: bytes, ct: str) -> None:
    ext = _CT_EXT.get((ct or "").split(";")[0].strip(), "jpg")
    try:
        _IMG_CACHE.mkdir(parents=True, exist_ok=True)
        tmp = _IMG_CACHE / f"{key}.{ext}.tmp"
        tmp.write_bytes(data)
        tmp.replace(_IMG_CACHE / f"{key}.{ext}")  # atomic publish
    except OSError:
        pass  # cache is best-effort


def _img_response(data: bytes, ct: str) -> Response:
    return Response(content=data, media_type=ct, headers={"Cache-Control": "public, max-age=86400"})


def _require_session(request: Request) -> int:
    """Lightweight auth for the high-fan-out image proxies: validate the signed session cookie WITHOUT
    a DB user-load, so these endpoints never hold a pool connection for the auth. Poster images are
    low-sensitivity; a valid session is enough."""
    uid, _ = resolved_user_id(request)
    if not uid:
        raise HTTPException(status_code=401, detail="Not authenticated")
    return uid


@router.get("/image/{rating_key}")
def image(rating_key: str, request: Request, variant: str = "thumb") -> Response:
    """Proxy a Plex poster/art image (keeps the admin token server-side; no image duplication). A
    thin on-disk cache serves repeat views without re-hitting Plex.

    ⚠️ This endpoint is called in BURSTS (a collections page fires 100+ concurrent poster requests),
    so it must NOT hold a DB connection during the slow Plex fetch — doing so exhausted the pool
    (QueuePool timeout, broken images on prod). Warm path (cache hit) touches NO DB; cold path opens a
    brief session only to resolve the image key + read the Plex config, then RELEASES it before the
    fetch. Auth is the signed-session check (no DB user-load)."""
    _require_session(request)
    v = "art" if variant == "art" else "thumb"
    cache_key = f"item_{rating_key}_{v}"
    hit = _img_cache_read(cache_key)
    if hit:
        return _img_response(*hit)
    # Cold miss: short-lived session for the key + Plex client config, released before the slow fetch.
    try:
        with SessionLocal() as db:
            key = _image_key(db, str(rating_key), v)
            if not key:
                raise HTTPException(status_code=404, detail="No image")
            plex = PlexClient(db)  # reads base URL + token now; the fetch below needs no DB
    except PlexNotConfigured as e:
        raise HTTPException(status_code=400, detail=str(e))
    try:
        with plex:
            data, ct = plex.image_bytes(key)
    except PlexError as e:
        raise HTTPException(status_code=502, detail=f"Plex image fetch failed: {e}")
    _img_cache_write(cache_key, data, ct)
    return _img_response(data, ct)


# --- Playback (local file → direct 206 or on-the-fly HLS remux; P2) ----------------------------


def _item_or_404(db: Session, rating_key: str) -> PlexItem:
    it = db.query(PlexItem).filter_by(rating_key=str(rating_key)).first()
    if it is None:
        raise HTTPException(status_code=404, detail="Unknown Plex item")
    return it


# Progress thresholds (mirror the YouTube player): ignore the first few seconds, and treat the
# last stretch as "finished" (mark watched, clear the resume point).
_PROGRESS_MIN_S = 5
_FINISH_MARGIN_S = 30


# Cast/crew photos come from Plex's PUBLIC metadata CDN (no token). We proxy them (host-whitelisted)
# so the user's browser makes no third-party request (privacy + self-host CSP friendliness).
_PERSON_IMG_HOST = "https://metadata-static.plex.tv/"


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
    cast = []
    for role in (meta.get("Role") or [])[:24]:
        name = role.get("tag")
        if not name:
            continue
        thumb = str(role.get("thumb") or "")
        cast.append(
            {
                "name": name,
                "role": role.get("role"),
                "thumb": f"/api/plex/person-image?u={quote(thumb, safe='')}"
                if thumb.startswith(_PERSON_IMG_HOST)
                else None,
            }
        )
    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": cast,
    }


@router.get("/person-image")
def person_image(u: str, request: Request) -> Response:
    """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed).
    Signed-session auth (no DB) so it, like /image, never holds a pool connection during the fetch."""
    _require_session(request)
    if not u.startswith(_PERSON_IMG_HOST):
        raise HTTPException(status_code=400, detail="Unsupported image host")
    cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest()
    hit = _img_cache_read(cache_key)
    if hit:
        return _img_response(*hit)
    try:
        with httpx.Client(timeout=10.0, follow_redirects=True) as c:
            r = c.get(u)
            r.raise_for_status()
    except httpx.HTTPError:
        raise HTTPException(status_code=502, detail="Image fetch failed")
    ct = r.headers.get("content-type", "image/jpeg")
    _img_cache_write(cache_key, r.content, ct)
    return _img_response(r.content, ct)


# Known external list providers that seed Plex collections (via plextraktsync et al). A collection's
# "source" buckets the info-page strips so the user can show/hide each type independently (e.g. hide
# the noisy 'IMDb Popular'/'TMDb Trending' auto-lists but keep genuine franchise collections). Derived
# from the already-synced title/smart — no schema change, no re-sync. Unknown providers stay "collection".
_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.query(PlexCollection).filter(PlexCollection.rating_key.in_(keys))}
    # 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.query(PlexItem)
            .filter(
                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)
            .all()
        )
        ids = [m.id for m in members]
        states = {
            s.item_id: s
            for s in db.query(PlexState).filter(
                PlexState.user_id == user_id, PlexState.item_id.in_(ids or [0])
            )
        }
        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 = [
        r[0]
        for r in db.query(PlexItem.rating_key)
        .filter_by(show_id=it.show_id, kind="episode")
        .order_by(PlexItem.season_number.asc().nullsfirst(), PlexItem.episode_number.asc().nullsfirst())
        .all()
    ]
    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.query(PlexState).filter_by(user_id=user.id, item_id=it.id).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:
                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"),
                    }
                )
    except (PlexError, PlexNotConfigured):
        pass  # metadata extras are best-effort; playback works without them

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


@router.post("/item/{rating_key}/progress")
def item_progress(
    rating_key: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Checkpoint the resume position while playing (near the end → mark watched + clear it)."""
    it = _item_or_404(db, rating_key)
    try:
        position = max(0, int(payload.get("position_seconds") or 0))
        duration = int(payload.get("duration_seconds") or 0)
    except (TypeError, ValueError):
        raise HTTPException(status_code=400, detail="position_seconds must be an integer")
    now = datetime.now(timezone.utc)
    st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()

    near_end = duration > 0 and position > duration - _FINISH_MARGIN_S
    if near_end:
        if st is None:
            st = PlexState(user_id=user.id, item_id=it.id)
            db.add(st)
        st.status = "watched"
        st.watched_at = now
        st.position_seconds = 0
        st.progress_updated_at = now
        db.commit()
        return {"status": "watched", "position_seconds": 0}

    if position < _PROGRESS_MIN_S:
        if st is not None and st.status == "new":
            db.delete(st)
        elif st is not None:
            st.position_seconds = 0
            st.progress_updated_at = now
        db.commit()
        return {"position_seconds": 0}

    if st is None:
        st = PlexState(user_id=user.id, item_id=it.id)
        db.add(st)
    st.position_seconds = position
    st.progress_updated_at = now
    db.commit()
    return {"position_seconds": position}


@router.post("/item/{rating_key}/state")
def item_state(
    rating_key: str,
    payload: dict,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Set the per-user watch status (new|watched|hidden)."""
    it = _item_or_404(db, rating_key)
    status = payload.get("status")
    if status not in ("new", "watched", "hidden"):
        raise HTTPException(status_code=400, detail="Invalid status")
    st = db.query(PlexState).filter_by(user_id=user.id, item_id=it.id).first()
    if status == "new":
        if st is not None:
            db.delete(st)
            db.commit()
        return {"status": "new"}
    if st is None:
        st = PlexState(user_id=user.id, item_id=it.id)
        db.add(st)
    st.status = status
    if status == "watched":
        st.watched_at = datetime.now(timezone.utc)
        st.position_seconds = 0
    db.commit()
    return {"status": status}


@router.post("/stream/{rating_key}/session")
def stream_session(
    rating_key: str,
    start: float = 0.0,
    audio: int | None = None,
    subtitle: int | None = None,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Prepare playback from `start` seconds. Direct-playable files are served raw; remux files
    (re)start an HLS session at the offset (seek-restart); transcode (HEVC/VP9) is P3. Selecting a
    non-default audio/subtitle track (`audio`/`subtitle` = stream ordinals) forces the HLS path so
    the chosen tracks can be muxed in."""
    plex_stream.reap_idle()  # opportunistic cleanup of idle sessions on each new playback
    it = _item_or_404(db, rating_key)
    if it.playable == "transcode":
        raise HTTPException(status_code=501, detail="This format needs transcoding (a later phase).")
    select = audio is not None or subtitle is not None
    if it.playable == "direct" and not select:
        return {"mode": "direct", "url": f"/api/plex/stream/{it.rating_key}/file", "start_s": 0.0}
    if plex_paths.local_media_path(db, it.file_path) is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    s = plex_stream.start_session(db, it, start, audio, subtitle)
    if s is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    return {"mode": "hls", "start_s": s.start_s, "url": f"/api/plex/stream/{it.rating_key}/hls/{s.entry}"}


@router.get("/stream/{rating_key}/file")
def stream_file(
    rating_key: str,
    download: bool = False,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> FileResponse:
    """Serve the raw local media file with HTTP range support. Used both for direct-play (inline)
    and for the player's "download original" button (`?download=1` → attachment). The download is
    the untouched physical file — no re-encode/repackage."""
    it = _item_or_404(db, rating_key)
    src = plex_paths.local_media_path(db, it.file_path)
    if src is None:
        raise HTTPException(status_code=404, detail="Media file not found on disk")
    if download:
        # A clean, safe download name from the title + the real file extension.
        ext = src.suffix or (f".{it.container}" if it.container else "")
        base = "".join(c for c in (it.title or "video") if c.isalnum() or c in " -_.").strip() or "video"
        return FileResponse(str(src), filename=f"{base}{ext}")
    return FileResponse(str(src))  # Starlette honours the Range header (206)


_HLS_CT = {".m3u8": "application/vnd.apple.mpegurl", ".ts": "video/mp2t", ".vtt": "text/vtt"}


@router.get("/stream/{rating_key}/hls/{filename}")
def stream_hls_file(
    rating_key: str,
    filename: str,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> Response:
    """Serve a file (playlist / segment / subtitle) from the current HLS session directory. The
    playlist references its segments/subtitles by relative name, all resolving back here."""
    # Filename must be a plain hls artifact — no path separators, only the expected extensions.
    ext = "." + filename.rsplit(".", 1)[-1] if "." in filename else ""
    if ext not in _HLS_CT or "/" in filename or "\\" in filename or ".." in filename:
        raise HTTPException(status_code=404, detail="Not found")
    s = plex_stream.current_session(str(rating_key))
    if s is None:
        # No session yet: only the entry playlist for a remux item may auto-start one at 0.
        it = _item_or_404(db, rating_key)
        if filename != "index.m3u8" or it.playable != "remux":
            raise HTTPException(status_code=404, detail="No active playback session")
        s = plex_stream.start_session(db, it, 0.0)
        if s is None:
            raise HTTPException(status_code=404, detail="Media file not found on disk")
    path = s.dir / filename
    timeout = 30 if ext == ".m3u8" else 20
    if not plex_stream.wait_for(path, timeout):
        raise HTTPException(status_code=404, detail="Not available yet")
    return Response(
        content=path.read_bytes(), media_type=_HLS_CT[ext], headers={"Cache-Control": "no-store"}
    )