jellyfin-fetch-subs.py

jellyfin-fetch-subs.py · 9.5 KiB · Python Raw

#!/usr/bin/env python3
"""Fetch missing subtitles for a Jellyfin series/season via Jellyfin's own
RemoteSearch/Subtitles API (backed by OpenSubtitles).

Built 2026-08-31 after doing this by hand for Mission: Impossible (1988) S1 —
generalized so any show/season can get the same treatment.

Usage:
    python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1
    python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1 --year 1988
    python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1 --lang eng --force
    python3 bin/jellyfin-fetch-subs.py --list-seasons "Mission: Impossible"

Series lookup goes through /Search/Hints (Type=Series) rather than /Items —
plain /Items search came back empty against this server for reasons not fully
chased down; Search/Hints works reliably. When a title matches more than one
series (e.g. two "Mission: Impossible" — 1966 and 1988), pass --year to
disambiguate, or the script lists the candidates and exits.

Credentials, in priority order: --jellyfin-url/--jellyfin-key flags, then
JELLYFIN_URL/JELLYFIN_API_KEY env vars, then secrets/credentials.yaml (Evie's
own default — needs the project .venv for PyYAML). The flag/env path needs
nothing beyond stdlib, so anyone with their own Jellyfin API key can run this
standalone with plain system python3:
    JELLYFIN_URL=http://jellyfin.lan.1qaz.ca:8096 JELLYFIN_API_KEY=<key> \\
        python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1
"""
import argparse
import json
import os
import sys
import time
import urllib.parse
import urllib.request
from pathlib import Path

BASE = None
HEADERS = None

def _configure(url: str | None, key: str | None) -> None:
    """Resolve BASE/HEADERS from (in order) explicit args, env vars, or
    secrets/credentials.yaml. The credentials.yaml path is imported lazily,
    right here, so anyone supplying their own url/key never needs PyYAML or
    the project .venv at all — a plain system python3 works."""
    global BASE, HEADERS
    url = url or os.environ.get("JELLYFIN_URL")
    key = key or os.environ.get("JELLYFIN_API_KEY")
    if not url or not key:
        sys.path.insert(0, str(Path(__file__).parent))
        from credentials import service  # noqa: E402
        jf = service("jellyfin")
        url = url or jf["url"]
        key = key or jf["api_key"]
    BASE = url.rstrip("/")
    HEADERS = {"X-Emby-Token": key}

def _get(path, params=None):
    url = f"{BASE}{path}"
    if params:
        url += "?" + urllib.parse.urlencode(params)
    req = urllib.request.Request(url, headers=HEADERS)
    with urllib.request.urlopen(req) as resp:
        body = resp.read()
    return json.loads(body) if body else None

def _post(path):
    req = urllib.request.Request(url=f"{BASE}{path}", headers=HEADERS, method="POST")
    with urllib.request.urlopen(req) as resp:
        return resp.status

def _first_user_id():
    users = _get("/Users")
    if not users:
        raise RuntimeError("no Jellyfin users found — can't check subtitle state")
    return users[0]["Id"]

def find_series(title, year=None):
    hints = _get("/Search/Hints", {"SearchTerm": title, "IncludeItemTypes": "Series"})
    candidates = [h for h in hints.get("SearchHints", []) if h.get("Type") == "Series"]
    if year:
        candidates = [c for c in candidates if c.get("ProductionYear") == year]
    if len(candidates) == 1:
        return candidates[0]
    if not candidates:
        raise SystemExit(f"no series found matching '{title}'" + (f" ({year})" if year else ""))
    print(f"multiple series match '{title}' — pass --year to disambiguate:", file=sys.stderr)
    for c in candidates:
        print(f"  {c.get('ProductionYear', '?')} — {c['Name']} (id={c['Id']})", file=sys.stderr)
    raise SystemExit(2)

def list_seasons(series_id):
    return _get(f"/Shows/{series_id}/Seasons")["Items"]

