297 lines
11 KiB
Python
297 lines
11 KiB
Python
#!/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") |