siftlode/backend/app/routes/plex/people.py

"""Plex people: cast/crew filter-chip cards + a single person's filmography page."""

import logging

from fastapi import APIRouter, Depends
from sqlalchemy import func, or_, select
from sqlalchemy.orm import Session

from app.auth import current_user
from app.db import get_db
from app.models import (
    PlexItem,
    PlexPerson,
    PlexShow,
    PlexState,
    User,
)
from app.plex import rollup
from app.plex import tmdb as plex_tmdb
from app.plex.cards import movie_card, show_card
from app.plex.query import split_csv
from app.routes.plex._shared import _content_libs

router = APIRouter(prefix="/api/plex", tags=["plex"])
log = logging.getLogger("siftlode.plex")


def _resolve_known_for(
    db: Session,
    known_for: list | None,
    uid: int,
    movie_lib_ids: list[int],
    show_lib_ids: list[int],
) -> list[dict]:
    """The person's TMDB "Known for" credits (top by popularity) that are IN our library, as cards, in
    popularity order. Matched by the canonical guid `tmdb:<id>`; out-of-library credits are dropped."""
    if not known_for:
        return []
    # Bucket by media_type: TMDB movie ids and tv ids collide numerically, so a movie credit must only
    # match a library movie and a tv credit only a library show (mirrors tmdb.resolve_recs). Matching
    # blind to media_type would render a tv credit as an unrelated movie whose tmdb id happens to match.
    movie_guids = [
        f"tmdb:{k['tmdb_id']}" for k in known_for if k.get("tmdb_id") and k.get("media_type") != "tv"
    ]
    tv_guids = [
        f"tmdb:{k['tmdb_id']}" for k in known_for if k.get("tmdb_id") and k.get("media_type") == "tv"
    ]
    if not movie_guids and not tv_guids:
        return []
    movies: dict[str, PlexItem] = {}
    if movie_lib_ids and movie_guids:
        for m in db.execute(
            select(PlexItem).where(
                PlexItem.kind == "movie",
                PlexItem.library_id.in_(movie_lib_ids),
                PlexItem.guid.in_(movie_guids),
            )
        ).scalars():
            movies.setdefault(m.guid, m)
    shows: dict[str, PlexShow] = {}
    if show_lib_ids and tv_guids:
        for s in db.execute(
            select(PlexShow).where(
                PlexShow.library_id.in_(show_lib_ids), PlexShow.guid.in_(tv_guids)
            )
        ).scalars():
            shows.setdefault(s.guid, s)
    m_states = {
        s.item_id: s
        for s in db.execute(
            select(PlexState).where(
                PlexState.user_id == uid,
                PlexState.item_id.in_([m.id for m in movies.values()] or [0]),
            )
        ).scalars()
    }
    show_statuses = rollup.state_map(db, uid, [s.id for s in shows.values()])
    out: list[dict] = []
    for k in known_for:
        g = f"tmdb:{k.get('tmdb_id')}"
        # A credit resolves against its OWN media_type's bucket only.
        if k.get("media_type") == "tv":
            if g in shows:
                out.append(show_card(shows[g], show_statuses.get(shows[g].id, "new")))
        elif g in movies:
            out.append(movie_card(movies[g], m_states.get(movies[g].id)))
    return out


@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.

    E5 S6: the first time a page is opened, the person's TMDB bio + "Known for" credits are lazily
    fetched (from the S2-linked `tmdb_person_id`) and persisted, then served from the row on every open.
    """
    person = db.execute(select(PlexPerson).where(PlexPerson.name == name)).scalars().first()
    # Lazy one-time bio enrich — best-effort, never fail the page if TMDB is down / the id is stale.
    if person is not None and person.tmdb_person_id and person.tmdb_enriched_at is None:
        try:
            plex_tmdb.enrich_person(db, person)
        except Exception:  # noqa: BLE001 — enrichment is optional; the filmography still renders
            # A failed enrich_person commit leaves the session in a needs-rollback state; without this
            # the following SELECTs would raise PendingRollbackError and 500 the page (SQLAlchemy 2.x
            # deactivates the transaction on a commit-time flush failure and does NOT auto-reset it).
            db.rollback()
            log.warning("person bio enrich failed for %s", name, exc_info=True)
    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]

    known_for = _resolve_known_for(
        db, person.known_for if person else None, user.id, movie_lib_ids, show_lib_ids
    )
    return {
        "name": name,
        "photo": person.photo_thumb if person else None,
        "biography": person.biography if person else None,
        "birthday": person.birthday if person else None,
        "deathday": person.deathday if person else None,
        "place_of_birth": person.place_of_birth if person else None,
        "known_for_department": person.known_for_department if person else None,
        "also_known_as": (person.also_known_as if person else None) or [],
        "known_for": known_for,
        "movies": movies,
        "shows": shows,
    }