feat(plex): the people are local — bios and portraits fetched ahead, not on the click (E30 S6)

A person's TMDB bio was fetched only when their page was first opened, off the request path, so the first
open showed no bio and the page never asked again (2 of 18,202 linked people were ever enriched).

- people_enrich scheduler job (every 10 min, 300 per run): linked people's bios from TMDB and their
  portraits into the image cache, the most-credited people first. A failed TMDB lookup ends the batch.
- Portraits are resized through Plex's photo transcoder to a 320 box (~15 KB instead of ~170 KB; the
  person-image proxy uses the same cache key, and a full-size original is removed as its small twin lands).
- The person response says `enriching` while a bio is on its way (a person the job hasn't reached yet).
- "Known for" is ranked by vote count, without guest visits as themselves (talk shows had crowded out the
  films: Tom Cruise had 14 of 15), with directed works, from a 40-deep pool; the page shows up to 15.
  Migration 0093 un-stamps enriched people so the job refetches them.
- A phased job result (backfill, people_enrich) now counts as a change for the audit trail; before, an
  automatic backfill run that added videos was never audited.
- HU: the watch reconcile job reads "egyeztetés" (a leftover from the S5 sweep).

backend/alembic/versions/0093_plex_person_known_for_votes.py

@@ -0,0 +1,29 @@
+"""re-fetch linked people — "Known for" ranked by votes, without guest visits (E30 S6)
+
+Revision ID: 0093_plex_person_known_for_votes
+Revises: 0092_plex_person_biography_hu
+Create Date: 2026-09-30
+
+A person's "Known for" pool was their credits ranked by TMDB popularity, so talk-show visits as
+themselves crowded out the films (Tom Cruise: 14 of 15 were talk shows, none in the library). The pool is
+now ranked by vote count, without guest visits, with directed works, and deeper. Already-enriched people
+are un-stamped so the people_enrich job fetches them again; their biography stays on the row meanwhile.
+"""
+
+from alembic import op
+
+revision: str = "0093_plex_person_known_for_votes"
+down_revision: str | None = "0092_plex_person_biography_hu"
+branch_labels: str | None = None
+depends_on: str | None = None
+
+
+def upgrade() -> None:
+    op.execute(
+        "UPDATE plex_people SET tmdb_enriched_at = NULL "
+        "WHERE tmdb_person_id IS NOT NULL AND tmdb_enriched_at IS NOT NULL"
+    )
+
+
+def downgrade() -> None:
+    pass  # the watermark only says "fetched"; a refetch under the old ranking is not worth undoing to

backend/app/config.py

@@ -366,6 +366,11 @@ class Settings(BaseSettings):
     tmdb_enrich_interval_min: int = 120
     # Titles fetched per TMDB enrichment run (TMDB is fast + generously rate-limited).
     tmdb_enrich_batch: int = 100
+    # How often the people job (E30 S6) fetches a batch of person bios from TMDB and portraits through Plex,
+    # in minutes, and how many of each per run: ~18k linked people at 300 per 10 min is about ten hours
+    # once, then only the newcomers.
+    people_enrich_interval_min: int = 10
+    people_enrich_batch: int = 300
 
     @property
     def allowed_email_set(self) -> set[str]:

backend/app/models.py

@@ -1417,7 +1417,8 @@ class PlexPerson(Base, TimestampMixin, UpdatedAtMixin):
     top few cast. Rows whose name no longer appears in ANY item's cast_names/directors (a library
     removal orphaned them) are pruned by the maintenance sweep, mirroring how Plex drops orphaned
     metadata. The filterable cast lives in plex_items/plex_shows.cast_names (self-cleaning with the
-    row); this table is ONLY the headshot cache."""
+    row); this table is the people's own facts: the headshot link, the TMDB link and the TMDB bio
+    (the headshot FILE lives in the image cache, filled by the people_enrich job)."""
 
     __tablename__ = "plex_people"
 
@@ -1427,8 +1428,8 @@ class PlexPerson(Base, TimestampMixin, UpdatedAtMixin):
     # TMDB person id, linked by name from a title's TMDB credits (E5 S2 lays the foundation; the /person
     # biography fetch + person-page render land in S6). NULL until a TMDB-enriched title names this person.
     tmdb_person_id: Mapped[int | None] = mapped_column(Integer)