def find_season(series_id, season_num):
    seasons = list_seasons(series_id)
    for s in seasons:
        if s.get("IndexNumber") == season_num:
            return s
    available = ", ".join(str(s.get("IndexNumber")) for s in seasons)
    raise SystemExit(f"season {season_num} not found — available: {available}")

def list_episodes(series_id, season_id):
    eps = _get(f"/Shows/{series_id}/Episodes", {"SeasonId": season_id})["Items"]
    return sorted(eps, key=lambda e: e.get("IndexNumber") or 0)

# Jellyfin falls back to writing subtitles here when it can't write next to
# the media file (e.g. a permission problem on the share) — silently, no
# error, just a stream Plex (or anything else reading the raw files) will
# never see. Found 2026-08-31/09-01 the hard way: HTTP 204 on every download
# the whole time, subtitle genuinely present in Jellyfin, Plex rescans forever
# finding nothing, because the file was living in here instead of on the NAS.
INTERNAL_SUB_PREFIX = "/var/lib/jellyfin/metadata"

def subtitle_state(user_id, item_id):
    """'external' (real sidecar file, visible to Plex etc), 'internal' (only in
    Jellyfin's own store — needs refetching once write access is fixed), or
    'none'."""
    item = _get(f"/Users/{user_id}/Items/{item_id}")
    subs = [s for s in item.get("MediaStreams", []) if s["Type"] == "Subtitle"]
    if not subs:
        return "none"
    if any(not (s.get("Path") or "").startswith(INTERNAL_SUB_PREFIX) for s in subs):
        return "external"
    return "internal"

def fetch_subtitle(item_id, lang):
    results = _get(f"/Items/{item_id}/RemoteSearch/Subtitles/{lang}")
    if not results:
        return None, "no results"
    hi = [r for r in results if r.get("HearingImpaired")]
    pick = hi[0] if hi else results[0]
    status = _post(f"/Items/{item_id}/RemoteSearch/Subtitles/{urllib.parse.quote(pick['Id'], safe='')}")
    return pick, status

def main():
    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("title", nargs="?", help="series title (fuzzy match via Jellyfin search)")
    ap.add_argument("--season", type=int, help="season number")
    ap.add_argument("--year", type=int, help="production year, to disambiguate same-titled series")
    ap.add_argument("--lang", default="eng", help="ISO 639-2 language code (default: eng)")
    ap.add_argument("--force", action="store_true", help="refetch even an existing external (real sidecar) subtitle")
    ap.add_argument("--list-seasons", action="store_true", help="just list seasons for the matched series and exit")
    ap.add_argument("--jellyfin-url", help="Jellyfin base URL (overrides JELLYFIN_URL env var and credentials.yaml)")
    ap.add_argument("--jellyfin-key", help="Jellyfin API key (overrides JELLYFIN_API_KEY env var and credentials.yaml)")
    args = ap.parse_args()

if not args.title:
        ap.error("title is required")

_configure(args.jellyfin_url, args.jellyfin_key)

series = find_series(args.title, args.year)
    print(f"series: {series['Name']} ({series.get('ProductionYear', '?')}) id={series['Id']}")

if args.list_seasons:
        for s in list_seasons(series["Id"]):
            print(f"  season {s.get('IndexNumber')}: {s['Name']} id={s['Id']}")
        return

if args.season is None:
        ap.error("--season is required (or use --list-seasons)")

season = find_season(series["Id"], args.season)
    episodes = list_episodes(series["Id"], season["Id"])
    print(f"season {args.season}: {len(episodes)} episodes")

user_id = _first_user_id()

fetched, skipped, failed, healed = 0, 0, 0, 0
    for ep in episodes:
        idx = ep.get("IndexNumber")
        name = ep.get("Name")
        item_id = ep["Id"]

state = subtitle_state(user_id, item_id)
        if state == "external" and not args.force:
            print(f"  ep{idx} '{name}': already has a real external sub, skipping")
            skipped += 1
            continue
        was_internal_only = state == "internal"

