sbcrumb
2e6d8a35e7
Local Docker Build (Dev) / build-dev (push) Successful in 19s
When Sonarr lookup fails to find a series or episode data, the system now falls back to external APIs (TMDB, OMDb) to get episode air dates. This ensures episodes still get proper dateadded values even when not found in Sonarr. Changes: - Add ExternalClientManager integration to TVSeriesProcessor - Implement _get_episode_airdate_from_external_apis method - Support TMDB TV episode lookups with IMDb->TMDB ID conversion - Support OMDb episode season lookups with date parsing - Version bump to 2.0.3-external-fallback 🤖 Generated with [Claude Code](https://claude.ai/code) Co-Authored-By: Claude <noreply@anthropic.com>
329 lines
16 KiB
Python
329 lines
16 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
TV Series Processor - Clean implementation for TV episode processing
|
|
Handles manual scans and webhook processing with proper NFO filename matching
|
|
"""
|
|
from pathlib import Path
|
|
from typing import Optional, Dict, Any, List, Tuple
|
|
from datetime import datetime, timezone
|
|
import re
|
|
|
|
from core.database import NFOGuardDatabase
|
|
from core.episode_nfo_manager import EpisodeNFOManager
|
|
from core.nfo_manager import NFOManager
|
|
from core.path_mapper import PathMapper
|
|
from clients.sonarr_client import SonarrClient
|
|
from clients.external_clients import ExternalClientManager
|
|
from core.logging import _log, convert_utc_to_local
|
|
|
|
|
|
class TVSeriesProcessor:
|
|
"""Clean TV series processor with video filename matching"""
|
|
|
|
def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager,
|
|
path_mapper: PathMapper, sonarr_client: SonarrClient):
|
|
self.db = db
|
|
self.nfo_manager = nfo_manager # Keep for series/season NFOs
|
|
self.episode_nfo_manager = EpisodeNFOManager()
|
|
self.path_mapper = path_mapper
|
|
self.sonarr = sonarr_client
|
|
self.external_manager = ExternalClientManager()
|
|
|
|
def process_series_manual_scan(self, series_path: Path) -> bool:
|
|
"""Process a TV series during manual scan"""
|
|
_log("INFO", f"Processing TV series: {series_path.name}")
|
|
|
|
# Extract IMDb ID
|
|
imdb_id = self._extract_imdb_id(series_path)
|
|
if not imdb_id:
|
|
_log("ERROR", f"No IMDb ID found for series: {series_path.name}")
|
|
return False
|
|
|
|
# Find all episodes on disk
|
|
episodes_on_disk = self._find_episodes_on_disk(series_path)
|
|
if not episodes_on_disk:
|
|
_log("WARNING", f"No episodes found on disk for: {series_path.name}")
|
|
return False
|
|
|
|
_log("INFO", f"Found {len(episodes_on_disk)} episodes on disk")
|
|
|
|
# Process each episode
|
|
episodes_processed = 0
|
|
for (season_num, episode_num), video_files in episodes_on_disk.items():
|
|
if self._process_episode_manual_scan(series_path, imdb_id, season_num, episode_num):
|
|
episodes_processed += 1
|
|
|
|
# Create series-level NFOs if any episodes were processed
|
|
if episodes_processed > 0:
|
|
self._create_series_nfos(series_path, imdb_id)
|
|
|
|
_log("INFO", f"Completed processing series: {series_path.name} ({episodes_processed} episodes)")
|
|
return episodes_processed > 0
|
|
|
|
def process_episode_webhook(self, webhook_data: Dict[str, Any]) -> bool:
|
|
"""Process a single episode from Sonarr webhook"""
|
|
# TODO: Parse webhook data and extract episode info
|
|
# This will be implemented when we add webhook support
|
|
pass
|
|
|
|
def _extract_imdb_id(self, series_path: Path) -> Optional[str]:
|
|
"""Extract IMDb ID from series directory or files"""
|
|
# Try directory name first
|
|
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
|
|
if imdb_id:
|
|
return imdb_id
|
|
|
|
# Try tvshow.nfo if it exists
|
|
tvshow_nfo = series_path / "tvshow.nfo"
|
|
if tvshow_nfo.exists():
|
|
imdb_id = self.nfo_manager.parse_imdb_from_nfo(tvshow_nfo)
|
|
if imdb_id:
|
|
return imdb_id
|
|
|
|
# Try any existing episode NFO files
|
|
for season_dir in series_path.iterdir():
|
|
if season_dir.is_dir() and self._is_season_directory(season_dir.name):
|
|
for nfo_file in season_dir.glob("*.nfo"):
|
|
imdb_id = self.nfo_manager.parse_imdb_from_nfo(nfo_file)
|
|
if imdb_id:
|
|
return imdb_id
|
|
|
|
return None
|
|
|
|
def _is_season_directory(self, dirname: str) -> bool:
|
|
"""Check if directory name matches season pattern"""
|
|
return bool(re.match(r'^[Ss]eason\s+\d+$', dirname, re.IGNORECASE))
|
|
|
|
def _find_episodes_on_disk(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
|
|
"""Find all episodes on disk, grouped by (season, episode)"""
|
|
episodes = {}
|
|
|
|
for season_dir in series_path.iterdir():
|
|
if season_dir.is_dir() and self._is_season_directory(season_dir.name):
|
|
season_num = self._extract_season_number(season_dir.name)
|
|
if season_num is None:
|
|
continue
|
|
|
|
# Find video files in this season
|
|
season_episodes = self.episode_nfo_manager.find_video_files_for_season(season_dir)
|
|
|
|
# Add season directory info to episodes
|
|
for (s_num, e_num), video_files in season_episodes.items():
|
|
if s_num == season_num: # Verify season matches directory
|
|
episodes[(s_num, e_num)] = video_files
|
|
|
|
return episodes
|
|
|
|
def _extract_season_number(self, dirname: str) -> Optional[int]:
|
|
"""Extract season number from directory name"""
|
|
match = re.search(r'[Ss]eason\s+(\d+)', dirname, re.IGNORECASE)
|
|
if match:
|
|
return int(match.group(1))
|
|
return None
|
|
|
|
def _process_episode_manual_scan(self, series_path: Path, imdb_id: str,
|
|
season_num: int, episode_num: int) -> bool:
|
|
"""Process a single episode during manual scan"""
|
|
season_dir = series_path / f"Season {season_num:02d}"
|
|
if not season_dir.exists():
|
|
# Try alternate format
|
|
season_dir = series_path / f"Season {season_num}"
|
|
if not season_dir.exists():
|
|
_log("ERROR", f"Season directory not found for S{season_num:02d}E{episode_num:02d}")
|
|
return False
|
|
|
|
_log("DEBUG", f"Processing episode S{season_num:02d}E{episode_num:02d}")
|
|
|
|
# Step 1: Check for existing NFOGuard data
|
|
existing_nfo = self.episode_nfo_manager.find_nfo_for_episode(season_dir, season_num, episode_num)
|
|
if existing_nfo:
|
|
nfo_data = self.episode_nfo_manager.extract_nfoguard_data(existing_nfo)
|
|
if nfo_data:
|
|
# Verify against database
|
|
db_data = self.db.get_episode_date(imdb_id, season_num, episode_num)
|
|
if db_data and db_data.get("dateadded") == nfo_data.get("dateadded"):
|
|
_log("DEBUG", f"Episode S{season_num:02d}E{episode_num:02d} already up to date")
|
|
# Still migrate filename if needed
|
|
self.episode_nfo_manager.migrate_nfo_to_video_filename(season_dir, season_num, episode_num)
|
|
return True
|
|
|
|
# Step 2: Check database
|
|
db_data = self.db.get_episode_date(imdb_id, season_num, episode_num)
|
|
if db_data and db_data.get("dateadded"):
|
|
_log("DEBUG", f"Using database data for S{season_num:02d}E{episode_num:02d}")
|
|
aired = db_data.get("aired")
|
|
dateadded = db_data.get("dateadded")
|
|
source = db_data.get("source", "database")
|
|
else:
|
|
# Step 3: Query Sonarr for episode data
|
|
aired, dateadded, source = self._get_episode_dates_from_sonarr(imdb_id, season_num, episode_num)
|
|
|
|
# Step 4: Create/update NFO and database
|
|
if dateadded:
|
|
# Get episode metadata for title/plot
|
|
title, plot = self._get_episode_metadata_from_sonarr(imdb_id, season_num, episode_num)
|
|
|
|
# Create/update NFO with video filename
|
|
success = self.episode_nfo_manager.create_episode_nfo(
|
|
season_dir, season_num, episode_num, aired, dateadded, source, title, plot
|
|
)
|
|
|
|
if success:
|
|
# Update database
|
|
self.db.upsert_episode_date(imdb_id, season_num, episode_num, aired, dateadded, source, True)
|
|
return True
|
|
|
|
_log("WARNING", f"Could not get dates for episode S{season_num:02d}E{episode_num:02d}")
|
|
return False
|
|
|
|
def _get_episode_dates_from_sonarr(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], Optional[str], str]:
|
|
"""Get episode dates from Sonarr API with proper fallback"""
|
|
aired = None
|
|
dateadded = None
|
|
source = "no_data_found"
|
|
|
|
if not self.sonarr.enabled:
|
|
_log("WARNING", "Sonarr not enabled, cannot get episode dates")
|
|
return aired, dateadded, source
|
|
|
|
try:
|
|
# Find series in Sonarr
|
|
series = self.sonarr.series_by_imdb(imdb_id)
|
|
if not series:
|
|
_log("WARNING", f"Series not found in Sonarr for IMDb: {imdb_id}")
|
|
return aired, dateadded, source
|
|
|
|
# Get episodes for series
|
|
episodes = self.sonarr.episodes_for_series(series["id"])
|
|
target_episode = None
|
|
|
|
for episode in episodes:
|
|
if (episode.get("seasonNumber") == season_num and
|
|
episode.get("episodeNumber") == episode_num):
|
|
target_episode = episode
|
|
break
|
|
|
|
if not target_episode:
|
|
_log("WARNING", f"Episode S{season_num:02d}E{episode_num:02d} not found in Sonarr")
|
|
return aired, dateadded, source
|
|
|
|
# Get airdate
|
|
aired = target_episode.get("airDateUtc")
|
|
|
|
# Try to get import history
|
|
episode_id = target_episode.get("id")
|
|
if episode_id:
|
|
import_date = self.sonarr.get_episode_import_history(episode_id)
|
|
if import_date:
|
|
dateadded = convert_utc_to_local(import_date)
|
|
source = "sonarr:history.import"
|
|
_log("INFO", f"Found import date for S{season_num:02d}E{episode_num:02d}: {dateadded}")
|
|
return aired, dateadded, source
|
|
|
|
# Fallback to airdate if no import history
|
|
if aired:
|
|
dateadded = convert_utc_to_local(aired)
|
|
source = "sonarr:episode.airDateUtc"
|
|
_log("WARNING", f"No import history for S{season_num:02d}E{episode_num:02d}, using airdate: {dateadded}")
|
|
return aired, dateadded, source
|
|
|
|
except Exception as e:
|
|
_log("ERROR", f"Sonarr API error for S{season_num:02d}E{episode_num:02d}: {e}")
|
|
|
|
# Try external APIs for episode airdate
|
|
_log("INFO", f"Trying external APIs for episode S{season_num:02d}E{episode_num:02d} airdate")
|
|
aired, source = self._get_episode_airdate_from_external_apis(imdb_id, season_num, episode_num)
|
|
if aired:
|
|
dateadded = convert_utc_to_local(aired)
|
|
return aired, dateadded, source
|
|
|
|
_log("ERROR", f"Could not get any date information for S{season_num:02d}E{episode_num:02d}")
|
|
return aired, dateadded, source
|
|
|
|
def _get_episode_metadata_from_sonarr(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], Optional[str]]:
|
|
"""Get episode title and plot from Sonarr"""
|
|
if not self.sonarr.enabled:
|
|
return None, None
|
|
|
|
try:
|
|
series = self.sonarr.series_by_imdb(imdb_id)
|
|
if series:
|
|
episodes = self.sonarr.episodes_for_series(series["id"])
|
|
for episode in episodes:
|
|
if (episode.get("seasonNumber") == season_num and
|
|
episode.get("episodeNumber") == episode_num):
|
|
title = episode.get("title")
|
|
plot = episode.get("overview")
|
|
return title, plot
|
|
except Exception as e:
|
|
_log("DEBUG", f"Could not get metadata for S{season_num:02d}E{episode_num:02d}: {e}")
|
|
|
|
return None, None
|
|
|
|
def _get_episode_airdate_from_external_apis(self, imdb_id: str, season_num: int, episode_num: int) -> Tuple[Optional[str], str]:
|
|
"""Get episode airdate from external APIs (TMDB, OMDb) as fallback"""
|
|
# Try TMDB first
|
|
if self.external_manager.tmdb.enabled:
|
|
try:
|
|
_log("DEBUG", f"Trying TMDB for episode S{season_num:02d}E{episode_num:02d} airdate")
|
|
|
|
# First convert IMDb to TMDB TV ID
|
|
tv_search = self.external_manager.tmdb._get(f"/find/{imdb_id}", {"external_source": "imdb_id"})
|
|
if tv_search and tv_search.get("tv_results"):
|
|
tv_id = tv_search["tv_results"][0].get("id")
|
|
if tv_id:
|
|
_log("DEBUG", f"Found TMDB TV ID {tv_id} for {imdb_id}")
|
|
|
|
# Get episode details
|
|
episode_data = self.external_manager.tmdb._get(f"/tv/{tv_id}/season/{season_num}/episode/{episode_num}")
|
|
if episode_data and episode_data.get("air_date"):
|
|
airdate = episode_data["air_date"]
|
|
# Convert to ISO format with UTC timezone
|
|
iso_airdate = f"{airdate}T00:00:00Z"
|
|
_log("INFO", f"Found TMDB airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}")
|
|
return iso_airdate, "tmdb:episode.air_date"
|
|
except Exception as e:
|
|
_log("WARNING", f"TMDB episode lookup failed for S{season_num:02d}E{episode_num:02d}: {e}")
|
|
|
|
# Try OMDb as fallback
|
|
if self.external_manager.omdb.enabled:
|
|
try:
|
|
_log("DEBUG", f"Trying OMDb for episode S{season_num:02d}E{episode_num:02d} airdate")
|
|
episode_dates = self.external_manager.omdb.get_tv_season_episodes(imdb_id, season_num)
|
|
if episode_num in episode_dates:
|
|
airdate = episode_dates[episode_num]
|
|
# Convert to ISO format
|
|
from datetime import datetime, timezone
|
|
try:
|
|
# Try to parse OMDb date format (usually DD MMM YYYY)
|
|
dt = datetime.strptime(airdate, "%d %b %Y").replace(tzinfo=timezone.utc)
|
|
iso_airdate = dt.isoformat(timespec="seconds")
|
|
_log("INFO", f"Found OMDb airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}")
|
|
return iso_airdate, "omdb:episode.released"
|
|
except ValueError:
|
|
# Try other common formats
|
|
for fmt in ["%Y-%m-%d", "%d %B %Y"]:
|
|
try:
|
|
dt = datetime.strptime(airdate, fmt).replace(tzinfo=timezone.utc)
|
|
iso_airdate = dt.isoformat(timespec="seconds")
|
|
_log("INFO", f"Found OMDb airdate for S{season_num:02d}E{episode_num:02d}: {iso_airdate}")
|
|
return iso_airdate, "omdb:episode.released"
|
|
except ValueError:
|
|
continue
|
|
except Exception as e:
|
|
_log("WARNING", f"OMDb episode lookup failed for S{season_num:02d}E{episode_num:02d}: {e}")
|
|
|
|
_log("WARNING", f"No external API airdate found for S{season_num:02d}E{episode_num:02d}")
|
|
return None, "no_external_data"
|
|
|
|
def _create_series_nfos(self, series_path: Path, imdb_id: str):
|
|
"""Create tvshow.nfo and season.nfo files"""
|
|
# Create tvshow.nfo
|
|
self.nfo_manager.create_tvshow_nfo(series_path, imdb_id)
|
|
|
|
# Create season.nfo for each season directory
|
|
for season_dir in series_path.iterdir():
|
|
if season_dir.is_dir() and self._is_season_directory(season_dir.name):
|
|
season_num = self._extract_season_number(season_dir.name)
|
|
if season_num is not None:
|
|
self.nfo_manager.create_season_nfo(season_dir, season_num) |