feat(cameo): detect recognised actors not credited in a title
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.
This commit is contained in:
@@ -0,0 +1,318 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user