-    # --- TMDB person enrichment (E5 S6): fetched from /person/{tmdb_person_id} the first time the person
-    #     page is opened, then SERVED FROM THE ROW. `tmdb_enriched_at` is the watermark (set even on a
+    # --- TMDB person enrichment (E5 S6): fetched from /person/{tmdb_person_id} by the people_enrich job
+    #     (E30 S6), or by the first person-page open the job hasn't reached, then SERVED FROM THE ROW. `tmdb_enriched_at` is the watermark (set even on a
     #     no-op fetch so it isn't retried every open). `known_for` = top combined-credits [{tmdb_id,
     #     media_type, title}], resolved to library rows by guid at serve time (like recommendations). ---
     biography: Mapped[str | None] = mapped_column(Text)

backend/app/plex/imgcache.py

@@ -56,6 +56,28 @@ def read(key: str) -> tuple[bytes, str] | None:
     return None
 
 
+def has(key: str) -> bool:
+    """A cached, non-empty file exists for `key` (a prefetch's cheap check — no read)."""
+    for ext, _ in _EXT_CT:
+        try:
+            if (CACHE_DIR / f"{key}.{ext}").stat().st_size > 0:
+                return True
+        # silent-ok: a miss, as in read().
+        except OSError:
+            pass
+    return False
+
+
+def drop(key: str) -> None:
+    """Remove `key`'s file, whatever its extension (a superseded variant). Best-effort."""
+    for ext, _ in _EXT_CT:
+        try:
+            (CACHE_DIR / f"{key}.{ext}").unlink(missing_ok=True)
+        # silent-ok: a leftover original only costs disk; the small twin is already written.
+        except OSError:
+            pass
+
+
 def write(key: str, data: bytes, ct: str) -> None:
     ext = _CT_EXT.get((ct or "").split(";")[0].strip(), "jpg")
     try:

backend/app/plex/people.py

