dtourolle
152c34b1f4
Consolidate copy-pasted logic across the gallery/run scripts into shared modules: - sae_env.py — zero-dependency .env loader (populates os.environ) - sae_tmdb.py — TMDB API helpers (tmdb_get, person images, id lookups) - sae_jellyfin.py— Jellyfin API helpers (jf_get, id/URL normalisation) - sae_gallery.py — image download + gallery.json writing make_gallery, make_jellyfin_gallery and filter_gallery now import these instead of carrying their own near-identical copies.
120 lines
4.6 KiB
Python
120 lines
4.6 KiB
Python
"""Shared Jellyfin API helpers for the gallery/run scripts.
|
|
|
|
Consolidates the jf_get/find_item_id/fetch_cast_person_ids helpers that were
|
|
previously copy-pasted across filter_gallery.py, make_jellyfin_gallery.py and
|
|
run_from_jellyfin.py, plus the small gallery-id and URL-normalisation helpers.
|
|
"""
|
|
|
|
import sys
|
|
|
|
import requests
|
|
|
|
|
|
def jf_get(base_url: str, api_key: str, path: str, **params) -> dict:
|
|
url = base_url.rstrip("/") + path
|
|
headers = {"X-Emby-Token": api_key, "Accept": "application/json"}
|
|
r = requests.get(url, params=params, headers=headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
def normalize_jellyfin_url(url: str) -> str:
|
|
"""Strip any query string and a trailing /Items from a base URL.
|
|
|
|
Defends against a base URL that accidentally includes an API path, e.g.
|
|
"https://host/Items?" — so we don't build double-nested, query-mangled URLs.
|
|
"""
|
|
url = url.split("?", 1)[0].rstrip("/")
|
|
if url.endswith("/Items"):
|
|
url = url[: -len("/Items")]
|
|
return url
|
|
|
|
|
|
def actor_jellyfin_id(actor: dict) -> str | None:
|
|
"""Return an actor's Jellyfin person id, tolerating the older key name.
|
|
|
|
current make_jellyfin_gallery.py writes "jellyfin_id"; older galleries used
|
|
"jellyfin_person_id" (see gallery_store.cpp's fallback for the same pair).
|
|
"""
|
|
return actor.get("jellyfin_id") or actor.get("jellyfin_person_id")
|
|
|
|
|
|
def find_item_id(base_url: str, api_key: str, title: str, item_types: list[str]) -> str:
|
|
data = jf_get(
|
|
base_url, api_key, "/Items",
|
|
Recursive="true",
|
|
IncludeItemTypes=",".join(item_types),
|
|
SearchTerm=title,
|
|
Limit=10,
|
|
)
|
|
items = data.get("Items", [])
|
|
if not items:
|
|
raise ValueError(f"No item found matching title {title!r}")
|
|
if len(items) > 1:
|
|
print("Multiple matches found:", file=sys.stderr)
|
|
for it in items:
|
|
print(f" {it['Id']} {it.get('Type')} {it.get('Name')} ({it.get('ProductionYear')})", file=sys.stderr)
|
|
print(f"Using first match: {items[0]['Name']}", file=sys.stderr)
|
|
return items[0]["Id"]
|
|
|
|
|
|
def fetch_episode_info(base_url: str, api_key: str, item_id: str) -> dict | None:
|
|
"""Return episode locator info, or None if the item isn't an Episode.
|
|
|
|
Pulls the season/episode numbers off the Episode item and the series'
|
|
TMDB/IMDB ProviderIds off the parent Series — TMDB's episode-credits
|
|
endpoint is keyed by (series id, season, episode). Returns a dict with
|
|
keys: series_id, season, episode, series_tmdb, series_imdb (any of the
|
|
id/number values may be None if Jellyfin hasn't populated them).
|
|
"""
|
|
data = jf_get(base_url, api_key, "/Items", Ids=item_id,
|
|
Fields="ParentIndexNumber,IndexNumber", Recursive="true")
|
|
items = data.get("Items", [])
|
|
if not items:
|
|
return None
|
|
item = items[0]
|
|
if item.get("Type") != "Episode":
|
|
return None
|
|
|
|
series_id = item.get("SeriesId")
|
|
series_tmdb = series_imdb = None
|
|
if series_id:
|
|
series_data = jf_get(base_url, api_key, "/Items", Ids=series_id,
|
|
Fields="ProviderIds", Recursive="true")
|
|
series_items = series_data.get("Items", [])
|
|
if series_items:
|
|
providers = series_items[0].get("ProviderIds", {})
|
|
series_tmdb = providers.get("Tmdb")
|
|
series_imdb = providers.get("Imdb")
|
|
|
|
return {
|
|
"series_id": series_id,
|
|
"season": item.get("ParentIndexNumber"),
|
|
"episode": item.get("IndexNumber"),
|
|
"series_tmdb": series_tmdb,
|
|
"series_imdb": series_imdb,
|
|
}
|
|
|
|
|
|
def fetch_cast_person_ids(base_url: str, api_key: str, item_id: str) -> set[str]:
|
|
# /Items/{id} (no user context) returns 400 on recent Jellyfin servers;
|
|
# /Items?Ids=... works with an API key alone.
|
|
data = jf_get(base_url, api_key, "/Items", Ids=item_id, Fields="People", Recursive="true")
|
|
items = data.get("Items", [])
|
|
if not items:
|
|
raise ValueError(f"Item {item_id} not found")
|
|
item = items[0]
|
|
cast_ids = {p["Id"] for p in item.get("People", []) if p.get("Type") == "Actor"}
|
|
|
|
# Episodes generally only carry their own guest stars in "People" — the
|
|
# regular/recurring cast lives on the parent Series item, so pull that
|
|
# in too or the filtered gallery ends up missing the main cast.
|
|
series_id = item.get("SeriesId")
|
|
if series_id:
|
|
series_data = jf_get(base_url, api_key, "/Items", Ids=series_id, Fields="People", Recursive="true")
|
|
series_items = series_data.get("Items", [])
|
|
if series_items:
|
|
cast_ids |= {p["Id"] for p in series_items[0].get("People", []) if p.get("Type") == "Actor"}
|
|
|
|
return cast_ids
|