scene-actor-extraction/scripts/run_from_jellyfin.py

#!/usr/bin/env python3
"""run_from_jellyfin.py — resolve a Jellyfin title to its media file and run scene_analyze.

Looks up a Movie/Episode in Jellyfin, reads its on-disk Path (Jellyfin and this
tool must share the same media mount), filters the gallery down to that
title's credited cast (via filter_gallery's logic, fewer look-alike
mismatches), and runs scene_analyze against the resolved file.

Usage:
    python scripts/run_from_jellyfin.py \\
        --jellyfin-url http://jellyfin.local:8096 \\
        --api-key YOUR_API_KEY \\
        --title "The Matrix" \\
        --gallery whole_gallery.h5 \\
        -- --fps 5 --verbosity 2

Anything after "--" is passed through unchanged to scene_analyze.

Add --preview to open the live OpenCV display window: it resolves the media path
from Jellyfin exactly as normal, then launches build/scene_preview instead of the
headless binary (implies --no-push). Works with --worker or a single title:

    python scripts/run_from_jellyfin.py ... --worker --preview -- --fps 5

Worker mode (--worker) polls the JRay plugin's Tasks/Pending endpoint for a
random batch of items with no truth data yet, processing each in turn:

    python scripts/run_from_jellyfin.py \\
        --jellyfin-url http://jellyfin.local:8096 \\
        --api-key YOUR_API_KEY \\
        --gallery whole_gallery.h5 \\
        --worker \\
        -- --fps 5
"""

import argparse
import json
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path

import requests

sys.path.insert(0, str(Path(__file__).resolve().parent))
import sae_env  # noqa: F401  — loads .env into os.environ on import
from sae_gallery import load_gallery_hdf5, save_gallery_hdf5
from sae_jellyfin import (
    jf_get, find_item_id, fetch_cast_person_ids, actor_jellyfin_id, fetch_episode_info,
)
from sae_tmdb import tmdb_episode_cast, tmdb_tv_id_from_imdb


def fetch_item_path(base_url: str, api_key: str, item_id: str) -> tuple[str, str]:
    """Return (Name, Path) for a Jellyfin item."""
    data = jf_get(base_url, api_key, "/Items", Ids=item_id, Fields="Path", Recursive="true")
    items = data.get("Items", [])
    if not items:
        raise ValueError(f"Item {item_id} not found")
    path = items[0].get("Path")
    if not path:
        raise ValueError(f"Item {item_id} has no Path (not a single media file?)")
    return items[0].get("Name", item_id), path


def fetch_pending_tasks(base_url: str, api_key: str, limit: int = 10) -> list[dict]:
    """GET /Plugins/JRay/Tasks/Pending — a random batch of items with no truth data yet.

    Returns a list of {"item_id", "path", "name"}; an empty list means
    there's nothing left to do (for now).
    """
    return jf_get(base_url, api_key, "/Plugins/JRay/Tasks/Pending", limit=limit)


def push_truth(base_url: str, api_key: str, item_id: str, output_path: str) -> None:
    """PUT the scene_analyze output JSON to the JRay plugin's Truth endpoint.

    Requires an Administrator API key (reuses --api-key).
    """
    url = base_url.rstrip("/") + f"/Plugins/JRay/Items/{item_id}/Truth"
    headers = {"X-Emby-Token": api_key, "Content-Type": "application/json"}

    # --verbosity 1 (standard) adds a "frames" array with per-frame bbox/
    # similarity data for local debugging — the Truth schema only needs
    # schema_version/movie/sample_fps/anneal_sec/actors, and "frames" can
    # be tens of MB for a full episode, well past typical proxy body limits.
    payload = json.loads(Path(output_path).read_text())
    if payload.pop("frames", None) is not None:
        print(f"Stripped per-frame data before push ({output_path} keeps it locally)", file=sys.stderr)
    body = json.dumps(payload)

    r = requests.put(url, headers=headers, data=body, timeout=30)
    if r.status_code == 204:
        print(f"Pushed {output_path} -> {url}", file=sys.stderr)
        return
    if r.status_code in (401, 403):
        raise RuntimeError(f"Push failed ({r.status_code}): API key needs Administrator rights for {url}")
    if r.status_code == 400:
        raise RuntimeError(f"Push failed (400 Bad Request): {r.text}")
    r.raise_for_status()


def _norm_name(name: str) -> str:
    """Casefold + collapse whitespace so 'Robert  Downey Jr.' matches 'robert downey jr.'."""
    return " ".join((name or "").split()).casefold()


