400 lines
17 KiB
Python
400 lines
17 KiB
Python
#!/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: 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
|
|
|
|
# Analyze for real import
|
|
is_real, reason, date_iso = self._analyze_event_for_import(event, movie_info)
|
|
|
|
if is_real and date_iso:
|
|
# Extract all possible identifiers
|
|
source_path = (event.get("data", {}).get("sourcePath", "") or
|
|
event.get("data", {}).get("droppedPath", "") or
|
|
event.get("sourcePath", "") or "").lower()
|
|
source_title = (event.get("data", {}).get("sourceTitle", "") or
|
|
event.get("sourceTitle", "") or "").lower()
|
|
|
|
# Get movie identifiers
|
|
movie_imdb = (movie_info.get("imdbId", "") or "").lower()
|
|
movie_title = (movie_info.get("title", "") or "").lower()
|
|
|
|
_log("DEBUG", f"Validating import - Path: {source_path}")
|
|
_log("DEBUG", f"Validating import - Title: {source_title}")
|
|
_log("DEBUG", f"Validating against - IMDb: {movie_imdb}, Title: {movie_title}")
|
|
|
|
# More flexible validation
|
|
valid = False
|
|
if movie_imdb and (movie_imdb in source_path or movie_imdb in source_title):
|
|
valid = True
|
|
_log("DEBUG", "✅ Validated by IMDb ID")
|
|
elif movie_title and (movie_title in source_path or movie_title in source_title):
|
|
valid = True
|
|
_log("DEBUG", "✅ Validated by title")
|
|
|
|
if valid:
|
|
_log("INFO", f"✅ FOUND REAL IMPORT: {event_type} at {date_iso} - {reason} (path validated)")
|
|
earliest_real_import = date_iso
|
|
break # Stop processing more pages
|
|
else:
|
|
_log("DEBUG", f"⚠️ Skipping import - path validation failed")
|
|
# 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") |