"""Shared Plex people helpers: parse cast/crew from rich metadata, persist the full cast onto a movie
or show (cast_names + the people_text search blob + the plex_people headshot cache), and prune
orphaned people. Used by BOTH the info-page lazy enrichment (routes.plex) and the background rich
re-sync (plex.sync), so the two paths never diverge."""

import hashlib
from typing import Literal
from urllib.parse import quote, unquote

import httpx
from sqlalchemy import select, text
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.orm import Session

from app import logs
from app.models import PlexPerson
from app.plex import imgcache
from app.plex.client import PlexClient, PlexError, PlexNotConfigured

log = logs.get_logger(__name__)

# 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 person_thumb_url(thumb: str | None) -> str | None:
    """Proxied /person-image URL for a Plex CDN headshot, or None when it's missing / off-CDN."""
    t = str(thumb or "")
    return (
        f"/api/plex/person-image?u={quote(t, safe='')}" if t.startswith(PERSON_IMG_HOST) else None
    )


def cast_from_meta(meta: dict, limit: int = 24) -> list[dict]:
    """[{name, role, thumb}] for the cast (Role tags) in a full Plex metadata dict — thumb is the
    proxied person-image URL or None. Shared by the info-page response, the lazy enrichment, and the
    re-sync so the parsing lives in one place."""
    out: list[dict] = []
    for role in (meta.get("Role") or [])[:limit]:
        name = role.get("tag")
        if not name:
            continue
        out.append(
            {"name": name, "role": role.get("role"), "thumb": person_thumb_url(role.get("thumb"))}
        )
    return out


def enrich_row_people(db: Session, row, cast: list[dict]) -> bool:
    """Persist the FULLER cast onto the row's `cast_names` + `people_text`, and cache each cast
    member's headshot in plex_people. The cheap section listing that drives the main sync only carries
    the top ~3 cast, so this — run whenever a rich fetch happens (info-page view + the rich re-sync) —
    is what makes the full displayed cast filterable. Returns True if anything was written (the caller
    commits). Best-effort by contract: never the reason a read 500s."""
    names = list(dict.fromkeys(c["name"] for c in cast if c.get("name")))[:24]
    if not names:
        return False
    dirty = False
    # Grow/refresh cast_names only when the rich set adds names the stored (listing top-3) set lacks —
    # avoid churn and never downgrade a fuller set back to the thin listing.
    if set(names) - set(row.cast_names or []):
        row.cast_names = names
        row.people_text = " ".join(dict.fromkeys(list(row.directors or []) + names)) or None
        dirty = True
    # Headshot cache: upsert name -> proxied thumb (a later fetch fills a prior NULL; names without a
    # whitelisted photo still get a row so a card can show a placeholder).
    existing = {
        p.name: p
        for p in db.execute(select(PlexPerson).where(PlexPerson.name.in_(names))).scalars()
    }
    missing: dict[str, str | None] = {}
    for c in cast:
        nm, th = c.get("name"), c.get("thumb")
        if not nm:
            continue
        p = existing.get(nm)
        if p is None:
            # Not yet cached — queue an insert, deduped by name (keep the first non-null thumb) so a
            # single INSERT statement can't hit the same conflict target twice.
            missing[nm] = missing.get(nm) or th
        elif th and not p.photo_thumb:
            p.photo_thumb = th
            dirty = True
    if missing:
        # on_conflict_do_nothing keeps a cold sync from crashing when one person appears across two
        # items in the same batch: SessionLocal is autoflush=False, so the ORM query above can't see a
        # same-batch pending insert, and this path also races the concurrent info-page enrichment. The
        # Core execute runs immediately (not deferred to the caller's commit), so a later row in the
        # batch sees the row and skips it. Mirrors app/sync/videos.py:insert_stubs.
        db.execute(
            pg_insert(PlexPerson)
            .values([{"name": nm, "photo_thumb": th} for nm, th in missing.items()])
            .on_conflict_do_nothing(index_elements=[PlexPerson.name])
        )
        dirty = True
    return dirty


# The person names still referenced by ANY movie/show (cast OR directors). plex_people rows outside
# this set are orphans left by a library removal.
_REFERENCED_PEOPLE_SQL = """
    SELECT jsonb_array_elements_text(cast_names) FROM plex_items WHERE jsonb_typeof(cast_names) = 'array'
    UNION SELECT jsonb_array_elements_text(directors) FROM plex_items WHERE jsonb_typeof(directors) = 'array'
    UNION SELECT jsonb_array_elements_text(cast_names) FROM plex_shows WHERE jsonb_typeof(cast_names) = 'array'
    UNION SELECT jsonb_array_elements_text(directors) FROM plex_shows WHERE jsonb_typeof(directors) = 'array'
"""


def prune_orphan_people(db: Session) -> int:
    """Delete plex_people (headshot cache) rows whose name no longer appears in any movie/show cast or
    directors — orphaned by a library removal. Returns the number pruned. Commits its own work."""
    # `WHERE ref IS NOT NULL` matters: a JSON-null array element surfaces as a SQL NULL in the
    # referenced set, and `name NOT IN (…, NULL)` is NULL (never TRUE) for every row — the classic
    # NOT IN trap that would silently prune nothing. Filtering NULLs out keeps the delete correct.
    res = db.execute(
        text(
            "DELETE FROM plex_people WHERE name NOT IN "
            f"(SELECT ref FROM ({_REFERENCED_PEOPLE_SQL}) AS refs(ref) WHERE ref IS NOT NULL)"
        )
    )
    db.commit()
    return res.rowcount or 0


