#!/usr/bin/env python3 """ External API clients for TMDB, OMDb, and Jellyseerr """ import json import os import time from datetime import datetime, timezone from typing import Dict, Any, List, Optional, Tuple from urllib.parse import urlencode, quote from urllib.request import Request as UrlRequest, urlopen from urllib.error import URLError, HTTPError def _log(level: str, msg: str): """Placeholder logging function - replace with your actual logger""" print(f"[{datetime.now().isoformat()}] {level}: {msg}") def _get_json(url: str, timeout: int = 20, headers: Dict[str, str] = None) -> Optional[Dict[str, Any]]: """Make GET request and return JSON""" try: req = UrlRequest(url, headers=headers or {"Accept": "application/json"}) with urlopen(req, timeout=timeout) as resp: return json.loads(resp.read().decode("utf-8")) except Exception as e: _log("WARNING", f"GET {url} failed: {e}") return None def _parse_date_to_iso(date_str: str) -> Optional[str]: """Parse various date formats to ISO string""" if not date_str or date_str == "N/A": return None try: if len(date_str) == 10 and date_str[4] == "-": # YYYY-MM-DD dt = datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc) else: dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc) return dt.isoformat(timespec="seconds") except Exception: return None class TMDBClient: """The Movie Database API client""" def __init__(self, api_key: str = None, primary_country: str = "US"): self.api_key = api_key or os.environ.get("TMDB_API_KEY", "") self.primary_country = primary_country.upper() self.enabled = bool(self.api_key) def _get(self, path: str, params: Dict[str, Any] = None) -> Optional[Dict[str, Any]]: """Make GET request to TMDB API""" if not self.enabled: return None params = params or {} params["api_key"] = self.api_key url = f"https://api.themoviedb.org/3{path}?{urlencode(params)}" return _get_json(url, timeout=20) def find_by_imdb(self, imdb_id: str) -> Optional[Dict[str, Any]]: """Find movie by IMDb ID""" result = self._get(f"/find/{quote(imdb_id)}", {"external_source": "imdb_id"}) if result and result.get("movie_results"): return result["movie_results"][0] return None def get_movie_details(self, tmdb_id: int) -> Optional[Dict[str, Any]]: """Get detailed movie information""" return self._get(f"/movie/{tmdb_id}") def get_digital_release_date(self, imdb_id: str) -> Optional[str]: """Get digital release date for a movie""" movie = self.find_by_imdb(imdb_id) if not movie: return None tmdb_id = movie.get("id") if not tmdb_id: return None release_dates = self._get(f"/movie/{tmdb_id}/release_dates") if not release_dates: return None for entry in release_dates.get("results", []): if entry.get("iso_3166_1", "").upper() != self.primary_country: continue for release in entry.get("release_dates", []): if release.get("type") == 4 and release.get("release_date"): # Digital release return _parse_date_to_iso(release["release_date"]) return None def get_tv_season_episodes(self, tv_id: int, season_number: int) -> Dict[int, str]: """Get episode air dates for a TV season""" result = self._get(f"/tv/{tv_id}/season/{season_number}") episodes = {} if result: for episode in result.get("episodes", []): ep_num = episode.get("episode_number") air_date = episode.get("air_date") if isinstance(ep_num, int) and air_date: episodes[ep_num] = air_date return episodes class OMDbClient: """Open Movie Database API client""" def __init__(self, api_key: str = None): self.api_key = api_key or os.environ.get("OMDB_API_KEY", "") self.enabled = bool(self.api_key) def get_movie_details(self, imdb_id: str) -> Optional[Dict[str, Any]]: """Get movie details from OMDb""" if not self.enabled: return None params = {"i": imdb_id, "apikey": self.api_key} url = f"http://www.omdbapi.com/?{urlencode(params)}" result = _get_json(url, timeout=15) if result and result.get("Response") == "True": return result return None def get_dvd_release_date(self, imdb_id: str) -> Optional[str]: """Get DVD/digital release date""" details = self.get_movie_details(imdb_id) if not details: return None dvd_date = details.get("DVD") or details.get("Released") if not dvd_date or dvd_date == "N/A": return None # Try to parse various date formats for fmt in ("%d %b %Y", "%d %B %Y", "%Y-%m-%d"): try: dt = datetime.strptime(dvd_date, fmt).replace(tzinfo=timezone.utc) return dt.isoformat(timespec="seconds") except Exception: continue return None def get_tv_season_episodes(self, imdb_id: str, season_number: int) -> Dict[int, str]: """Get episode release dates for a TV season""" if not self.enabled: return {} params = {"i": imdb_id, "Season": str(season_number), "apikey": self.api_key} url = f"http://www.omdbapi.com/?{urlencode(params)}" result = _get_json(url, timeout=15) episodes = {} if result and result.get("Response") == "True": for episode in result.get("Episodes", []): try: ep_num = int(episode.get("Episode", 0)) released = episode.get("Released") if ep_num and released and released != "N/A": episodes[ep_num] = released except Exception: continue return episodes class JellyseerrClient: """Jellyseerr API client""" def __init__(self, base_url: str = None, api_key: str = None): self.base_url = (base_url or os.environ.get("JELLYSEERR_URL", "")).rstrip("/") self.api_key = api_key or os.environ.get("JELLYSEERR_API_KEY", "") self.enabled = bool(self.base_url and self.api_key) def _get(self, path: str) -> Optional[Dict[str, Any]]: """Make GET request to Jellyseerr API""" if not self.enabled: return None url = f"{self.base_url}/api/v1{path}" headers = {"X-Api-Key": self.api_key, "Accept": "application/json"} return _get_json(url, timeout=20, headers=headers) def get_movie_details(self, tmdb_id: int) -> Optional[Dict[str, Any]]: """Get movie details from Jellyseerr""" return self._get(f"/movie/{tmdb_id}") def get_digital_release_dates(self, tmdb_id: int) -> List[str]: """Get digital release date candidates from Jellyseerr""" details = self.get_movie_details(tmdb_id) if not details: return [] candidates = [] # Check direct fields for field in ("digitalReleaseDate", "physicalReleaseDate", "vodReleaseDate"): value = details.get(field) if value: iso_date = _parse_date_to_iso(value) if iso_date: candidates.append(iso_date) # Check release dates array for array_field in ("releaseDates", "releases", "dates"): release_array = details.get(array_field) if not isinstance(release_array, list): continue for release in release_array: if not isinstance(release, dict): continue release_type = (release.get("type") or release.get("label") or "").lower() release_date = release.get("date") or release.get("releaseDate") if release_date and ("digital" in release_type or "vod" in release_type or "stream" in release_type): iso_date = _parse_date_to_iso(release_date) if iso_date: candidates.append(iso_date) return candidates class ExternalClientManager: """Manager for all external API clients""" def __init__(self): self.tmdb = TMDBClient() self.omdb = OMDbClient() self.jellyseerr = JellyseerrClient() def get_digital_release_candidates(self, imdb_id: str) -> List[Tuple[str, str]]: """Get digital release date candidates from all sources""" candidates = [] # TMDB digital release if self.tmdb.enabled: tmdb_date = self.tmdb.get_digital_release_date(imdb_id) if tmdb_date: candidates.append((tmdb_date, "tmdb:digital")) # OMDb DVD/digital release if self.omdb.enabled: omdb_date = self.omdb.get_dvd_release_date(imdb_id) if omdb_date: candidates.append((omdb_date, "omdb:dvd")) # Jellyseerr digital releases if self.jellyseerr.enabled and self.tmdb.enabled: tmdb_movie = self.tmdb.find_by_imdb(imdb_id) if tmdb_movie: tmdb_id = tmdb_movie.get("id") if tmdb_id: jellyseerr_dates = self.jellyseerr.get_digital_release_dates(tmdb_id) for date in jellyseerr_dates: candidates.append((date, "jellyseerr:digital")) return candidates def get_earliest_digital_release(self, imdb_id: str) -> Optional[Tuple[str, str]]: """Get the earliest digital release date""" candidates = self.get_digital_release_candidates(imdb_id) if not candidates: return None try: return sorted( candidates, key=lambda x: datetime.fromisoformat(x[0].replace("Z", "+00:00")) )[0] except Exception: return candidates[0] if candidates else None if __name__ == "__main__": # Test the clients manager = ExternalClientManager() test_imdb = "tt1596343" # Example IMDb ID digital_candidates = manager.get_digital_release_candidates(test_imdb) print(f"Digital release candidates for {test_imdb}: {digital_candidates}") earliest = manager.get_earliest_digital_release(test_imdb) if earliest: print(f"Earliest digital release: {earliest[0]} ({earliest[1]})") else: print("No digital release dates found")