@@ -3,13 +3,21 @@ or show (cast_names + the people_text search blob + the plex_people headshot cac
 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."""
 
-from urllib.parse import quote
+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).
@@ -113,3 +121,119 @@ def prune_orphan_people(db: Session) -> int:
     )
     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 {})}

backend/app/plex/tmdb.py

@@ -10,6 +10,7 @@ One-parser rule (mirrors app/plex/detail.py): `parse_tmdb` is the pure parser, `
 the row. The enrich batch + scheduler entry live here too so the whole TMDB integration is one module.
 """
 
+import re
 import time
 from datetime import datetime, timezone
 from typing import cast
@@ -23,6 +24,7 @@ from app import logs, sysconfig
 from app.config import settings
 from app.models import PlexItem, PlexPerson, PlexShow
 from app.plex.cards import display_title
+from app.plex.people import people_by_credits
 
 log = logs.get_logger(__name__)
 # Politeness throttle: TMDB tolerates ~50 req/s; ~33/s leaves comfortable headroom for a shared host.
@@ -315,8 +317,8 @@ def _link_people(db: Session, links: dict[str, int]) -> None:
 
 
 def enrich_batch(db: Session, tmdb: TmdbClient, limit: int) -> int:
-    """Enrich up to `limit` not-yet-TMDB-enriched titles (movies first, shows fill the remainder —
-    mirrors _enrich_people_batch). A transient TmdbError leaves the watermark NULL for retry; a
+    """Enrich up to `limit` not-yet-TMDB-enriched titles (movies first, shows fill the
+    remainder). A transient TmdbError leaves the watermark NULL for retry; a
     definitive no-match (unresolvable guid / TMDB 404) still stamps the watermark so it isn't re-fetched
     every run. One commit at the end. Returns the number processed (matched OR definitively unmatched)."""
     movies = (
@@ -501,23 +503,51 @@ def serve_block(db: Session, row, *, lang: str) -> dict | None:
 
 
 # --- Person enrichment (E5 S6) -------------------------------------------------------------------
-# A person's bio + birth facts + a "Known for" credit pool, fetched once (the first person-page open)
-# and served from the plex_people row. `known_for` stores the top combined-credits ids; the person
+# A person's bio + birth facts + a "Known for" credit pool, fetched once — by the people_enrich job, or
+# by the first person-page open if the job has not reached them yet — and served from the plex_people row. `known_for` stores the top combined-credits ids; the person
 # endpoint resolves them to library rows by guid at serve time (like recommendations).
 
-_KNOWN_FOR_MAX = 15
+# The candidate pool kept per person. Only the titles that are in the library are shown (at most
+# routes.plex.people._KNOWN_FOR_SHOWN), so the pool is deeper than the rail: a star's 15 most-voted
+# credits are often not the 15 a given library holds.
+_KNOWN_FOR_MAX = 40
+# TMDB genre ids of the shows a person visits as a guest (News, Reality, Talk). Ranked by popularity,
+# talk-show visits crowded out the films (Tom Cruise: 14 of 15 were talk shows; E30 S6).
+_GUEST_GENRES = {10763, 10764, 10767}
+# A credit in which the person plays themselves ("Self", "Himself - Guest", "Herself (archive footage)").
+_SELF_CREDIT = re.compile(r"^\s*(self|himself|herself|themselves)\b", re.IGNORECASE)
+# The crew jobs that make a title the person's own work (a director is known for what they directed).
+_KNOWN_FOR_CREW_JOBS = {"Director", "Creator"}
+
+
+def _known_for_credit(c: dict, crew: bool) -> bool:
+    if crew:
+        return c.get("job") in _KNOWN_FOR_CREW_JOBS
+    if _SELF_CREDIT.match(c.get("character") or ""):
+        return False
+    return not _GUEST_GENRES.intersection(c.get("genre_ids") or [])
 
 
 def parse_person(payload: dict) -> dict:
     """TMDB /person/{id} (+combined_credits) → the persisted bio fields. `known_for` = the person's
-    most-popular movie/tv credits (deduped by id), top `_KNOWN_FOR_MAX`, resolved to library later."""
-    credits = (payload.get("combined_credits") or {}).get("cast") or []
+    most-voted movie/tv credits — roles and directed works, not guest visits as themselves — deduped,
+    the top `_KNOWN_FOR_MAX`, resolved to library rows at serve time. Ranked by vote count, as TMDB's
+    own "Known for" is: popularity is this week's attention, which a talk-show visit wins."""
+    combined = payload.get("combined_credits") or {}
+    credits = [c for c in combined.get("cast") or [] if _known_for_credit(c, crew=False)] + [
+        c for c in combined.get("crew") or [] if _known_for_credit(c, crew=True)
+    ]
     # Dedup on (id, media_type), not the bare id: TMDB movie ids and tv ids are independent sequences
     # (id 603 can be both a movie and a show — see resolve_library_guids), so a same-id movie+show pair
     # is two distinct credits and both must survive.
     seen: set[tuple[int, str]] = set()
     known: list[dict] = []
-    for c in sorted(credits, key=lambda c: c.get("popularity") or 0, reverse=True):
+    ranked = sorted(
+        credits,
+        key=lambda c: (c.get("vote_count") or 0, c.get("popularity") or 0),
+        reverse=True,
+    )
+    for c in ranked:
         tid, mt = c.get("id"), c.get("media_type")
         if not tid or mt not in ("movie", "tv") or (tid, mt) in seen:
             continue
@@ -557,11 +587,22 @@ def apply_person(row: PlexPerson, parsed: dict) -> None:
     row.tmdb_enriched_at = datetime.now(timezone.utc)
 
 
+def _fetch_person_into(tmdb: TmdbClient, person: PlexPerson) -> None:
+    """Fetch one person from TMDB and write the row (not committed). A 404 (deleted person) stamps the
+    watermark with empty fields so it isn't fetched again. Raises TmdbError on a transient failure,
+    leaving the row untouched and `tmdb_enriched_at` NULL for a retry."""
+    payload = tmdb.fetch_person(cast(int, person.tmdb_person_id))
+    if payload is None:
+        person.tmdb_enriched_at = datetime.now(timezone.utc)
+    else:
+        apply_person(person, parse_person(payload))
+
+
 def enrich_person(db: Session, person: PlexPerson) -> bool:
-    """Lazily fetch + persist one person's TMDB bio (first person-page open), committing the write so it
-    survives regardless of the read path. No-op (False) without a key or a `tmdb_person_id`; a transient
-    error leaves `tmdb_enriched_at` NULL so the next open retries. A 404 (deleted person) stamps the
-    watermark with empty fields so we don't refetch it every open."""
+    """Lazily fetch + persist one person's TMDB bio (a person page opened before the background job
+    reached them), committing the write so it survives regardless of the read path. No-op (False)
+    without a key or a `tmdb_person_id`; a transient error leaves `tmdb_enriched_at` NULL so the next
+    open retries."""
     if not person.tmdb_person_id or not sysconfig.get_str(db, "tmdb_api_key"):
         return False
     try:
@@ -569,7 +610,7 @@ def enrich_person(db: Session, person: PlexPerson) -> bool:
         # longer blocks a serve; still cap the outbound TMDB call with a short timeout so a hanging TMDB
         # doesn't tie up a threadpool worker for the default 30s.
         with TmdbClient(db, timeout=8.0) as tmdb:
-            payload = tmdb.fetch_person(person.tmdb_person_id)
+            _fetch_person_into(tmdb, person)
     except TmdbError as exc:
         # False leaves the person un-enriched and unstamped, so it is retried — but the page in
         # front of the user right now shows an empty biography with nothing to say why.
@@ -582,9 +623,33 @@ def enrich_person(db: Session, person: PlexPerson) -> bool:
             ),
         )
         return False