# How often each person is credited across movies, shows and episodes (cast OR director) — the order the
# people_enrich job works in, so the faces met most often are local first.
_CREDIT_COUNTS_SQL = """
    SELECT ref, count(*) AS n FROM (
        SELECT jsonb_array_elements_text(cast_names) FROM plex_items WHERE jsonb_typeof(cast_names) = 'array'
        UNION ALL SELECT jsonb_array_elements_text(directors) FROM plex_items WHERE jsonb_typeof(directors) = 'array'
        UNION ALL SELECT jsonb_array_elements_text(cast_names) FROM plex_shows WHERE jsonb_typeof(cast_names) = 'array'
        UNION ALL SELECT jsonb_array_elements_text(directors) FROM plex_shows WHERE jsonb_typeof(directors) = 'array'
    ) AS refs(ref) GROUP BY ref
"""
_NEEDS = {
    # linked to TMDB, bio never fetched
    "bio": "p.tmdb_person_id IS NOT NULL AND p.tmdb_enriched_at IS NULL",
    # has a Plex headshot (whether it is cached is the disk's answer, not the row's)
    "portrait": "p.photo_thumb IS NOT NULL",
}


def people_by_credits(
    db: Session, need: Literal["bio", "portrait"], limit: int | None
) -> list[int]:
    """plex_people ids that `need` something, the most-credited first (ties by id)."""
    rows = db.execute(
        text(
            f"SELECT p.id FROM plex_people p LEFT JOIN ({_CREDIT_COUNTS_SQL}) c ON c.ref = p.name "
            f"WHERE {_NEEDS[need]} ORDER BY coalesce(c.n, 0) DESC, p.id"
            + (" LIMIT :limit" if limit is not None else "")
        ),
        {"limit": limit} if limit is not None else {},
    )
    return [r[0] for r in rows]


# --- Portraits (E30 S6) ---------------------------------------------------------------------------
# A Plex CDN headshot is a full-size original (~170 KB on average, measured). Plex's photo transcoder
# fits it into this box (~13 KB), which covers the largest place a portrait is drawn (the person page's
# 6rem avatar) at 2x density and a raised text size. The box is part of the cache key, like a poster's:
# a new box re-caches instead of serving a stale size, and the full-size original under the old key is
# removed as its small twin lands.
PORTRAIT_BOX = 320


def _portrait_keys(url: str) -> tuple[str, str]:
    base = "person_" + hashlib.sha1(url.encode()).hexdigest()
    return f"{base}_{PORTRAIT_BOX}", base


def portrait_cached(url: str) -> tuple[bytes, str] | None:
    return imgcache.read(_portrait_keys(url)[0])


def cache_portrait(plex: PlexClient | None, url: str) -> tuple[bytes, str]:
    """Fetch one CDN headshot — resized through Plex when a Plex client is given, else the original
    straight from the public CDN — and cache it. Raises PlexError / httpx.HTTPError on a failed fetch."""
    key, legacy = _portrait_keys(url)
    if plex is not None:
        data, ct = plex.image_bytes(url, PORTRAIT_BOX, PORTRAIT_BOX)
    else:
        with httpx.Client(timeout=10.0, follow_redirects=True) as c:
            r = c.get(url)
            r.raise_for_status()
        data, ct = r.content, r.headers.get("content-type", "image/jpeg")
    imgcache.write(key, data, ct)
    if plex is not None:
        imgcache.drop(legacy)
    return data, ct


def cdn_url(photo_thumb: str) -> str | None:
    """The CDN url inside a stored proxied `/person-image?u=` value (see person_thumb_url)."""
    _, _, q = photo_thumb.partition("?u=")
    url = unquote(q)
    return url if url.startswith(PERSON_IMG_HOST) else None


def warm_portraits(db: Session, limit: int) -> dict:
    """Cache up to `limit` not-yet-cached headshots, the most-credited people first (E30 S6), so a
    person's face comes from local disk. Needs Plex (the resize); without it, nothing is prefetched —
    full-size originals for every person would cost gigabytes. A failed fetch is skipped; one failing
    Plex ends the run."""
    try:
        plex = PlexClient(db)
    # silent-ok: Plex off is a setting, not a failure; the run's result says "disabled".
    except PlexNotConfigured:
        return {"cached": 0, "skipped": "disabled"}
    photos: dict[int, str] = {
        pid: thumb
        for pid, thumb in db.execute(
            select(PlexPerson.id, PlexPerson.photo_thumb).where(PlexPerson.photo_thumb.is_not(None))
        ).tuples()
        if thumb
    }
    done = failed = 0
    with plex:
        for pid in people_by_credits(db, "portrait", None):
            if done + failed >= limit:
                break
            url = cdn_url(photos.get(pid) or "")
            if url is None or imgcache.has(_portrait_keys(url)[0]):
                continue
            try:
                cache_portrait(plex, url)
                done += 1
            except (PlexError, httpx.HTTPError) as exc:
                failed += 1
                log.warning(
                    "a portrait could not be cached",
                    extra=logs.event(
                        "plex.people.portrait_failed", person_id=pid, reason=type(exc).__name__
                    ),
                )
                if failed >= 5:
                    break
    return {"cached": done, **({"failed": failed} if failed else {})}