initial updates to mirror github files
This commit is contained in:
@@ -0,0 +1,286 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Enhanced Sonarr API client for TV show metadata and episode management
|
||||
"""
|
||||
import json
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Dict, Any, List, Optional
|
||||
from urllib.parse import urlencode
|
||||
from urllib.request import Request as UrlRequest, urlopen
|
||||
from urllib.error import URLError, HTTPError
|
||||
|
||||
from core.logging import _log
|
||||
|
||||
|
||||
class SonarrClient:
|
||||
"""Enhanced Sonarr API client for TV series and episode management"""
|
||||
|
||||
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)
|
||||
self.enabled = bool(self.base_url and self.api_key)
|
||||
|
||||
def _get(self, path: str, params: Dict[str, Any] = None) -> Optional[Any]:
|
||||
"""Make GET request to Sonarr API with retries"""
|
||||
if not self.enabled:
|
||||
return None
|
||||
|
||||
url = f"{self.base_url}/api/v3{path}"
|
||||
if params:
|
||||
url += "?" + urlencode(params)
|
||||
|
||||
headers = {"X-Api-Key": self.api_key}
|
||||
|
||||
for attempt in range(self.retries):
|
||||
try:
|
||||
_log("DEBUG", f"Sonarr API Request: {url}")
|
||||
req = UrlRequest(url, headers=headers)
|
||||
|
||||
with urlopen(req, timeout=self.timeout) as resp:
|
||||
data = resp.read().decode("utf-8")
|
||||
result = json.loads(data) if data else None
|
||||
return result
|
||||
|
||||
except HTTPError as e:
|
||||
if e.code == 401:
|
||||
_log("ERROR", "Sonarr authentication failed - check API key")
|
||||
return None
|
||||
elif e.code == 429:
|
||||
wait_time = (attempt + 1) * 2
|
||||
_log("WARNING", f"Sonarr rate limited, waiting {wait_time}s (attempt {attempt+1}/{self.retries})")
|
||||
time.sleep(wait_time)
|
||||
else:
|
||||
_log("WARNING", f"Sonarr HTTP {e.code} error on attempt {attempt+1}/{self.retries}: {e.reason}")
|
||||
|
||||
except Exception as e:
|
||||
_log("WARNING", f"Sonarr API attempt {attempt+1}/{self.retries} failed: {e}")
|
||||
|
||||
if attempt < self.retries - 1:
|
||||
time.sleep(0.5 * (attempt + 1))
|
||||
|
||||
_log("ERROR", f"Sonarr API failed after {self.retries} attempts: {url}")
|
||||
return None
|
||||
|
||||
def series_by_imdb(self, imdb_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""Find series by IMDb ID using lookup endpoint"""
|
||||
search_term = f"imdbid:{imdb_id}"
|
||||
_log("DEBUG", f"Searching Sonarr with term: {search_term}")
|
||||
|
||||
result = self._get("/series/lookup", {"term": search_term})
|
||||
if not result:
|
||||
_log("WARNING", f"No results from Sonarr lookup for: {search_term}")
|
||||
return None
|
||||
|
||||
_log("DEBUG", f"Sonarr lookup returned {len(result)} results")
|
||||
|
||||
# Log all results for debugging
|
||||
for i, series in enumerate(result):
|
||||
series_imdb = series.get("imdbId", "")
|
||||
series_title = series.get("title", "")
|
||||
series_id = series.get("id", "")
|
||||
_log("DEBUG", f"Result {i+1}: Title='{series_title}', IMDb='{series_imdb}', ID={series_id}")
|
||||
|
||||
# Find exact IMDb match (case insensitive)
|
||||
target_imdb = imdb_id.lower()
|
||||
for series in result:
|
||||
series_imdb = (series.get("imdbId") or "").lower()
|
||||
if series_imdb == target_imdb:
|
||||
_log("INFO", f"Found exact IMDb match: {series.get('title')} (ID: {series.get('id')})")
|
||||
return series
|
||||
|
||||
# Try partial match as fallback
|
||||
for series in result:
|
||||
series_imdb = (series.get("imdbId") or "").lower()
|
||||
if target_imdb in series_imdb or series_imdb in target_imdb:
|
||||
_log("WARNING", f"Found partial IMDb match: {series.get('title')} (Expected: {imdb_id}, Found: {series.get('imdbId')})")
|
||||
return series
|
||||
|
||||
_log("WARNING", f"No IMDb match found in {len(result)} results for {imdb_id}")
|
||||
return None
|
||||
|
||||
def series_by_title(self, title: str) -> Optional[Dict[str, Any]]:
|
||||
"""Search for series by title as fallback when IMDb lookup fails"""
|
||||
_log("DEBUG", f"Searching Sonarr by title: {title}")
|
||||
|
||||
result = self._get("/series/lookup", {"term": title})
|
||||
if not result:
|
||||
_log("WARNING", f"No results from Sonarr title search for: {title}")
|
||||
return None
|
||||
|
||||
_log("DEBUG", f"Sonarr title search returned {len(result)} results")
|
||||
|
||||
title_lower = title.lower()
|
||||
|
||||
# Look for exact title match
|
||||
for series in result:
|
||||
series_title = (series.get("title") or "").lower()
|
||||
if series_title == title_lower:
|
||||
_log("INFO", f"Found exact title match: {series.get('title')} (ID: {series.get('id')})")
|
||||
return series
|
||||
|
||||
# Look for partial title match
|
||||
for series in result:
|
||||
series_title = (series.get("title") or "").lower()
|
||||
if title_lower in series_title or series_title in title_lower:
|
||||
_log("INFO", f"Found partial title match: '{series.get('title')}' for search '{title}' (ID: {series.get('id')})")
|
||||
return series
|
||||
|
||||
_log("WARNING", f"No title match found for: {title}")
|
||||
return None
|
||||
|
||||
def get_all_series(self) -> List[Dict[str, Any]]:
|
||||
"""Get all series from Sonarr"""
|
||||
return self._get("/series") or []
|
||||
|
||||
def series_by_imdb_direct(self, imdb_id: str) -> Optional[Dict[str, Any]]:
|
||||
"""Find series by scanning all series for IMDb match (slower but more reliable)"""
|
||||
_log("DEBUG", f"Direct series lookup for IMDb: {imdb_id}")
|
||||
all_series = self.get_all_series()
|
||||
|
||||
target_imdb = imdb_id.lower()
|
||||
for series in all_series:
|
||||
series_imdb = (series.get("imdbId") or "").lower()
|
||||
if series_imdb == target_imdb:
|
||||
_log("INFO", f"Found series via direct lookup: {series.get('title')} (ID: {series.get('id')})")
|
||||
return series
|
||||
|
||||
_log("WARNING", f"No series found with IMDb ID via direct lookup: {imdb_id}")
|
||||
return None
|
||||
|
||||
def episodes_for_series(self, series_id: int) -> List[Dict[str, Any]]:
|
||||
"""Get all episodes for a series"""
|
||||
return self._get("/episode", {"seriesId": series_id}) or []
|
||||
|
||||
def episode_file(self, episode_file_id: int) -> Optional[Dict[str, Any]]:
|
||||
"""Get episode file details"""
|
||||
return self._get(f"/episodefile/{episode_file_id}")
|
||||
|
||||
def get_episode_import_history(self, episode_id: int) -> Optional[str]:
|
||||
"""
|
||||
Get the original import date from history with enhanced detection.
|
||||
Focuses on finding the earliest REAL import, not upgrades.
|
||||
"""
|
||||
all_records = []
|
||||
page = 1
|
||||
page_size = 100
|
||||
|
||||
# Collect all history records for this episode
|
||||
while True:
|
||||
history = self._get("/history", {
|
||||
"episodeId": episode_id,
|
||||
"sortKey": "date",
|
||||
"sortDir": "asc",
|
||||
"page": page,
|
||||
"pageSize": page_size
|
||||
})
|
||||
|
||||
if not history:
|
||||
break
|
||||
|
||||
records = history.get("records", []) if isinstance(history, dict) else []
|
||||
if not records:
|
||||
break
|
||||
|
||||
all_records.extend(records)
|
||||
|
||||
if len(records) < page_size:
|
||||
break
|
||||
|
||||
page += 1
|
||||
if page > 10: # Safety valve
|
||||
break
|
||||
|
||||
_log("DEBUG", f"Got {len(all_records)} history records for episode {episode_id}")
|
||||
|
||||
# Categorize events
|
||||
import_events = []
|
||||
grabbed_events = []
|
||||
rename_events = []
|
||||
|
||||
for event in all_records:
|
||||
event_type = event.get("eventType", "").lower()
|
||||
date = event.get("date")
|
||||
|
||||
if not date:
|
||||
continue
|
||||
|
||||
_log("DEBUG", f"History event: {event_type} at {date}")
|
||||
|
||||
if event_type == "downloadfolderimported":
|
||||
import_events.append({"date": date, "event": event})
|
||||
elif event_type == "grabbed":
|
||||
grabbed_events.append({"date": date, "event": event})
|
||||
elif event_type == "episodefilerenamed":
|
||||
rename_events.append({"date": date, "event": event})
|
||||
|
||||
# Use the earliest real import event
|
||||
if import_events:
|
||||
earliest_import = min(import_events, key=lambda x: x["date"])
|
||||
import_date = earliest_import["date"]
|
||||
_log("INFO", f"Found import date: {import_date} for episode {episode_id}")
|
||||
|
||||
# Check if this looks like an upgrade by comparing to renames
|
||||
if rename_events:
|
||||
earliest_rename = min(rename_events, key=lambda x: x["date"])
|
||||
rename_date = earliest_rename["date"]
|
||||
|
||||
try:
|
||||
import_dt = datetime.fromisoformat(import_date.replace("Z", "+00:00"))
|
||||
rename_dt = datetime.fromisoformat(rename_date.replace("Z", "+00:00"))
|
||||
days_diff = (import_dt - rename_dt).days
|
||||
|
||||
# If import is significantly after rename, prefer rename date
|
||||
if days_diff > 30:
|
||||
_log("WARNING", f"Import {import_date} is {days_diff} days after rename {rename_date} - using rename date")
|
||||
return rename_date
|
||||
|
||||
except Exception as e:
|
||||
_log("DEBUG", f"Error comparing dates: {e}")
|
||||
|
||||
return import_date
|
||||
|
||||
# Fallback to grab event
|
||||
if grabbed_events:
|
||||
earliest_grab = min(grabbed_events, key=lambda x: x["date"])
|
||||
_log("WARNING", f"No import events, using grab date: {earliest_grab['date']} for episode {episode_id}")
|
||||
return earliest_grab["date"]
|
||||
|
||||
_log("WARNING", f"No reliable import events found for episode {episode_id}")
|
||||
return None
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Test the client
|
||||
import os
|
||||
|
||||
base_url = os.environ.get("SONARR_URL", "")
|
||||
api_key = os.environ.get("SONARR_API_KEY", "")
|
||||
|
||||
if base_url and api_key:
|
||||
client = SonarrClient(base_url, api_key)
|
||||
|
||||
# Test with a known series
|
||||
test_imdb = "tt2085059" # Example
|
||||
series = client.series_by_imdb(test_imdb)
|
||||
|
||||
if series:
|
||||
series_id = series.get("id")
|
||||
print(f"Found series: {series.get('title')} (ID: {series_id})")
|
||||
|
||||
# Get episodes
|
||||
episodes = client.episodes_for_series(series_id)
|
||||
print(f"Found {len(episodes)} episodes")
|
||||
|
||||
# Test import date for first episode
|
||||
if episodes:
|
||||
first_episode = episodes[0]
|
||||
episode_id = first_episode.get("id")
|
||||
import_date = client.get_episode_import_history(episode_id)
|
||||
print(f"First episode import date: {import_date}")
|
||||
else:
|
||||
print(f"Series not found: {test_imdb}")
|
||||
else:
|
||||
print("Please set SONARR_URL and SONARR_API_KEY environment variables for testing")
|
||||
Reference in New Issue
Block a user