from __future__ import annotations import hashlib import html import json import os import re import socket import urllib.parse import urllib.parse import urllib.error import urllib.request from datetime import datetime, timezone from typing import Any from .core import ( BotRuntime, DEFAULT_JELLYFIN_SYNC_INTERVAL_SECONDS, MAX_DISCORD_EMBEDS, MEDIA_ITEMS_PER_EMBED, MediaItem, clean_error, ) from .discord_api import discord_delete_message, discord_multipart_request from .storage import ( channel_settings, jellyfin_settings, load_state, save_catalog_url_setting, save_media_channel_setting, save_state, validate_channel_id, ) TMDB_IMAGE_BASE_URL = "https://image.tmdb.org/t/p/w342" TMDB_POSTER_CACHE: dict[str, str | None] = {} def parse_int_text(value: str) -> int | None: digits = "".join(character for character in value if character.isdigit()) if not digits: return None try: return int(digits) except ValueError: return None def parse_year_text(value: str) -> str | None: match = re.search(r"(18|19|20|21)\d{2}", value) return match.group(0) if match else None def clean_media_text(value: str, limit: int = 220) -> str | None: cleaned = " ".join(str(value).replace("\r", " ").replace("\n", " ").split()) if not cleaned: return None if len(cleaned) <= limit: return cleaned return cleaned[: limit - 1].rstrip() + "…" def format_runtime(value: str) -> str | None: cleaned = clean_media_text(value, 48) if not cleaned: return None if not cleaned.isdigit(): return cleaned minutes = int(cleaned) if minutes <= 0: return None hours, remainder = divmod(minutes, 60) if hours and remainder: return f"{hours}h {remainder}m" if hours: return f"{hours}h" return f"{remainder}m" def media_item_to_jsonable(item: MediaItem) -> dict[str, Any]: data: dict[str, Any] = { "title": item.title, "mediaType": item.media_type, "sourceId": item.source_id or "", "tmdbId": item.tmdb_id or "", "imdbId": item.imdb_id or "", "year": item.year or "", "genres": item.genres or "", "rating": item.rating or "", "runtime": item.runtime or "", "summary": item.summary or "", } if item.media_type == "show": data["seasons"] = item.seasons if item.seasons is not None else "" data["episodes"] = item.episodes if item.episodes is not None else "" return data def media_item_from_data(data: Any, media_type: str) -> MediaItem: if not isinstance(data, dict): raise ValueError(f"{media_type.title()} entries must be objects") title = clean_media_text(str(data.get("title", "")), 120) if not title: raise ValueError(f"{media_type.title()} entries must include a title") year = parse_year_text(str(data.get("year", ""))) seasons = parse_int_text(str(data.get("seasons", ""))) if media_type == "show" else None episodes = parse_int_text(str(data.get("episodes", ""))) if media_type == "show" else None return MediaItem( title=title, media_type=media_type, source_id=clean_media_text(str(data.get("sourceId", "")), 120), tmdb_id=clean_media_text(str(data.get("tmdbId", "")), 120), imdb_id=clean_media_text(str(data.get("imdbId", "")), 120), year=year, genres=clean_media_text(str(data.get("genres", "")), 120), rating=clean_media_text(str(data.get("rating", "")), 32), runtime=format_runtime(str(data.get("runtime", ""))) if media_type == "movie" else None, summary=clean_media_text(str(data.get("summary", ""))), seasons=seasons, episodes=episodes, ) def media_items_from_data(raw_items: Any, media_type: str) -> list[MediaItem]: if raw_items is None: return [] if not isinstance(raw_items, list): raise ValueError(f"{media_type.title()} library must be a list") deduped: dict[tuple[str, str], MediaItem] = {} for raw_item in raw_items: item = media_item_from_data(raw_item, media_type) deduped[(item.title.casefold(), item.year or "")] = item return sorted(deduped.values(), key=lambda item: (item.title.casefold(), item.year or "")) def media_library_to_jsonable(movies: list[MediaItem], shows: list[MediaItem]) -> dict[str, Any]: return { "movies": [media_item_to_jsonable(item) for item in movies], "shows": [media_item_to_jsonable(item) for item in shows], } def load_media_library(runtime: BotRuntime) -> tuple[list[MediaItem], list[MediaItem]]: data = load_state(runtime.media_library_path) return ( media_items_from_data(data.get("movies", []), "movie"), media_items_from_data(data.get("shows", []), "show"), ) def save_media_library(runtime: BotRuntime, movies: list[MediaItem], shows: list[MediaItem]) -> dict[str, Any]: payload = media_library_to_jsonable(movies, shows) payload["updated_at"] = datetime.now(timezone.utc).isoformat() save_state(runtime.media_library_path, payload) return payload def reset_media_library(runtime: BotRuntime) -> dict[str, Any]: save_media_library(runtime, [], []) state = load_state(runtime.settings_path) for key in ( "jellyfin_last_sync_at", "jellyfin_last_sync_error", "jellyfin_last_published_at", "jellyfin_last_fingerprint", "jellyfin_last_published_fingerprint", "jellyfin_last_changes", ): state.pop(key, None) state["updated_at"] = datetime.now(timezone.utc).isoformat() save_state(runtime.settings_path, state) return { "library": media_library_to_jsonable([], []), "movieCount": 0, "showCount": 0, "changes": {"added": [], "removed": [], "addedCount": 0, "removedCount": 0}, "jellyfin": jellyfin_settings(runtime), } def jellyfin_runtime(runtime_ticks: Any) -> str | None: try: ticks = int(runtime_ticks or 0) except (TypeError, ValueError): return None if ticks <= 0: return None minutes = round(ticks / 10_000_000 / 60) if minutes <= 0: return None hours, remainder = divmod(minutes, 60) if hours and remainder: return f"{hours}h {remainder}m" if hours: return f"{hours}h" return f"{remainder}m" def jellyfin_item_year(item: dict[str, Any]) -> str | None: production_year = item.get("ProductionYear") if production_year: return parse_year_text(str(production_year)) return parse_year_text(str(item.get("PremiereDate", ""))) def jellyfin_item_summary(item: dict[str, Any]) -> str | None: return clean_media_text(str(item.get("Overview", "") or item.get("ShortOverview", ""))) def jellyfin_movie_from_item(item: dict[str, Any]) -> MediaItem: provider_ids = item.get("ProviderIds") if isinstance(item.get("ProviderIds"), dict) else {} return MediaItem( title=clean_media_text(str(item.get("Name", "")), 120) or "Untitled Movie", media_type="movie", source_id=clean_media_text(str(item.get("Id", "")), 120), tmdb_id=clean_media_text(str(provider_ids.get("Tmdb", "")), 120), imdb_id=clean_media_text(str(provider_ids.get("Imdb", "")), 120), year=jellyfin_item_year(item), genres=clean_media_text(", ".join(str(genre) for genre in item.get("Genres", []) if genre), 120), rating=clean_media_text(str(item.get("OfficialRating", "")), 32), runtime=jellyfin_runtime(item.get("RunTimeTicks")), summary=jellyfin_item_summary(item), ) def jellyfin_show_from_item(item: dict[str, Any]) -> MediaItem: provider_ids = item.get("ProviderIds") if isinstance(item.get("ProviderIds"), dict) else {} return MediaItem( title=clean_media_text(str(item.get("Name", "")), 120) or "Untitled Show", media_type="show", source_id=clean_media_text(str(item.get("Id", "")), 120), tmdb_id=clean_media_text(str(provider_ids.get("Tmdb", "")), 120), imdb_id=clean_media_text(str(provider_ids.get("Imdb", "")), 120), year=jellyfin_item_year(item), genres=clean_media_text(", ".join(str(genre) for genre in item.get("Genres", []) if genre), 120), rating=clean_media_text(str(item.get("OfficialRating", "")), 32), summary=jellyfin_item_summary(item), seasons=parse_int_text(str(item.get("ChildCount", ""))), episodes=parse_int_text(str(item.get("RecursiveItemCount", ""))), ) def jellyfin_request(settings: dict[str, Any], path: str, params: dict[str, Any]) -> dict[str, Any]: base_url = str(settings.get("url", "")).strip().rstrip("/") api_key = str(settings.get("apiKey", "")).strip() if not base_url or not api_key: raise ValueError("Jellyfin URL and API key are required") query = urllib.parse.urlencode({key: value for key, value in params.items() if value is not None}) url = f"{base_url}{path}" if query: url = f"{url}?{query}" request = urllib.request.Request( url, headers={ "Accept": "application/json", "User-Agent": "ArchiveStatusBot/1.0", "X-Emby-Token": api_key, }, method="GET", ) try: with urllib.request.urlopen(request, timeout=30) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore") raise RuntimeError(f"Jellyfin API failed: HTTP {exc.code} {detail}") from exc except (urllib.error.URLError, TimeoutError, socket.timeout) as exc: raise RuntimeError(f"Jellyfin API failed: {clean_error(exc)}") from exc def jellyfin_connection_settings(runtime: BotRuntime) -> dict[str, Any]: state = load_state(runtime.settings_path) return { "url": str(state.get("jellyfin_url", "")).strip(), "apiKey": str(state.get("jellyfin_api_key", "")).strip(), } def media_item_duration_seconds(item: MediaItem) -> int: runtime = str(item.runtime or "").strip().lower() if not runtime: return 0 hours = 0 minutes = 0 hour_match = re.search(r"(\d+)\s*h", runtime) minute_match = re.search(r"(\d+)\s*m", runtime) if hour_match: hours = int(hour_match.group(1)) if minute_match: minutes = int(minute_match.group(1)) elif runtime.isdigit(): minutes = int(runtime) return max(0, hours * 3600 + minutes * 60) def jellyfin_stream_url(base_url: str, item_id: str, api_key: str, subtitle_index: int | None = None) -> str: params = {"api_key": api_key} if subtitle_index is not None: params["SubtitleStreamIndex"] = str(subtitle_index) else: params["static"] = "true" query = urllib.parse.urlencode(params) return f"{base_url}/Videos/{urllib.parse.quote(item_id, safe='')}/stream?{query}" def jellyfin_download_url(base_url: str, item_id: str, api_key: str) -> str: query = urllib.parse.urlencode({"api_key": api_key}) return f"{base_url}/Items/{urllib.parse.quote(item_id, safe='')}/Download?{query}" def jellyfin_first_episode(settings: dict[str, Any], series_id: str) -> dict[str, Any] | None: data = jellyfin_request( settings, "/Items", { "ParentId": series_id, "Recursive": "true", "IncludeItemTypes": "Episode", "Fields": "RunTimeTicks,ParentIndexNumber,IndexNumber", "SortBy": "ParentIndexNumber,IndexNumber,SortName", "SortOrder": "Ascending", "Limit": 1, }, ) items = data.get("Items", []) if not isinstance(items, list): raise RuntimeError("Jellyfin returned an invalid episode payload") for item in items: if isinstance(item, dict) and str(item.get("Id", "")).strip(): return item return None def resolve_jellyfin_playback_source(runtime: BotRuntime, item: MediaItem, subtitle_index: int | None = None) -> dict[str, Any]: settings = jellyfin_connection_settings(runtime) base_url = str(settings.get("url", "")).strip().rstrip("/") api_key = str(settings.get("apiKey", "")).strip() if not base_url or not api_key: raise ValueError("Jellyfin playback is not configured") if not item.source_id: raise ValueError(f"Media item has no Jellyfin source ID: {item.title}") playback_item_id = item.source_id playback_title = item.title duration_seconds = media_item_duration_seconds(item) episode_meta: dict[str, Any] | None = None if item.media_type == "show": episode = jellyfin_first_episode(settings, item.source_id) if episode is None: raise ValueError(f"No playable episodes found for show: {item.title}") playback_item_id = clean_media_text(str(episode.get("Id", "")), 120) or item.source_id season_number = int(episode.get("ParentIndexNumber") or 0) episode_number = int(episode.get("IndexNumber") or 0) episode_title = clean_media_text(str(episode.get("Name", "")), 120) or "Episode" playback_title = f"{item.title} - S{season_number:02d}E{episode_number:02d} - {episode_title}" ticks = episode.get("RunTimeTicks") try: duration_seconds = max(0, round(int(ticks or 0) / 10_000_000)) except (TypeError, ValueError): duration_seconds = 0 episode_meta = { "seasonNumber": season_number, "episodeNumber": episode_number, "episodeTitle": episode_title, "seriesId": item.source_id, } return { "itemId": playback_item_id, "sourceId": item.source_id, "title": playback_title, "mediaType": item.media_type, "streamUrl": jellyfin_stream_url(base_url, playback_item_id, api_key, subtitle_index), "downloadUrl": jellyfin_download_url(base_url, playback_item_id, api_key), "durationSeconds": duration_seconds, "posterUrl": media_item_card_payload(item).get("posterUrl", ""), "year": item.year or "", "tmdbId": item.tmdb_id or "", "imdbId": item.imdb_id or "", "episode": episode_meta, } def fetch_jellyfin_library_ids(settings: dict[str, Any], library_names: list[str]) -> list[str]: if not library_names: return [] requested = {name.casefold() for name in library_names} data = jellyfin_request(settings, "/Library/MediaFolders", {}) items = data.get("Items", []) if not isinstance(items, list): raise RuntimeError("Jellyfin returned an invalid library folder payload") matched: list[str] = [] available: list[str] = [] for item in items: if not isinstance(item, dict): continue name = str(item.get("Name", "")).strip() folder_id = str(item.get("Id", "")).strip() if name: available.append(name) if name.casefold() in requested and folder_id: matched.append(folder_id) missing = sorted(set(library_names) - {name for name in available if name.casefold() in requested}) if missing: raise ValueError(f"Jellyfin library not found: {', '.join(missing)}") return matched def fetch_jellyfin_items(settings: dict[str, Any], item_type: str) -> list[dict[str, Any]]: parent_ids = settings.get("ParentIds") if isinstance(parent_ids, list) and parent_ids: all_items: list[dict[str, Any]] = [] for parent_id in parent_ids: scoped_settings = dict(settings) scoped_settings["ParentId"] = str(parent_id) scoped_settings.pop("ParentIds", None) all_items.extend(fetch_jellyfin_items(scoped_settings, item_type)) return all_items parent_id_value = str(settings.get("ParentId", "")).strip() or None items: list[dict[str, Any]] = [] start_index = 0 limit = 200 while True: data = jellyfin_request( settings, "/Items", { "Recursive": "true", "IncludeItemTypes": item_type, "Fields": "Genres,Overview,OfficialRating,RecursiveItemCount,ChildCount,PremiereDate,RunTimeTicks,ProviderIds", "SortBy": "SortName", "SortOrder": "Ascending", "EnableImages": "false", "ParentId": parent_id_value, "StartIndex": start_index, "Limit": limit, }, ) page = data.get("Items", []) if not isinstance(page, list): raise RuntimeError("Jellyfin returned an invalid Items payload") items.extend(item for item in page if isinstance(item, dict)) total = int(data.get("TotalRecordCount", len(items)) or len(items)) if not page or len(items) >= total: return items start_index += limit def jellyfin_item_dedupe_key(item: dict[str, Any], item_type: str) -> tuple[str, str]: provider_ids = item.get("ProviderIds") if isinstance(provider_ids, dict): for provider in ("Tmdb", "Imdb", "Tvdb"): value = str(provider_ids.get(provider, "")).strip().casefold() if value: return item_type.casefold(), f"{provider.casefold()}:{value}" title = clean_media_text(str(item.get("Name", "")), 120) or "" year = jellyfin_item_year(item) or "" return item_type.casefold(), f"title:{title.casefold()}:{year}" def dedupe_jellyfin_items(items: list[dict[str, Any]], item_type: str) -> list[dict[str, Any]]: provider_deduped: dict[tuple[str, str], dict[str, Any]] = {} for item in items: key = jellyfin_item_dedupe_key(item, item_type) current = provider_deduped.get(key) if current is None: provider_deduped[key] = item continue current_overview = str(current.get("Overview", "") or current.get("ShortOverview", "")) next_overview = str(item.get("Overview", "") or item.get("ShortOverview", "")) if len(next_overview) > len(current_overview): provider_deduped[key] = item title_deduped: dict[tuple[str, str], dict[str, Any]] = {} for item in provider_deduped.values(): title = clean_media_text(str(item.get("Name", "")), 120) or "" key = (title.casefold(), jellyfin_item_year(item) or "") current = title_deduped.get(key) if current is None: title_deduped[key] = item continue current_has_provider = bool(current.get("ProviderIds")) next_has_provider = bool(item.get("ProviderIds")) if next_has_provider and not current_has_provider: title_deduped[key] = item continue current_overview = str(current.get("Overview", "") or current.get("ShortOverview", "")) next_overview = str(item.get("Overview", "") or item.get("ShortOverview", "")) if len(next_overview) > len(current_overview): title_deduped[key] = item return sorted( title_deduped.values(), key=lambda item: (str(item.get("SortName", item.get("Name", ""))).casefold(), str(item.get("ProductionYear", ""))), ) def fetch_jellyfin_library(runtime: BotRuntime) -> tuple[list[MediaItem], list[MediaItem]]: state = load_state(runtime.settings_path) settings = { "url": str(state.get("jellyfin_url", "")).strip(), "apiKey": str(state.get("jellyfin_api_key", "")).strip(), } library_names = jellyfin_settings(runtime).get("libraryNames", []) if isinstance(library_names, list) and library_names: settings["ParentIds"] = fetch_jellyfin_library_ids(settings, [str(name) for name in library_names]) movie_items = dedupe_jellyfin_items(fetch_jellyfin_items(settings, "Movie"), "Movie") show_items = dedupe_jellyfin_items(fetch_jellyfin_items(settings, "Series"), "Series") movies = [jellyfin_movie_from_item(item) for item in movie_items] shows = [jellyfin_show_from_item(item) for item in show_items] return ( sorted(movies, key=lambda item: (item.title.casefold(), item.year or "")), sorted(shows, key=lambda item: (item.title.casefold(), item.year or "")), ) def media_library_fingerprint(movies: list[MediaItem], shows: list[MediaItem]) -> str: payload = media_library_to_jsonable(movies, shows) body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8") return hashlib.sha256(body).hexdigest() def media_item_tracking_key(item: MediaItem) -> str: return f"{item.media_type}:{item.title.casefold()}:{item.year or ''}" def media_change_entry(item: MediaItem) -> dict[str, Any]: return { "title": item.title, "year": item.year or "", "mediaType": item.media_type, } def catalog_poster_path(item: MediaItem) -> str: if not item.source_id: return "" return f"/catalog/poster/{urllib.parse.quote(item.source_id, safe='')}" def tmdb_auth_headers() -> dict[str, str]: bearer = os.getenv("TMDB_API_READ_TOKEN", "").strip() if bearer: return { "Authorization": f"Bearer {bearer}", "Accept": "application/json", } return {"Accept": "application/json"} def tmdb_api_key() -> str: return os.getenv("TMDB_API_KEY", "").strip() def tmdb_enabled() -> bool: return bool(os.getenv("TMDB_API_READ_TOKEN", "").strip() or tmdb_api_key()) def tmdb_request(path: str, params: dict[str, Any]) -> dict[str, Any]: if not tmdb_enabled(): raise RuntimeError("TMDB credentials are not configured") query_params = {key: value for key, value in params.items() if value not in {None, ""}} api_key = tmdb_api_key() if api_key: query_params["api_key"] = api_key query = urllib.parse.urlencode(query_params) url = f"https://api.themoviedb.org/3{path}" if query: url = f"{url}?{query}" request = urllib.request.Request( url, headers={ "User-Agent": "ArchiveStatusBot/1.0", **tmdb_auth_headers(), }, method="GET", ) try: with urllib.request.urlopen(request, timeout=20) as response: return json.loads(response.read().decode("utf-8")) except urllib.error.HTTPError as exc: detail = exc.read().decode("utf-8", errors="ignore") raise RuntimeError(f"TMDB API failed: HTTP {exc.code} {detail}") from exc except (urllib.error.URLError, TimeoutError, socket.timeout) as exc: raise RuntimeError(f"TMDB API failed: {clean_error(exc)}") from exc def tmdb_image_url(file_path: str | None) -> str | None: if not file_path: return None return f"{TMDB_IMAGE_BASE_URL}{file_path}" def tmdb_cache_key(item: MediaItem) -> str: return "|".join( [ item.media_type, item.source_id or "", item.tmdb_id or "", item.imdb_id or "", item.title.casefold(), item.year or "", ] ) def tmdb_search_poster_path(item: MediaItem) -> str | None: if item.tmdb_id: endpoint = "/movie" if item.media_type == "movie" else "/tv" data = tmdb_request(f"{endpoint}/{urllib.parse.quote(item.tmdb_id, safe='')}", {}) return data.get("poster_path") if item.imdb_id: data = tmdb_request( f"/find/{urllib.parse.quote(item.imdb_id, safe='')}", {"external_source": "imdb_id"}, ) results_key = "movie_results" if item.media_type == "movie" else "tv_results" results = data.get(results_key, []) if isinstance(results, list): for result in results: if isinstance(result, dict) and result.get("poster_path"): return str(result["poster_path"]) search_params: dict[str, Any] = {"query": item.title} year = media_item_year_number(item) if item.media_type == "movie": endpoint = "/search/movie" if year is not None: search_params["year"] = year else: endpoint = "/search/tv" if year is not None: search_params["first_air_date_year"] = year data = tmdb_request(endpoint, search_params) results = data.get("results", []) if not isinstance(results, list): return None for result in results: if isinstance(result, dict) and result.get("poster_path"): return str(result["poster_path"]) return None def tmdb_fallback_poster_url(item: MediaItem) -> str | None: if not tmdb_enabled(): return None cache_key = tmdb_cache_key(item) if cache_key in TMDB_POSTER_CACHE: return TMDB_POSTER_CACHE[cache_key] try: poster_path = tmdb_search_poster_path(item) except Exception: TMDB_POSTER_CACHE[cache_key] = None return None url = tmdb_image_url(poster_path) TMDB_POSTER_CACHE[cache_key] = url return url def fetch_jellyfin_image(runtime: BotRuntime, item_id: str) -> tuple[bytes, str] | None: if not item_id.strip(): return None state = load_state(runtime.settings_path) base_url = str(state.get("jellyfin_url", "")).strip().rstrip("/") api_key = str(state.get("jellyfin_api_key", "")).strip() if not base_url or not api_key: return None url = f"{base_url}/Items/{urllib.parse.quote(item_id, safe='')}/Images/Primary?maxHeight=720&maxWidth=480&quality=90" request = urllib.request.Request( url, headers={ "User-Agent": "ArchiveStatusBot/1.0", "X-Emby-Token": api_key, "Accept": "image/*", }, method="GET", ) try: with urllib.request.urlopen(request, timeout=30) as response: content_type = response.headers.get_content_type() or "image/jpeg" return response.read(), content_type except urllib.error.HTTPError as exc: if exc.code == 404: return None detail = exc.read().decode("utf-8", errors="ignore") raise RuntimeError(f"Jellyfin image fetch failed: HTTP {exc.code} {detail}") from exc except (urllib.error.URLError, TimeoutError, socket.timeout) as exc: raise RuntimeError(f"Jellyfin image fetch failed: {clean_error(exc)}") from exc def catalog_known_source_ids(runtime: BotRuntime) -> set[str]: movies, shows = load_media_library(runtime) return { item.source_id.strip() for item in [*movies, *shows] if item.source_id and item.source_id.strip() } def catalog_item_for_source_id(runtime: BotRuntime, item_id: str) -> MediaItem | None: movies, shows = load_media_library(runtime) for item in [*movies, *shows]: if item.source_id and item.source_id.strip() == item_id: return item return None def media_library_changes( previous_movies: list[MediaItem], previous_shows: list[MediaItem], current_movies: list[MediaItem], current_shows: list[MediaItem], ) -> dict[str, Any]: previous = {media_item_tracking_key(item): item for item in [*previous_movies, *previous_shows]} current = {media_item_tracking_key(item): item for item in [*current_movies, *current_shows]} added_keys = sorted(set(current) - set(previous), key=lambda key: (current[key].media_type, current[key].title.casefold())) removed_keys = sorted(set(previous) - set(current), key=lambda key: (previous[key].media_type, previous[key].title.casefold())) added = [media_change_entry(current[key]) for key in added_keys] removed = [media_change_entry(previous[key]) for key in removed_keys] return { "added": added, "removed": removed, "addedCount": len(added), "removedCount": len(removed), } def update_jellyfin_sync_state( runtime: BotRuntime, *, fingerprint: str | None = None, error: str | None = None, published: bool = False, changes: dict[str, Any] | None = None, ) -> None: state = load_state(runtime.settings_path) now = datetime.now(timezone.utc).isoformat() state["jellyfin_last_sync_at"] = now state["jellyfin_last_sync_error"] = error if fingerprint is not None: state["jellyfin_last_fingerprint"] = fingerprint if published: state["jellyfin_last_published_at"] = now if fingerprint is not None: state["jellyfin_last_published_fingerprint"] = fingerprint if changes is not None: state["jellyfin_last_changes"] = changes save_state(runtime.settings_path, state) def sync_jellyfin_library(runtime: BotRuntime, force_publish: bool = False) -> dict[str, Any]: previous_movies, previous_shows = load_media_library(runtime) movies, shows = fetch_jellyfin_library(runtime) changes = media_library_changes(previous_movies, previous_shows, movies, shows) fingerprint = media_library_fingerprint(movies, shows) settings_state = load_state(runtime.settings_path) changed = fingerprint != str(settings_state.get("jellyfin_last_fingerprint", "")) publish_changed = fingerprint != str(settings_state.get("jellyfin_last_published_fingerprint", "")) save_media_library(runtime, movies, shows) published = False result: dict[str, Any] = {} if (publish_changed or force_publish) and not runtime.dry_run: result = publish_media_items( runtime=runtime, channel_id=channel_settings(runtime)["mediaChannelId"], movies_all=movies, shows_all=shows, ) published = True update_jellyfin_sync_state(runtime, fingerprint=fingerprint, published=published, changes=changes) return { "changed": changed, "published": published, "changes": changes, "movieCount": len(movies), "showCount": len(shows), "libraryNames": jellyfin_settings(runtime).get("libraryNames", []), "library": media_library_to_jsonable(movies, shows), "publishResult": result, "jellyfin": jellyfin_settings(runtime), } def maybe_run_jellyfin_sync(runtime: BotRuntime) -> dict[str, Any] | None: settings = jellyfin_settings(runtime) if not settings["configured"] or not settings["autoSync"]: return None interval = int(os.getenv("JELLYFIN_SYNC_INTERVAL_SECONDS", str(DEFAULT_JELLYFIN_SYNC_INTERVAL_SECONDS))) state = load_state(runtime.settings_path) last_sync = str(state.get("jellyfin_last_sync_at", "")).strip() if last_sync: try: last = datetime.fromisoformat(last_sync) if (datetime.now(timezone.utc) - last).total_seconds() < interval: return None except ValueError: pass try: return sync_jellyfin_library(runtime) except Exception as exc: update_jellyfin_sync_state(runtime, error=str(exc)[:240]) raise def media_item_heading(item: MediaItem) -> str: if item.year: return f"{item.title} ({item.year})" return item.title def media_item_genres(item: MediaItem) -> list[str]: if not item.genres: return [] return [genre.strip() for genre in item.genres.split(",") if genre.strip()] def media_item_runtime_minutes(item: MediaItem) -> int | None: if not item.runtime or item.media_type != "movie": return None cleaned = item.runtime.lower().replace(" ", "") match = re.fullmatch(r"(?:(\d+)h)?(?:(\d+)m)?", cleaned) if match and (match.group(1) or match.group(2)): hours = int(match.group(1) or 0) minutes = int(match.group(2) or 0) return (hours * 60) + minutes if cleaned.isdigit(): return int(cleaned) return None def media_item_year_number(item: MediaItem) -> int | None: if not item.year: return None try: return int(item.year) except ValueError: return None def media_item_search_blob(item: MediaItem) -> str: fields = [ item.title, item.year or "", item.genres or "", item.rating or "", item.runtime or "", item.summary or "", ] return " ".join(fields).casefold() def recommendation_presets() -> list[dict[str, str]]: return [ {"id": "balanced", "label": "Balanced"}, {"id": "quick", "label": "Quick Watch"}, {"id": "deep", "label": "Deep Dive"}, {"id": "recent", "label": "Newer Picks"}, {"id": "comfort", "label": "Comfort Picks"}, ] def recommendation_reason(item: MediaItem, mode: str, genre: str) -> str: reasons: list[str] = [] if genre and genre.lower() != "all": reasons.append(f"matches {genre}") if mode == "quick" and item.media_type == "movie" and item.runtime: reasons.append(f"short runtime: {item.runtime}") elif mode == "deep": if item.media_type == "movie" and item.runtime: reasons.append(f"long runtime: {item.runtime}") elif item.media_type == "show" and item.seasons: reasons.append(f"{item.seasons} season{'s' if item.seasons != 1 else ''}") elif mode == "recent" and item.year: reasons.append(f"newer release: {item.year}") elif mode == "comfort": if item.genres: reasons.append(item.genres) if item.media_type == "show" and item.episodes: reasons.append(f"{item.episodes} episodes") else: if item.genres: reasons.append(item.genres) if item.media_type == "movie" and item.runtime: reasons.append(item.runtime) if item.media_type == "show" and item.episodes: reasons.append(f"{item.episodes} episodes") return " · ".join(reasons[:2]) or "from your library" def recommendation_score(item: MediaItem, mode: str, genre: str, search: str) -> float: score = 0.0 genres = {value.casefold() for value in media_item_genres(item)} runtime_minutes = media_item_runtime_minutes(item) year = media_item_year_number(item) blob = media_item_search_blob(item) if genre and genre.casefold() != "all": score += 40 if genre.casefold() in genres else -20 if search: score += 25 if search.casefold() in blob else -30 if mode == "quick": if item.media_type == "movie": if runtime_minutes is not None: score += max(0, 180 - runtime_minutes) / 2 else: score += 12 if item.episodes: score += max(0, 48 - item.episodes) / 3 elif mode == "deep": if item.media_type == "movie": if runtime_minutes is not None: score += min(runtime_minutes, 240) / 2 else: if item.seasons: score += min(item.seasons * 10, 50) if item.episodes: score += min(item.episodes / 3, 30) elif mode == "recent": if year is not None: score += max(0, year - 1990) elif mode == "comfort": if item.media_type == "show": score += 24 if item.episodes: score += min(item.episodes / 4, 25) if genres: score += 6 else: if genres: score += 8 if item.summary: score += 4 if item.media_type == "movie" and runtime_minutes is not None: score += max(0, 150 - abs(115 - runtime_minutes)) / 10 if item.media_type == "show" and item.episodes: score += min(item.episodes / 6, 16) score += (sum(ord(character) for character in item.title.casefold()) % 17) / 10 return score def available_media_genres(movies: list[MediaItem], shows: list[MediaItem]) -> list[str]: genres: set[str] = set() for item in [*movies, *shows]: genres.update(media_item_genres(item)) return sorted(genres, key=str.casefold) def media_item_card_payload(item: MediaItem, reason: str = "") -> dict[str, Any]: return { "title": item.title, "mediaType": item.media_type, "sourceId": item.source_id or "", "tmdbId": item.tmdb_id or "", "imdbId": item.imdb_id or "", "posterUrl": catalog_poster_path(item), "year": item.year or "", "genres": item.genres or "", "rating": item.rating or "", "runtime": item.runtime or "", "summary": item.summary or "", "seasons": item.seasons or 0, "episodes": item.episodes or 0, "reason": reason, } def recommend_media_items( movies: list[MediaItem], shows: list[MediaItem], *, media_type: str = "all", genre: str = "all", mode: str = "balanced", search: str = "", limit: int = 6, ) -> dict[str, Any]: pools = { "movie": movies, "show": shows, "all": [*movies, *shows], } selected_pool = pools.get(media_type, pools["all"]) if not selected_pool: return {"items": [], "summary": "No media available for recommendations."} ranked = sorted( selected_pool, key=lambda item: ( recommendation_score(item, mode, genre, search), media_item_year_number(item) or 0, item.title.casefold(), ), reverse=True, ) items = [ media_item_card_payload(item, recommendation_reason(item, mode, genre)) for item in ranked[: max(1, min(limit, 12))] ] media_label = {"all": "movies and shows", "movie": "movies", "show": "shows"}.get(media_type, "media") if search: summary = f"{mode.title()} picks in {media_label} for “{search}”." elif genre and genre.casefold() != "all": summary = f"{mode.title()} picks in {media_label} for {genre}." else: summary = f"{mode.title()} picks from your {media_label}." return {"items": items, "summary": summary} def media_item_value(item: MediaItem) -> str: details: list[str] = [] if item.genres: details.append(item.genres) if item.rating: details.append(item.rating) if item.runtime and item.media_type == "movie": details.append(item.runtime) if item.media_type == "show": counts = [] if item.seasons: counts.append(f"{item.seasons} season{'s' if item.seasons != 1 else ''}") if item.episodes: counts.append(f"{item.episodes} episode{'s' if item.episodes != 1 else ''}") if counts: details.append(" · ".join(counts)) lines = [] if details: lines.append(" • ".join(details)) if item.summary: lines.append(item.summary) return "\n".join(lines)[:1024] or "No details provided." def media_category_embeds( title: str, items: list[MediaItem], color: int, total_count: int, source_name: str, ) -> list[dict[str, Any]]: embeds: list[dict[str, Any]] = [] omitted = max(total_count - len(items), 0) page_count = max((len(items) + MEDIA_ITEMS_PER_EMBED - 1) // MEDIA_ITEMS_PER_EMBED, 1) for page_index in range(page_count): start = page_index * MEDIA_ITEMS_PER_EMBED chunk = items[start : start + MEDIA_ITEMS_PER_EMBED] description = f"{total_count} total from `{source_name}`" if omitted: description += f" · showing first {len(items)}" if page_count > 1: description += f" · page {page_index + 1}/{page_count}" fields = [ { "name": media_item_heading(item)[:256], "value": media_item_value(item), "inline": False, } for item in chunk ] embeds.append( { "title": title, "description": description, "color": color, "fields": fields, } ) return embeds def render_media_catalog_payloads( movies: list[MediaItem], shows: list[MediaItem], movie_total: int, show_total: int, movie_source: str, show_source: str, ) -> list[dict[str, Any]]: now = datetime.now(timezone.utc) embeds: list[dict[str, Any]] = [ { "title": "The Mithral Archive Media Catalog", "description": "Current movies and shows available in the archive.", "color": 0x5865F2, "fields": [ {"name": "Movies", "value": str(movie_total), "inline": True}, {"name": "Shows", "value": str(show_total), "inline": True}, {"name": "Updated", "value": now.strftime("%Y-%m-%d %H:%M UTC"), "inline": True}, ], "timestamp": now.isoformat(), } ] if movies: embeds.extend(media_category_embeds("Movies", movies, 0xF59E0B, movie_total, movie_source)) if shows: embeds.extend(media_category_embeds("Shows", shows, 0x10B981, show_total, show_source)) payloads = [] for index in range(0, len(embeds), MAX_DISCORD_EMBEDS): payloads.append( { "content": "**The Mithral Archive Media Catalog**" if index == 0 else "", "embeds": embeds[index : index + MAX_DISCORD_EMBEDS], } ) return payloads def media_markdown_line(item: MediaItem) -> str: title = media_item_heading(item) details: list[str] = [] if item.genres: details.append(item.genres) if item.rating: details.append(item.rating) if item.runtime and item.media_type == "movie": details.append(item.runtime) if item.media_type == "show": if item.seasons: details.append(f"{item.seasons} season{'s' if item.seasons != 1 else ''}") if item.episodes: details.append(f"{item.episodes} episode{'s' if item.episodes != 1 else ''}") suffix = f" - {'; '.join(details)}" if details else "" return f"- **{title}**{suffix}" def render_media_catalog_markdown(movies: list[MediaItem], shows: list[MediaItem]) -> str: now = datetime.now(timezone.utc) lines = [ "# The Mithral Archive Media Catalog", "", f"Updated: {now.strftime('%Y-%m-%d %H:%M UTC')}", f"Movies: {len(movies)}", f"Shows: {len(shows)}", "", ] if movies: lines.extend(["## Movies", ""]) for item in movies: lines.append(media_markdown_line(item)) if item.summary: lines.append(f" - {item.summary}") lines.append("") if shows: lines.extend(["## Shows", ""]) for item in shows: lines.append(media_markdown_line(item)) if item.summary: lines.append(f" - {item.summary}") lines.append("") return "\n".join(lines).strip() + "\n" def render_catalog_html(runtime: BotRuntime) -> bytes: movies, shows = load_media_library(runtime) genres = available_media_genres(movies, shows) default_recommendations = { "balanced": recommend_media_items(movies, shows, mode="balanced", limit=6), "quick": recommend_media_items(movies, shows, mode="quick", limit=6), "deep": recommend_media_items(movies, shows, mode="deep", limit=6), } updated_at = load_state(runtime.media_library_path).get("updated_at") updated = "Never" if updated_at: try: updated = datetime.fromisoformat(str(updated_at)).strftime("%Y-%m-%d %H:%M UTC") except ValueError: updated = str(updated_at) def item_block(item: MediaItem) -> str: meta = [] if item.year: meta.append(item.year) if item.genres: meta.append(item.genres) if item.rating: meta.append(item.rating) if item.runtime: meta.append(item.runtime) if item.media_type == "show": if item.seasons: meta.append(f"{item.seasons} season{'s' if item.seasons != 1 else ''}") if item.episodes: meta.append(f"{item.episodes} episode{'s' if item.episodes != 1 else ''}") summary = f"

{html.escape(item.summary)}

" if item.summary else "" poster = ( f'
{html.escape(item.title)} poster
' if item.source_id else '
No poster
' ) return ( f'
' f"{poster}" f'

{html.escape(item.title)}

' f'
{html.escape(" · ".join(meta))}
' f"{summary}
" ) items = "\n".join(item_block(item) for item in [*movies, *shows]) genre_options = "\n".join( f'' for genre in genres ) recommendations_json = json.dumps(default_recommendations) body = f""" The Mithral Archive Media Catalog

The Mithral Archive Media Catalog

{len(movies)}Movies
{len(shows)}Shows
{html.escape(updated)}Updated
{items}
""" return body.encode("utf-8") def publish_media_markdown_message( token: str, channel_id: str, movies: list[MediaItem], shows: list[MediaItem], catalog_url: str = "", ) -> str: markdown = render_media_catalog_markdown(movies, shows) payload: dict[str, Any] = { "content": ( "**The Mithral Archive Media Catalog**\n" f"{len(movies)} movies · {len(shows)} shows\n" "Attached as a compact Markdown list." ), "allowed_mentions": {"parse": []}, } if catalog_url: payload["components"] = [ { "type": 1, "components": [ { "type": 2, "style": 5, "label": "Open Catalog", "url": catalog_url, } ], } ] message = discord_multipart_request( "POST", token, f"/channels/{channel_id}/messages", payload, [("files[0]", "media-catalog.md", "text/markdown; charset=utf-8", markdown.encode("utf-8"))], ) message_id = str(message.get("id", "")).strip() if not message_id: raise RuntimeError("Discord did not return a message id for the media catalog") return message_id def publish_media_items( runtime: BotRuntime, channel_id: str, movies_all: list[MediaItem], shows_all: list[MediaItem], movie_source: str = "Dashboard library", show_source: str = "Dashboard library", ) -> dict[str, Any]: del movie_source, show_source if runtime.dry_run: raise RuntimeError("Discord dry run is enabled; media catalog was parsed but not sent") settings = channel_settings(runtime) channel = validate_channel_id(channel_id.strip() or settings["mediaChannelId"], "Media") if not movies_all and not shows_all: raise ValueError("Add at least one movie or show before publishing") state = load_state(runtime.media_state_path) old_channel = str(state.get("channel_id", "")).strip() existing_ids = [str(message_id) for message_id in state.get("message_ids", []) if str(message_id).strip()] delete_channel = old_channel or channel for old_id in existing_ids: discord_delete_message(runtime.token, delete_channel, old_id) message_id = publish_media_markdown_message( runtime.token, channel, movies_all, shows_all, catalog_url=settings.get("catalogUrl", ""), ) save_state( runtime.media_state_path, { "channel_id": channel, "message_ids": [message_id], "movie_count": len(movies_all), "show_count": len(shows_all), "format": "markdown", "published_at": datetime.now(timezone.utc).isoformat(), }, ) save_media_channel_setting(runtime, channel) return { "channelId": channel, "messageIds": [message_id], "movieCount": len(movies_all), "showCount": len(shows_all), "displayedMovieCount": len(movies_all), "displayedShowCount": len(shows_all), "format": "markdown", } def media_catalog_status(runtime: BotRuntime) -> dict[str, Any]: state = load_state(runtime.media_state_path) settings = channel_settings(runtime) movies, shows = load_media_library(runtime) return { "channelId": str(state.get("channel_id", "")).strip() or settings["mediaChannelId"], "channels": settings, "messageIds": state.get("message_ids", []) if isinstance(state.get("message_ids"), list) else [], "movieCount": state.get("movie_count"), "showCount": state.get("show_count"), "format": state.get("format", "markdown"), "publishedAt": state.get("published_at"), "changes": jellyfin_settings(runtime).get("lastChanges"), "genres": available_media_genres(movies, shows), "recommendations": { "balanced": recommend_media_items(movies, shows, mode="balanced", limit=5), "quick": recommend_media_items(movies, shows, mode="quick", limit=5), "deep": recommend_media_items(movies, shows, mode="deep", limit=5), }, "library": media_library_to_jsonable(movies, shows), }