pick, status = fetch_subtitle(item_id, args.lang)
        if pick is None:
            print(f"  ep{idx} '{name}': {status}")
            failed += 1
            continue
        ok = status == 204
        if not ok:
            print(f"  ep{idx} '{name}': {pick['ProviderName']} '{pick['Name']}' -> HTTP {status} (FAILED)")
            failed += 1
            continue

# HTTP 204 alone isn't proof — Jellyfin can silently no-op a rate-limited
        # download and still return 204 (found 2026-09-01). Re-check the real
        # state, but the download+write is itself async on Jellyfin's side, so
        # a check made immediately after the POST returns can race ahead of the
        # write finishing (also found 2026-09-01 — every "failure" in a batch
        # turned out to be a real success once re-checked a few seconds later).
        # Poll briefly before calling it a failure.
        new_state = "none"
        for _ in range(5):
            new_state = subtitle_state(user_id, item_id)
            if new_state == "external":
                break
            time.sleep(1.5)
        if new_state == "external":
            tag = "healed (was internal-only)" if was_internal_only else "ok"
            print(f"  ep{idx} '{name}': {pick['ProviderName']} '{pick['Name']}' -> external sub confirmed ({tag})")
            fetched += 1
            if was_internal_only:
                healed += 1
        else:
            print(f"  ep{idx} '{name}': HTTP 204 but no external sub landed (state={new_state}) — likely a silent failure, not written to disk")
            failed += 1

print(f"\ndone — fetched {fetched} (of which {healed} healed from internal-only), skipped {skipped}, failed {failed}")

if __name__ == "__main__":
    main()