-    if payload is None:
-        person.tmdb_enriched_at = datetime.now(timezone.utc)
-    else:
-        apply_person(person, parse_person(payload))
     db.commit()
     return True
+
+
+def enrich_people_batch(db: Session, tmdb: TmdbClient, limit: int) -> dict:
+    """Fetch up to `limit` linked, not-yet-enriched people (E30 S6), the most-credited first, so a
+    person page reads its bio from the row instead of waiting on TMDB. Each person is committed as it
+    lands: a page open during the run already sees it. A transient TmdbError ends the batch (TMDB is
+    degraded; hammering it helps nobody) and the rest wait for the next run."""
+    ids = people_by_credits(db, "bio", limit)
+    done = 0
+    for pid in ids:
+        person = db.get(PlexPerson, pid)
+        # A page open may have enriched this person since the select: skip, don't fetch twice.
+        if person is None or person.tmdb_enriched_at is not None:
+            continue
+        try:
+            _fetch_person_into(tmdb, person)
+        except TmdbError as exc:
+            db.rollback()
+            log.warning(
+                "the TMDB person batch stopped on a failed lookup",
+                extra=logs.event(
+                    "plex.tmdb.people_batch_failed", person_id=pid, reason=type(exc).__name__
+                ),
+            )
+            return {"enriched": done, "failed": 1}
+        db.commit()
+        done += 1
+    return {"enriched": done}

backend/app/routes/plex/people.py

@@ -8,7 +8,7 @@ from sqlalchemy import func, or_, select
 from sqlalchemy.orm import Session
 
 from app.collation import collated
