dtourolle
96b1c22194
Add two cameo hunters that flag actors recognised in a title but absent from its cast: - cameo_jellyfin.py — pure-Jellyfin cast-membership check (no id cross-walk) - cameo_hunt.py — TMDB filmography check (actor's combined_credits) run_from_jellyfin.py now stamps the analysed title's Jellyfin item GUID into the output JSON as top-level 'jellyfin_item_id' (scene_analyze can't know it), which cameo_jellyfin.py uses to look up the cast in Jellyfin's own id space. Document that field in the result-sink output schema header.
319 lines
13 KiB
Python
319 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
"""cameo_hunt.py — find actors recognized in titles that aren't on their TMDB resumé.
|
||
|
||
For each scene_analyze output JSON, every recognized actor is checked against
|
||
their own TMDB filmography (combined_credits). If the title an actor was
|
||
recognized in is NOT among the series/movies TMDB lists for them, it's flagged
|
||
as a cameo candidate — an uncredited/surprise appearance (or, with no
|
||
confidence gating, possibly a face misidentification: the two look identical in
|
||
this data, so treat the list as leads, not proof).
|
||
|
||
Resolution is all via TMDB (key from .env), no per-file Jellyfin calls:
|
||
* each title's series/movie is resolved by name (+year) parsed from the
|
||
JSON's "movie" path, cached per unique title;
|
||
* each actor's resumé is fetched once via /person/{id}/combined_credits,
|
||
cached per unique tmdb_id.
|
||
|
||
Usage:
|
||
python scripts/cameo_hunt.py *.json
|
||
python scripts/cameo_hunt.py --min-seconds 5 --json out.json *.json
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import re
|
||
import sys
|
||
from collections import defaultdict
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent))
|
||
import sae_env # noqa: F401 — loads .env (TMDB_API_KEY, JELLYFIN_*)
|
||
from sae_tmdb import tmdb_get
|
||
from sae_jellyfin import jf_get, normalize_jellyfin_url
|
||
|
||
import requests
|
||
|
||
# Strip episode markers, a "(year)", and release/quality/codec tags to recover a
|
||
# searchable series/movie name from a media path.
|
||
# Episode markers: "S01E13", "1x03".
|
||
_SXXEXX = re.compile(r"\b([sS]\d{1,2}[eE]\d{1,3}|\d{1,2}x\d{1,3})\b")
|
||
_YEAR = re.compile(r"[.\s(](\d{4})[.\s)]")
|
||
# Common scene-release tags; everything from the first one onward is noise. Match
|
||
# as whole tokens so we can also truncate the name at the first tag hit.
|
||
_TAG_WORDS = (
|
||
r"\d{3,4}p|2160p|4k|uhd|hdr(10)?(plus)?|web[\s.\-]?dl|web[\s.\-]?rip|"
|
||
r"bluray|blu[\s.\-]?ray|brrip|bdrip|hdtv|dvdrip|remux|"
|
||
r"x26[45]|h\.?26[45]|hevc|xvid|divx|aac|ac3|dts|ddp?5\.1|"
|
||
r"internal|theatrical|extended|uncut|repack|proper"
|
||
)
|
||
_TAG = re.compile(rf"(?i)\b({_TAG_WORDS})\b")
|
||
|
||
|
||
def parse_title(movie_path: str) -> tuple[str, int | None, bool]:
|
||
"""Return (search_name, year, is_tv) parsed from a media file path.
|
||
|
||
TV episodes look like ".../<Series> (year)/Season N/<Series> SxxExx ...".
|
||
For those we want the *series* directory name; for movies, the filename.
|
||
"""
|
||
p = Path(movie_path)
|
||
parts = p.parts
|
||
has_season_dir = any(re.match(r"(?i)season\s*\d+", x) for x in parts)
|
||
has_ep_marker = bool(_SXXEXX.search(p.stem))
|
||
is_tv = has_season_dir or has_ep_marker
|
||
|
||
if has_season_dir:
|
||
# The series folder is the part above the "Season N" dir (…/Series/Season N/file).
|
||
season_idx = next(i for i, x in enumerate(parts) if re.match(r"(?i)season\s*\d+", x))
|
||
raw = parts[season_idx - 1] if season_idx > 0 else p.stem
|
||
else:
|
||
raw = p.stem
|
||
|
||
# Pull a 4-digit year (1900–2099) before stripping, if present.
|
||
year = None
|
||
for m in _YEAR.finditer(" " + raw + " "):
|
||
y = int(m.group(1))
|
||
if 1900 <= y <= 2099:
|
||
year = y
|
||
break
|
||
|
||
# Truncate at the first release tag (everything after is noise), then strip
|
||
# episode markers / year / separators from what remains.
|
||
tag = _TAG.search(raw)
|
||
if tag:
|
||
raw = raw[: tag.start()]
|
||
name = _SXXEXX.sub(" ", raw)
|
||
name = re.sub(r"[.\s(]\d{4}[.\s)]?", " ", " " + name + " ") # drop the year token
|
||
name = name.replace(".", " ").replace("_", " ")
|
||
name = re.sub(r"\s+", " ", name).strip(" -")
|
||
return name, year, is_tv
|
||
|
||
|
||
def jf_title_tmdb_id(movie_path: str, jf_url: str, jf_key: str, is_tv: bool,
|
||
timeout: float = 8.0) -> int | None:
|
||
"""Resolve a media path to its (series, for TV) TMDB id via Jellyfin, or None.
|
||
|
||
Looks the item up by Path, then reads ProviderIds.Tmdb off the parent Series
|
||
(TV) or the item itself (movie). Uses a short timeout so a slow Path query
|
||
degrades to the TMDB fallback instead of stalling the whole run.
|
||
"""
|
||
if not (jf_url and jf_key):
|
||
return None
|
||
base = jf_url.rstrip("/")
|
||
headers = {"X-Emby-Token": jf_key, "Accept": "application/json"}
|
||
try:
|
||
r = requests.get(base + "/Items", headers=headers, timeout=timeout,
|
||
params={"Path": movie_path, "Recursive": "true",
|
||
"IncludeItemTypes": "Episode,Movie",
|
||
"Fields": "ProviderIds", "Limit": 1})
|
||
r.raise_for_status()
|
||
items = r.json().get("Items", [])
|
||
if not items:
|
||
return None
|
||
item = items[0]
|
||
if is_tv:
|
||
# The path says TV episode (SxxExx + Season dir). If Jellyfin matched
|
||
# it to a Movie item or an episode with no SeriesId, its Tmdb id is
|
||
# for the wrong title (Jellyfin mis-identified the file) — don't trust
|
||
# it; return None so the caller falls back to TMDB name search.
|
||
if item.get("Type") != "Episode" or not item.get("SeriesId"):
|
||
return None
|
||
target_id = item.get("SeriesId") if is_tv else item.get("Id")
|
||
providers = item.get("ProviderIds", {})
|
||
if is_tv and target_id:
|
||
# ProviderIds on the episode are the episode's; fetch the series'.
|
||
r2 = requests.get(base + "/Items", headers=headers, timeout=timeout,
|
||
params={"Ids": target_id, "Fields": "ProviderIds",
|
||
"Recursive": "true", "Limit": 1})
|
||
r2.raise_for_status()
|
||
sitems = r2.json().get("Items", [])
|
||
providers = sitems[0].get("ProviderIds", {}) if sitems else {}
|
||
tmdb = providers.get("Tmdb")
|
||
return int(tmdb) if tmdb else None
|
||
except (requests.RequestException, ValueError):
|
||
return None
|
||
|
||
|
||
def resolve_title_id(name: str, year: int | None, is_tv: bool, key: str,
|
||
cache: dict) -> int | None:
|
||
"""Resolve a series/movie name to its TMDB id (cached per (name, is_tv))."""
|
||
ck = (name, is_tv)
|
||
if ck in cache:
|
||
return cache[ck]
|
||
endpoint = "/search/tv" if is_tv else "/search/movie"
|
||
yparam = {"first_air_date_year": year} if (is_tv and year) else (
|
||
{"year": year} if year else {})
|
||
try:
|
||
data = tmdb_get(endpoint, key, query=name, **yparam)
|
||
except requests.RequestException as e:
|
||
print(f" [warn] TMDB search failed for {name!r}: {e}", file=sys.stderr)
|
||
cache[ck] = None
|
||
return None
|
||
results = data.get("results", [])
|
||
tid = results[0]["id"] if results else None
|
||
if tid is None:
|
||
print(f" [warn] no TMDB match for {name!r} ({'tv' if is_tv else 'movie'})", file=sys.stderr)
|
||
cache[ck] = tid
|
||
return tid
|
||
|
||
|
||
TMDB_ANIMATION_GENRE_ID = 16 # "Animation" — same id for /movie and /tv on TMDB
|
||
|
||
|
||
def is_animation(title_id: int, is_tv: bool, key: str, cache: dict) -> bool:
|
||
"""Return True if the title is tagged with TMDB's Animation genre (cached).
|
||
|
||
Fetches /movie/{id} or /tv/{id} details (which include `genres`). On a fetch
|
||
failure we return False (don't drop the title on a transient API error).
|
||
"""
|
||
ck = (title_id, is_tv)
|
||
if ck in cache:
|
||
return cache[ck]
|
||
endpoint = f"/tv/{title_id}" if is_tv else f"/movie/{title_id}"
|
||
try:
|
||
data = tmdb_get(endpoint, key)
|
||
except requests.RequestException as e:
|
||
print(f" [warn] genre fetch failed for title {title_id}: {e}", file=sys.stderr)
|
||
cache[ck] = False
|
||
return False
|
||
animated = any(g.get("id") == TMDB_ANIMATION_GENRE_ID for g in data.get("genres", []))
|
||
cache[ck] = animated
|
||
return animated
|
||
|
||
|
||
def actor_resume(tmdb_id: str, key: str, cache: dict) -> set[int] | None:
|
||
"""Return the set of TMDB title ids on an actor's resumé (cached), or None on failure."""
|
||
if tmdb_id in cache:
|
||
return cache[tmdb_id]
|
||
try:
|
||
data = tmdb_get(f"/person/{tmdb_id}/combined_credits", key)
|
||
except requests.RequestException as e:
|
||
print(f" [warn] resumé fetch failed for tmdb person {tmdb_id}: {e}", file=sys.stderr)
|
||
cache[tmdb_id] = None
|
||
return None
|
||
ids = {c["id"] for c in data.get("cast", []) if "id" in c}
|
||
ids |= {c["id"] for c in data.get("crew", []) if "id" in c}
|
||
cache[tmdb_id] = ids
|
||
return ids
|
||
|
||
|
||
def total_seconds(scenes: list) -> float:
|
||
return sum(e - s for s, e in (x for x in scenes if len(x) == 2))
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description=__doc__,
|
||
formatter_class=argparse.RawDescriptionHelpFormatter)
|
||
ap.add_argument("files", nargs="+", help="scene_analyze output JSON files")
|
||
ap.add_argument("--min-seconds", type=float, default=0.0,
|
||
help="only flag a cameo if the actor's total on-screen time in the "
|
||
"title is at least this many seconds (default: 0 — flag all)")
|
||
ap.add_argument("--json", default=None, help="also write findings to this JSON file")
|
||
ap.add_argument("--no-skip-animation", dest="skip_animation", action="store_false",
|
||
help="do not skip titles tagged with TMDB's Animation genre "
|
||
"(animated titles are skipped by default)")
|
||
args = ap.parse_args()
|
||
|
||
key = os.environ.get("TMDB_API_KEY")
|
||
if not key:
|
||
sys.exit("TMDB_API_KEY not set (expected in .env)")
|
||
jf_url = normalize_jellyfin_url(os.environ.get("JELLYFIN_URL", "")) or ""
|
||
jf_key = os.environ.get("JELLYFIN_API_KEY", "")
|
||
|
||
title_cache: dict = {} # (name, is_tv) -> tmdb id, for TMDB name search
|
||
jf_cache: dict = {} # series folder / movie path -> tmdb id, for Jellyfin
|
||
resume_cache: dict = {}
|
||
genre_cache: dict = {} # (title_id, is_tv) -> is-animation bool
|
||
cameos: list[dict] = []
|
||
skipped_no_tmdb = 0
|
||
skipped_animation = 0
|
||
n_via_jf = 0
|
||
|
||
paths = [Path(f) for f in args.files]
|
||
for i, path in enumerate(paths, 1):
|
||
try:
|
||
data = json.loads(path.read_text())
|
||
except (OSError, ValueError):
|
||
continue
|
||
if not isinstance(data, dict) or "actors" not in data:
|
||
continue
|
||
movie = data.get("movie", "")
|
||
name, year, is_tv = parse_title(movie)
|
||
|
||
# Jellyfin first (authoritative TMDB id), cached per series folder so a
|
||
# show is only looked up once across all its episodes; TMDB name search
|
||
# as fallback on timeout/miss.
|
||
title_id = None
|
||
if jf_url and jf_key:
|
||
mp = Path(movie)
|
||
ck = str(mp.parent.parent) if is_tv else movie # series folder vs movie file
|
||
if ck in jf_cache:
|
||
title_id = jf_cache[ck]
|
||
else:
|
||
title_id = jf_title_tmdb_id(movie, jf_url, jf_key, is_tv)
|
||
jf_cache[ck] = title_id
|
||
if title_id is not None:
|
||
n_via_jf += 1
|
||
if title_id is None:
|
||
title_id = resolve_title_id(name, year, is_tv, key, title_cache)
|
||
if title_id is None:
|
||
continue # can't decide cameo without the title's id
|
||
|
||
if args.skip_animation and is_animation(title_id, is_tv, key, genre_cache):
|
||
skipped_animation += 1
|
||
continue
|
||
|
||
for actor in data["actors"]:
|
||
tmdb_id = actor.get("tmdb_id")
|
||
if not tmdb_id:
|
||
skipped_no_tmdb += 1
|
||
continue
|
||
secs = total_seconds(actor.get("scenes", []))
|
||
if secs < args.min_seconds:
|
||
continue
|
||
resume = actor_resume(tmdb_id, key, resume_cache)
|
||
if resume is None:
|
||
continue
|
||
if title_id not in resume:
|
||
cameos.append({
|
||
"actor": actor.get("name", ""),
|
||
"tmdb_id": tmdb_id,
|
||
"title": name,
|
||
"title_id": title_id,
|
||
"is_tv": is_tv,
|
||
"file": path.name,
|
||
"seconds": round(secs, 1),
|
||
"scenes": len(actor.get("scenes", [])),
|
||
})
|
||
if i % 50 == 0:
|
||
print(f" …{i}/{len(paths)} files, {len(cameos)} candidates so far", file=sys.stderr)
|
||
|
||
# Report, grouped by actor, most on-screen time first.
|
||
cameos.sort(key=lambda c: (c["actor"], -c["seconds"]))
|
||
by_actor = defaultdict(list)
|
||
for c in cameos:
|
||
by_actor[c["actor"]].append(c)
|
||
|
||
print(f"\n(resolved {n_via_jf} title lookup(s) via Jellyfin, rest via TMDB search)",
|
||
file=sys.stderr)
|
||
print(f"\n=== {len(cameos)} cameo candidate(s) across {len(by_actor)} actor(s) ===\n")
|
||
for actor in sorted(by_actor):
|
||
rows = by_actor[actor]
|
||
print(f"{actor}:")
|
||
for c in rows:
|
||
kind = "TV" if c["is_tv"] else "film"
|
||
print(f" {c['title']!r} ({kind}) {c['seconds']}s / {c['scenes']} scene(s) [{c['file']}]")
|
||
if skipped_animation:
|
||
print(f"\n({skipped_animation} animated title file(s) skipped)", file=sys.stderr)
|
||
if skipped_no_tmdb:
|
||
print(f"\n({skipped_no_tmdb} recognized actor(s) had no tmdb_id and were skipped)",
|
||
file=sys.stderr)
|
||
|
||
if args.json:
|
||
Path(args.json).write_text(json.dumps(cameos, indent=2) + "\n")
|
||
print(f"\nWrote {len(cameos)} candidate(s) to {args.json}", file=sys.stderr)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|