#!/usr/bin/env python3 """ Marvel timeline synchronizer for Jellyfin + Seerr. Features -------- - Resolves TMDB IDs through the local Seerr API (no separate TMDB key required). - Finds movies and series in Jellyfin by TMDB provider ID. - Expands TV entries into individual Jellyfin episodes in season/episode order. - Requests missing movies or TV seasons through Seerr. - Creates or replaces chronologically ordered Jellyfin video playlists. - Uses only the Python standard library. Default profile: mcu-complete Timeline revision: 2026-08-03 """ from __future__ import annotations import argparse import difflib import json import os import re import ssl import sys import time import unicodedata from dataclasses import dataclass, field from pathlib import Path from typing import Any, Iterable, Optional from urllib.error import HTTPError, URLError from urllib.parse import quote, urlencode from urllib.request import Request, urlopen VERSION = "2026.08.03-r3" # --------------------------------------------------------------------------- # Timeline data # --------------------------------------------------------------------------- def movie(title: str, year: int, *, aliases: Iterable[str] = (), tier: str = "core") -> dict[str, Any]: return { "kind": "movie", "title": title, "year": year, "aliases": list(aliases), "tier": tier, } def tv( title: str, year: int, seasons: Iterable[int], *, aliases: Iterable[str] = (), tier: str = "core", ) -> dict[str, Any]: return { "kind": "tv", "title": title, "year": year, "seasons": list(seasons), "aliases": list(aliases), "tier": tier, } # Story-chronological MCU timeline. "complete" entries add the Defenders Saga, # Agent Carter and Marvel One-Shots to the Marvel Studios core timeline. MCU_TIMELINE: list[dict[str, Any]] = [ tv("Eyes of Wakanda", 2025, [1]), movie("Captain America: The First Avenger", 2011), movie("Marvel One-Shot: Agent Carter", 2013, aliases=["Agent Carter"], tier="complete"), tv("Marvel's Agent Carter", 2015, [1, 2], aliases=["Agent Carter"], tier="complete"), movie("Captain Marvel", 2019), movie("Iron Man", 2008), movie("Iron Man 2", 2010), movie("The Incredible Hulk", 2008), movie("A Funny Thing Happened on the Way to Thor's Hammer", 2011, tier="complete"), movie("Thor", 2011), movie("The Consultant", 2011, aliases=["Marvel One-Shot: The Consultant"], tier="complete"), movie("The Avengers", 2012, aliases=["Marvel's The Avengers", "Avengers" ]), movie("Item 47", 2012, aliases=["Marvel One-Shot: Item 47"], tier="complete"), movie("Thor: The Dark World", 2013), movie("Iron Man 3", 2013), movie("All Hail the King", 2014, aliases=["Marvel One-Shot: All Hail the King"], tier="complete"), movie("Captain America: The Winter Soldier", 2014), movie("Guardians of the Galaxy", 2014), movie("Guardians of the Galaxy Vol. 2", 2017), tv("I Am Groot", 2022, [1]), tv("I Am Groot", 2022, [2]), tv("Marvel's Daredevil", 2015, [1], aliases=["Daredevil"], tier="complete"), tv("Marvel's Jessica Jones", 2015, [1], aliases=["Jessica Jones"], tier="complete"), movie("Avengers: Age of Ultron", 2015), movie("Ant-Man", 2015), tv("Marvel's Daredevil", 2015, [2], aliases=["Daredevil"], tier="complete"), tv("Marvel's Luke Cage", 2016, [1], aliases=["Luke Cage"], tier="complete"), tv("Marvel's Iron Fist", 2017, [1], aliases=["Iron Fist"], tier="complete"), tv("Marvel's The Defenders", 2017, [1], aliases=["The Defenders"], tier="complete"), movie("Captain America: Civil War", 2016), movie("Black Widow", 2021), movie("Black Panther", 2018), movie("Spider-Man: Homecoming", 2017), tv("Marvel's The Punisher", 2017, [1], aliases=["The Punisher"], tier="complete"), movie("Doctor Strange", 2016), tv("Marvel's Jessica Jones", 2015, [2], aliases=["Jessica Jones"], tier="complete"), tv("Marvel's Luke Cage", 2016, [2], aliases=["Luke Cage"], tier="complete"), tv("Marvel's Iron Fist", 2017, [2], aliases=["Iron Fist"], tier="complete"), tv("Marvel's Daredevil", 2015, [3], aliases=["Daredevil"], tier="complete"), movie("Thor: Ragnarok", 2017), tv("Marvel's The Punisher", 2017, [2], aliases=["The Punisher"], tier="complete"), tv("Marvel's Jessica Jones", 2015, [3], aliases=["Jessica Jones"], tier="complete"), movie("Ant-Man and the Wasp", 2018), movie("Avengers: Infinity War", 2018), movie("Avengers: Endgame", 2019), tv("Loki", 2021, [1]), tv("What If...?", 2021, [1], aliases=["What If" ]), tv("Marvel Zombies", 2025, [1]), tv("WandaVision", 2021, [1]), movie("Shang-Chi and the Legend of the Ten Rings", 2021), tv("The Falcon and the Winter Soldier", 2021, [1]), movie("Spider-Man: Far From Home", 2019), movie("Eternals", 2021), movie("Spider-Man: No Way Home", 2021), movie("Doctor Strange in the Multiverse of Madness", 2022), tv("Hawkeye", 2021, [1]), tv("Moon Knight", 2022, [1]), movie("Black Panther: Wakanda Forever", 2022), tv("Echo", 2024, [1]), tv("She-Hulk: Attorney at Law", 2022, [1], aliases=["She-Hulk" ]), tv("Ms. Marvel", 2022, [1]), movie("Thor: Love and Thunder", 2022), tv("Ironheart", 2025, [1]), movie("Werewolf by Night", 2022), movie("The Guardians of the Galaxy Holiday Special", 2022, aliases=["Guardians of the Galaxy Holiday Special"]), movie("Ant-Man and the Wasp: Quantumania", 2023), movie("Guardians of the Galaxy Vol. 3", 2023), tv("Secret Invasion", 2023, [1]), movie("The Marvels", 2023), tv("Loki", 2021, [2]), tv("What If...?", 2021, [2], aliases=["What If" ]), movie("Deadpool & Wolverine", 2024, aliases=["Deadpool and Wolverine"]), tv("Agatha All Along", 2024, [1]), tv("What If...?", 2021, [3], aliases=["What If" ]), tv("Daredevil: Born Again", 2025, [1]), movie("Captain America: Brave New World", 2025), movie("Thunderbolts*", 2025, aliases=["Thunderbolts" ]), movie("The Fantastic Four: First Steps", 2025, aliases=["Fantastic Four: First Steps"]), tv("Daredevil: Born Again", 2025, [2]), movie( "The Punisher: One Last Kill", 2026, aliases=["Punisher: One Last Kill", "Marvel Studios' Special Presentation: The Punisher"], ), tv("Wonder Man", 2026, [1]), movie("Spider-Man: Brand New Day", 2026), ] # Separate multiverse/legacy preparation list. It is deliberately kept separate # because the Fox X-Men timelines branch and cannot be merged into one strictly # chronological MCU timeline without contradictions. LEGACY_TIMELINE: list[dict[str, Any]] = [ movie("X-Men: First Class", 2011), movie("X-Men: Days of Future Past", 2014), movie("X-Men Origins: Wolverine", 2009), movie("X-Men: Apocalypse", 2016), movie("Dark Phoenix", 2019, aliases=["X-Men: Dark Phoenix"]), movie("X-Men", 2000), movie("X2", 2003, aliases=["X2: X-Men United"]), movie("X-Men: The Last Stand", 2006), movie("The Wolverine", 2013), movie("Deadpool", 2016), movie("Deadpool 2", 2018), movie("The New Mutants", 2020), movie("Logan", 2017), movie("Spider-Man", 2002), movie("Spider-Man 2", 2004), movie("Spider-Man 3", 2007), movie("The Amazing Spider-Man", 2012), movie("The Amazing Spider-Man 2", 2014), movie("Fantastic Four", 2005), movie("Fantastic Four: Rise of the Silver Surfer", 2007), movie("Fantastic Four", 2015, aliases=["Fant4stic"]), ] @dataclass(frozen=True) class TimelineEntry: kind: str title: str year: int seasons: tuple[int, ...] = () aliases: tuple[str, ...] = () tier: str = "core" @classmethod def from_dict(cls, value: dict[str, Any]) -> "TimelineEntry": return cls( kind=str(value["kind"]), title=str(value["title"]), year=int(value["year"]), seasons=tuple(int(x) for x in value.get("seasons", [])), aliases=tuple(str(x) for x in value.get("aliases", [])), tier=str(value.get("tier", "core")), ) @property def label(self) -> str: if self.kind == "tv": seasons = ", ".join(f"S{x}" for x in self.seasons) return f"{self.title} ({self.year}) [{seasons}]" return f"{self.title} ({self.year})" @dataclass class ResolvedEntry: entry: TimelineEntry tmdb_id: Optional[int] = None jellyfin_ids: list[str] = field(default_factory=list) missing_seasons: list[int] = field(default_factory=list) local_episode_counts: dict[int, int] = field(default_factory=dict) expected_episode_counts: dict[int, int] = field(default_factory=dict) seerr_status: Optional[int] = None note: str = "" # --------------------------------------------------------------------------- # Utility functions # --------------------------------------------------------------------------- class ApiError(RuntimeError): def __init__(self, method: str, url: str, status: int, body: str): self.method = method self.url = url self.status = status self.body = body super().__init__(f"{method} {url} -> HTTP {status}: {body[:500]}") class HttpClient: def __init__( self, base_url: str, headers: Optional[dict[str, str]] = None, timeout: int = 30, verify_ssl: bool = True, ) -> None: self.base_url = base_url.rstrip("/") self.headers = headers or {} self.timeout = timeout self.ssl_context = None if verify_ssl else ssl._create_unverified_context() # noqa: SLF001 def request( self, method: str, path: str, *, params: Optional[dict[str, Any]] = None, json_body: Any = None, expected: Iterable[int] = (200, 201, 204), ) -> Any: url = f"{self.base_url}{path}" if params: clean_params: dict[str, Any] = {} for key, value in params.items(): if value is None: continue if isinstance(value, bool): clean_params[key] = "true" if value else "false" elif isinstance(value, (list, tuple)): clean_params[key] = ",".join(str(x) for x in value) else: clean_params[key] = value # Seerr rejects form-style "+" encoding for spaces. Use strict # RFC 3986 percent-encoding for every query-parameter value so # reserved characters such as :, &, ?, #, apostrophes, *, / and # non-ASCII characters are encoded consistently. query = urlencode(clean_params, quote_via=quote, safe="") if query: url = f"{url}?{query}" headers = {"Accept": "application/json", **self.headers} data = None if json_body is not None: data = json.dumps(json_body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" req = Request(url, data=data, headers=headers, method=method.upper()) try: with urlopen(req, timeout=self.timeout, context=self.ssl_context) as response: status = int(response.status) raw = response.read() except HTTPError as exc: raw = exc.read() body = raw.decode("utf-8", errors="replace") raise ApiError(method.upper(), url, int(exc.code), body) from exc except URLError as exc: raise RuntimeError(f"Verbindung fehlgeschlagen: {method.upper()} {url}: {exc}") from exc if status not in set(expected): body = raw.decode("utf-8", errors="replace") raise ApiError(method.upper(), url, status, body) if not raw: return None text = raw.decode("utf-8", errors="replace") try: return json.loads(text) except json.JSONDecodeError: return text def get(self, path: str, **kwargs: Any) -> Any: return self.request("GET", path, **kwargs) def post(self, path: str, **kwargs: Any) -> Any: return self.request("POST", path, **kwargs) def delete(self, path: str, **kwargs: Any) -> Any: return self.request("DELETE", path, **kwargs) def load_env(path: Path) -> None: if not path.exists(): return for raw_line in path.read_text(encoding="utf-8").splitlines(): line = raw_line.strip() if not line or line.startswith("#") or "=" not in line: continue key, value = line.split("=", 1) key = key.strip() value = value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in {'"', "'"}: value = value[1:-1] os.environ.setdefault(key, value) def env_bool(name: str, default: bool = True) -> bool: value = os.environ.get(name) if value is None: return default return value.strip().lower() not in {"0", "false", "no", "off"} def normalize_title(value: str) -> str: value = unicodedata.normalize("NFKD", value or "") value = "".join(ch for ch in value if not unicodedata.combining(ch)) value = value.casefold().replace("&", " and ") value = re.sub(r"\bmarvel(?:'s| studios)?\b", " ", value) value = re.sub(r"[^a-z0-9]+", " ", value) return " ".join(value.split()) def item_year(item: dict[str, Any]) -> Optional[int]: year = item.get("ProductionYear") if isinstance(year, int): return year date = item.get("PremiereDate") or item.get("ProductionYear") if isinstance(date, str) and len(date) >= 4 and date[:4].isdigit(): return int(date[:4]) return None def result_year(item: dict[str, Any]) -> Optional[int]: for key in ("releaseDate", "firstAirDate"): value = item.get(key) if isinstance(value, str) and len(value) >= 4 and value[:4].isdigit(): return int(value[:4]) return None def provider_tmdb_id(item: dict[str, Any]) -> Optional[int]: providers = item.get("ProviderIds") or {} for key, value in providers.items(): if str(key).casefold() == "tmdb" and str(value).isdigit(): return int(value) return None def compact_error(exc: Exception) -> str: if isinstance(exc, ApiError): body = exc.body.strip().replace("\n", " ") try: parsed = json.loads(exc.body) body = str(parsed.get("message") or parsed.get("title") or parsed) except (json.JSONDecodeError, AttributeError): pass return f"HTTP {exc.status}: {body[:240]}" return str(exc) def chunks(values: list[str], size: int) -> Iterable[list[str]]: for index in range(0, len(values), size): yield values[index : index + size] # --------------------------------------------------------------------------- # Seerr integration # --------------------------------------------------------------------------- class Seerr: STATUS = { 1: "unbekannt", 2: "angefragt", 3: "in Bearbeitung", 4: "teilweise vorhanden", 5: "vorhanden", 6: "gelöscht", } def __init__(self, client: HttpClient, cache_file: Path) -> None: self.client = client self.cache_file = cache_file self.cache: dict[str, int] = {} if cache_file.exists(): try: raw = json.loads(cache_file.read_text(encoding="utf-8")) self.cache = {str(k): int(v) for k, v in raw.items()} except (OSError, ValueError, TypeError, json.JSONDecodeError): self.cache = {} def save_cache(self) -> None: self.cache_file.parent.mkdir(parents=True, exist_ok=True) self.cache_file.write_text( json.dumps(self.cache, indent=2, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8", ) @staticmethod def cache_key(entry: TimelineEntry) -> str: return f"{entry.kind}:{entry.year}:{normalize_title(entry.title)}" def resolve_tmdb(self, entry: TimelineEntry) -> Optional[int]: key = self.cache_key(entry) if key in self.cache: return self.cache[key] queries = [entry.title, *entry.aliases] candidates: list[tuple[float, dict[str, Any]]] = [] wanted_names = {normalize_title(name) for name in queries} for query in queries: payload = self.client.get( "/api/v1/search", params={"query": query, "page": 1, "language": "en"}, ) for result in payload.get("results", []) if isinstance(payload, dict) else []: if result.get("mediaType") != entry.kind: continue names = [ result.get("title"), result.get("originalTitle"), result.get("name"), result.get("originalName"), ] normalized = [normalize_title(str(name)) for name in names if name] if not normalized: continue best_name_score = max( difflib.SequenceMatcher(None, wanted, candidate).ratio() for wanted in wanted_names for candidate in normalized ) year = result_year(result) year_delta = abs(year - entry.year) if year is not None else 9 year_score = 1.0 if year_delta == 0 else 0.75 if year_delta == 1 else 0.0 exact_bonus = 0.25 if any(name in wanted_names for name in normalized) else 0.0 score = best_name_score * 0.75 + year_score * 0.25 + exact_bonus candidates.append((score, result)) if candidates and max(score for score, _ in candidates) >= 1.10: break if not candidates: return None candidates.sort(key=lambda pair: pair[0], reverse=True) best_score, best = candidates[0] best_year = result_year(best) if best_score < 0.82: return None if best_year is not None and abs(best_year - entry.year) > 1: return None tmdb_id = best.get("id") if not isinstance(tmdb_id, int): return None self.cache[key] = tmdb_id return tmdb_id def details(self, entry: TimelineEntry, tmdb_id: int) -> dict[str, Any]: path = f"/api/v1/{'movie' if entry.kind == 'movie' else 'tv'}/{tmdb_id}" payload = self.client.get(path) return payload if isinstance(payload, dict) else {} @staticmethod def media_status(details: dict[str, Any]) -> Optional[int]: media_info = details.get("mediaInfo") or {} value = media_info.get("status") try: return int(value) except (TypeError, ValueError): return None @staticmethod def expected_season_counts(details: dict[str, Any], wanted: Iterable[int]) -> dict[int, int]: wanted_set = set(wanted) counts: dict[int, int] = {} for season in details.get("seasons", []) or []: try: number = int(season.get("seasonNumber")) count = int(season.get("episodeCount")) except (TypeError, ValueError, AttributeError): continue if number in wanted_set and count >= 0: counts[number] = count return counts @staticmethod def requested_seasons(details: dict[str, Any]) -> set[int]: media_info = details.get("mediaInfo") or {} requested: set[int] = set() for req in media_info.get("requests", []) or []: seasons = req.get("seasons") or [] for season in seasons: if isinstance(season, int): requested.add(season) continue if isinstance(season, dict): for key in ("seasonNumber", "number"): try: requested.add(int(season[key])) break except (KeyError, TypeError, ValueError): continue return requested def request_movie(self, tmdb_id: int) -> dict[str, Any]: payload = {"mediaType": "movie", "mediaId": tmdb_id, "is4k": False} result = self.client.post("/api/v1/request", json_body=payload) return result if isinstance(result, dict) else {} def request_tv(self, tmdb_id: int, seasons: list[int]) -> dict[str, Any]: payload = { "mediaType": "tv", "mediaId": tmdb_id, "seasons": seasons, "is4k": False, } result = self.client.post("/api/v1/request", json_body=payload) return result if isinstance(result, dict) else {} # --------------------------------------------------------------------------- # Jellyfin integration # --------------------------------------------------------------------------- class Jellyfin: def __init__(self, client: HttpClient, username: Optional[str], user_id: Optional[str]) -> None: self.client = client self.username = username self.user_id = user_id def resolve_user(self) -> str: if self.user_id: return self.user_id users = self.client.get("/Users") if not isinstance(users, list): raise RuntimeError("Jellyfin /Users hat keine Benutzerliste zurückgegeben.") if self.username: for user in users: if str(user.get("Name", "")).casefold() == self.username.casefold(): self.user_id = str(user["Id"]) return self.user_id available = ", ".join(str(user.get("Name")) for user in users) raise RuntimeError(f"Jellyfin-Benutzer '{self.username}' nicht gefunden. Vorhanden: {available}") if len(users) == 1: self.user_id = str(users[0]["Id"]) return self.user_id available = ", ".join(str(user.get("Name")) for user in users) raise RuntimeError( "Mehrere Jellyfin-Benutzer gefunden. Setze JELLYFIN_USERNAME oder " f"JELLYFIN_USER_ID. Vorhanden: {available}" ) def paged_user_items(self, **params: Any) -> list[dict[str, Any]]: user_id = self.resolve_user() result: list[dict[str, Any]] = [] start = 0 limit = 500 while True: query = {**params, "StartIndex": start, "Limit": limit} payload = self.client.get(f"/Users/{user_id}/Items", params=query) if not isinstance(payload, dict): break items = payload.get("Items", []) or [] result.extend(items) total = int(payload.get("TotalRecordCount", len(result))) start += len(items) if not items or start >= total: break return result def movies(self) -> list[dict[str, Any]]: return self.paged_user_items( Recursive=True, IncludeItemTypes="Movie", Fields="ProviderIds,ProductionYear,PremiereDate,OriginalTitle,Path", EnableImages=False, ) def series(self) -> list[dict[str, Any]]: return self.paged_user_items( Recursive=True, IncludeItemTypes="Series", Fields="ProviderIds,ProductionYear,PremiereDate,OriginalTitle,Path", EnableImages=False, ) def episodes(self, series_id: str, seasons: Iterable[int]) -> list[dict[str, Any]]: wanted = set(seasons) items = self.paged_user_items( ParentId=series_id, Recursive=True, IncludeItemTypes="Episode", Fields="ProviderIds,ParentIndexNumber,IndexNumber,PremiereDate,Path,SeriesName", EnableImages=False, ) filtered = [] for item in items: try: season = int(item.get("ParentIndexNumber")) except (TypeError, ValueError): continue if season in wanted: filtered.append(item) filtered.sort( key=lambda item: ( int(item.get("ParentIndexNumber") or 0), int(item.get("IndexNumber") or 0), str(item.get("SortName") or item.get("Name") or ""), ) ) return filtered @staticmethod def match_item( items: list[dict[str, Any]], entry: TimelineEntry, tmdb_id: Optional[int], ) -> Optional[dict[str, Any]]: if tmdb_id is not None: for item in items: if provider_tmdb_id(item) == tmdb_id: return item wanted = {normalize_title(entry.title), *(normalize_title(x) for x in entry.aliases)} candidates: list[tuple[float, dict[str, Any]]] = [] for item in items: names = [item.get("Name"), item.get("OriginalTitle")] names_normalized = [normalize_title(str(x)) for x in names if x] if not names_normalized: continue title_score = max( difflib.SequenceMatcher(None, target, candidate).ratio() for target in wanted for candidate in names_normalized ) year = item_year(item) year_score = 1.0 if year == entry.year else 0.7 if year and abs(year - entry.year) == 1 else 0.0 score = title_score * 0.8 + year_score * 0.2 if score >= 0.82: candidates.append((score, item)) if not candidates: return None candidates.sort(key=lambda pair: pair[0], reverse=True) return candidates[0][1] def find_playlist(self, name: str) -> Optional[dict[str, Any]]: playlists = self.paged_user_items( Recursive=True, IncludeItemTypes="Playlist", Fields="Path", EnableImages=False, ) for playlist in playlists: if str(playlist.get("Name", "")).casefold() == name.casefold(): return playlist return None def playlist_items(self, playlist_id: str) -> list[dict[str, Any]]: payload = self.client.get( f"/Playlists/{playlist_id}/Items", params={"UserId": self.resolve_user(), "Fields": "ProviderIds"}, ) if not isinstance(payload, dict): return [] return list(payload.get("Items", []) or []) def create_playlist(self, name: str, item_ids: list[str]) -> dict[str, Any]: body = { "Name": name, "Ids": item_ids, "UserId": self.resolve_user(), "MediaType": "Video", "IsPublic": False, } result = self.client.post("/Playlists", json_body=body) return result if isinstance(result, dict) else {} def clear_playlist(self, playlist_id: str) -> None: existing = self.playlist_items(playlist_id) if not existing: return entry_ids = [ str(item.get("PlaylistItemId") or item.get("Id")) for item in existing if item.get("PlaylistItemId") or item.get("Id") ] for group in chunks(entry_ids, 100): self.client.delete( f"/Playlists/{playlist_id}/Items", params={"EntryIds": group}, ) # Some older Jellyfin versions returned 204 without deleting. Verify. if self.playlist_items(playlist_id): raise RuntimeError("Jellyfin meldet Erfolg, aber die Playlist ist nach dem Leeren nicht leer.") def add_playlist_items(self, playlist_id: str, item_ids: list[str]) -> None: for group in chunks(item_ids, 100): self.client.post( f"/Playlists/{playlist_id}/Items", params={"Ids": group, "UserId": self.resolve_user()}, ) def delete_playlist(self, playlist_id: str) -> None: self.client.delete(f"/Items/{playlist_id}") def sync_playlist(self, name: str, item_ids: list[str]) -> str: if not item_ids: raise RuntimeError("Keine verfügbaren Jellyfin-Medien für die Playlist gefunden.") existing = self.find_playlist(name) if existing: playlist_id = str(existing["Id"]) old_ids = [str(item.get("Id")) for item in self.playlist_items(playlist_id)] if old_ids == item_ids: return "unverändert" try: self.clear_playlist(playlist_id) self.add_playlist_items(playlist_id, item_ids) return "aktualisiert" except Exception as clear_error: # Fallback for known playlist-removal quirks in some Jellyfin releases. try: self.delete_playlist(playlist_id) self.create_playlist(name, item_ids) return f"neu erstellt (Fallback nach: {compact_error(clear_error)})" except Exception as recreate_error: raise RuntimeError( "Playlist konnte weder aktualisiert noch neu erstellt werden. " f"Update: {compact_error(clear_error)}; Neu: {compact_error(recreate_error)}" ) from recreate_error self.create_playlist(name, item_ids) return "erstellt" # --------------------------------------------------------------------------- # Orchestration # --------------------------------------------------------------------------- @dataclass class Profile: key: str playlist_name: str entries: list[TimelineEntry] def build_profiles(profile_name: str) -> list[Profile]: complete = [TimelineEntry.from_dict(x) for x in MCU_TIMELINE] core = [entry for entry in complete if entry.tier == "core"] legacy = [TimelineEntry.from_dict(x) for x in LEGACY_TIMELINE] mcu_complete_name = os.environ.get( "JELLYFIN_PLAYLIST_MCU_COMPLETE", "Marvel – MCU komplett chronologisch bis Doomsday", ) mcu_core_name = os.environ.get( "JELLYFIN_PLAYLIST_MCU_CORE", "Marvel – MCU Core chronologisch bis Doomsday", ) legacy_name = os.environ.get( "JELLYFIN_PLAYLIST_LEGACY", "Marvel – Doomsday Legacy Story Order", ) if profile_name == "mcu-complete": return [Profile("mcu-complete", mcu_complete_name, complete)] if profile_name == "mcu-core": return [Profile("mcu-core", mcu_core_name, core)] if profile_name == "legacy": return [Profile("legacy", legacy_name, legacy)] if profile_name == "all": return [ Profile("mcu-complete", mcu_complete_name, complete), Profile("legacy", legacy_name, legacy), ] raise ValueError(f"Unbekanntes Profil: {profile_name}") def inspect_profile( profile: Profile, jellyfin: Jellyfin, seerr: Seerr, movies: list[dict[str, Any]], series: list[dict[str, Any]], delay: float, ) -> list[ResolvedEntry]: resolved: list[ResolvedEntry] = [] for index, entry in enumerate(profile.entries, start=1): print(f"[{index:02d}/{len(profile.entries):02d}] Prüfe {entry.label}") result = ResolvedEntry(entry=entry) try: result.tmdb_id = seerr.resolve_tmdb(entry) except Exception as exc: result.note = f"Seerr-Suche fehlgeschlagen: {compact_error(exc)}" resolved.append(result) print(f" ! {result.note}") continue if result.tmdb_id is None: result.note = "Kein eindeutiger TMDB-Treffer über Seerr" resolved.append(result) print(f" ! {result.note}") continue try: details = seerr.details(entry, result.tmdb_id) result.seerr_status = seerr.media_status(details) except Exception as exc: details = {} result.note = f"Seerr-Details fehlgeschlagen: {compact_error(exc)}" if entry.kind == "movie": local = jellyfin.match_item(movies, entry, result.tmdb_id) if local: result.jellyfin_ids = [str(local["Id"])] print(f" + Jellyfin: vorhanden | TMDB {result.tmdb_id}") else: status_text = Seerr.STATUS.get(result.seerr_status or 0, "nicht vorhanden") print(f" - Jellyfin: fehlt | Seerr: {status_text} | TMDB {result.tmdb_id}") else: local_series = jellyfin.match_item(series, entry, result.tmdb_id) result.expected_episode_counts = seerr.expected_season_counts(details, entry.seasons) if local_series: episodes = jellyfin.episodes(str(local_series["Id"]), entry.seasons) counts: dict[int, int] = {season: 0 for season in entry.seasons} for episode in episodes: season = int(episode.get("ParentIndexNumber") or 0) counts[season] = counts.get(season, 0) + 1 result.local_episode_counts = counts result.jellyfin_ids = [str(episode["Id"]) for episode in episodes] for season in entry.seasons: local_count = counts.get(season, 0) expected = result.expected_episode_counts.get(season) if local_count == 0 or (expected is not None and local_count < expected): result.missing_seasons.append(season) count_text = ", ".join( f"S{season}: {counts.get(season, 0)}/{result.expected_episode_counts.get(season, '?')}" for season in entry.seasons ) if result.missing_seasons: print(f" ~ Jellyfin: unvollständig ({count_text}) | TMDB {result.tmdb_id}") else: print(f" + Jellyfin: vorhanden ({count_text}) | TMDB {result.tmdb_id}") else: result.missing_seasons = list(entry.seasons) status_text = Seerr.STATUS.get(result.seerr_status or 0, "nicht vorhanden") print( f" - Jellyfin: Serie fehlt (S{','.join(map(str, entry.seasons))}) " f"| Seerr: {status_text} | TMDB {result.tmdb_id}" ) resolved.append(result) if delay > 0: time.sleep(delay) seerr.save_cache() return resolved def request_missing(resolved: list[ResolvedEntry], seerr: Seerr, dry_run: bool) -> tuple[int, int, int]: requested = 0 skipped = 0 failed = 0 print("\n=== Seerr-Requests ===") for result in resolved: entry = result.entry if result.tmdb_id is None: print(f"[FEHLER] {entry.label}: keine TMDB-ID") failed += 1 continue if entry.kind == "movie": if result.jellyfin_ids: continue if result.seerr_status in {2, 3, 5}: print( f"[SKIP] {entry.label}: Seerr-Status " f"{Seerr.STATUS.get(result.seerr_status, result.seerr_status)}" ) skipped += 1 continue if dry_run: print(f"[DRY] Film anfragen: {entry.label} (TMDB {result.tmdb_id})") requested += 1 continue try: seerr.request_movie(result.tmdb_id) print(f"[REQUEST] {entry.label}") requested += 1 except Exception as exc: message = compact_error(exc) if isinstance(exc, ApiError) and exc.status in {400, 409}: print(f"[SKIP] {entry.label}: {message}") skipped += 1 else: print(f"[FEHLER] {entry.label}: {message}") failed += 1 continue missing = list(result.missing_seasons) if not missing: continue # Avoid duplicate season requests where Seerr exposes them in mediaInfo. try: details = seerr.details(entry, result.tmdb_id) already = seerr.requested_seasons(details) status = seerr.media_status(details) except Exception: already = set() status = result.seerr_status if status in {2, 3} and not already: print( f"[SKIP] {entry.label}: Serie bereits " f"{Seerr.STATUS.get(status, status)}" ) skipped += 1 continue wanted = [season for season in missing if season not in already] if not wanted: print(f"[SKIP] {entry.title}: fehlende Staffeln bereits angefragt") skipped += 1 continue if dry_run: print(f"[DRY] Serie anfragen: {entry.title}, Staffeln {wanted} (TMDB {result.tmdb_id})") requested += 1 continue try: seerr.request_tv(result.tmdb_id, wanted) print(f"[REQUEST] {entry.title}: Staffeln {wanted}") requested += 1 except Exception as exc: message = compact_error(exc) if isinstance(exc, ApiError) and exc.status in {400, 409}: print(f"[SKIP] {entry.title}: {message}") skipped += 1 else: print(f"[FEHLER] {entry.title}: {message}") failed += 1 return requested, skipped, failed def sync_playlist(profile: Profile, resolved: list[ResolvedEntry], jellyfin: Jellyfin, dry_run: bool) -> tuple[int, int]: item_ids: list[str] = [] missing_entries = 0 for result in resolved: item_ids.extend(result.jellyfin_ids) if result.entry.kind == "movie" and not result.jellyfin_ids: missing_entries += 1 elif result.entry.kind == "tv" and result.missing_seasons: missing_entries += 1 # Preserve order but remove accidental duplicate IDs. seen: set[str] = set() ordered_ids: list[str] = [] for item_id in item_ids: if item_id not in seen: seen.add(item_id) ordered_ids.append(item_id) print("\n=== Jellyfin-Playlist ===") print(f"Name: {profile.playlist_name}") print(f"Einträge: {len(ordered_ids)} Jellyfin-Videos") print(f"Unvollst.: {missing_entries} Timeline-Blöcke") if dry_run: print("[DRY] Playlist wird nicht verändert.") return len(ordered_ids), missing_entries try: state = jellyfin.sync_playlist(profile.playlist_name, ordered_ids) print(f"[OK] Playlist {state}.") except Exception as exc: message = compact_error(exc) if "parentFolder" in message: message += ( " | Bekannter Jellyfin-Fehler: Prüfe, ob im Jellyfin-Datenverzeichnis " "der Ordner 'Playlists' existiert und dem Jellyfin-Benutzer gehört." ) raise RuntimeError(message) from exc return len(ordered_ids), missing_entries def print_profile(profile: Profile) -> None: print(f"\n# {profile.playlist_name} ({len(profile.entries)} Blöcke)") for index, entry in enumerate(profile.entries, start=1): print(f"{index:02d}. {entry.label}") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Marvel-Timeline in Jellyfin synchronisieren und fehlende Medien über Seerr anfragen.", formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) parser.add_argument("--version", action="version", version=f"%(prog)s {VERSION}") parser.add_argument( "--env-file", default="marvel.env", help="Konfigurationsdatei im KEY=VALUE-Format", ) parser.add_argument( "--profile", choices=["mcu-complete", "mcu-core", "legacy", "all"], default="mcu-complete", help="Welche Playlist(s) verarbeitet werden", ) parser.add_argument("--list", action="store_true", help="Timeline nur ausgeben") parser.add_argument("--dry-run", action="store_true", help="Nur prüfen; nichts anfragen oder ändern") parser.add_argument("--request-missing", action="store_true", help="Fehlende Medien über Seerr anfragen") parser.add_argument("--sync-playlist", action="store_true", help="Jellyfin-Playlist erstellen/aktualisieren") parser.add_argument( "--cache-file", default=".cache/marvel-playlist/tmdb.json", help="Lokaler Cache für über Seerr aufgelöste TMDB-IDs", ) parser.add_argument( "--search-delay", type=float, default=0.05, help="Pause zwischen Timeline-Einträgen in Sekunden", ) return parser.parse_args() def required_env(name: str) -> str: value = os.environ.get(name, "").strip() if not value: raise RuntimeError(f"Pflichtwert fehlt: {name}") return value def main() -> int: args = parse_args() load_env(Path(args.env_file).expanduser()) profiles = build_profiles(args.profile) if args.list: for profile in profiles: print_profile(profile) return 0 # With no write action, behave as a useful status-only dry run. dry_run = args.dry_run or not (args.request_missing or args.sync_playlist) try: jellyfin_url = required_env("JELLYFIN_URL") jellyfin_api_key = required_env("JELLYFIN_API_KEY") seerr_url = required_env("SEERR_URL") seerr_api_key = required_env("SEERR_API_KEY") except RuntimeError as exc: print(f"Konfigurationsfehler: {exc}", file=sys.stderr) print(f"Lege '{args.env_file}' anhand von marvel.env.example an.", file=sys.stderr) return 2 timeout = int(os.environ.get("HTTP_TIMEOUT", "30")) verify_ssl = env_bool("VERIFY_SSL", True) jellyfin_client = HttpClient( jellyfin_url, headers={"X-Emby-Token": jellyfin_api_key}, timeout=timeout, verify_ssl=verify_ssl, ) seerr_client = HttpClient( seerr_url, headers={"X-Api-Key": seerr_api_key}, timeout=timeout, verify_ssl=verify_ssl, ) jellyfin = Jellyfin( jellyfin_client, username=os.environ.get("JELLYFIN_USERNAME"), user_id=os.environ.get("JELLYFIN_USER_ID"), ) seerr = Seerr(seerr_client, Path(args.cache_file).expanduser()) try: user_id = jellyfin.resolve_user() jellyfin_client.get("/System/Info/Public") seerr_client.get("/api/v1/status") except Exception as exc: print(f"Verbindungstest fehlgeschlagen: {compact_error(exc)}", file=sys.stderr) return 3 print(f"Marvel Playlist Sync {VERSION}") print(f"Jellyfin-Benutzer-ID: {user_id}") print(f"Modus: {'DRY-RUN' if dry_run else 'SCHREIBEND'}") try: movies = jellyfin.movies() series = jellyfin.series() except Exception as exc: print(f"Jellyfin-Bibliothek konnte nicht gelesen werden: {compact_error(exc)}", file=sys.stderr) return 4 print(f"Jellyfin-Inventar: {len(movies)} Filme, {len(series)} Serien") total_requests = total_skips = total_failures = 0 total_playlist_items = total_missing_blocks = 0 for profile in profiles: print(f"\n{'=' * 72}\nProfil: {profile.playlist_name}\n{'=' * 72}") resolved = inspect_profile(profile, jellyfin, seerr, movies, series, args.search_delay) if args.request_missing or dry_run: req, skip, fail = request_missing(resolved, seerr, dry_run=dry_run) total_requests += req total_skips += skip total_failures += fail if args.sync_playlist or dry_run: try: count, missing = sync_playlist(profile, resolved, jellyfin, dry_run=dry_run) total_playlist_items += count total_missing_blocks += missing except Exception as exc: print(f"Playlist-Fehler: {compact_error(exc)}", file=sys.stderr) total_failures += 1 print("\n=== Zusammenfassung ===") print(f"Playlist-Einträge: {total_playlist_items}") print(f"Unvollständige Blöcke: {total_missing_blocks}") print(f"Requests/DRY-Aktionen: {total_requests}") print(f"Übersprungen: {total_skips}") print(f"Fehler: {total_failures}") return 1 if total_failures else 0 if __name__ == "__main__": raise SystemExit(main())