-from app import logs
+from app import logs, sysconfig
 from app.auth import current_user
 from app.db import SessionLocal, get_db
 from app.models import (
@@ -26,6 +26,10 @@ router = APIRouter(prefix="/api/plex", tags=["plex"])
 log = logs.get_logger(__name__)
 
 
+# How many "Known for" titles the person page shows (the stored pool is deeper; see plex.tmdb).
+_KNOWN_FOR_SHOWN = 15
+
+
 def _resolve_known_for(
     db: Session,
     known_for: list | None,
@@ -34,8 +38,9 @@ def _resolve_known_for(
     show_lib_ids: list[int],
     lang: str,
 ) -> 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."""
+    """The person's TMDB "Known for" credits (most-voted first) that are IN our library, as cards, in
+    that order, at most _KNOWN_FOR_SHOWN. Matched by the canonical guid `tmdb:<id>`; out-of-library
+    credits are dropped."""
     if not known_for:
         return []
     # Match each credit to a library row by guid, bucketed by media_type (movie ids and tv ids collide
@@ -55,7 +60,7 @@ def _resolve_known_for(
                 out.append(show_card(shows[g], show_statuses.get(shows[g].id, "new"), lang=lang))
         elif g in movies:
             out.append(movie_card(movies[g], m_states.get(movies[g].id), lang=lang))
-    return out
+    return out[:_KNOWN_FOR_SHOWN]
 
 
 @router.get("/people/cards")
@@ -174,19 +179,24 @@ def person_detail(
     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.
     The fetch runs OFF the request path (a background task); the first open serves whatever's in the row
-    (bio possibly still null) and the bio lands from the row by the next open.
+    with `enriching` set, and the page asks again until the bio lands. E30 S6: the people_enrich job
+    fetches everyone ahead of time, so this lazy path is only for a person the job hasn't reached yet.
     """
     lang = norm_lang(lang)
     person = db.execute(select(PlexPerson).where(PlexPerson.name == name)).scalars().first()
     # Lazy one-time bio enrich, scheduled AFTER the response so the TMDB call is never on the serve path.
     # Gated by a per-person cooldown so concurrent/rapid opens (or repeated opens while TMDB is degraded)
     # can't pile background tasks onto the shared threadpool.
-    if (
+    # `enriching` tells the page the bio is on its way (E30 S6): it asks again shortly instead of showing
+    # an empty page until the next open. True while a fetch is pending, including one an earlier open
+    # already reserved.
+    enriching = (
         person is not None
-        and person.tmdb_person_id
+        and bool(person.tmdb_person_id)
         and person.tmdb_enriched_at is None
-        and _should_schedule_enrich(person.id)
-    ):
+        and bool(sysconfig.get_str(db, "tmdb_api_key"))
+    )
+    if enriching and person is not None and _should_schedule_enrich(person.id):
         background_tasks.add_task(_enrich_person_bg, person.id)
     movie_lib_ids, show_lib_ids = content_libs(db)
 
@@ -253,6 +263,7 @@ def person_detail(
         "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,
+        "enriching": enriching,
         "movies": movies,
         "shows": shows,
     }

backend/app/routes/plex/playback.py

@@ -1,6 +1,5 @@
 """Plex playback: image/subtitle proxies, HLS streaming, and watch-state writes."""
 
-import hashlib
 import threading
 from datetime import datetime, timedelta, timezone
 
@@ -214,24 +213,30 @@ def _resume_push_superseded(user_id: int, item_id: int, intent: datetime) -> boo
 
 @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."""
+    """Proxy a cast/crew photo from Plex's public metadata CDN (host-whitelisted; no token needed),
+    resized through Plex and cached (plex.people.cache_portrait; the people_enrich job fills the same
+    cache ahead of time). Signed-session auth; like /image, the DB session is released before the fetch."""
     _require_session(request)
     if not u.startswith(plex_people.PERSON_IMG_HOST):
         raise refused(400, "image_host")
-    cache_key = "person_" + hashlib.sha1(u.encode()).hexdigest()
-    hit = imgcache.read(cache_key)
+    hit = plex_people.portrait_cached(u)
     if hit:
         return imgcache.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 refused(502, "image_fetch_failed")
-    ct = r.headers.get("content-type", "image/jpeg")
-    imgcache.write(cache_key, r.content, ct)
-    return imgcache.response(r.content, ct)
+        with SessionLocal() as db:
+            plex: PlexClient | None = PlexClient(db)
+    # silent-ok: Plex off is a setting; without its resize the original comes straight from the CDN.
+    except PlexNotConfigured:
+        plex = None
+    try:
+        if plex is None:
+            data, ct = plex_people.cache_portrait(None, u)
+        else:
+            with plex:
+                data, ct = plex_people.cache_portrait(plex, u)
+    except (PlexError, httpx.HTTPError):
+        raise refused(502, "image_fetch_failed") from None
+    return imgcache.response(data, ct)
 
 
 def _push_resume_guarded(

backend/app/scheduler.py

@@ -69,6 +69,7 @@ JOB_INTERVALS: dict[str, int] = {
     "plex_watch_sync": settings.plex_watch_sync_interval_min,
     "plex_watch_reconcile": settings.plex_watch_reconcile_interval_min,
     "tmdb_enrich": settings.tmdb_enrich_interval_min,
+    "people_enrich": settings.people_enrich_interval_min,
     "audit_gc": settings.audit_gc_minutes,
     "log_gc": settings.log_gc_minutes,
 }
@@ -142,7 +143,12 @@ def _run_changed(result) -> bool:
     if result is None:
         return False
     if isinstance(result, dict):
-        return any(isinstance(v, (int, float)) and v for v in result.values())
+        # A phased job (backfill's recent/deep, people_enrich's bios/portraits) nests a result per
+        # phase: a count inside a phase is a change too.
+        return any(
+            _run_changed(v) if isinstance(v, dict) else isinstance(v, (int, float)) and bool(v)
+            for v in result.values()
+        )
     if isinstance(result, (int, float)):
         return result != 0
     return bool(result)
@@ -403,6 +409,26 @@ def _tmdb_enrich_job() -> None:
     _job("tmdb_enrich", run_tmdb_enrich)
 
 
+def _people_enrich_job() -> None:
+    # Make the people local (E30 S6): a batch of person bios from TMDB (the most-credited first) and a
+    # batch of portraits resized through Plex into the image cache, so a person page and a cast strip
+    # never wait on an outside service. Each half is a no-op without its key / without Plex.
+    def work(db):
+        from app import sysconfig
+        from app.plex import people as plex_people
+        from app.plex import tmdb as plex_tmdb
+
+        batch = settings.people_enrich_batch
+        if sysconfig.get_str(db, "tmdb_api_key"):
+            with plex_tmdb.TmdbClient(db) as tmdb:
+                bios = plex_tmdb.enrich_people_batch(db, tmdb, batch)
+        else:
+            bios = {"skipped": "tmdb_api_key not set"}
+        return {"bios": bios, "portraits": plex_people.warm_portraits(db, batch)}
+
+    _job("people_enrich", work)
+
+
 def _audit_gc_job() -> None:
     # Prune audit-log rows older than the admin-set retention (audit_retention_days; 0 = keep
     # forever). Pure DB, no quota.
@@ -449,6 +475,7 @@ JOB_FUNCS: dict[str, Callable[[], None]] = {
     "plex_watch_sync": _plex_watch_sync_job,
     "plex_watch_reconcile": _plex_watch_reconcile_job,
     "tmdb_enrich": _tmdb_enrich_job,
+    "people_enrich": _people_enrich_job,
     "audit_gc": _audit_gc_job,
     "log_gc": _log_gc_job,
 }

backend/tests/test_plex_people_enrich.py

@@ -0,0 +1,278 @@
+"""E30 S6: the people are made local ahead of time. The people_enrich job fetches linked people's TMDB
+bios (the most-credited first) and caches their portraits resized through Plex; the person page says
+when a bio is still on its way (`enriching`), so it can ask again instead of staying empty.
+DB lane (Postgres); the portrait cache runs in a temp directory.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+
+from app import scheduler
+from app.models import PlexItem, PlexLibrary, PlexPerson, User
+from app.plex import imgcache
+from app.plex import people as plex_people
+from app.plex import tmdb as plex_tmdb
+
+CDN = plex_people.PERSON_IMG_HOST + "people/"
+DONE = datetime(2024, 1, 1, tzinfo=timezone.utc)
+
+
+def _movie(db, lib, key, cast):
+    db.add(PlexItem(rating_key=key, library_id=lib.id, kind="movie", title=key, cast_names=cast))
+
+
+def _library(db):
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    return lib
+
+
+class _Tmdb:
+    """A TmdbClient stand-in: records the ids asked for, fails on `fail_on`."""
+
+    def __init__(self, fail_on=None):
+        self.asked: list[int] = []
+        self.fail_on = fail_on
+
+    def fetch_person(self, pid):
+        self.asked.append(pid)
+        if pid == self.fail_on:
+            raise plex_tmdb.TmdbError("429")
+        return {"biography": f"Bio {pid}."}
+
+
+def test_the_batch_takes_the_most_credited_linked_people_first(db):
+    lib = _library(db)
+    _movie(db, lib, "a", ["Often", "Sometimes"])
+    _movie(db, lib, "b", ["Often", "Sometimes"])
+    _movie(db, lib, "c", ["Often", "Unlinked"])
+    db.add_all(
+        [
+            PlexPerson(name="Sometimes", tmdb_person_id=2),
+            PlexPerson(name="Often", tmdb_person_id=1),
+            PlexPerson(name="Unlinked"),  # no TMDB id: nothing to fetch
+            PlexPerson(name="Done", tmdb_person_id=3, tmdb_enriched_at=DONE),
+            PlexPerson(name="Rare", tmdb_person_id=4),  # credited nowhere: last
+        ]
+    )
+    db.commit()
+    tmdb = _Tmdb()
+    assert plex_tmdb.enrich_people_batch(db, tmdb, 2) == {"enriched": 2}  # type: ignore[arg-type]
+    assert tmdb.asked == [1, 2]
+    db.expire_all()
+    often = db.query(PlexPerson).filter_by(name="Often").one()
+    assert often.biography == "Bio 1." and often.tmdb_enriched_at is not None
+    assert db.query(PlexPerson).filter_by(name="Rare").one().tmdb_enriched_at is None  # next run
+
+
+def test_a_failed_lookup_ends_the_batch_and_keeps_what_landed(db):
+    db.add_all([PlexPerson(name="A", tmdb_person_id=1), PlexPerson(name="B", tmdb_person_id=2)])
+    db.add(PlexPerson(name="C", tmdb_person_id=3))
+    db.commit()
+    tmdb = _Tmdb(fail_on=2)
+    assert plex_tmdb.enrich_people_batch(db, tmdb, 10) == {"enriched": 1, "failed": 1}  # type: ignore[arg-type]
+    assert tmdb.asked == [1, 2]  # C is not asked: TMDB is struggling
+    db.expire_all()
+    stamped = {p.name: p.tmdb_enriched_at is not None for p in db.query(PlexPerson)}
+    assert stamped == {"A": True, "B": False, "C": False}
+
+
+class _Plex:
+    def __init__(self, fail=False):
+        self.asked: list[tuple[str, int | None]] = []
+        self.fail = fail
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+    def image_bytes(self, url, width=None, height=None):
+        self.asked.append((url, width))
+        if self.fail:
+            raise plex_people.PlexError("down")
+        return b"small", "image/jpeg"
+
+
+@pytest.fixture
+def cache_dir(tmp_path, monkeypatch):
+    monkeypatch.setattr(imgcache, "CACHE_DIR", tmp_path)
+    return tmp_path
+
+
+def test_portraits_are_cached_small_and_replace_the_full_size_original(db, cache_dir, monkeypatch):
+    lib = _library(db)
+    _movie(db, lib, "a", ["Often"])
+    _movie(db, lib, "b", ["Often", "Cached"])
+    db.add_all(
+        [
+            PlexPerson(name="Often", photo_thumb=plex_people.person_thumb_url(CDN + "often.jpg")),
+            PlexPerson(name="Cached", photo_thumb=plex_people.person_thumb_url(CDN + "cached.jpg")),
+            PlexPerson(name="Faceless"),
+        ]
+    )
+    db.commit()
+    small, legacy = plex_people._portrait_keys(CDN + "often.jpg")
+    imgcache.write(legacy, b"a big original", "image/jpeg")  # from before E30 S6
+    imgcache.write(plex_people._portrait_keys(CDN + "cached.jpg")[0], b"x", "image/jpeg")
+    plex = _Plex()
+    monkeypatch.setattr(plex_people, "PlexClient", lambda d: plex)
+
+    assert plex_people.warm_portraits(db, 10) == {"cached": 1}
+    assert plex.asked == [
+        (CDN + "often.jpg", plex_people.PORTRAIT_BOX)
+    ]  # resized, the cached one skipped
+    assert imgcache.read(small) == (b"small", "image/jpeg")
+    assert not imgcache.has(legacy)
+
+
+def test_portraits_wait_for_plex(db, cache_dir, monkeypatch):
+    db.add(PlexPerson(name="P", photo_thumb=plex_people.person_thumb_url(CDN + "p.jpg")))
+    db.commit()
+
+    def not_configured(d):
+        raise plex_people.PlexNotConfigured()
+
+    monkeypatch.setattr(plex_people, "PlexClient", not_configured)
+    assert plex_people.warm_portraits(db, 10) == {"cached": 0, "skipped": "disabled"}
+    assert list(cache_dir.iterdir()) == []  # no full-size originals behind Plex's back
+
+
+def test_a_failing_plex_ends_the_portrait_run(db, cache_dir, monkeypatch):
+    db.add_all(
+        [
+            PlexPerson(name=f"P{i}", photo_thumb=plex_people.person_thumb_url(CDN + f"{i}.jpg"))
+            for i in range(8)
+        ]
+    )
+    db.commit()
+    plex = _Plex(fail=True)
+    monkeypatch.setattr(plex_people, "PlexClient", lambda d: plex)
+    assert plex_people.warm_portraits(db, 100) == {"cached": 0, "failed": 5}
+    assert len(plex.asked) == 5
+
+
+def test_a_phase_count_is_a_change_worth_an_audit_row():
+    assert scheduler._run_changed({"bios": {"enriched": 3}, "portraits": {"cached": 0}})
+    assert scheduler._run_changed({"recent": {"videos_added": 2}, "deep": {}})
+    assert not scheduler._run_changed({"bios": {"skipped": "x"}, "portraits": {"cached": 0}})
+
+
+def test_the_person_page_says_a_bio_is_on_its_way(db, monkeypatch, override_dep):
+    from fastapi.testclient import TestClient
+
+    import app.state as state
+    from app.auth import current_user
+    from app.db import get_db
+    from app.main import app
+    from app.routes.plex import people
+
+    db.add_all(
+        [
+            PlexPerson(name="Pending", tmdb_person_id=1),
+            PlexPerson(name="Known", tmdb_person_id=2, tmdb_enriched_at=DONE, biography="Bio."),
+            PlexPerson(name="Nobody"),
+        ]
+    )
+    user = User(email="enriching@example.com")
+    db.add(user)
+    db.commit()
+    monkeypatch.setattr(state, "_configured_cache", True)
+    override_dep(get_db, db.info["request"])
+    override_dep(current_user, lambda: user)
+    monkeypatch.setattr(people, "_enrich_person_bg", lambda pid: None)
+    key = {"value": "key"}
+    monkeypatch.setattr(people.sysconfig, "get_str", lambda d, k: key["value"])
+
+    def enriching(name):
+        resp = TestClient(app).get(f"/api/plex/person/{name}")
+        assert resp.status_code == 200, resp.text
+        return resp.json()["enriching"]
+
+    assert enriching("Pending") is True
+    assert enriching("Known") is False
+    assert enriching("Nobody") is False
+    key["value"] = ""  # no TMDB key: nothing will ever arrive, so the page must not wait
+    assert enriching("Pending") is False
+
+
+def test_known_for_is_the_most_voted_roles_not_guest_visits():
+    payload = {
+        "combined_credits": {
+            "cast": [
+                # Popular this week, but a visit as himself: not what he is known for.
+                {
+                    "id": 1,
+                    "media_type": "tv",
+                    "name": "Late Show",
+                    "character": "Himself - Guest",
+                    "popularity": 900,
+                    "vote_count": 50,
+                },
+                {
+                    "id": 2,
+                    "media_type": "tv",
+                    "name": "Talk Hour",
+                    "character": "",
+                    "genre_ids": [10767],
+                    "popularity": 800,
+                    "vote_count": 400,
+                },
+                {
+                    "id": 3,
+                    "media_type": "movie",
+                    "title": "Small Film",
+                    "character": "Bob",
+                    "popularity": 90,
+                    "vote_count": 300,
+                },
+                {
+                    "id": 4,
+                    "media_type": "movie",
+                    "title": "Big Film",
+                    "character": "Hero",
+                    "popularity": 10,
+                    "vote_count": 9000,
+                },
+            ],
+            "crew": [
+                {
+                    "id": 5,
+                    "media_type": "movie",
+                    "title": "His Own Film",
+                    "job": "Director",
+                    "vote_count": 1000,
+                },
+                {
+                    "id": 6,
+                    "media_type": "movie",
+                    "title": "Thanked In",
+                    "job": "Thanks",
+                    "vote_count": 99999,
+                },
+            ],
+        }
+    }
+    kf = plex_tmdb.parse_person(payload)["known_for"]
+    assert [k["title"] for k in kf] == ["Big Film", "His Own Film", "Small Film"]
+
+
+def test_the_page_shows_at_most_fifteen_known_for_titles(db):
+    from app.routes.plex import people
+
+    lib = PlexLibrary(plex_key="1", title="Movies", kind="movie", enabled=True)
+    db.add(lib)
+    db.flush()
+    for i in range(1, 21):  # TMDB ids start at 1
+        db.add(
+            PlexItem(
+                rating_key=f"k{i}", library_id=lib.id, kind="movie", title=f"T{i}", guid=f"tmdb:{i}"
+            )
+        )
+    db.commit()
+    pool = [{"tmdb_id": i, "media_type": "movie", "title": f"T{i}"} for i in range(1, 21)]
+    cards = people._resolve_known_for(db, pool, 0, [lib.id], [], "en")
+    assert [c["title"] for c in cards] == [f"T{i}" for i in range(1, 16)]

backend/tests/test_plex_person.py

@@ -133,6 +133,7 @@ def test_person_detail_unknown_name_is_empty(db, monkeypatch, override_dep):
         "place_of_birth": None,
         "known_for_department": None,
         "also_known_as": [],
+        "enriching": False,
         "known_for": [],
         "movies": [],
         "shows": [],

backend/tests/test_plex_person_bio.py

@@ -191,11 +191,12 @@ def test_person_detail_schedules_enrich_off_request_path(db, monkeypatch, overri
 
     scheduled: list[int] = []
     monkeypatch.setattr(people, "_enrich_person_bg", lambda pid: scheduled.append(pid))
+    monkeypatch.setattr(people.sysconfig, "get_str", lambda d, k: "key")
     people._enrich_inflight.clear()  # ensure the per-person cooldown guard doesn't suppress this open
 
     resp = _get(db, user, monkeypatch, override_dep, "/api/plex/person/Lazy P")
     assert resp.status_code == 200, resp.text
-    assert resp.json()["biography"] is None  # not enriched yet — the bio lands by the next open
+    assert resp.json()["biography"] is None  # not enriched yet — the page asks again (`enriching`)
     assert len(scheduled) == 1  # the enrich was scheduled off the request path