#!/usr/bin/env python3 """ Enhanced Radarr API client with improved history event detection and path mapping """ import json import time from datetime import datetime, timezone from pathlib import Path from typing import Dict, Any, List, Optional, Tuple from urllib.parse import urlencode, urljoin from urllib.request import Request as UrlRequest, urlopen from urllib.error import URLError, HTTPError # Import path mapper for proper path handling try: from ..core.path_mapper import path_mapper except ImportError: # Fallback for standalone testing class DummyPathMapper: def analyze_import_source_path(self, path): return "/downloads/" in path.lower(), "basic_check" path_mapper = DummyPathMapper() def _log(level: str, msg: str): """Placeholder logging function - replace with your actual logger""" print(f"[{datetime.now().isoformat()}] {level}: {msg}") class RadarrClient: """Enhanced Radarr API client with improved import date detection""" # Known Radarr event types that indicate actual imports REAL_IMPORT_EVENTS = { # v4+ events "downloadfolderimported", "moviefileimported", "imported", "downloadimported", "fileimported", "movieimported", # v3 events "downloadedepisode", "downloadedmovie", } # These are now handled by path_mapper, but keeping for backward compatibility DOWNLOAD_PATH_INDICATORS = [ '/downloads/', '/download/', '/completed/', '/importing/', '/nzbs/', '/torrents/', '/temp/', '/tmp/', 'sabnzbd', 'nzbget', 'deluge', 'qbittorrent', 'transmission', 'usenet', 'torrent', 'radarr', 'completed', 'processing' ] def __init__(self, base_url: str, api_key: str, timeout: int = 45, retries: int = 3): self.base_url = base_url.rstrip("/") self.api_key = api_key self.timeout = timeout self.retries = max(0, retries) def _get(self, path: str, params: Dict[str, Any] = None) -> Optional[Any]: """Make GET request to Radarr API with retries""" if not self.api_key: return None attempt = 0 last_err = None while attempt <= self.retries: try: params = params or {} params["apikey"] = self.api_key url = urljoin(f"{self.base_url}/", path.lstrip("/")) if params: url = url + ("&" if "?" in url else "?") + urlencode(params) _log("DEBUG", f"Radarr API Request: {url}") req = UrlRequest(url, headers={"Accept": "application/json"}) with urlopen(req, timeout=self.timeout) as resp: data = resp.read().decode("utf-8") result = json.loads(data) return result except (URLError, HTTPError, json.JSONDecodeError) as e: last_err = e _log("DEBUG", f"Radarr API attempt {attempt + 1} failed: {e}") time.sleep(min(2 ** attempt, 5)) # Exponential backoff attempt += 1 _log("WARNING", f"Radarr GET {path} failed after {self.retries + 1} attempts: {last_err}") return None def movie_by_imdb(self, imdb_id: str) -> Optional[Dict[str, Any]]: """Find movie by IMDb ID with enhanced validation""" imdb_id = imdb_id if imdb_id.startswith("tt") else f"tt{imdb_id}" _log("DEBUG", f"Looking up movie by IMDb ID: {imdb_id}") # Method 1: Get all movies and filter by IMDb (most reliable) all_movies = self._get("/api/v3/movie") if isinstance(all_movies, list): _log("DEBUG", f"Got {len(all_movies)} total movies, filtering by IMDb ID") for movie in all_movies: movie_imdb = movie.get("imdbId", "").lower() if movie_imdb == imdb_id.lower(): _log("INFO", f"Found exact match: {movie.get('title')} (ID: {movie.get('id')})") return movie # Method 2: Direct API search with validation result = self._get("/api/v3/movie", {"imdbId": imdb_id}) if isinstance(result, list) and result: for movie in result: movie_imdb = movie.get("imdbId", "").lower() if movie_imdb == imdb_id.lower(): _log("INFO", f"Found via search: {movie.get('title')} (ID: {movie.get('id')})") return movie # Method 3: Lookup endpoint lookup_result = self._get("/api/v3/movie/lookup/imdb", {"imdbId": imdb_id}) if isinstance(lookup_result, dict) and lookup_result.get("imdbId", "").lower() == imdb_id.lower(): _log("INFO", f"Found via lookup: {lookup_result.get('title')} (ID: {lookup_result.get('id')})") return lookup_result _log("ERROR", f"No movie found for IMDb ID: {imdb_id}") return None def _analyze_event_for_import(self, event: Dict[str, Any]) -> Tuple[bool, str, Optional[str]]: """ Analyze a history event to determine if it's a real import. Returns: (is_real_import, reason, date_iso) """ event_type = (event.get("eventType") or event.get("type") or "").lower() date_str = event.get("date") event_data = event.get("data", {}) # Parse date date_iso = None if date_str: try: date_iso = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc).isoformat(timespec="seconds") except Exception: date_iso = None if not date_iso: return False, "no_valid_date", None # Check if event type indicates import if event_type not in self.REAL_IMPORT_EVENTS: return False, f"event_type_not_import({event_type})", date_iso # Extract source path information source_path = ( event_data.get('droppedPath', '') or event_data.get('sourcePath', '') or event_data.get('path', '') or event_data.get('sourceTitle', '') or event.get('sourcePath', '') or event.get('sourceTitle', '') ).lower() if not source_path: return False, "no_source_path", date_iso # Use path mapper for more sophisticated path analysis is_from_downloads, path_reason = path_mapper.analyze_import_source_path(source_path) if not is_from_downloads: return False, f"not_from_downloads({path_reason})", date_iso return True, f"real_import_from_downloads({path_reason})", date_iso def earliest_import_event_optimized(self, movie_id: int) -> Optional[str]: """ Find earliest real import event with optimized querying. Stops as soon as we find valid import events instead of loading everything. """ _log("INFO", f"Finding earliest import for movie_id {movie_id}") earliest_real_import = None first_grab = None page = 1 page_size = 50 # Smaller pages for faster iteration total_processed = 0 while page <= 20: # Safety limit data = self._get("/api/v3/history", { "movieId": str(movie_id), # Ensure it's a string "page": page, "pageSize": page_size, "sortKey": "date", "sortDirection": "ascending" # Start from oldest }) if not data: break items = data if isinstance(data, list) else data.get("records", []) if not items: break _log("DEBUG", f"Page {page}: Processing {len(items)} events") for event in items: total_processed += 1 event_type = (event.get("eventType") or "").lower() # Check for grab events (fallback) if event_type == "grabbed" and not first_grab: if event.get("date"): try: first_grab = datetime.fromisoformat(event["date"].replace("Z", "+00:00")).astimezone(timezone.utc).isoformat(timespec="seconds") _log("DEBUG", f"Found grab event at {first_grab}") except Exception: pass # Analyze for real import is_real, reason, date_iso = self._analyze_event_for_import(event) if is_real and date_iso: _log("INFO", f"✅ FOUND REAL IMPORT: {event_type} at {date_iso} - {reason}") earliest_real_import = date_iso # We found a real import - we can stop here since events are chronologically ordered break elif event_type in self.REAL_IMPORT_EVENTS: _log("DEBUG", f"⚠️ Skipped import event: {reason}") # If we found a real import, no need to continue if earliest_real_import: break # If we got less than page size, we've seen all events if len(items) < page_size: break page += 1 _log("INFO", f"Processed {total_processed} events across {page-1} pages") if earliest_real_import: _log("INFO", f"✅ Using earliest real import: {earliest_real_import}") return earliest_real_import if first_grab: _log("WARNING", f"⚠️ No real imports found, using grab date: {first_grab}") return first_grab _log("ERROR", f"❌ No import or grab events found for movie_id {movie_id}") return None def movie_files(self, movie_id: int) -> List[Dict[str, Any]]: """Get movie files for a movie""" result = self._get("/api/v3/moviefile", {"movieId": movie_id}) return result if isinstance(result, list) else [] def earliest_file_dateadded(self, movie_id: int) -> Optional[str]: """Get earliest file dateAdded as fallback""" files = self.movie_files(movie_id) if not files: return None earliest = None for file_obj in files: date_added = file_obj.get("dateAdded") if not date_added: continue try: dt = datetime.fromisoformat(date_added.replace("Z", "+00:00")) if earliest is None or dt < earliest: earliest = dt except Exception: continue if earliest: return earliest.astimezone(timezone.utc).isoformat(timespec="seconds") return None def get_movie_import_date(self, movie_id: int, fallback_to_file_date: bool = True) -> Tuple[Optional[str], str]: """ Get the best import date for a movie. Returns: (date_iso, source_description) """ # Try history first import_date = self.earliest_import_event_optimized(movie_id) if import_date: return import_date, "radarr:history.import" # Fallback to file dateAdded if requested if fallback_to_file_date: file_date = self.earliest_file_dateadded(movie_id) if file_date: _log("WARNING", f"Using file dateAdded as fallback for movie_id {movie_id}") return file_date, "radarr:file.dateAdded" return None, "radarr:no_date_found" if __name__ == "__main__": # Test the client import os base_url = os.environ.get("RADARR_URL", "") api_key = os.environ.get("RADARR_API_KEY", "") if base_url and api_key: client = RadarrClient(base_url, api_key) # Test with a known movie test_imdb = "tt1596343" # Example movie = client.movie_by_imdb(test_imdb) if movie: movie_id = movie.get("id") print(f"Found movie: {movie.get('title')} (ID: {movie_id})") import_date, source = client.get_movie_import_date(movie_id) print(f"Import date: {import_date} ({source})") else: print(f"Movie not found: {test_imdb}") else: print("Please set RADARR_URL and RADARR_API_KEY environment variables for testing")