def episode_cast_actors(args, item_id: str, gallery: dict) -> list[dict] | None:
    """Filter the gallery to one episode's TMDB cast, or None to fall back to series-wide.

    Returns None (so the caller uses the existing series-wide jellyfin_id
    filter) whenever the item isn't an episode, the series TMDB id or
    season/episode numbers can't be resolved, TMDB returns no cast, or the
    match against the gallery comes up empty. Matches TMDB people to gallery
    actors by tmdb_id first, then by normalized name for actors whose gallery
    entry has no tmdb_id.
    """
    if args.episode_cast == "series":
        return None
    if not args.tmdb_key:
        return None

    info = fetch_episode_info(args.jellyfin_url, args.api_key, item_id)
    if not info or info["season"] is None or info["episode"] is None:
        return None

    tv_id = info["series_tmdb"]
    if not tv_id and info["series_imdb"]:
        try:
            tv_id = tmdb_tv_id_from_imdb(info["series_imdb"], args.tmdb_key)
        except requests.RequestException as e:
            print(f"  [warn] TMDB series lookup failed: {e}", file=sys.stderr)
    if not tv_id:
        print("  No TMDB series id for this episode; using series-wide cast", file=sys.stderr)
        return None

    try:
        cast = tmdb_episode_cast(tv_id, info["season"], info["episode"], args.tmdb_key)
    except requests.RequestException as e:
        print(f"  [warn] TMDB episode credits failed (S{info['season']}E{info['episode']}): {e}; "
              f"using series-wide cast", file=sys.stderr)
        return None
    if not cast:
        return None

    # The episode-credits endpoint gives a TMDB person id + name per cast
    # member (no imdb id), so match on the gallery's tmdb_id first — populated
    # by make_jellyfin_gallery.py when built with a TMDB key — then fall back
    # to normalized name for actors whose gallery entry has no tmdb_id.
    cast_tmdb = {str(p["id"]) for p in cast if p.get("id") is not None}
    cast_names = {_norm_name(p.get("name", "")) for p in cast}

    actors = [a for a in gallery.get("actors", [])
              if (a.get("tmdb_id") and str(a["tmdb_id"]) in cast_tmdb)
              or _norm_name(a.get("name", "")) in cast_names]
    if not actors:
        print("  TMDB episode cast matched no gallery actors; using series-wide cast",
              file=sys.stderr)
        return None

    print(f"Filtered gallery to {len(actors)}/{len(gallery.get('actors', []))} "
          f"actor(s) in TMDB S{info['season']}E{info['episode']} cast", file=sys.stderr)
    return actors


def process_item(args, item_id: str, extra: list[str]) -> None:
    """Resolve, analyze, and (unless --no-push) push truth for one Jellyfin item."""
    name, movie_path = fetch_item_path(args.jellyfin_url, args.api_key, item_id)
    print(f"Resolved {name!r} -> {movie_path}", file=sys.stderr)
    if not Path(movie_path).is_file():
        raise RuntimeError(f"Resolved path does not exist on this filesystem: {movie_path} "
                            f"(this tool must share Jellyfin's media mount)")

    output = args.output or f"{name}.json"

    gallery_path = args.gallery
    filtered_file = None
    if not args.no_filter:
        gallery = load_gallery_hdf5(Path(args.gallery))
        actors = episode_cast_actors(args, item_id, gallery)
        if actors is None:
            cast_ids = fetch_cast_person_ids(args.jellyfin_url, args.api_key, item_id)
            actors = [a for a in gallery.get("actors", [])
                      if actor_jellyfin_id(a) in cast_ids]
            print(f"Filtered gallery to {len(actors)}/{len(gallery.get('actors', []))} "
                  f"actor(s) credited in {name!r}", file=sys.stderr)
        filtered_file = tempfile.NamedTemporaryFile(
            suffix=".h5", prefix="sae_gallery_", delete=False)
        filtered_file.close()
        save_gallery_hdf5({"actors": actors}, Path(filtered_file.name))
        gallery_path = filtered_file.name

    try:
        # Default to minimal output (just actor/scene windows, no per-frame
        # data) — that's all the Truth push needs. Placed before *extra so a
        # debug run can override with `-- --verbosity 1` (last flag wins).
        cmd = [args.bin, "--movie", movie_path, "--gallery", gallery_path,
               "--output", output, "--verbosity", "0", *extra]
        print("Running:", " ".join(cmd), file=sys.stderr)

        if args.dry_run:
            return

        subprocess.run(cmd, check=True)
    finally:
        if filtered_file is not None:
            Path(filtered_file.name).unlink(missing_ok=True)

    # Stamp the title's Jellyfin item-id into the output so downstream tools
    # (e.g. cameo_hunt) can check cast membership in Jellyfin's own id space
    # without an unreliable Path lookup. scene_analyze doesn't know the id, so
    # patch the JSON it just wrote.
    if not args.dry_run:
        try:
            out = json.loads(Path(output).read_text())
            if isinstance(out, dict):
                out["jellyfin_item_id"] = item_id
                Path(output).write_text(json.dumps(out))
        except (OSError, ValueError) as e:
            print(f"  [warn] could not stamp jellyfin_item_id into {output}: {e}",
                  file=sys.stderr)

    if not args.no_push:
        push_truth(args.jellyfin_url, args.api_key, item_id, output)


