#!/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" def is_download_path(self, path): return "/downloads/" in str(path).lower() or "/completed/" in str(path).lower() 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 = { # Primary events to check first "moviefileimported", "importevent", # Secondary events "downloadfolderimported", "imported", "downloadimported", "fileimported", "movieimported", # Legacy events "downloadedepisode", "downloadedmovie", } # Events that should be checked first PRIMARY_IMPORT_EVENTS = { "moviefileimported", "importevent" } # 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: Direct lookup by ID movie = self._get(f"/api/v3/movie/lookup/imdb/{imdb_id}") if isinstance(movie, dict) and movie.get("imdbId", "").lower() == imdb_id.lower(): _log("INFO", f"Found via direct lookup: {movie.get('title')} (ID: {movie.get('id')})") return movie # Method 2: Get all movies and filter by IMDb (fallback) 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], movie_info: Dict[str, Any] = None) -> Tuple[bool, str, Optional[str]]: """ Analyze a history event to determine if it's a real import. Args: event: The history event to analyze movie_info: Optional movie information to validate paths against 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 # Get all possible source paths/titles source_items = [] # Get both sourcePath and importedPath if available if event_data: for key in ['sourcePath', 'droppedPath', 'path', 'sourceTitle', 'importedPath']: if event_data.get(key): source_items.append(event_data[key]) # Also check event root for these fields for key in ['sourcePath', 'sourceTitle', 'importedPath']: if event.get(key): source_items.append(event[key]) # Clean up and make unique source_items = [str(s).lower().strip() for s in source_items if s] source_items = list(set(source_items)) # Remove duplicates if not source_items: return False, "no_source_paths", date_iso # If we have movie info, look for title/year match if movie_info: movie_title = movie_info.get('title', '').lower().replace(':', '.').replace(' ', '.') movie_year = str(movie_info.get('year', '')) for source in source_items: # Clean up source text for comparison source_clean = source.replace(' ', '.').replace('_', '.').replace('-', '.') # Check if both title and year are in the source if movie_title and movie_year: if movie_title in source_clean and movie_year in source_clean: _log("DEBUG", f"✅ Match found - Title: {movie_title}, Year: {movie_year}") return True, "matched_title_and_year", date_iso # Also check for downloads path as secondary validation if path_mapper.is_download_path(source): _log("DEBUG", f"Source is from downloads: {source}") return True, "from_downloads_path", date_iso _log("DEBUG", f"⚠️ No match found in sources: {source_items}") return False, "no_title_year_match", date_iso # Fallback to basic path validation if no movie info for source in source_items: if path_mapper.is_download_path(source): return True, "basic_download_path_match", date_iso return False, "no_download_path_match", 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}") # Get movie info for path validation movie_info = self._get(f"/api/v3/movie/{movie_id}") if not movie_info or not isinstance(movie_info, dict): _log("ERROR", f"Could not get movie info for ID {movie_id}") return None 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 # For moviefileimported events, handle specially if event_type == "moviefileimported": imported_path = (event.get("data", {}).get("importedPath", "") or event.get("data", {}).get("path", "") or event.get("importedPath", "")).lower() if imported_path: _log("DEBUG", f"Found moviefileimported event with path: {imported_path}") # Check for IMDb ID match movie_imdb = (movie_info.get("imdbId", "") or "").lower() if movie_imdb and ( f"[imdb-{movie_imdb}]" in imported_path or f"[{movie_imdb}]" in imported_path ): date_iso = datetime.fromisoformat(event["date"].replace("Z", "+00:00")).astimezone(timezone.utc).isoformat(timespec="seconds") _log("INFO", f"✅ FOUND IMPORT: exact IMDb match at {date_iso}") earliest_real_import = date_iso break # Stop processing more pages # Fallback to normal import analysis is_real, reason, date_iso = self._analyze_event_for_import(event, movie_info) if is_real and date_iso: source_path = (event.get("data", {}).get("sourcePath", "") or event.get("data", {}).get("droppedPath", "") or event.get("sourcePath", "") or event.get("data", {}).get("importedPath", "") or event.get("importedPath", "") or "").lower() if source_path: # Check for path match movie_imdb = (movie_info.get("imdbId", "") or "").lower() if movie_imdb and ( f"[imdb-{movie_imdb}]" in source_path or f"[{movie_imdb}]" in source_path or movie_imdb in source_path ): _log("INFO", f"✅ FOUND IMPORT: IMDb match in path at {date_iso}") earliest_real_import = date_iso break # Check for title/year match movie_title = (movie_info.get("title", "") or "").lower().replace(":", ".").replace(" ", ".") movie_year = str(movie_info.get("year", "")) if movie_title and movie_year and movie_title in source_path and movie_year in source_path: _log("INFO", f"✅ FOUND IMPORT: Title/year match at {date_iso}") earliest_real_import = date_iso 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")