1	#!/usr/bin/env python3
2	"""Fetch missing subtitles for a Jellyfin series/season via Jellyfin's own
3	RemoteSearch/Subtitles API (backed by OpenSubtitles).
4
5	Built 2026-08-31 after doing this by hand for Mission: Impossible (1988) S1 —
6	generalized so any show/season can get the same treatment.
7
8	Usage:
9	python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1
10	python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1 --year 1988
11	python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1 --lang eng --force
12	python3 bin/jellyfin-fetch-subs.py --list-seasons "Mission: Impossible"
13
14	Series lookup goes through /Search/Hints (Type=Series) rather than /Items —
15	plain /Items search came back empty against this server for reasons not fully
16	chased down; Search/Hints works reliably. When a title matches more than one
17	series (e.g. two "Mission: Impossible" — 1966 and 1988), pass --year to
18	disambiguate, or the script lists the candidates and exits.
19
20	Credentials, in priority order: --jellyfin-url/--jellyfin-key flags, then
21	JELLYFIN_URL/JELLYFIN_API_KEY env vars, then secrets/credentials.yaml (Evie's
22	own default — needs the project .venv for PyYAML). The flag/env path needs
23	nothing beyond stdlib, so anyone with their own Jellyfin API key can run this
24	standalone with plain system python3:
25	JELLYFIN_URL=http://jellyfin.lan.1qaz.ca:8096 JELLYFIN_API_KEY=<key> \\
26	python3 bin/jellyfin-fetch-subs.py "Mission: Impossible" --season 1
27	"""
28	import argparse
29	import json
30	import os
31	import sys
32	import time
33	import urllib.parse
34	import urllib.request
35	from pathlib import Path
36
37	BASE = None
38	HEADERS = None
39
40
41	def _configure(url: str \| None, key: str \| None) -> None:
42	"""Resolve BASE/HEADERS from (in order) explicit args, env vars, or
43	secrets/credentials.yaml. The credentials.yaml path is imported lazily,
44	right here, so anyone supplying their own url/key never needs PyYAML or
45	the project .venv at all — a plain system python3 works."""
46	global BASE, HEADERS
47	url = url or os.environ.get("JELLYFIN_URL")
48	key = key or os.environ.get("JELLYFIN_API_KEY")
49	if not url or not key:
50	sys.path.insert(0, str(Path(__file__).parent))
51	from credentials import service # noqa: E402
52	jf = service("jellyfin")
53	url = url or jf["url"]
54	key = key or jf["api_key"]
55	BASE = url.rstrip("/")
56	HEADERS = {"X-Emby-Token": key}
57
58
59	def _get(path, params=None):
60	url = f"{BASE}{path}"
61	if params:
62	url += "?" + urllib.parse.urlencode(params)
63	req = urllib.request.Request(url, headers=HEADERS)
64	with urllib.request.urlopen(req) as resp:
65	body = resp.read()
66	return json.loads(body) if body else None
67
68
69	def _post(path):
70	req = urllib.request.Request(url=f"{BASE}{path}", headers=HEADERS, method="POST")
71	with urllib.request.urlopen(req) as resp:
72	return resp.status
73
74
75	def _first_user_id():
76	users = _get("/Users")
77	if not users:
78	raise RuntimeError("no Jellyfin users found — can't check subtitle state")
79	return users[0]["Id"]
80
81
82	def find_series(title, year=None):
83	hints = _get("/Search/Hints", {"SearchTerm": title, "IncludeItemTypes": "Series"})
84	candidates = [h for h in hints.get("SearchHints", []) if h.get("Type") == "Series"]
85	if year:
86	candidates = [c for c in candidates if c.get("ProductionYear") == year]
87	if len(candidates) == 1:
88	return candidates[0]
89	if not candidates:
90	raise SystemExit(f"no series found matching '{title}'" + (f" ({year})" if year else ""))
91	print(f"multiple series match '{title}' — pass --year to disambiguate:", file=sys.stderr)
92	for c in candidates:
93	print(f" {c.get('ProductionYear', '?')} — {c['Name']} (id={c['Id']})", file=sys.stderr)
94	raise SystemExit(2)
95
96
97	def list_seasons(series_id):
98	return _get(f"/Shows/{series_id}/Seasons")["Items"]
99
100
101	def find_season(series_id, season_num):
102	seasons = list_seasons(series_id)
103	for s in seasons:
104	if s.get("IndexNumber") == season_num:
105	return s
106	available = ", ".join(str(s.get("IndexNumber")) for s in seasons)
107	raise SystemExit(f"season {season_num} not found — available: {available}")
108
109
110	def list_episodes(series_id, season_id):
111	eps = _get(f"/Shows/{series_id}/Episodes", {"SeasonId": season_id})["Items"]
112	return sorted(eps, key=lambda e: e.get("IndexNumber") or 0)
113
114
115	# Jellyfin falls back to writing subtitles here when it can't write next to
116	# the media file (e.g. a permission problem on the share) — silently, no
117	# error, just a stream Plex (or anything else reading the raw files) will
118	# never see. Found 2026-08-31/09-01 the hard way: HTTP 204 on every download
119	# the whole time, subtitle genuinely present in Jellyfin, Plex rescans forever
120	# finding nothing, because the file was living in here instead of on the NAS.
121	INTERNAL_SUB_PREFIX = "/var/lib/jellyfin/metadata"
122
123
124	def subtitle_state(user_id, item_id):
125	"""'external' (real sidecar file, visible to Plex etc), 'internal' (only in
126	Jellyfin's own store — needs refetching once write access is fixed), or
127	'none'."""
128	item = _get(f"/Users/{user_id}/Items/{item_id}")
129	subs = [s for s in item.get("MediaStreams", []) if s["Type"] == "Subtitle"]
130	if not subs:
131	return "none"
132	if any(not (s.get("Path") or "").startswith(INTERNAL_SUB_PREFIX) for s in subs):
133	return "external"
134	return "internal"
135
136
137	def fetch_subtitle(item_id, lang):
138	results = _get(f"/Items/{item_id}/RemoteSearch/Subtitles/{lang}")
139	if not results:
140	return None, "no results"
141	hi = [r for r in results if r.get("HearingImpaired")]
142	pick = hi[0] if hi else results[0]
143	status = _post(f"/Items/{item_id}/RemoteSearch/Subtitles/{urllib.parse.quote(pick['Id'], safe='')}")
144	return pick, status
145
146
147	def main():
148	ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
149	ap.add_argument("title", nargs="?", help="series title (fuzzy match via Jellyfin search)")
150	ap.add_argument("--season", type=int, help="season number")
151	ap.add_argument("--year", type=int, help="production year, to disambiguate same-titled series")
152	ap.add_argument("--lang", default="eng", help="ISO 639-2 language code (default: eng)")
153	ap.add_argument("--force", action="store_true", help="refetch even an existing external (real sidecar) subtitle")
154	ap.add_argument("--list-seasons", action="store_true", help="just list seasons for the matched series and exit")
155	ap.add_argument("--jellyfin-url", help="Jellyfin base URL (overrides JELLYFIN_URL env var and credentials.yaml)")
156	ap.add_argument("--jellyfin-key", help="Jellyfin API key (overrides JELLYFIN_API_KEY env var and credentials.yaml)")
157	args = ap.parse_args()
158
159	if not args.title:
160	ap.error("title is required")
161
162	_configure(args.jellyfin_url, args.jellyfin_key)
163
164	series = find_series(args.title, args.year)
165	print(f"series: {series['Name']} ({series.get('ProductionYear', '?')}) id={series['Id']}")
166
167	if args.list_seasons:
168	for s in list_seasons(series["Id"]):
169	print(f" season {s.get('IndexNumber')}: {s['Name']} id={s['Id']}")
170	return
171
172	if args.season is None:
173	ap.error("--season is required (or use --list-seasons)")
174
175	season = find_season(series["Id"], args.season)
176	episodes = list_episodes(series["Id"], season["Id"])
177	print(f"season {args.season}: {len(episodes)} episodes")
178
179	user_id = _first_user_id()
180
181	fetched, skipped, failed, healed = 0, 0, 0, 0
182	for ep in episodes:
183	idx = ep.get("IndexNumber")
184	name = ep.get("Name")
185	item_id = ep["Id"]
186
187	state = subtitle_state(user_id, item_id)
188	if state == "external" and not args.force:
189	print(f" ep{idx} '{name}': already has a real external sub, skipping")
190	skipped += 1
191	continue
192	was_internal_only = state == "internal"
193
194	pick, status = fetch_subtitle(item_id, args.lang)
195	if pick is None:
196	print(f" ep{idx} '{name}': {status}")
197	failed += 1
198	continue
199	ok = status == 204
200	if not ok:
201	print(f" ep{idx} '{name}': {pick['ProviderName']} '{pick['Name']}' -> HTTP {status} (FAILED)")
202	failed += 1
203	continue
204
205	# HTTP 204 alone isn't proof — Jellyfin can silently no-op a rate-limited
206	# download and still return 204 (found 2026-09-01). Re-check the real
207	# state, but the download+write is itself async on Jellyfin's side, so
208	# a check made immediately after the POST returns can race ahead of the
209	# write finishing (also found 2026-09-01 — every "failure" in a batch
210	# turned out to be a real success once re-checked a few seconds later).
211	# Poll briefly before calling it a failure.
212	new_state = "none"
213	for _ in range(5):
214	new_state = subtitle_state(user_id, item_id)
215	if new_state == "external":
216	break
217	time.sleep(1.5)
218	if new_state == "external":
219	tag = "healed (was internal-only)" if was_internal_only else "ok"
220	print(f" ep{idx} '{name}': {pick['ProviderName']} '{pick['Name']}' -> external sub confirmed ({tag})")
221	fetched += 1
222	if was_internal_only:
223	healed += 1
224	else:
225	print(f" ep{idx} '{name}': HTTP 204 but no external sub landed (state={new_state}) — likely a silent failure, not written to disk")
226	failed += 1
227
228	print(f"\ndone — fetched {fetched} (of which {healed} healed from internal-only), skipped {skipped}, failed {failed}")
229
230
231	if __name__ == "__main__":
232	main()
233