def run_worker(args, extra: list[str]) -> None:
    """Poll /Plugins/JRay/Tasks/Pending and process items until the backlog is empty.

    The endpoint returns a random sample, so repeated polling naturally
    spreads work across the backlog without server-side task tracking. An
    empty array means there's nothing left to do (for now).
    """
    while True:
        try:
            tasks = fetch_pending_tasks(args.jellyfin_url, args.api_key, args.poll_limit)
        except requests.HTTPError as e:
            print(f"Failed to fetch pending tasks: {e}", file=sys.stderr)
            tasks = []

        if not tasks:
            print("No pending tasks.", file=sys.stderr)
            if args.once:
                return
            time.sleep(args.poll_interval)
            continue

        for task in tasks:
            item_id = task["item_id"]
            label = task.get("name") or task.get("path") or item_id
            print(f"=== {label} ({item_id}) ===", file=sys.stderr)
            try:
                process_item(args, item_id, extra)
            except Exception as e:
                print(f"Failed to process {label!r} ({item_id}): {e}", file=sys.stderr)

        if args.once:
            return


def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter,
    )
    parser.add_argument("--jellyfin-url", default=os.environ.get("JELLYFIN_URL"),
                        required=not os.environ.get("JELLYFIN_URL"),
                        help="Jellyfin base URL. Env: JELLYFIN_URL")
    parser.add_argument("--api-key", default=os.environ.get("JELLYFIN_API_KEY"),
                        required=not os.environ.get("JELLYFIN_API_KEY"),
                        help="Jellyfin API key. Env: JELLYFIN_API_KEY")
    group = parser.add_mutually_exclusive_group()
    group.add_argument("--item-id", help="Jellyfin item id of the title")
    group.add_argument("--title", help="Title to search for (uses first match)")
    group.add_argument("--worker", action="store_true",
                        help="Poll /Plugins/JRay/Tasks/Pending for work instead of "
                             "processing a single --item-id/--title")
    parser.add_argument("--poll-limit", type=int, default=10,
                         help="Tasks/Pending batch size for --worker (default: 10, max 100)")
    parser.add_argument("--poll-interval", type=float, default=60,
                         help="Seconds to wait between polls when --worker finds nothing "
                              "(default: 60)")
    parser.add_argument("--once", action="store_true",
                         help="With --worker, process one batch and exit instead of "
                              "looping forever")
    parser.add_argument("--item-types", default="Movie,Episode",
                         help="Item types to search when using --title (default: Movie,Episode)")
    parser.add_argument("--gallery", required=True,
                         help="Global gallery.h5 built by make_jellyfin_gallery.py")
    parser.add_argument("--no-filter", action="store_true",
                         help="Skip per-title cast filtering and pass --gallery through as-is")
    parser.add_argument("--episode-cast", choices=("tmdb", "series"), default="tmdb",
                         help="For Episode items, how to pick the cast to filter the gallery: "
                              "'tmdb' uses TMDB per-episode credits (cast + guest stars), "
                              "falling back to series-wide if unavailable; 'series' always uses "
                              "Jellyfin's series-wide cast (default: tmdb)")
    parser.add_argument("--tmdb-key", default=os.environ.get("TMDB_API_KEY"),
                         help="TMDB API key/bearer token, used for --episode-cast tmdb. "
                              "Env: TMDB_API_KEY (also read from .env)")
    parser.add_argument("--output", default=None,
                         help="scene_analyze output JSON (default: <title>.json; "
                              "ignored with --worker, which always uses <title>.json)")
    parser.add_argument("--bin", default="build/scene_analyze",
                         help="Path to scene_analyze binary (default: build/scene_analyze)")
    parser.add_argument("--preview", action="store_true",
                         help="Launch the scene_preview binary (live OpenCV display window) "
                              "instead of headless scene_analyze. Implies --no-push. Use "
                              "--preview-bin to override its path. Press q/Esc to close.")
    parser.add_argument("--preview-bin", default="build/scene_preview",
                         help="Path to scene_preview binary (default: build/scene_preview)")
    parser.add_argument("--dry-run", action="store_true",
                         help="Resolve and print the scene_analyze command without running it")
    parser.add_argument("--no-push", action="store_true",
                         help="Don't push results to the JRay plugin after the run "
                              "(local-only/debug; --output is still written)")
    args, extra = parser.parse_known_args()
    if extra and extra[0] == "--":
        extra = extra[1:]

    # --preview swaps in the display binary and disables pushing truth (a preview
    # run is interactive/debug, not a truth-producing analysis).
    if args.preview:
        args.bin = args.preview_bin
        args.no_push = True

    if args.worker:
        if args.output:
            sys.exit("--output is incompatible with --worker (each item needs its own file)")
        run_worker(args, extra)
        return

    if not args.item_id and not args.title:
        sys.exit("one of --item-id, --title, or --worker is required")

    item_id = args.item_id
    if item_id is None:
        item_id = find_item_id(args.jellyfin_url, args.api_key, args.title, args.item_types.split(","))
        print(f"Resolved title to item id {item_id}", file=sys.stderr)

    try:
        process_item(args, item_id, extra)
    except RuntimeError as e:
        sys.exit(str(e))


if __name__ == "__main__":
    main()