siftlode/backend/app/plex/people.py

"""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
import re
from datetime import timedelta
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, sysconfig
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 is_portrait_url(t) 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


# Every credit reference: one person name per (movie/show, credit column). The ONE list of credit
# columns, so the orphan prune and the enrichment order never disagree about who is credited.
_CREDIT_COLUMNS = [
    (table, col) for table in ("plex_items", "plex_shows") for col in ("cast_names", "directors")
]


def _credit_refs_sql(union: Literal["UNION", "UNION ALL"]) -> str:
    return f" {union} ".join(
        f"SELECT jsonb_array_elements_text({col}) FROM {table} WHERE jsonb_typeof({col}) = 'array'"
        for table, col in _CREDIT_COLUMNS
    )


# 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 = _credit_refs_sql("UNION")


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 = (
    f"SELECT ref, count(*) AS n FROM ({_credit_refs_sql('UNION ALL')}) 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 ---------------------------------------------------------------------------
# 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. Only a resized image is ever written under it —
# an original fetched without Plex keeps its own key, and goes once its small twin lands.
PORTRAIT_BOX = 320
# A headshot Plex answered "not found" for (a dead CDN file) is not asked for again this long, so a
# few dead faces never hold up the rest of the prefetch.
PORTRAIT_RETRY = timedelta(days=7)
# The one shape a Plex CDN headshot has (…/7/people/<hash>.jpg): plain path segments, no query, no
# fragment, no dot segment or empty segment (an HTTP client folds `a/../b` and the CDN `a//b` onto one
# file). The person-image proxy takes nothing else, so it cannot be made to cache endless spellings of
# one file (the image cache has no eviction). Used where a stored headshot is made, served and warmed.
_SEGMENT = r"[A-Za-z0-9_-]+(?:\.[A-Za-z0-9_-]+)*"  # a name, dots only between its parts
_PORTRAIT_URL = re.compile(re.escape(PERSON_IMG_HOST) + _SEGMENT + rf"(?:/{_SEGMENT})*")


def is_portrait_url(url: str) -> bool:
    return _PORTRAIT_URL.fullmatch(url) is not None


def _portrait_keys(url: str) -> tuple[str, str]:
    """(the resized portrait's key, the full-size original's key)."""
    base = "person_" + hashlib.sha1(url.encode()).hexdigest()
    return f"{base}_{PORTRAIT_BOX}", base


def portrait_cached(url: str) -> tuple[bytes, str] | None:
    """The cached portrait: the resized one, else an original kept from a time without Plex."""
    small, original = _portrait_keys(url)
    return imgcache.read(small) or imgcache.read(original)


def cache_portrait(plex: PlexClient | None, url: str) -> tuple[bytes, str]:
    """Fetch one CDN headshot and cache it. With a Plex client it is resized (the box key; the original
    is removed); without one the original comes straight from the public CDN and is cached under its own
    key, so the job still resizes it once Plex is there. Raises PlexError / httpx.HTTPError."""
    small, original = _portrait_keys(url)
    if plex is None:
        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(original, data, ct)
        return data, ct
    data, ct = plex.image_bytes(url, PORTRAIT_BOX, PORTRAIT_BOX)
    imgcache.write(small, data, ct)
    imgcache.drop(original)
    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 is_portrait_url(url) else None


# Plex failing this many headshots in a row ends a run: it is Plex that is failing, not the files.
_PLEX_DOWN_AFTER = 3
# The statuses that mean the one file is gone.
_DEAD = frozenset({404, 410})


def warm_portraits(db: Session, limit: int) -> dict:
    """Cache up to `limit` not-yet-cached headshots, the most-credited people first, so a
    person's face comes from local disk. Needs Plex (the resize) and the Plex module on; without them
    nothing is prefetched — full-size originals for every person would cost gigabytes. A headshot Plex
    answers "not found" for is marked and left for PORTRAIT_RETRY; Plex failing in a row ends the run."""
    if not sysconfig.get_bool(db, "plex_enabled"):
        return {"cached": 0, "skipped": "disabled"}
    try:
        plex = PlexClient(db)
    # silent-ok: no Plex server set up 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
    }
    # `down` counts Plex failing in a row (any answer from Plex breaks it), `plex_failed` in all
    done = dead = plex_failed = down = 0
    with plex:
        for pid in people_by_credits(db, "portrait", None):
            if done + dead >= limit or down >= _PLEX_DOWN_AFTER:
                break
            url = cdn_url(photos.get(pid) or "")
            if url is None:
                continue
            key = _portrait_keys(url)[0]
            if imgcache.has(key) or imgcache.missed_within(key, PORTRAIT_RETRY):
                continue
            try:
                cache_portrait(plex, url)
                done, down = done + 1, 0
            except PlexError as exc:
                # "Not found" is Plex's answer about this one file (a dead CDN headshot: measured 404,
                # the CDN itself says 403). Anything else — no answer, a refused token, a 5xx — is
                # Plex failing, which must neither mark the file nor keep the run going.
                cause = exc.__cause__
                if isinstance(cause, httpx.HTTPStatusError) and cause.response.status_code in _DEAD:
                    dead, down = dead + 1, 0  # Plex answered: it is up, the streak is broken
                    imgcache.mark_miss(key)
                else:
                    plex_failed, down = plex_failed + 1, down + 1
                log.warning(
                    "a portrait could not be cached",
                    extra=logs.event(
                        "plex.people.portrait_failed",
                        person_id=pid,
                        reason=type(exc.__cause__).__name__,
                    ),
                )
    failed = dead + plex_failed
    return {"cached": done, **({"failed": failed} if failed else {})}