"""Plex people: cast/crew filter-chip cards + a single person's filmography page.""" 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.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"]) @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}