Files
nfoguard/nfoguard.py
T
2025-09-10 08:29:04 -04:00

1451 lines
60 KiB
Python

#!/usr/bin/env python3
"""
NFOGuard - Consolidated webhook server for TV and Movie import date management
Replaces webhook_server.py and media_date_cache.py with modular architecture
"""
import os
import json
import asyncio
import glob
import re
import logging
import logging.handlers
from pathlib import Path
from datetime import datetime, timezone, timedelta
from fastapi import FastAPI, HTTPException, BackgroundTasks, Request
from pydantic import BaseModel
from typing import Optional, Dict, Any, List, Set, Tuple
import threading
from concurrent.futures import ThreadPoolExecutor
import uvicorn
# Import our new modular components
from core.database import NFOGuardDatabase
from core.nfo_manager import NFOManager
from core.path_mapper import PathMapper
from clients.radarr_client import RadarrClient
from clients.sonarr_client import SonarrClient
from clients.external_clients import ExternalClientManager
# ---------------------------
# Configuration & Logging
# ---------------------------
def _setup_file_logging():
"""Setup file logging for NFOGuard"""
log_dir = Path("/app/data/logs")
log_dir.mkdir(parents=True, exist_ok=True)
logger = logging.getLogger("NFOGuard")
logger.setLevel(logging.DEBUG)
file_handler = logging.handlers.RotatingFileHandler(
log_dir / "nfoguard.log", maxBytes=50*1024*1024, backupCount=3
)
formatter = logging.Formatter(
'[%(asctime)s] %(levelname)s: %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S+00:00'
)
file_handler.setFormatter(formatter)
logger.addHandler(file_handler)
return logger
def _mask_sensitive_data(msg: str) -> str:
"""Mask API keys and other sensitive data in log messages"""
import re
# List of patterns to mask
sensitive_patterns = [
(r'api_key=([a-zA-Z0-9_\-]+)', r'api_key=***masked***'),
(r'password=([^\s&]+)', r'password=***masked***'),
(r'token=([a-zA-Z0-9_\-]+)', r'token=***masked***'),
(r'key=([a-zA-Z0-9_\-]{8,})', r'key=***masked***'), # Keys longer than 8 chars
(r'([a-zA-Z0-9]{32,})', lambda m: m.group(1)[:8] + '***masked***' if len(m.group(1)) > 16 else m.group(1)) # Long strings likely to be keys
]
masked_msg = msg
for pattern, replacement in sensitive_patterns:
if isinstance(replacement, str):
masked_msg = re.sub(pattern, replacement, masked_msg, flags=re.IGNORECASE)
else:
masked_msg = re.sub(pattern, replacement, masked_msg, flags=re.IGNORECASE)
return masked_msg
def _log(level: str, msg: str):
"""Enhanced logging that writes to both console and file with sensitive data masking"""
masked_msg = _mask_sensitive_data(msg)
print(f"[{datetime.now(timezone.utc).isoformat(timespec='seconds')}] {level}: {masked_msg}")
try:
file_logger = _setup_file_logging()
getattr(file_logger, level.lower(), file_logger.info)(masked_msg)
except Exception as e:
print(f"File logging error: {e}")
# Initialize logging
_setup_file_logging()
def _load_environment_files():
"""Load environment variables from .env and optionally .env.secrets"""
from pathlib import Path
# Try to load from python-dotenv if available
try:
from dotenv import load_dotenv
# Load main .env file
env_file = Path(".env")
if env_file.exists():
load_dotenv(env_file)
_log("INFO", f"Loaded environment from {env_file}")
# Load secrets file if it exists
secrets_file = Path(".env.secrets")
if secrets_file.exists():
load_dotenv(secrets_file)
_log("INFO", f"Loaded secrets from {secrets_file}")
except ImportError:
_log("WARNING", "python-dotenv not available - environment files not loaded")
# Load environment files at startup
_load_environment_files()
def _bool_env(name: str, default: bool) -> bool:
v = os.environ.get(name)
if v is None:
return default
return v.lower() in ("1", "true", "yes", "y", "on")
# ---------------------------
# Configuration
# ---------------------------
class NFOGuardConfig:
def __init__(self):
# Paths
self.tv_paths = [Path(p.strip()) for p in os.environ.get("TV_PATHS", "/media/tv,/media/tv6").split(",") if p.strip()]
self.movie_paths = [Path(p.strip()) for p in os.environ.get("MOVIE_PATHS", "/media/movies,/media/movies6").split(",") if p.strip()]
# Core settings
self.manage_nfo = _bool_env("MANAGE_NFO", True)
self.fix_dir_mtimes = _bool_env("FIX_DIR_MTIMES", True)
self.lock_metadata = _bool_env("LOCK_METADATA", True)
self.debug = _bool_env("DEBUG", False)
self.manager_brand = os.environ.get("MANAGER_BRAND", "NFOGuard")
# Batching
self.batch_delay = float(os.environ.get("BATCH_DELAY", "5.0"))
self.max_concurrent = int(os.environ.get("MAX_CONCURRENT_SERIES", "3"))
# Database
self.db_path = Path(os.environ.get("DB_PATH", "/app/data/media_dates.db"))
# Movie processing
self.movie_priority = os.environ.get("MOVIE_PRIORITY", "import_then_digital").lower()
self.prefer_release_dates_over_file_dates = _bool_env("PREFER_RELEASE_DATES_OVER_FILE_DATES", True)
self.allow_file_date_fallback = _bool_env("ALLOW_FILE_DATE_FALLBACK", False)
self.release_date_priority = [p.strip() for p in os.environ.get("RELEASE_DATE_PRIORITY", "digital,physical,theatrical").split(",")]
self.enable_smart_date_validation = _bool_env("ENABLE_SMART_DATE_VALIDATION", True)
self.max_release_date_gap_years = int(os.environ.get("MAX_RELEASE_DATE_GAP_YEARS", "10"))
self.movie_poll_mode = os.environ.get("MOVIE_POLL_MODE", "always").lower()
self.movie_update_mode = os.environ.get("MOVIE_DATE_UPDATE_MODE", "backfill_only").lower()
# TV processing
self.tv_season_dir_format = os.environ.get("TV_SEASON_DIR_FORMAT", "Season {season:02d}")
self.tv_season_dir_pattern = os.environ.get("TV_SEASON_DIR_PATTERN", "season ").lower()
config = NFOGuardConfig()
# ---------------------------
# Models
# ---------------------------
class SonarrWebhook(BaseModel):
eventType: str
series: Optional[Dict[str, Any]] = None
episodes: Optional[list] = []
episodeFile: Optional[Dict[str, Any]] = None
isUpgrade: Optional[bool] = False
class Config:
extra = "allow"
class RadarrWebhook(BaseModel):
eventType: str
movie: Optional[Dict[str, Any]] = None
movieFile: Optional[Dict[str, Any]] = None
isUpgrade: Optional[bool] = False
deletedFiles: Optional[list] = []
remoteMovie: Optional[Dict[str, Any]] = None
renamedMovieFiles: Optional[List[Dict[str, Any]]] = None
class Config:
extra = "allow"
class HealthResponse(BaseModel):
status: str
version: str
uptime: str
database_status: str
radarr_database: Optional[Dict[str, Any]] = None
# ---------------------------
# Core Processing
# ---------------------------
class TVProcessor:
"""Handles TV series processing"""
def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper):
self.db = db
self.nfo_manager = nfo_manager
self.path_mapper = path_mapper
self.sonarr = SonarrClient(
os.environ.get("SONARR_URL", ""),
os.environ.get("SONARR_API_KEY", "")
)
self.external_clients = ExternalClientManager()
def find_series_path(self, series_title: str, imdb_id: str, sonarr_path: str = None) -> Optional[Path]:
"""Find series directory path"""
# Try webhook path first
if sonarr_path:
container_path = self.path_mapper.sonarr_path_to_container_path(sonarr_path)
path_obj = Path(container_path)
if path_obj.exists():
return path_obj
# Search by IMDb ID or title
for media_path in config.tv_paths:
if not media_path.exists():
continue
# Search by IMDb ID
if imdb_id:
pattern = str(media_path / f"*[imdb-{imdb_id}]*")
matches = glob.glob(pattern)
if matches:
return Path(matches[0])
# Search by title
if series_title:
title_clean = series_title.lower().replace(" ", "").replace("-", "")
for item in media_path.iterdir():
if item.is_dir() and "[imdb-" in item.name.lower():
item_clean = item.name.lower().replace(" ", "").replace("-", "")
if title_clean in item_clean:
return item
return None
def process_series(self, series_path: Path) -> None:
"""Process a TV series directory"""
imdb_id = self.nfo_manager.parse_imdb_from_path(series_path)
if not imdb_id:
_log("ERROR", f"No IMDb ID found in series path: {series_path}")
return
_log("INFO", f"Processing TV series: {series_path.name}")
# Update database
self.db.upsert_series(imdb_id, str(series_path))
# Find video files
disk_episodes = self._find_disk_episodes(series_path)
_log("INFO", f"Found {len(disk_episodes)} episodes on disk")
# Get episode dates
episode_dates = self._gather_episode_dates(series_path, imdb_id, disk_episodes)
# Process episodes
for (season, episode), (aired, dateadded, source) in episode_dates.items():
if (season, episode) in disk_episodes:
# Create NFO
if config.manage_nfo:
season_dir = series_path / config.tv_season_dir_format.format(season=season)
self.nfo_manager.create_episode_nfo(
season_dir,
season, episode, aired, dateadded, source, config.lock_metadata
)
# Update file mtimes
if config.fix_dir_mtimes and dateadded:
video_files = disk_episodes[(season, episode)]
for video_file in video_files:
self.nfo_manager.set_file_mtime(video_file, dateadded)
# Save to database
self.db.upsert_episode_date(imdb_id, season, episode, aired, dateadded, source, True)
# Create season/tvshow NFOs
if config.manage_nfo:
seasons_processed = set()
for (season, episode) in disk_episodes.keys():
if season not in seasons_processed:
season_dir = series_path / config.tv_season_dir_format.format(season=season)
self.nfo_manager.create_season_nfo(season_dir, season)
seasons_processed.add(season)
self.nfo_manager.create_tvshow_nfo(series_path, imdb_id)
_log("INFO", f"Completed processing TV series: {series_path.name}")
def _find_disk_episodes(self, series_path: Path) -> Dict[Tuple[int, int], List[Path]]:
"""Find all episode video files on disk"""
disk_episodes = {}
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
for season_dir in series_path.iterdir():
if not (season_dir.is_dir() and season_dir.name.lower().startswith(config.tv_season_dir_pattern)):
continue
try:
# Extract season number from directory name
# Handle formats like "Season 01", "S01", "Season01", etc.
dir_name = season_dir.name.lower()
if config.tv_season_dir_pattern in dir_name:
# Extract everything after the pattern
season_part = dir_name[len(config.tv_season_dir_pattern):].strip()
else:
continue
season_num = int(season_part)
except (ValueError, IndexError):
continue
for video_file in season_dir.iterdir():
if video_file.is_file() and video_file.suffix.lower() in video_exts:
match = re.search(r"S(\d{2})E(\d{2})", video_file.name, re.IGNORECASE)
if match:
file_season, file_episode = int(match.group(1)), int(match.group(2))
key = (season_num, file_episode) # Use directory season number
if key not in disk_episodes:
disk_episodes[key] = []
disk_episodes[key].append(video_file)
return disk_episodes
def _gather_episode_dates(self, series_path: Path, imdb_id: str, disk_episodes: Dict) -> Dict:
"""Gather episode dates from various sources"""
episode_dates = {}
# Check cache first
cached_episodes = self.db.get_series_episodes(imdb_id, has_video_file_only=True)
for ep in cached_episodes:
key = (ep["season"], ep["episode"])
if key in disk_episodes: # Only use cached data for episodes we have
episode_dates[key] = (ep["aired"], ep["dateadded"], ep["source"])
# Find missing episodes
cached_keys = set(episode_dates.keys())
missing_keys = set(disk_episodes.keys()) - cached_keys
if not missing_keys:
_log("INFO", "All episodes found in cache")
return episode_dates
_log("INFO", f"Querying APIs for {len(missing_keys)} missing episodes")
# Query Sonarr for missing episodes
if self.sonarr.enabled:
series = self.sonarr.series_by_imdb(imdb_id)
if series:
series_id = series.get("id")
if series_id:
episodes = self.sonarr.episodes_for_series(series_id)
for ep in episodes:
season_num = ep.get("seasonNumber")
episode_num = ep.get("episodeNumber")
if not isinstance(season_num, int) or not isinstance(episode_num, int):
continue
key = (season_num, episode_num)
if key not in missing_keys:
continue
# Get dates
aired = self._parse_date_to_iso(ep.get("airDateUtc"))
dateadded = None
source = "sonarr:episode.airDateUtc"
# Try to get import history
episode_id = ep.get("id")
if episode_id:
import_date = self.sonarr.get_episode_import_history(episode_id)
if import_date:
dateadded = self._parse_date_to_iso(import_date)
source = "sonarr:history.import"
if not dateadded:
dateadded = aired
if aired or dateadded:
episode_dates[key] = (aired, dateadded, source)
# Fill remaining gaps with external APIs
remaining_keys = missing_keys - set(episode_dates.keys())
if remaining_keys and self.external_clients.tmdb.enabled:
tmdb_movie = self.external_clients.tmdb.find_by_imdb(imdb_id)
if tmdb_movie:
tv_id = tmdb_movie.get("id")
if tv_id:
seasons_needed = set(season for season, episode in remaining_keys)
for season_num in seasons_needed:
tmdb_episodes = self.external_clients.tmdb.get_tv_season_episodes(tv_id, season_num)
for ep_num, air_date in tmdb_episodes.items():
key = (season_num, ep_num)
if key in remaining_keys:
aired = self._parse_date_to_iso(air_date)
episode_dates[key] = (aired, aired, "tmdb:air_date")
return episode_dates
def _parse_date_to_iso(self, date_str: str) -> Optional[str]:
"""Parse date string to ISO format"""
if not date_str:
return None
try:
if len(date_str) == 10 and date_str[4] == "-":
dt = datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc)
else:
dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc)
return dt.isoformat(timespec="seconds")
except Exception:
return None
class MovieProcessor:
"""Handles movie processing"""
def __init__(self, db: NFOGuardDatabase, nfo_manager: NFOManager, path_mapper: PathMapper):
self.db = db
self.nfo_manager = nfo_manager
self.path_mapper = path_mapper
self.radarr = RadarrClient(
os.environ.get("RADARR_URL", ""),
os.environ.get("RADARR_API_KEY", "")
)
self.external_clients = ExternalClientManager()
def find_movie_path(self, movie_title: str, imdb_id: str, radarr_path: str = None) -> Optional[Path]:
"""Find movie directory path"""
# Try webhook path first
if radarr_path:
container_path = self.path_mapper.radarr_path_to_container_path(radarr_path)
path_obj = Path(container_path)
if path_obj.exists():
return path_obj
# Search by IMDb ID or title
for media_path in config.movie_paths:
if not media_path.exists():
continue
# Search by IMDb ID
if imdb_id:
pattern = str(media_path / f"*[imdb-{imdb_id}]*")
matches = glob.glob(pattern)
if matches:
return Path(matches[0])
# Search by title
if movie_title:
title_clean = movie_title.lower().replace(" ", "").replace("-", "")
for item in media_path.iterdir():
if item.is_dir() and "[imdb-" in item.name.lower():
item_clean = item.name.lower().replace(" ", "").replace("-", "")
if title_clean in item_clean:
return item
return None
def process_movie(self, movie_path: Path) -> None:
"""Process a movie directory"""
imdb_id = self.nfo_manager.parse_imdb_from_path(movie_path)
if not imdb_id:
_log("ERROR", f"No IMDb ID found in movie path: {movie_path}")
return
_log("INFO", f"Processing movie: {movie_path.name}")
# Update database
self.db.upsert_movie(imdb_id, str(movie_path))
# Check for video files
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
has_video = any(f.is_file() and f.suffix.lower() in video_exts for f in movie_path.iterdir())
if not has_video:
_log("WARNING", f"No video files found in: {movie_path}")
self.db.upsert_movie_dates(imdb_id, None, None, None, False)
return
# Get existing dates
existing = self.db.get_movie_dates(imdb_id)
# Determine if we should query APIs
should_query = (
config.movie_poll_mode == "always" or
(config.movie_poll_mode == "if_missing" and not existing) or
(config.movie_poll_mode == "if_missing" and existing and existing.get("source") == "file:mtime")
)
# Decide dates
dateadded, source, released = self._decide_movie_dates(imdb_id, movie_path, should_query, existing)
# Skip processing if no valid date found and file dates disabled
if dateadded is None:
_log("WARNING", f"Skipping movie {movie_path.name} - no valid date source available")
self.db.upsert_movie_dates(imdb_id, released, None, source, True)
return
# Create NFO
if config.manage_nfo:
self.nfo_manager.create_movie_nfo(
movie_path, imdb_id, dateadded, released, source, config.lock_metadata
)
# Update file mtimes
if config.fix_dir_mtimes and dateadded and dateadded != "MANUAL_REVIEW_NEEDED":
self.nfo_manager.update_movie_files_mtime(movie_path, dateadded)
# Save to database
self.db.upsert_movie_dates(imdb_id, released, dateadded, source, True)
_log("INFO", f"Completed processing movie: {movie_path.name} (source: {source})")
def _decide_movie_dates(self, imdb_id: str, movie_path: Path, should_query: bool, existing: Optional[Dict]) -> Tuple[str, str, Optional[str]]:
"""Decide movie dates based on configuration and available data"""
if not should_query and existing:
return existing["dateadded"], existing["source"], existing.get("released")
# Query Radarr for movie info
radarr_movie = None
if should_query and self.radarr.api_key:
radarr_movie = self.radarr.movie_by_imdb(imdb_id)
released = None
if radarr_movie:
released = self._parse_date_to_iso(radarr_movie.get("inCinemas"))
# Try import history first if configured
if config.movie_priority == "import_then_digital":
import_date, import_source = None, None
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = self.radarr.get_movie_import_date(movie_id, fallback_to_file_date=config.allow_file_date_fallback)
_log("INFO", f"Movie {imdb_id}: Radarr import result: date={import_date}, source={import_source}")
# Check for special case: rename-first scenario (should prefer release dates)
if import_source == "radarr:db.prefer_release_dates":
_log("INFO", f"🎯 Movie {imdb_id} has rename-first history - skipping import, preferring release dates")
# Fall through to release date logic below
# Check if we got a real import date or just file date fallback
elif import_date and import_source != "radarr:db.file.dateAdded":
_log("INFO", f"✅ Movie {imdb_id}: Using import date {import_date} from {import_source}")
return import_date, import_source, released
# Get digital release date for comparison/fallback
_log("INFO", f"🔍 Movie {imdb_id}: Trying digital release date fallback...")
digital_date, digital_source = self._get_digital_release_date(imdb_id)
_log("INFO", f"Movie {imdb_id}: Digital release result: date={digital_date}, source={digital_source}")
# If we only have file date and release date exists, prefer it if reasonable and enabled
if import_date and import_source == "radarr:db.file.dateAdded" and digital_date and config.prefer_release_dates_over_file_dates:
# Compare dates - prefer release date if it's reasonable
if self._should_prefer_release_over_file_date(digital_date, digital_source, released, imdb_id):
_log("INFO", f"✅ Movie {imdb_id}: Preferring digital release date {digital_date} over file date")
return digital_date, digital_source, released
else:
_log("INFO", f"✅ Movie {imdb_id}: Keeping file date {import_date} - digital date not reasonable")
return import_date, import_source, released
# Use whichever we have
if import_date:
_log("INFO", f"✅ Movie {imdb_id}: Using import date {import_date} from {import_source}")
return import_date, import_source, released
elif digital_date:
_log("INFO", f"✅ Movie {imdb_id}: Using digital release date {digital_date} from {digital_source}")
return digital_date, digital_source, released
else:
_log("WARNING", f"⚠️ Movie {imdb_id}: No import date OR digital release date found - will try other fallbacks")
else: # digital_then_import
# Try digital release first
digital_date, digital_source = self._get_digital_release_date(imdb_id)
if digital_date:
return digital_date, digital_source, released
# Fall back to import history
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = self.radarr.get_movie_import_date(movie_id, fallback_to_file_date=config.allow_file_date_fallback)
if import_date:
return import_date, import_source, released
# Last resort: file mtime (if allowed)
if config.allow_file_date_fallback:
return self._get_file_mtime_date(movie_path)
else:
_log("INFO", f"No valid dates found for {imdb_id} and file date fallback disabled - skipping NFO creation")
return None, "no_valid_date_source", None
def _get_digital_release_date(self, imdb_id: str) -> Tuple[Optional[str], str]:
"""Get release date from external sources using configured priority"""
release_result = self.external_clients.get_release_date_by_priority(
imdb_id,
config.release_date_priority,
enable_smart_validation=config.enable_smart_date_validation
)
if release_result:
return release_result[0], release_result[1]
return None, "release:none"
def _get_file_mtime_date(self, movie_path: Path) -> Tuple[str, str, Optional[str]]:
"""Get date from file modification time as last resort"""
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
newest_mtime = None
for file_path in movie_path.iterdir():
if file_path.is_file() and file_path.suffix.lower() in video_exts:
try:
mtime = file_path.stat().st_mtime
if newest_mtime is None or mtime > newest_mtime:
newest_mtime = mtime
except Exception:
continue
if newest_mtime:
try:
iso_date = datetime.fromtimestamp(newest_mtime, tz=timezone.utc).isoformat(timespec="seconds")
return iso_date, "file:mtime", None
except Exception:
pass
return "MANUAL_REVIEW_NEEDED", "manual_review_required", None
def _should_prefer_release_over_file_date(self, release_date: str, release_source: str, theatrical_release: Optional[str], imdb_id: str) -> bool:
"""
Decide if release date should be preferred over file date
Logic:
- For theatrical dates: Always prefer over file dates (they're authoritative)
- For physical dates: Usually prefer over file dates
- For digital dates: Prefer if reasonable (not decades before theatrical)
"""
try:
release_dt = datetime.fromisoformat(release_date.replace("Z", "+00:00"))
# Always prefer theatrical and physical releases over file dates
if any(release_type in release_source for release_type in ["theatrical", "physical"]):
_log("INFO", f"Release date {release_date} ({release_source}) for {imdb_id}, preferring over file date")
return True
# If we have theatrical release date, compare digital against it
if theatrical_release:
theatrical_dt = datetime.fromisoformat(theatrical_release.replace("Z", "+00:00"))
year_diff = release_dt.year - theatrical_dt.year
# If digital is more than 10 years before theatrical, it's probably wrong
if year_diff < -10:
_log("INFO", f"Release date {release_date} is {abs(year_diff)} years before theatrical {theatrical_release} for {imdb_id}, using file date instead")
return False
# If digital is within reasonable range (theatrical to +20 years), use it
if -2 <= year_diff <= 20:
_log("INFO", f"Release date {release_date} is reasonable for {imdb_id} (theatrical: {theatrical_release}), preferring over file date")
return True
# If no theatrical date, use digital if it's not absurdly old
if release_dt.year >= 1990: # Reasonable minimum for digital releases
_log("INFO", f"Release date {release_date} seems reasonable for {imdb_id}, preferring over file date")
return True
_log("INFO", f"Release date {release_date} seems too old for {imdb_id}, using file date instead")
return False
except Exception as e:
_log("WARNING", f"Error comparing dates for {imdb_id}: {e}")
return False
def _parse_date_to_iso(self, date_str: str) -> Optional[str]:
"""Parse date string to ISO format"""
if not date_str:
return None
try:
if len(date_str) == 10 and date_str[4] == "-":
dt = datetime.fromisoformat(date_str).replace(tzinfo=timezone.utc)
else:
dt = datetime.fromisoformat(date_str.replace("Z", "+00:00")).astimezone(timezone.utc)
return dt.isoformat(timespec="seconds")
except Exception:
return None
# ---------------------------
# Batching System
# ---------------------------
class WebhookBatcher:
"""Batches webhook events to avoid processing storms"""
def __init__(self):
self.pending: Dict[str, Dict] = {}
self.timers: Dict[str, threading.Timer] = {}
self.processing: Set[str] = set()
self.lock = threading.Lock()
self.executor = ThreadPoolExecutor(max_workers=config.max_concurrent)
def add_webhook(self, key: str, webhook_data: Dict, media_type: str):
"""Add webhook to batch queue"""
with self.lock:
if key in self.timers:
self.timers[key].cancel()
webhook_data['media_type'] = media_type
self.pending[key] = webhook_data
_log("INFO", f"Batched {media_type} webhook for {key}")
timer = threading.Timer(config.batch_delay, self._process_item, args=[key])
self.timers[key] = timer
timer.start()
def _process_item(self, key: str):
"""Process a batched item"""
with self.lock:
if key in self.processing or key not in self.pending:
return
self.processing.add(key)
webhook_data = self.pending.pop(key)
self.timers.pop(key, None)
try:
self.executor.submit(self._process_sync, key, webhook_data)
except Exception as e:
_log("ERROR", f"Error submitting processing for {key}: {e}")
with self.lock:
self.processing.discard(key)
def _process_sync(self, key: str, webhook_data: Dict):
"""Synchronous processing of webhook data"""
try:
media_type = webhook_data.get('media_type')
path_str = webhook_data.get('path')
if not path_str:
_log("ERROR", f"No path found for {media_type} {key}")
return
path_obj = Path(path_str)
if not path_obj.exists():
_log("ERROR", f"Path does not exist: {path_obj}")
return
# Process based on media type
if media_type == 'tv':
tv_processor.process_series(path_obj)
elif media_type == 'movie':
movie_processor.process_movie(path_obj)
else:
_log("ERROR", f"Unknown media type: {media_type}")
except Exception as e:
_log("ERROR", f"Error processing {media_type} {key}: {e}")
finally:
with self.lock:
self.processing.discard(key)
def get_status(self) -> Dict:
"""Get batch queue status"""
with self.lock:
return {
"pending_items": list(self.pending.keys()),
"processing_items": list(self.processing),
"pending_count": len(self.pending),
"processing_count": len(self.processing)
}
# ---------------------------
# FastAPI Application
# ---------------------------
# Get version
try:
version = (Path(__file__).parent / "VERSION").read_text().strip()
except:
version = "0.1.0"
# Check if running in dev environment (set during Docker build)
build_type = os.environ.get('BUILD_TYPE', 'production')
image_tag = os.environ.get('IMAGE_TAG', 'latest')
if (build_type == 'dev' or 'dev' in image_tag.lower()):
version = f"{version}-dev"
app = FastAPI(
title="NFOGuard",
description="Webhook server for preserving media import dates",
version=version
)
start_time = datetime.now(timezone.utc)
# Initialize components
db = NFOGuardDatabase(config.db_path)
nfo_manager = NFOManager(config.manager_brand)
path_mapper = PathMapper()
tv_processor = TVProcessor(db, nfo_manager, path_mapper)
movie_processor = MovieProcessor(db, nfo_manager, path_mapper)
batcher = WebhookBatcher()
# ---------------------------
# Webhook Handlers
# ---------------------------
async def _read_payload(request: Request) -> dict:
"""Read webhook payload from request"""
content_type = (request.headers.get("content-type") or "").lower()
try:
if "application/json" in content_type:
return await request.json()
form = await request.form()
if "payload" in form:
return json.loads(form["payload"])
return dict(form)
except Exception as e:
_log("ERROR", f"Failed to read webhook payload: {e}")
return {}
@app.post("/webhook/sonarr")
async def sonarr_webhook(request: Request, background_tasks: BackgroundTasks):
"""Handle Sonarr webhooks"""
try:
payload = await _read_payload(request)
if not payload:
raise HTTPException(status_code=422, detail="Empty Sonarr payload")
webhook = SonarrWebhook(**payload)
_log("INFO", f"Received Sonarr webhook: {webhook.eventType}")
if webhook.eventType not in ["Download", "Upgrade", "Rename"]:
return {"status": "ignored", "reason": f"Event type {webhook.eventType} not processed"}
if not webhook.series:
return {"status": "ignored", "reason": "No series data"}
series_info = webhook.series
series_title = series_info.get("title", "")
imdb_id = series_info.get("imdbId", "").replace("tt", "").strip()
if imdb_id:
imdb_id = f"tt{imdb_id}"
sonarr_path = series_info.get("path", "")
if not imdb_id:
_log("ERROR", f"No IMDb ID for series: {series_title}")
return {"status": "error", "reason": "No IMDb ID"}
# Find series path
series_path = tv_processor.find_series_path(series_title, imdb_id, sonarr_path)
if not series_path:
_log("ERROR", f"Could not find series directory: {series_title} ({imdb_id})")
return {"status": "error", "reason": "Series directory not found"}
# Add to batch queue
webhook_dict = {
'path': str(series_path),
'series_info': series_info,
'event_type': webhook.eventType
}
batcher.add_webhook(imdb_id, webhook_dict, 'tv')
return {"status": "accepted", "message": "Sonarr webhook queued"}
except Exception as e:
_log("ERROR", f"Sonarr webhook error: {e}")
raise HTTPException(status_code=422, detail=f"Invalid webhook: {e}")
@app.post("/webhook/radarr")
async def radarr_webhook(request: Request, background_tasks: BackgroundTasks):
"""Handle Radarr webhooks"""
try:
payload = await _read_payload(request)
if not payload:
raise HTTPException(status_code=422, detail="Empty Radarr payload")
webhook = RadarrWebhook(**payload)
_log("INFO", f"Received Radarr webhook: {webhook.eventType}")
if webhook.eventType not in ["Download", "Upgrade", "Rename", "Test"]:
return {"status": "ignored", "reason": f"Event type {webhook.eventType} not processed"}
if webhook.eventType == "Test":
return {"status": "ok", "message": "Test received"}
if not webhook.movie:
return {"status": "ignored", "reason": "No movie data"}
movie_info = webhook.movie
movie_title = movie_info.get("title", "")
imdb_id = (movie_info.get("imdbId") or "").strip()
if imdb_id:
imdb_id = f"tt{imdb_id.replace('tt','')}"
radarr_path = movie_info.get("folderPath") or movie_info.get("path", "")
# Find movie path
movie_path = movie_processor.find_movie_path(movie_title, imdb_id, radarr_path)
if not movie_path:
_log("ERROR", f"Could not find movie directory: {movie_title} ({imdb_id})")
return {"status": "error", "reason": "Movie directory not found"}
# Extract IMDb ID from path if not in webhook
if not imdb_id:
imdb_id = nfo_manager.parse_imdb_from_path(movie_path)
if not imdb_id:
_log("ERROR", f"No IMDb ID available for movie: {movie_title}")
return {"status": "error", "reason": "No IMDb ID"}
# Add to batch queue
webhook_dict = {
'path': str(movie_path),
'movie_info': movie_info,
'event_type': webhook.eventType
}
batcher.add_webhook(imdb_id, webhook_dict, 'movie')
return {"status": "accepted", "message": "Radarr webhook queued"}
except Exception as e:
_log("ERROR", f"Radarr webhook error: {e}")
raise HTTPException(status_code=422, detail=f"Invalid webhook: {e}")
# ---------------------------
# API Endpoints
# ---------------------------
@app.get("/health")
async def health() -> HealthResponse:
"""Health check endpoint with Radarr database status"""
uptime = datetime.now(timezone.utc) - start_time
# Check NFOGuard database
try:
with db.get_connection() as conn:
conn.execute("SELECT 1").fetchone()
db_status = "healthy"
except Exception as e:
db_status = f"error: {e}"
# Check Radarr database if available
radarr_db_health = None
overall_status = "healthy" if db_status == "healthy" else "degraded"
# Get Radarr client with database access from movie processor
try:
if hasattr(movie_processor, 'radarr') and movie_processor.radarr:
radarr_client = movie_processor.radarr
if hasattr(radarr_client, 'db_client') and radarr_client.db_client:
try:
radarr_db_health = radarr_client.db_client.health_check()
if radarr_db_health["status"] != "healthy":
overall_status = "degraded"
except Exception as e:
radarr_db_health = {
"status": "error",
"error": str(e),
"tested_at": datetime.now(timezone.utc).isoformat(timespec="seconds")
}
overall_status = "degraded"
except Exception as e:
# If movie processor isn't available, skip database health check
_log("DEBUG", f"Skipping Radarr database health check: {e}")
return HealthResponse(
status=overall_status,
version=version,
uptime=str(uptime),
database_status=db_status,
radarr_database=radarr_db_health
)
@app.get("/stats")
async def get_stats():
"""Get database statistics"""
try:
return db.get_stats()
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/batch/status")
async def batch_status():
"""Get batch queue status"""
return batcher.get_status()
@app.get("/debug/movie/{imdb_id}")
async def debug_movie_import_date(imdb_id: str):
"""Debug endpoint to analyze movie import date detection"""
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
_log("INFO", f"=== DEBUG MOVIE IMPORT DATE: {imdb_id} ===")
if not (os.environ.get("RADARR_URL") and os.environ.get("RADARR_API_KEY")):
return {
"error": "Radarr not configured",
"imdb_id": imdb_id,
"radarr_configured": False
}
# Create Radarr client
from clients.radarr_client import RadarrClient
radarr_client = RadarrClient(
os.environ.get("RADARR_URL"),
os.environ.get("RADARR_API_KEY")
)
# Look up movie
movie_obj = radarr_client.movie_by_imdb(imdb_id)
if not movie_obj:
return {
"error": f"Movie not found in Radarr for IMDb ID {imdb_id}",
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": False
}
movie_id = movie_obj.get("id")
movie_title = movie_obj.get("title")
_log("INFO", f"Found movie: {movie_title} (Radarr ID: {movie_id})")
# Get import date with detailed logging - DATABASE ONLY
import_date, source = radarr_client.get_movie_import_date(movie_id, fallback_to_file_date=True)
# Movie files info - DATABASE ONLY mode (minimal info)
file_info = []
if import_date:
file_info.append({
"note": "DATABASE ONLY MODE - File details from database queries",
"import_date_used": import_date,
"source": source
})
return {
"imdb_id": imdb_id,
"radarr_configured": True,
"movie_found": True,
"movie_title": movie_title,
"movie_id": movie_id,
"detected_import_date": import_date,
"import_source": source,
"movie_files": file_info,
"debug_info": {
"radarr_url": os.environ.get("RADARR_URL"),
"movie_digital_release": movie_obj.get("digitalRelease"),
"movie_in_cinemas": movie_obj.get("inCinemas"),
"movie_physical_release": movie_obj.get("physicalRelease"),
"movie_folder_path": movie_obj.get("folderPath")
}
}
except Exception as e:
_log("ERROR", f"Debug endpoint error for {imdb_id}: {e}")
return {
"error": str(e),
"imdb_id": imdb_id,
"success": False
}
@app.get("/debug/movie/{imdb_id}/history")
async def debug_movie_history(imdb_id: str):
"""Detailed history analysis for a movie"""
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
_log("INFO", f"=== DETAILED HISTORY ANALYSIS: {imdb_id} ===")
# Use database-only mode for consistency
if not movie_processor.radarr.db_client:
return {"error": "Radarr database not configured"}
radarr_client = movie_processor.radarr
# Look up movie
movie_obj = radarr_client.movie_by_imdb(imdb_id)
if not movie_obj:
return {"error": f"Movie not found for {imdb_id}"}
movie_id = movie_obj.get("id")
movie_title = movie_obj.get("title")
# Get history from database instead of API
if not radarr_client.db_client:
return {"error": "Database-only mode required"}
# TODO: Implement database-only history retrieval
return {
"error": "History endpoint temporarily disabled - use /debug/movie/{imdb_id}/priority for date analysis",
"imdb_id": imdb_id,
"movie_id": movie_id,
"movie_title": movie_title,
"note": "This endpoint needs database-only implementation to avoid showing wrong movie events"
}
# Analyze each event
analyzed_events = []
for event in all_history:
event_type = event.get("eventType", "")
date_str = event.get("date", "")
event_data = event.get("data", {})
# Get source path info
source_path = (
event_data.get('droppedPath', '') or
event_data.get('sourcePath', '') or
event_data.get('path', '') or
event_data.get('sourceTitle', '')
)
# Analyze if this is a real import
is_real, reason, date_iso = radarr_client._analyze_event_for_import(event)
analyzed_events.append({
"event_type": event_type,
"date": date_str,
"source_path": source_path,
"is_real_import": is_real,
"analysis_reason": reason,
"parsed_date": date_iso,
"full_data": event_data
})
# Find what our algorithm would pick - DATABASE ONLY
picked_date, _ = radarr_client.get_movie_import_date(movie_id, fallback_to_file_date=True)
return {
"imdb_id": imdb_id,
"movie_title": movie_title,
"movie_id": movie_id,
"total_history_events": len(all_history),
"our_algorithm_picked": picked_date,
"all_events": analyzed_events,
"expected_july_date": "2025-07-07" in (picked_date or "")
}
except Exception as e:
_log("ERROR", f"History debug error for {imdb_id}: {e}")
return {"error": str(e)}
@app.post("/manual/scan")
async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = None, scan_type: str = "both"):
"""Manual scan endpoint"""
if scan_type not in ["both", "tv", "movies"]:
raise HTTPException(status_code=400, detail="scan_type must be 'both', 'tv', or 'movies'")
async def run_scan():
paths_to_scan = []
if path:
paths_to_scan = [Path(path)]
else:
if scan_type in ["both", "tv"]:
paths_to_scan.extend(config.tv_paths)
if scan_type in ["both", "movies"]:
paths_to_scan.extend(config.movie_paths)
for scan_path in paths_to_scan:
if not scan_path.exists():
continue
if scan_type in ["both", "tv"] and scan_path in config.tv_paths:
for item in scan_path.iterdir():
if item.is_dir() and nfo_manager.parse_imdb_from_path(item):
try:
tv_processor.process_series(item)
except Exception as e:
_log("ERROR", f"Failed processing TV series {item}: {e}")
if scan_type in ["both", "movies"] and scan_path in config.movie_paths:
_log("INFO", f"Scanning movies in: {scan_path}")
movie_count = 0
for item in scan_path.iterdir():
if item.is_dir() and nfo_manager.parse_imdb_from_path(item):
movie_count += 1
_log("INFO", f"Processing movie: {item.name}")
try:
movie_processor.process_movie(item)
except Exception as e:
_log("ERROR", f"Failed processing movie {item}: {e}")
_log("INFO", f"Completed movie scan: {movie_count} movies processed in {scan_path}")
background_tasks.add_task(run_scan)
return {"status": "started", "message": f"Manual {scan_type} scan started"}
@app.post("/test/bulk-update")
async def test_bulk_update():
"""Test bulk update functionality without modifying data"""
try:
from clients.radarr_db_client import RadarrDbClient
# Test Radarr database
radarr_db = RadarrDbClient.from_env()
if not radarr_db:
return {"status": "error", "message": "Radarr database connection failed"}
# Test query execution
query = 'SELECT COUNT(*) FROM "Movies" m JOIN "MovieMetadata" mm ON m."MovieMetadataId" = mm."Id" WHERE mm."ImdbId" IS NOT NULL'
with radarr_db._get_connection() as conn:
cursor = conn.cursor()
cursor.execute(query)
movie_count = cursor.fetchone()[0]
return {
"status": "success",
"message": "Bulk update test passed",
"movies_with_imdb": movie_count,
"database_type": radarr_db.db_type
}
except Exception as e:
return {"status": "error", "message": f"Bulk update test failed: {e}"}
@app.post("/test/movie-scan")
async def test_movie_scan():
"""Test movie directory scanning logic"""
try:
results = []
for path in config.movie_paths:
path_result = {
"path": str(path),
"exists": path.exists(),
"movies_found": 0
}
if path.exists():
for item in path.iterdir():
if item.is_dir() and nfo_manager.parse_imdb_from_path(item):
path_result["movies_found"] += 1
results.append(path_result)
total_movies = sum(r["movies_found"] for r in results)
return {
"status": "success",
"message": f"Movie scan test found {total_movies} movies",
"path_results": results
}
except Exception as e:
return {"status": "error", "message": f"Movie scan test failed: {e}"}
@app.post("/bulk/update")
async def trigger_bulk_update(background_tasks: BackgroundTasks):
"""Trigger bulk update of all movies"""
async def run_bulk_update():
try:
from bulk_update_movies import bulk_update_all_movies
success = bulk_update_all_movies()
_log("INFO", f"Bulk update completed: {'success' if success else 'failed'}")
except Exception as e:
_log("ERROR", f"Bulk update error: {e}")
background_tasks.add_task(run_bulk_update)
return {"status": "started", "message": "Bulk update started"}
@app.get("/debug/movie/{imdb_id}/priority")
async def debug_movie_priority_logic(imdb_id: str):
"""Debug endpoint showing how MOVIE_PRIORITY affects date selection"""
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
result = {
"imdb_id": imdb_id,
"movie_priority": config.movie_priority,
"release_date_priority": config.release_date_priority,
"priority_explanation": "",
"date_sources": {},
"selected_date": None,
"selected_source": None
}
# Get Radarr import date
if movie_processor.radarr.api_key:
radarr_movie = movie_processor.radarr.movie_by_imdb(imdb_id)
if radarr_movie:
movie_id = radarr_movie.get("id")
if movie_id:
import_date, import_source = movie_processor.radarr.get_movie_import_date(movie_id)
if import_date:
result["date_sources"]["radarr_import"] = {
"date": import_date,
"source": import_source
}
# Get digital release dates with detailed logging
digital_date, digital_source = movie_processor._get_digital_release_date(imdb_id)
if digital_date:
result["date_sources"]["digital_release"] = {
"date": digital_date,
"source": digital_source
}
else:
# Add debug info about why digital date wasn't found
candidates = movie_processor.external_clients.get_digital_release_candidates(imdb_id)
result["date_sources"]["digital_release_debug"] = {
"candidates_found": len(candidates),
"candidates": candidates[:3] if candidates else [], # Show first 3
"reason": digital_source if digital_source else "no_digital_dates_found"
}
# Show priority logic
if config.movie_priority == "import_then_digital":
priority_list = "".join(config.release_date_priority)
result["priority_explanation"] = f"1st: Radarr import history, 2nd: Release dates ({priority_list}), 3rd: file mtime. Note: If import is only file date, prefer reasonable release dates."
radarr_import = result["date_sources"].get("radarr_import")
digital_release = result["date_sources"].get("digital_release")
# Check for file date fallback logic
if radarr_import and radarr_import["source"] == "radarr:db.file.dateAdded" and digital_release:
# Test the smart logic
would_prefer_digital = movie_processor._should_prefer_release_over_file_date(
digital_release["date"],
digital_release["source"],
None, # We don't have theatrical date in this debug context
imdb_id
)
result["file_date_detected"] = True
result["would_prefer_digital"] = would_prefer_digital
if would_prefer_digital:
result["selected_date"] = digital_release["date"]
result["selected_source"] = digital_release["source"] + " (preferred over file date)"
else:
result["selected_date"] = radarr_import["date"]
result["selected_source"] = radarr_import["source"] + " (digital too old)"
elif radarr_import and radarr_import["source"] != "radarr:db.file.dateAdded":
result["selected_date"] = radarr_import["date"]
result["selected_source"] = radarr_import["source"]
elif digital_release:
result["selected_date"] = digital_release["date"]
result["selected_source"] = digital_release["source"]
else: # digital_then_import
result["priority_explanation"] = "1st: TMDB/OMDb digital release, 2nd: Radarr import history, 3rd: file mtime"
if result["date_sources"].get("digital_release"):
result["selected_date"] = result["date_sources"]["digital_release"]["date"]
result["selected_source"] = result["date_sources"]["digital_release"]["source"]
elif result["date_sources"].get("radarr_import"):
result["selected_date"] = result["date_sources"]["radarr_import"]["date"]
result["selected_source"] = result["date_sources"]["radarr_import"]["source"]
# Show external API status
result["external_apis"] = {
"tmdb_enabled": movie_processor.external_clients.tmdb.enabled,
"omdb_enabled": movie_processor.external_clients.omdb.enabled,
"jellyseerr_enabled": movie_processor.external_clients.jellyseerr.enabled
}
return result
except Exception as e:
return {"error": str(e), "imdb_id": imdb_id}
@app.get("/debug/tmdb/{imdb_id}")
async def debug_tmdb_lookup(imdb_id: str):
"""Debug TMDB API lookup for a specific movie"""
try:
if not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
result = {
"imdb_id": imdb_id,
"tmdb_api_enabled": movie_processor.external_clients.tmdb.enabled,
"tmdb_api_key_configured": bool(movie_processor.external_clients.tmdb.api_key),
"steps": {}
}
if not movie_processor.external_clients.tmdb.enabled:
result["error"] = "TMDB API not enabled - check TMDB_API_KEY environment variable"
return result
# Step 1: Find movie by IMDb ID
_log("INFO", f"TMDB Debug: Looking up {imdb_id}")
tmdb_movie = movie_processor.external_clients.tmdb.find_by_imdb(imdb_id)
result["steps"]["1_find_by_imdb"] = {
"found": bool(tmdb_movie),
"tmdb_movie": tmdb_movie if tmdb_movie else None
}
if not tmdb_movie:
result["error"] = f"Movie {imdb_id} not found in TMDB"
return result
tmdb_id = tmdb_movie.get("id")
result["tmdb_id"] = tmdb_id
# Step 2: Get release dates
if tmdb_id:
_log("INFO", f"TMDB Debug: Getting release dates for TMDB ID {tmdb_id}")
release_dates_result = movie_processor.external_clients.tmdb._get(f"/movie/{tmdb_id}/release_dates")
result["steps"]["2_release_dates"] = {
"raw_response": release_dates_result,
"has_results": bool(release_dates_result and release_dates_result.get("results"))
}
# Step 3: Look for US digital releases
if release_dates_result and release_dates_result.get("results"):
us_releases = []
for country_data in release_dates_result["results"]:
if country_data.get("iso_3166_1") == "US":
us_releases = country_data.get("release_dates", [])
break
result["steps"]["3_us_releases"] = {
"found_us_data": bool(us_releases),
"us_releases": us_releases
}
# Step 4: Look for digital releases (type 4)
digital_releases = [r for r in us_releases if r.get("type") == 4]
result["steps"]["4_digital_releases"] = {
"digital_count": len(digital_releases),
"digital_releases": digital_releases
}
# Step 5: Test the full digital release function
digital_date = movie_processor.external_clients.tmdb.get_digital_release_date(imdb_id)
result["steps"]["5_final_result"] = {
"digital_date": digital_date,
"success": bool(digital_date)
}
return result
except Exception as e:
return {"error": str(e), "imdb_id": imdb_id, "traceback": str(e)}
# ---------------------------
# Main
# ---------------------------
if __name__ == "__main__":
_log("INFO", "Starting NFOGuard")
_log("INFO", f"Version: {version}")
_log("INFO", f"TV paths: {[str(p) for p in config.tv_paths]}")
_log("INFO", f"Movie paths: {[str(p) for p in config.movie_paths]}")
_log("INFO", f"Database: {config.db_path}")
_log("INFO", f"Config: manage_nfo={config.manage_nfo}, fix_mtimes={config.fix_dir_mtimes}")
_log("INFO", f"Movie priority: {config.movie_priority}")
uvicorn.run(
app,
host="0.0.0.0",
port=int(os.environ.get("PORT", "8080")),
reload=False
)