sbcrumb/nfoguard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
c3d1f50cfbe5c84bc19d8590155e2cb947d2f289
nfoguard/webhook_server.py
T
sbcrumb c3d1f50cfb update
2025-09-07 21:13:46 -04:00

1781 lines
70 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# app.py - NFOGuard Webhook Server (TV + Movies)
#
# Features:
# - TV via existing media_date_cache.apply_series (Sonarr webhooks + manual scan)
# - Movies with robust <dateadded> logic and NFO upsert:
# * Priority configurable: import_then_digital | digital_then_import
# * Poll policy: always | if_missing | never
# * Digital date candidates: Radarr, TMDB, Jellyseerr (optional), OMDb (optional) -> pick earliest
# * Digital sanity check (re-release guard): compare to cinema/import/min and fallback
# * Update policy: backfill_only | overwrite (+ preserve flags in backfill mode)
# * Pretty XML, ensures <dateadded>, <lockdata>, <premiered>, <year>, footer comments, uniqueid imdb
# * Optional mtimes update to match chosen <dateadded>
#
# Endpoints:
# - POST /webhook/sonarr
# - POST /webhook/radarr
# - POST /manual/scan (scan_type=both|tv|movies, optional path=...)
# - POST /manual/scan/tv
# - POST /manual/scan/movies
# - GET /health
# - GET /stats
# - GET /batch/status
#
# Env (highlights):
# MANAGER_BRAND (default NFOGuard)
# MOVIE_PRIORITY=import_then_digital|digital_then_import
# MOVIE_POLL_MODE=always|if_missing|never
# MOVIE_DATE_UPDATE_MODE=backfill_only|overwrite
# MOVIE_PRESERVE_RADARR_IMPORT=true|false
# MOVIE_PRESERVE_EXISTING=true|false
# MOVIE_DIGITAL_SANITY_CHECK=true|false
# MOVIE_DIGITAL_SANITY_REF=cinema|import|min
# MOVIE_DIGITAL_MAX_LAG_DAYS=365
# MOVIE_DIGITAL_INVALID_FALLBACK=import|cinema
# TMDB_API_KEY / TMDB_PRIMARY_COUNTRY
# OMDB_API_KEY
# JELLYSEERR_URL / JELLYSEERR_API_KEY
# RADARR_URL / RADARR_API_KEY / RADARR_TIMEOUT / RADARR_RETRIES
# SONARR_URL / SONARR_API_KEY (TV pathing only; series handled in media_date_cache)
# TV_PATHS, MOVIE_PATHS
# MANAGE_NFO, FIX_DIR_MTIMES, LOCK_METADATA, DEBUG
# BATCH_DELAY, MAX_CONCURRENT_SERIES
#
# Run: uvicorn app:app --host 0.0.0.0 --port 8080
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 sqlite3
import uvicorn
import threading
from concurrent.futures import ThreadPoolExecutor
import xml.etree.ElementTree as ET
from urllib.parse import urlencode, urljoin, quote
from urllib.request import Request as UrlRequest, urlopen
from urllib.error import URLError, HTTPError
import time
# Import TV helpers from your existing module
from media_date_cache import (
db_connect, apply_series, parse_imdb_from_path,
_log as _original_log, _iso_now, SonarrClient
)
# ---------------------------
# Enhanced logging with file support
# ---------------------------
# Global file logger
_file_logger = None
def _setup_file_logging():
"""Setup file logging for NFOGuard"""
global _file_logger
if _file_logger is not None:
return _file_logger
# Create logs directory
log_dir = Path("/app/data/logs")
log_dir.mkdir(parents=True, exist_ok=True)
# Setup rotating file handler
log_file = log_dir / "nfoguard.log"
_file_logger = logging.getLogger("NFOGuard")
_file_logger.setLevel(logging.DEBUG)
# Rotating file handler (10MB max, 5 backups)
file_handler = logging.handlers.RotatingFileHandler(
log_file, maxBytes=10*1024*1024, backupCount=5
)
# Detailed format for file logging
formatter = logging.Formatter(
'[%(asctime)s] %(levelname)s: %(message)s',
datefmt='%Y-%m-%dT%H:%M:%S+00:00'
)
file_handler.setFormatter(formatter)
_file_logger.addHandler(file_handler)
return _file_logger
def _log(level: str, msg: str):
"""Enhanced logging that writes to both console and file"""
# Call original log for console output
_original_log(level, msg)
# Also log to file
try:
file_logger = _setup_file_logging()
if level == "DEBUG":
file_logger.debug(msg)
elif level == "INFO":
file_logger.info(msg)
elif level == "WARNING":
file_logger.warning(msg)
elif level == "ERROR":
file_logger.error(msg)
else:
file_logger.info(f"{level}: {msg}")
except Exception as e:
# Don't let logging errors break the app
print(f"File logging error: {e}")
# Initialize file logging on startup
_setup_file_logging()
# ---------------------------
# 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
movie: Optional[Dict[str, Any]] = None
remoteMovie: Optional[Dict[str, Any]] = None
deletedFiles: Optional[list] = []
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 # on Rename
class Config:
extra = "allow"
class HealthResponse(BaseModel):
status: str
version: str
uptime: str
database_status: str
# ---------------------------
# App & Config
# ---------------------------
app = FastAPI(
title="NFOGuard Webhook Server",
description="Locks true import dates; manages NFOs and mtimes for TV & Movies",
version="1.7.0",
)
start_time = datetime.now(timezone.utc)
db_path = Path(os.environ.get("DB_PATH", "/app/data/media_dates.db"))
def _norm(s: str) -> str:
return (s or "").strip().lower()
def _bool_env(name: str, default: bool) -> bool:
v = os.environ.get(name)
if v is None:
return default
return _norm(v) in ("1", "true", "yes", "y", "on")
class WebhookConfig:
def __init__(self):
tv_paths_str = os.environ.get("TV_PATHS", os.environ.get("MEDIA_PATH", "/media/tv,/media/tv6"))
self.tv_paths = [Path(p.strip()) for p in tv_paths_str.split(",") if p.strip()]
movie_paths_str = os.environ.get("MOVIE_PATHS", "/media/movies,/media/movies6")
self.movie_paths = [Path(p.strip()) for p in movie_paths_str.split(",") if p.strip()]
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.batch_delay = float(os.environ.get("BATCH_DELAY", "5.0"))
self.max_concurrent_series = int(os.environ.get("MAX_CONCURRENT_SERIES", "3"))
self.manager_brand = os.environ.get("MANAGER_BRAND", "NFOGuard")
# Radarr
self.radarr_url = (os.environ.get("RADARR_URL", "") or "").rstrip("/")
self.radarr_api_key = os.environ.get("RADARR_API_KEY", "") or ""
self.radarr_timeout = int(os.environ.get("RADARR_TIMEOUT", "45"))
self.radarr_retries = int(os.environ.get("RADARR_RETRIES", "3"))
# Optional externals
self.tmdb_api_key = os.environ.get("TMDB_API_KEY") or ""
self.tmdb_country = (os.environ.get("TMDB_PRIMARY_COUNTRY") or "US").upper()
self.omdb_api_key = os.environ.get("OMDB_API_KEY") or ""
self.jelly_url = (os.environ.get("JELLYSEERR_URL") or "").rstrip("/")
self.jelly_api = os.environ.get("JELLYSEERR_API_KEY") or ""
# Legacy (logged only)
raw_strategy = _norm(os.environ.get("MOVIE_DATE_STRATEGY", "radarr_import"))
aliases = {
"import": "radarr_import",
"radarr": "radarr_import",
"digital_release": "digital",
"physical_release": "physical",
"in_cinemas": "cinema",
"theatrical": "cinema",
"mtime": "file_mtime",
}
self.movie_date_strategy = aliases.get(raw_strategy, raw_strategy)
# Priority
self.movie_priority = _norm(os.environ.get("MOVIE_PRIORITY", "import_then_digital"))
if self.movie_priority not in ("import_then_digital", "digital_then_import"):
self.movie_priority = "import_then_digital"
# Poll policy
self.movie_poll_mode = _norm(os.environ.get("MOVIE_POLL_MODE", "always")) # always|if_missing|never
if self.movie_poll_mode not in ("always", "if_missing", "never"):
self.movie_poll_mode = "always"
# Update policy
self.movie_date_update_mode = _norm(os.environ.get("MOVIE_DATE_UPDATE_MODE", "backfill_only"))
self.movie_preserve_radarr_import = _bool_env("MOVIE_PRESERVE_RADARR_IMPORT", True)
self.movie_preserve_existing = _bool_env("MOVIE_PRESERVE_EXISTING", True)
# Digital sanity check
self.movie_digital_sanity_check = _bool_env("MOVIE_DIGITAL_SANITY_CHECK", False)
self.movie_digital_max_lag_days = int(os.environ.get("MOVIE_DIGITAL_MAX_LAG_DAYS", "365"))
self.movie_digital_sanity_ref = _norm(os.environ.get("MOVIE_DIGITAL_SANITY_REF", "cinema")) # cinema|import|min
if self.movie_digital_sanity_ref not in ("cinema", "import", "min"):
self.movie_digital_sanity_ref = "cinema"
self.movie_digital_invalid_fallback = _norm(os.environ.get("MOVIE_DIGITAL_INVALID_FALLBACK", "import")) # import|cinema
if self.movie_digital_invalid_fallback not in ("import", "cinema"):
self.movie_digital_invalid_fallback = "import"
config = WebhookConfig()
# ---------------------------
# DB schema (movies)
# ---------------------------
MOVIE_SCHEMA = """
CREATE TABLE IF NOT EXISTS movies (
imdb_id TEXT PRIMARY KEY,
movie_path TEXT NOT NULL,
last_applied TEXT
);
CREATE TABLE IF NOT EXISTS movie_dates (
imdb_id TEXT PRIMARY KEY,
released TEXT,
dateadded TEXT,
source TEXT,
has_video_file INTEGER DEFAULT 0
);
"""
def ensure_movie_tables(conn: sqlite3.Connection):
try:
result = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='movies'").fetchone()
if not result:
_log("INFO", "Creating movie tables in database")
conn.executescript(MOVIE_SCHEMA)
conn.commit()
except sqlite3.Error as e:
_log("ERROR", f"Failed to create movie tables: {e}")
# ---------------------------
# HTTP helper
# ---------------------------
def _get_json(url: str, timeout: int = 20, headers: Dict[str, str] = None) -> Optional[Any]:
try:
req = UrlRequest(url, headers=headers or {"Accept": "application/json"})
with urlopen(req, timeout=timeout) as resp:
return json.loads(resp.read().decode("utf-8"))
except Exception as e:
_log("WARNING", f"GET {url} failed: {e}")
return None
# ---------------------------
# Radarr client
# ---------------------------
class RadarrClient:
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]:
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(self.base_url, path.lstrip("?").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)
_log("DEBUG", f"Radarr API Response: {json.dumps(result, indent=2) if isinstance(result, (dict, list)) else result}")
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))
attempt += 1
_log("WARNING", f"Radarr GET {path} failed after retries: {last_err}")
return None
def movie_by_imdb(self, imdb_id: str) -> Optional[Dict[str, Any]]:
imdb_id = imdb_id if imdb_id.startswith("tt") else f"tt{imdb_id}"
_log("DEBUG", f"Radarr API: Looking up movie by IMDb ID: {imdb_id}")
# Radarr's IMDb filtering is unreliable, so fetch all movies and filter client-side
res = self._get("/api/v3/movie")
if isinstance(res, list):
# Filter results to find exact IMDb ID match
for movie in res:
if movie.get("imdbId") == imdb_id:
_log("DEBUG", f"Radarr API: Found exact match: {movie.get('title')} (ID: {movie.get('id')})")
return movie
_log("WARNING", f"Radarr API: Searched {len(res)} movies but none match IMDb {imdb_id}")
# Fallback: Try lookup endpoint
res = self._get("/api/v3/movie/lookup/imdb", {"imdbId": imdb_id})
if isinstance(res, dict) and res.get("id") and res.get("imdbId") == imdb_id:
_log("DEBUG", f"Radarr API: Found movie via lookup: {res.get('title')} (ID: {res.get('id')})")
return res
_log("DEBUG", f"Radarr API: No movie found for IMDb ID: {imdb_id}")
return None
def movie_files(self, movie_id: int) -> List[Dict[str, Any]]:
res = self._get("/api/v3/moviefile", {"movieId": movie_id})
return res if isinstance(res, list) else []
def earliest_dateadded(self, movie_id: int) -> Optional[str]:
files = self.movie_files(movie_id)
best = None
for f in files:
da = f.get("dateAdded")
if not da:
continue
try:
dt = datetime.fromisoformat(da.replace("Z", "+00:00"))
if best is None or dt < best:
best = dt
except Exception:
continue
if best:
return best.astimezone(timezone.utc).isoformat(timespec="seconds")
return None
def earliest_import_event(self, movie_id: int) -> Optional[str]:
# Collect all history pages to ensure we don't miss early import events
all_items = []
page = 1
page_size = 1000
while True:
data = self._get("/api/v3/history", {
"movieId": movie_id, "page": page, "pageSize": page_size,
"sortKey": "date", "sortDirection": "ascending"
})
if not data:
break
items = data if isinstance(data, list) else data.get("records") or []
if not items:
break
all_items.extend(items)
_log("DEBUG", f"Page {page}: Found {len(items)} history items for movie_id {movie_id}")
# If we got less than pageSize, we've reached the end
if len(items) < page_size:
break
page += 1
# Safety valve - don't go crazy with pagination
if page > 10:
_log("WARNING", f"Stopped pagination at page 10 for movie_id {movie_id} - may have missed some history")
break
_log("DEBUG", f"Found {len(all_items)} total history items across {page} pages for movie_id {movie_id}")
items = all_items
first_grab = None
for it in items:
ev = (it.get("eventType") or it.get("type") or "").lower()
date_iso = None
if it.get("date"):
try:
date_iso = datetime.fromisoformat(it["date"].replace("Z", "+00:00")).astimezone(timezone.utc).isoformat(timespec="seconds")
except Exception:
date_iso = None
_log("DEBUG", f"History event: type='{ev}', date={date_iso}, data={it.get('data', {})}")
# Expanded event types for Radarr v4/v5 compatibility
import_events = {
"downloadfolderimported", "moviefileimported", "imported",
"downloadimported", "fileimported", "movieimported"
}
if ev in import_events:
# Check if this is a real import (from downloads) vs existing file rename
event_data = it.get('data', {})
source_path = event_data.get('sourcePath', '') or event_data.get('path', '')
# Real imports typically come from download folders
is_real_import = any(download_indicator in source_path.lower() for download_indicator in [
'/downloads/', '/download/', '/completed/', '/nzbs/', '/torrents/',
'sabnzbd', 'nzbget', 'deluge', 'qbittorrent', 'transmission'
])
if is_real_import and date_iso:
_log("INFO", f"Found REAL import event '{ev}' from downloads at {date_iso} (source: {source_path})")
return date_iso
elif date_iso:
_log("DEBUG", f"Skipping non-import event '{ev}' (rename/existing file) from {source_path}")
if ev == "grabbed" and date_iso and first_grab is None:
first_grab = date_iso
if first_grab:
_log("INFO", f"Using grab date {first_grab} as fallback for movie_id {movie_id}")
return first_grab
_log("WARNING", f"No real import/grab events found for movie_id {movie_id} (only renames/existing files)")
return None
# ---------------------------
# TMDB / OMDb / Jellyseerr helpers
# ---------------------------
def _tmdb_id_from_imdb(imdb_id: str) -> Optional[int]:
if not config.tmdb_api_key:
return None
try:
url_find = f"https://api.themoviedb.org/3/find/{quote(imdb_id)}?{urlencode({'api_key': config.tmdb_api_key, 'external_source':'imdb_id'})}"
j = _get_json(url_find, timeout=20)
res = (j or {}).get("movie_results") or []
if res:
return res[0].get("id")
return None
except Exception:
return None
def tmdb_digital_date_from_imdb(imdb_id: str) -> Optional[str]:
if not config.tmdb_api_key:
return None
try:
tmdb_id = _tmdb_id_from_imdb(imdb_id)
if not tmdb_id:
return None
url_rel = f"https://api.themoviedb.org/3/movie/{tmdb_id}/release_dates?{urlencode({'api_key': config.tmdb_api_key})}"
k = _get_json(url_rel, timeout=20)
if not k:
return None
for entry in (k.get("results") or []):
if (entry.get("iso_3166_1") or "").upper() != config.tmdb_country:
continue
for rd in (entry.get("release_dates") or []):
if rd.get("type") == 4 and rd.get("release_date"): # 4 = Digital
try:
dt = datetime.fromisoformat(rd["release_date"].replace("Z", "+00:00"))
return dt.astimezone(timezone.utc).isoformat(timespec="seconds")
except Exception:
continue
return None
except Exception as e:
_log("WARNING", f"TMDB lookup failed for {imdb_id}: {e}")
return None
def jellyseerr_digital_date_from_tmdb(tmdb_id: int) -> Optional[str]:
base = config.jelly_url
api = config.jelly_api
if not (base and api and tmdb_id):
return None
try:
url = base.rstrip("/") + f"/api/v1/movie/{tmdb_id}"
j = _get_json(url, timeout=20, headers={"X-Api-Key": api, "Accept": "application/json"})
if not j:
return None
candidates = []
for k in ("digitalReleaseDate", "physicalReleaseDate", "vodReleaseDate"):
v = j.get(k)
if v:
iso = _parse_date_to_iso(v)
if iso:
candidates.append(iso)
for arr_key in ("releaseDates", "releases", "dates"):
arr_data = j.get(arr_key)
if not isinstance(arr_data, list):
continue
for rd in arr_data:
if not isinstance(rd, dict):
continue
name = (rd.get("type") or rd.get("label") or "").lower()
v = rd.get("date") or rd.get("releaseDate")
if v and ("digital" in name or "vod" in name or "stream" in name):
iso = _parse_date_to_iso(v)
if iso:
candidates.append(iso)
if not candidates:
return None
return sorted(
candidates,
key=lambda s: datetime.fromisoformat(s.replace("Z", "+00:00"))
)[0]
except Exception as e:
_log("WARNING", f"Jellyseerr lookup failed for tmdb:{tmdb_id}: {e}")
return None
def omdb_dvd_date(imdb_id: str) -> Optional[str]:
if not config.omdb_api_key:
return None
try:
url = f"http://www.omdbapi.com/?{urlencode({'i': imdb_id, 'apikey': config.omdb_api_key})}"
j = _get_json(url, timeout=15)
if not j or _norm(j.get("Response")) != "true":
return None
dvd = j.get("DVD") or j.get("Released")
if not dvd:
return None
for fmt in ("%d %b %Y", "%d %B %Y", "%Y-%m-%d"):
try:
dt = datetime.strptime(dvd, fmt).replace(tzinfo=timezone.utc)
return dt.isoformat(timespec="seconds")
except Exception:
continue
return None
except Exception as e:
_log("WARNING", f"OMDb lookup failed for {imdb_id}: {e}")
return None
# ---------------------------
# XML helpers (pretty + upsert)
# ---------------------------
def _ensure_child_text(parent: ET.Element, tag: str, text: str, overwrite: bool = True) -> ET.Element:
child = parent.find(tag)
if child is None:
child = ET.SubElement(parent, tag)
child.text = text
else:
if overwrite or (child.text is None or str(child.text).strip() == ""):
child.text = text
return child
def _ensure_uniqueid_imdb(root: ET.Element, imdb_id: str):
imdb_elem = None
for uid in root.findall("uniqueid"):
if (uid.attrib.get("type") or "").lower() == "imdb":
imdb_elem = uid
break
if imdb_elem is None:
imdb_elem = ET.SubElement(root, "uniqueid", {"type": "imdb", "default": "true"})
else:
if "default" not in imdb_elem.attrib:
imdb_elem.set("default", "true")
imdb_elem.text = imdb_id
def _indent(elem: ET.Element, level: int = 0, space: str = " "):
i = "\n" + level * space
if len(elem):
if not elem.text or not elem.text.strip():
elem.text = i + space
for idx, e in enumerate(list(elem)):
_indent(e, level + 1, space)
if not e.tail or not e.tail.strip():
e.tail = i + (space if idx < len(elem) - 1 else "")
if not elem.tail or not elem.tail.strip():
elem.tail = "\n" + (level - 1) * space if level else "\n"
else:
if not elem.text or not elem.text.strip():
elem.text = ""
if not elem.tail or not elem.tail.strip():
elem.tail = "\n" + (level - 1) * space if level else "\n"
def _strip_existing_footer_comments(root: ET.Element):
for node in list(root):
if isinstance(node.tag, str):
continue
txt = (node.text or "").strip().lower()
if "source:" in txt or "managed by" in txt:
root.remove(node)
def upsert_movie_nfo(
movie_dir: Path,
imdb_id: str,
released: Optional[str],
dateadded: Optional[str],
source_detail: str,
lock_metadata: bool = True,
media_label: str = "Movie",
brand: str = "NFOGuard",
):
canonical_path = movie_dir / "movie.nfo"
alt_path = movie_dir / f"{movie_dir.name}.nfo"
root = None
for p in (canonical_path, alt_path):
if p.exists():
try:
t = ET.parse(str(p))
r = t.getroot()
if r is not None and r.tag.lower() == "movie":
root = r
break
except Exception:
pass
if root is None:
root = ET.Element("movie")
if imdb_id and not imdb_id.lower().startswith("tt"):
imdb_id = f"tt{imdb_id}"
if imdb_id:
_ensure_uniqueid_imdb(root, imdb_id)
if released:
ymd = released.split("T")[0]
_ensure_child_text(root, "premiered", ymd, overwrite=False)
_ensure_child_text(root, "year", ymd.split("-")[0], overwrite=False)
if dateadded:
if dateadded == "MANUAL_REVIEW_NEEDED":
_ensure_child_text(root, "dateadded", "MANUAL_REVIEW_NEEDED", overwrite=True)
_log("WARNING", f"Set MANUAL_REVIEW_NEEDED marker in NFO: {movie_dir}/movie.nfo")
else:
_ensure_child_text(root, "dateadded", dateadded, overwrite=True)
if lock_metadata:
_ensure_child_text(root, "lockdata", "true", overwrite=True)
_strip_existing_footer_comments(root)
root.append(ET.Comment(f" source: {media_label}; {source_detail} "))
root.append(ET.Comment(f" managed by {brand} "))
_indent(root)
for target in [canonical_path] + ([alt_path] if alt_path.exists() else []):
try:
target.parent.mkdir(parents=True, exist_ok=True)
ET.ElementTree(root).write(str(target), encoding="utf-8", xml_declaration=True)
_log("DEBUG", f"Upserted NFO: {target}")
except Exception as e:
_log("ERROR", f"Failed writing {target}: {e}")
# ---------------------------
# Date selection
# ---------------------------
def _parse_date_to_iso(d: str) -> Optional[str]:
if not d:
return None
try:
if re.fullmatch(r"\d{4}-\d{2}-\d{2}", d):
dt = datetime.fromisoformat(d).replace(tzinfo=timezone.utc)
else:
dt = datetime.fromisoformat(d.replace("Z", "+00:00")).astimezone(timezone.utc)
# Reject dates that are too recent (prevent bogus timestamps)
now = datetime.now(timezone.utc)
if (now - dt).days < 1:
_log("WARNING", f"Rejecting recent timestamp (less than 1 day old): {d} -> {dt.isoformat()}")
return None
return dt.isoformat(timespec="seconds")
except Exception:
return None
def _radarr_movie_by_imdb(imdb_id: str) -> Optional[Dict[str, Any]]:
if not (config.radarr_url and config.radarr_api_key):
return None
rc = RadarrClient(config.radarr_url, config.radarr_api_key, config.radarr_timeout, config.radarr_retries)
return rc.movie_by_imdb(imdb_id)
def _radarr_earliest_import_history_iso(movie_id: int) -> Optional[str]:
rc = RadarrClient(config.radarr_url, config.radarr_api_key, config.radarr_timeout, config.radarr_retries)
return rc.earliest_import_event(movie_id)
def _radarr_earliest_file_dateadded_iso(movie_id: int) -> Optional[str]:
rc = RadarrClient(config.radarr_url, config.radarr_api_key, config.radarr_timeout, config.radarr_retries)
return rc.earliest_dateadded(movie_id)
def _strategy_movie_date_from_movie_obj(movie_obj: Dict[str, Any], which: str) -> Optional[str]:
key_map = {
"digital": ["digitalRelease", "digitalReleaseDate"],
"physical": ["physicalRelease", "physicalReleaseDate"],
"cinema": ["inCinemas", "inCinema", "theatricalRelease"],
}
for k in key_map.get(which, []):
val = movie_obj.get(k)
iso = _parse_date_to_iso(val) if val else None
if iso:
return iso
return None
def _collect_digital_candidates(imdb_id: str, movie_obj: Optional[Dict[str, Any]]) -> List[Tuple[str, str]]:
cands: List[Tuple[str, str]] = []
_log("DEBUG", f"Collecting digital dates for {imdb_id}")
if movie_obj:
iso = _strategy_movie_date_from_movie_obj(movie_obj, "digital")
if iso:
cands.append((iso, "radarr:digitalRelease"))
_log("DEBUG", f"Found Radarr digital date: {iso}")
iso = tmdb_digital_date_from_imdb(imdb_id)
if iso:
cands.append((iso, "tmdb:digital"))
_log("DEBUG", f"Found TMDB digital date: {iso}")
tmdb_id = _tmdb_id_from_imdb(imdb_id)
if tmdb_id:
iso = jellyseerr_digital_date_from_tmdb(tmdb_id)
if iso:
cands.append((iso, "jellyseerr:digital"))
_log("DEBUG", f"Found Jellyseerr digital date: {iso}")
iso = omdb_dvd_date(imdb_id)
if iso:
cands.append((iso, "omdb:dvd"))
_log("DEBUG", f"Found OMDb DVD date: {iso}")
_log("INFO", f"Found {len(cands)} digital date candidates: {[f'{src}={date}' for date, src in cands]}")
return cands
def _pick_earliest(cands: List[Tuple[str, str]]) -> Optional[Tuple[str, str]]:
if not cands:
return None
try:
return sorted(
cands,
key=lambda t: datetime.fromisoformat(t[0].replace("Z", "+00:00"))
)[0]
except Exception:
return cands[0]
def _pick_digital_closest_to_cinema(cands: List[Tuple[str, str]], cinema_date: Optional[str]) -> Optional[Tuple[str, str]]:
"""Pick digital release date closest to cinema date, with TMDB preferred over Radarr as tiebreaker"""
if not cands:
return None
if not cinema_date:
return _pick_earliest(cands) # Fall back to earliest if no cinema date
try:
cinema_dt = datetime.fromisoformat(cinema_date.replace("Z", "+00:00"))
# Source preference: TMDB > Jellyseerr > OMDb > Radarr
source_priority = {
"tmdb:digital": 0,
"jellyseerr:digital": 1,
"omdb:dvd": 2,
"radarr:digitalRelease": 3
}
# Sort candidates by cinema proximity first, then by source preference
def sort_key(candidate):
try:
cand_dt = datetime.fromisoformat(candidate[0].replace("Z", "+00:00"))
days_diff = abs((cand_dt - cinema_dt).days)
# Primary sort: distance from cinema (within 5 years preferred)
if days_diff <= 1825: # 5 years * 365 days
distance_score = days_diff
else:
# Heavily penalize dates far from cinema release
distance_score = days_diff + 10000
# Secondary sort: source preference (lower number = higher preference)
source_score = source_priority.get(candidate[1], 99)
# Return tuple for sorting: (distance, source_priority)
return (distance_score, source_score)
except Exception:
return (999999, 99) # Invalid dates/sources go to the end
sorted_cands = sorted(cands, key=sort_key)
selected = sorted_cands[0]
# Log the decision for debugging
try:
selected_dt = datetime.fromisoformat(selected[0].replace("Z", "+00:00"))
days_diff = (selected_dt - cinema_dt).days
_log("DEBUG", f"Selected digital date {selected[0]} ({selected[1]}) - {days_diff} days from cinema date {cinema_date}")
# Show other candidates if multiple exist
if len(cands) > 1:
other_cands = [f"{c[1]}={c[0]}" for c in sorted_cands[1:3]] # Show top 2 alternatives
_log("DEBUG", f"Other candidates considered: {other_cands}")
except Exception:
pass
return selected
except Exception:
_log("WARNING", "Failed to parse cinema date for digital selection, using earliest")
return _pick_earliest(cands)
def _decide_movie_date(imdb_id: str, movie_dir: Path, poll_external: bool) -> Tuple[Optional[str], str, Optional[str]]:
"""
Decide <dateadded> and supporting 'released' (premiered/year) for NFO.
Returns: (dateadded_iso, source_detail, released_iso)
"""
# Initialize fresh variables for each movie
released_iso: Optional[str] = None
movie_obj: Optional[Dict[str, Any]] = None
movie_id: Optional[int] = None
_log("DEBUG", f"Starting date selection for {imdb_id} - fresh state initialized")
def from_file_mtime() -> Tuple[Optional[str], str]:
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
newest = None
for f in movie_dir.iterdir():
if f.is_file() and f.suffix.lower() in video_exts:
try:
mt = f.stat().st_mtime
if newest is None or mt > newest:
newest = mt
except Exception:
continue
if newest:
try:
iso = datetime.fromtimestamp(newest, tz=timezone.utc).isoformat(timespec="seconds")
return iso, "file:mtime"
except Exception:
pass
return None, "file:mtime"
def ensure_movie():
nonlocal movie_obj, movie_id, released_iso
if movie_obj is None and poll_external and config.radarr_url and config.radarr_api_key:
_log("DEBUG", f"Looking up Radarr movie for IMDb ID: {imdb_id}")
movie_obj = _radarr_movie_by_imdb(imdb_id)
if movie_obj:
movie_id = movie_obj.get("id")
movie_title = movie_obj.get("title", "Unknown")
movie_imdb = movie_obj.get("imdbId", "Unknown")
# CRITICAL: Verify returned movie matches requested IMDb ID
if movie_imdb.lower() != imdb_id.lower():
_log("ERROR", f"Radarr API returned WRONG movie! Requested: {imdb_id}, Got: {movie_imdb} ({movie_title}). Ignoring this result.")
movie_obj = None
movie_id = None
_log("WARNING", f"Movie {imdb_id} not properly found in Radarr - will use digital/cinema dates instead")
else:
_log("INFO", f"Found Radarr movie: {movie_title} (ID: {movie_id}, IMDb: {movie_imdb}) for requested {imdb_id}")
released_iso = _parse_date_to_iso(movie_obj.get("inCinemas") or movie_obj.get("theatricalRelease"))
else:
movie_id = None # Explicitly set to None when movie not found
movie_obj = None
_log("WARNING", f"No Radarr movie found for IMDb ID: {imdb_id} - will skip import history")
def from_import_history() -> Tuple[Optional[str], str]:
ensure_movie()
if not poll_external:
_log("DEBUG", f"Skipping import history: poll_external={poll_external}")
return None, "radarr:history.skipped"
if not movie_id:
_log("DEBUG", f"Skipping import history: movie not found in Radarr for {imdb_id}")
return None, "radarr:movie.not_found"
# Try import history events first (most reliable)
iso = _radarr_earliest_import_history_iso(movie_id)
if iso:
_log("INFO", f"Using Radarr import history date: {iso} for movie_id {movie_id} (IMDb: {imdb_id})")
return iso, "radarr:history.import"
# If no real import events found, return None to continue to digital/cinema dates
# instead of falling back to arbitrary moviefile.dateAdded
_log("DEBUG", f"No real import events found for movie_id {movie_id}, will try digital/cinema dates")
return None, "radarr:history.none"
def from_radarr_field(which: str) -> Tuple[Optional[str], str]:
ensure_movie()
label_map = {"digital": "radarr:digitalRelease", "physical": "radarr:physicalRelease", "cinema": "radarr:inCinemas"}
if poll_external and movie_obj:
iso = _strategy_movie_date_from_movie_obj(movie_obj, which)
return iso, label_map.get(which, "radarr:field")
return None, label_map.get(which, "radarr:field")
def from_external_digital_bundle() -> Tuple[Optional[str], str]:
ensure_movie()
cands = _collect_digital_candidates(imdb_id, movie_obj)
# Use cinema date to pick the most appropriate digital release
cinema_date = None
if movie_obj:
cinema_date = movie_obj.get("inCinemas") or movie_obj.get("theatricalRelease")
picked = _pick_digital_closest_to_cinema(cands, cinema_date)
return picked if picked else (None, "digital:none")
def from_moviefile_dateadded() -> Tuple[Optional[str], str]:
"""Last resort fallback for existing files that were just renamed/scanned"""
ensure_movie()
if not movie_id:
return None, "radarr:moviefile.not_found"
iso = _radarr_earliest_file_dateadded_iso(movie_id)
if iso:
_log("INFO", f"Using Radarr moviefile dateAdded as fallback: {iso} for movie_id {movie_id}")
return iso, "radarr:moviefile.dateAdded"
return None, "radarr:moviefile.none"
def maybe_sanitize_digital(digi_iso: Optional[str]) -> Tuple[Optional[str], str]:
if not digi_iso or not config.movie_digital_sanity_check:
return digi_iso, "radarr:digitalRelease"
ref_dates: List[datetime] = []
if config.movie_digital_sanity_ref in ("cinema", "min"):
cine_iso, _ = from_radarr_field("cinema")
if cine_iso:
try:
ref_dates.append(datetime.fromisoformat(cine_iso.replace("Z", "+00:00")))
except Exception:
pass
if config.movie_digital_sanity_ref in ("import", "min"):
imp_iso, _ = from_import_history()
if imp_iso:
try:
ref_dates.append(datetime.fromisoformat(imp_iso.replace("Z", "+00:00")))
except Exception:
pass
if not ref_dates:
return digi_iso, "radarr:digitalRelease"
try:
d_digi = datetime.fromisoformat(digi_iso.replace("Z", "+00:00"))
ref_dt = min(ref_dates)
lag = (d_digi - ref_dt).days
if lag > config.movie_digital_max_lag_days:
if config.movie_digital_invalid_fallback == "import":
imp_iso2, imp_src2 = from_import_history()
if imp_iso2:
return imp_iso2, "radarr:history.import"
cine_iso2, _ = from_radarr_field("cinema")
if cine_iso2:
return cine_iso2, "radarr:inCinemas"
except Exception:
pass
return digi_iso, "radarr:digitalRelease"
# Priority flows
_log("INFO", f"Date selection for {imdb_id}: priority={config.movie_priority}, poll_external={poll_external}")
if config.movie_priority == "import_then_digital":
_log("DEBUG", "Trying import history first (import_then_digital mode)")
imp_iso, imp_src = from_import_history()
if imp_iso:
_log("INFO", f"Selected import date: {imp_iso} ({imp_src})")
return imp_iso, imp_src, released_iso
_log("DEBUG", "No import date found, trying digital release dates")
digi_iso, digi_src = from_external_digital_bundle()
digi_iso, digi_src = maybe_sanitize_digital(digi_iso)
if digi_iso:
_log("INFO", f"Selected digital date: {digi_iso} ({digi_src})")
return digi_iso, digi_src, released_iso
_log("DEBUG", "No digital date found, trying cinema release")
cine_iso, cine_src = from_radarr_field("cinema")
if cine_iso:
_log("INFO", f"Selected cinema date: {cine_iso} ({cine_src})")
return cine_iso, cine_src, released_iso or cine_iso
_log("DEBUG", "No external dates found, trying moviefile dateAdded as fallback")
moviefile_iso, moviefile_src = from_moviefile_dateadded()
if moviefile_iso:
_log("WARNING", f"Using moviefile dateAdded for existing/renamed file: {moviefile_iso} ({moviefile_src})")
return moviefile_iso, moviefile_src, released_iso
_log("DEBUG", "No moviefile date found, falling back to file mtime")
mtime_iso, mtime_src = from_file_mtime()
if mtime_iso:
_log("WARNING", f"Using file mtime as last resort: {mtime_iso} ({mtime_src})")
return mtime_iso, mtime_src, released_iso
# No valid date found - create manual review marker
_log("ERROR", f"No valid dateadded found for {imdb_id} - marking for MANUAL_REVIEW_NEEDED")
return "MANUAL_REVIEW_NEEDED", "manual_review_required", released_iso
# digital_then_import
_log("DEBUG", "Trying digital release dates first (digital_then_import mode)")
digi_iso, digi_src = from_external_digital_bundle()
digi_iso, digi_src = maybe_sanitize_digital(digi_iso)
if digi_iso:
_log("INFO", f"Selected digital date: {digi_iso} ({digi_src})")
return digi_iso, digi_src, released_iso
_log("DEBUG", "No digital date found, trying import history")
imp_iso, imp_src = from_import_history()
if imp_iso:
_log("INFO", f"Selected import date: {imp_iso} ({imp_src})")
return imp_iso, imp_src, released_iso
_log("DEBUG", "No import date found, trying cinema release")
cine_iso, cine_src = from_radarr_field("cinema")
if cine_iso:
_log("INFO", f"Selected cinema date: {cine_iso} ({cine_src})")
return cine_iso, cine_src, released_iso or cine_iso
_log("DEBUG", "No external dates found, falling back to file mtime")
mtime_iso, mtime_src = from_file_mtime()
if mtime_iso:
_log("WARNING", f"Using file mtime as last resort: {mtime_iso} ({mtime_src})")
return mtime_iso, mtime_src, released_iso
# No valid date found - create manual review marker
_log("ERROR", f"No valid dateadded found for {imdb_id} - marking for MANUAL_REVIEW_NEEDED")
return "MANUAL_REVIEW_NEEDED", "manual_review_required", released_iso
# ---------------------------
# Movie apply
# ---------------------------
def parse_imdb_from_movie_path(p: Path) -> Optional[str]:
m = re.search(r'\[imdb-(tt\d+)\]', p.name, re.IGNORECASE)
imdb_id = m.group(1).lower() if m else None
_log("DEBUG", f"Parsed IMDb ID from path '{p.name}': {imdb_id}")
return imdb_id
def is_movie_dir(p: Path) -> bool:
return p.is_dir() and parse_imdb_from_movie_path(p) is not None
def set_movie_file_mtime(movie_dir: Path, dateadded_iso: str):
if not dateadded_iso:
return
try:
ts = datetime.fromisoformat(dateadded_iso.replace("Z", "+00:00")).timestamp()
# Update ALL files in the directory to the correct timestamp
files_updated = 0
for f in movie_dir.iterdir():
if f.is_file():
try:
os.utime(f, (ts, ts), follow_symlinks=False)
files_updated += 1
_log("DEBUG", f"Updated file mtime: {f.name} -> {dateadded_iso}")
except Exception as e:
_log("WARNING", f"Failed to update mtime on {f.name}: {e}")
# Update directory timestamp
try:
os.utime(movie_dir, (ts, ts), follow_symlinks=False)
_log("INFO", f"Updated {files_updated} files and directory mtime: {movie_dir.name} -> {dateadded_iso}")
except Exception as e:
_log("WARNING", f"Failed to update mtime on directory {movie_dir}: {e}")
except Exception as e:
_log("WARNING", f"Bad timestamp for movie mtimes: {dateadded_iso}: {e}")
def apply_movie(movie_dir: Path, conn: sqlite3.Connection):
imdb_id = parse_imdb_from_movie_path(movie_dir)
if not imdb_id:
_log("ERROR", f"Movie path does not contain an IMDb id [imdb-tt...]: {movie_dir}")
return
ensure_movie_tables(conn)
conn.execute(
"INSERT INTO movies(imdb_id, movie_path, last_applied) VALUES(?,?,?) "
"ON CONFLICT(imdb_id) DO UPDATE SET movie_path=excluded.movie_path, last_applied=excluded.last_applied",
(imdb_id, str(movie_dir), _iso_now()),
)
video_exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v")
has_video_files = any(f.is_file() and f.suffix.lower() in video_exts for f in movie_dir.iterdir())
if not has_video_files:
_log("WARNING", f"No video files found in movie directory: {movie_dir}")
conn.execute(
"INSERT INTO movie_dates(imdb_id, released, dateadded, source, has_video_file) VALUES(?,?,?,?,?) "
"ON CONFLICT(imdb_id) DO UPDATE SET has_video_file=excluded.has_video_file",
(imdb_id, None, None, None, 0),
)
conn.commit()
return
row = conn.execute(
"SELECT released, dateadded, source FROM movie_dates WHERE imdb_id=?", (imdb_id,)
).fetchone()
existing_released = row[0] if row else None
existing_dateadded = row[1] if row else None
existing_source = row[2] if row else None
# poll policy
if config.movie_poll_mode == "never":
poll_external = False
elif config.movie_poll_mode == "if_missing":
poll_external = not existing_dateadded or (existing_source or "") == "file:mtime"
else:
poll_external = True
preserve = False
if config.movie_date_update_mode == "backfill_only":
if existing_dateadded:
if (config.movie_preserve_radarr_import and (existing_source or "").startswith("radarr:moviefile")) \
or (config.movie_preserve_existing and (existing_source or "") != "file:mtime"):
preserve = True
if preserve:
dateadded_iso = existing_dateadded
src_detail = existing_source or "unknown"
if not existing_released and poll_external and (config.radarr_url and config.radarr_api_key):
movie_obj = _radarr_movie_by_imdb(imdb_id)
if movie_obj:
existing_released = _parse_date_to_iso(movie_obj.get("inCinemas") or movie_obj.get("theatricalRelease"))
chosen_released = existing_released
_log("INFO", f"Preserving existing movie date for {movie_dir.name} ({dateadded_iso}; {src_detail})")
else:
dateadded_iso, src_detail, decided_released = _decide_movie_date(imdb_id, movie_dir, poll_external=poll_external)
chosen_released = existing_released or decided_released
conn.execute(
"INSERT INTO movie_dates(imdb_id, released, dateadded, source, has_video_file) VALUES(?,?,?,?,?) "
"ON CONFLICT(imdb_id) DO UPDATE SET released=excluded.released, dateadded=excluded.dateadded, "
"source=excluded.source, has_video_file=excluded.has_video_file",
(imdb_id, chosen_released, dateadded_iso, src_detail, 1),
)
if config.manage_nfo:
upsert_movie_nfo(
movie_dir,
imdb_id,
chosen_released,
dateadded_iso,
source_detail=src_detail or "unknown",
lock_metadata=config.lock_metadata,
media_label="Movie",
brand=config.manager_brand,
)
if config.fix_dir_mtimes and dateadded_iso and dateadded_iso != "MANUAL_REVIEW_NEEDED":
set_movie_file_mtime(movie_dir, dateadded_iso)
conn.commit()
_log("INFO", f"Processed movie: {movie_dir.name} (source={src_detail or 'unknown'}; preserved={preserve})")
# Add detailed logging for debugging with searchable format
_log("INFO", f"MOVIE_SUMMARY: imdb={imdb_id} title='{movie_dir.name}' dateadded={dateadded_iso} source={src_detail} preserved={preserve}")
# ---------------------------
# Batching
# ---------------------------
class WebhookBatcher:
def __init__(self):
self.pending_items: 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_series, thread_name_prefix="media-processor")
def add_webhook(self, key: str, webhook_data: Dict, media_type: str):
with self.lock:
if key in self.timers:
self.timers[key].cancel()
_log("DEBUG", f"Cancelled existing timer for {key}")
webhook_data['media_type'] = media_type
self.pending_items[key] = webhook_data
_log("INFO", f"Batched {media_type} webhook for {key} (delay: {config.batch_delay}s)")
timer = threading.Timer(config.batch_delay, self._process_item, args=[key])
self.timers[key] = timer
timer.start()
def _process_item(self, key: str):
with self.lock:
if key in self.processing or key not in self.pending_items:
return
self.processing.add(key)
webhook_data = self.pending_items.pop(key)
self.timers.pop(key, None)
try:
media_type = webhook_data.get('media_type', 'unknown')
_log("INFO", f"Processing batched {media_type}: {key}")
self.executor.submit(self._process_item_sync, key, webhook_data)
except Exception as e:
_log("ERROR", f"Error submitting {media_type} processing for {key}: {e}")
with self.lock:
self.processing.discard(key)
def _process_item_sync(self, key: str, webhook_data: Dict):
try:
media_type = webhook_data.get('media_type', 'unknown')
if media_type == 'tv':
self._process_tv_sync(key, webhook_data)
elif media_type == 'movie':
self._process_movie_sync(key, webhook_data)
else:
_log("ERROR", f"Unknown media type: {media_type}")
finally:
with self.lock:
self.processing.discard(key)
def _process_tv_sync(self, imdb_id: str, webhook_data: Dict):
try:
series_info = webhook_data.get('series', {})
series_title = series_info.get("title", "Unknown")
series_path_str = webhook_data.get('series_path')
if not series_path_str:
_log("ERROR", f"No series path found for {imdb_id}")
return
series_path = Path(series_path_str)
if not series_path.exists():
_log("ERROR", f"Series path does not exist: {series_path}")
return
conn = None
try:
conn = db_connect(db_path)
conn.execute("PRAGMA busy_timeout = 30000")
conn.execute("PRAGMA journal_mode = WAL")
class MockArgs:
def __init__(self):
self.manage_nfo = config.manage_nfo
self.fix_dir_mtimes = config.fix_dir_mtimes
self.lock_metadata = config.lock_metadata
self.use_cache = True
self.force_refresh = False
self.skip_if_current = False
self.debug = config.debug
self.update_video_mtimes = True
_log("INFO", f"Processing TV series: {series_title} at {series_path}")
apply_series(series_path, MockArgs(), conn)
_log("INFO", f"Successfully processed batched TV series: {series_title}")
except Exception as e:
_log("ERROR", f"Error processing TV series {series_title}: {e}")
finally:
if conn:
try:
conn.close()
except:
pass
except Exception as e:
_log("ERROR", f"Error in TV processing: {e}")
def _process_movie_sync(self, imdb_id: str, webhook_data: Dict):
try:
movie_info = webhook_data.get('movie', {})
movie_title = movie_info.get("title", "Unknown")
movie_path_str = webhook_data.get('movie_path')
if not movie_path_str:
_log("ERROR", f"No movie path found for {imdb_id}")
return
movie_path = Path(movie_path_str)
if not movie_path.exists():
_log("ERROR", f"Movie path does not exist: {movie_path}")
return
conn = None
try:
conn = db_connect(db_path)
conn.execute("PRAGMA busy_timeout = 30000")
conn.execute("PRAGMA journal_mode = WAL")
_log("INFO", f"Processing movie: {movie_title} at {movie_path}")
apply_movie(movie_path, conn)
_log("INFO", f"Successfully processed batched movie: {movie_title}")
except Exception as e:
_log("ERROR", f"Error processing movie {movie_title}: {e}")
finally:
if conn:
try:
conn.close()
except:
pass
except Exception as e:
_log("ERROR", f"Error in movie processing: {e}")
def get_status(self) -> Dict:
with self.lock:
return {
"pending_items": list(self.pending_items.keys()),
"processing_items": list(self.processing),
"pending_count": len(self.pending_items),
"processing_count": len(self.processing),
"executor_active_threads": getattr(self.executor, '_threads', 0)
}
batcher = WebhookBatcher()
# ---------------------------
# Path helpers
# ---------------------------
def find_series_path_from_sonarr_path(sonarr_path: str) -> Optional[Path]:
if not sonarr_path:
return None
if sonarr_path.startswith("/mnt/unionfs/Media/TV/"):
container_path = sonarr_path.replace("/mnt/unionfs/Media/TV/", "/media/")
path = Path(container_path)
if path.exists():
_log("INFO", f"Found series path via Sonarr path: {path}")
return path
return None
def find_movie_path_from_radarr_path(radarr_path: str) -> Optional[Path]:
if not radarr_path:
return None
if radarr_path.startswith("/mnt/unionfs/Media/Movies/"):
container_path = radarr_path.replace("/mnt/unionfs/Media/Movies/", "/media/")
path = Path(container_path)
if path.exists():
_log("INFO", f"Found movie path via Radarr path: {path}")
return path
return None
def find_series_path(series_title: str, imdb_id: str, sonarr_path: str = None) -> Optional[Path]:
if sonarr_path:
series_path = find_series_path_from_sonarr_path(sonarr_path)
if series_path:
return series_path
if imdb_id and not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
_log("DEBUG", f"Searching for series: {series_title} with IMDb ID: {imdb_id}")
for media_path in config.tv_paths:
if not media_path.exists():
_log("WARNING", f"TV path does not exist: {media_path}")
continue
if imdb_id:
pattern = str(media_path / f"*[imdb-{imdb_id}]*")
matches = glob.glob(pattern)
if matches:
found_path = Path(matches[0])
_log("INFO", f"Found series directory by IMDb ID: {found_path}")
return found_path
if series_title:
series_title_clean = series_title.lower().replace(" ", "").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("-", "").replace(":", "")
if series_title_clean in item_clean:
_log("INFO", f"Found series directory by title: {item}")
return item
_log("WARNING", f"Could not find directory for series: {series_title} ({imdb_id})")
return None
def find_movie_path(movie_title: str, imdb_id: str, radarr_path: str = None) -> Optional[Path]:
if radarr_path:
movie_path = find_movie_path_from_radarr_path(radarr_path)
if movie_path:
return movie_path
if imdb_id and not imdb_id.startswith("tt"):
imdb_id = f"tt{imdb_id}"
_log("DEBUG", f"Searching for movie: {movie_title} with IMDb ID: {imdb_id}")
for media_path in config.movie_paths:
if not media_path.exists():
_log("WARNING", f"Movie path does not exist: {media_path}")
continue
if imdb_id:
pattern = str(media_path / f"*[imdb-{imdb_id}]*")
matches = glob.glob(pattern)
if matches:
found_path = Path(matches[0])
_log("INFO", f"Found movie directory by IMDb ID: {found_path}")
return found_path
if movie_title:
movie_title_clean = movie_title.lower().replace(" ", "").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("-", "").replace(":", "")
if movie_title_clean in item_clean:
_log("INFO", f"Found movie directory by title: {item}")
return item
_log("WARNING", f"Could not find directory for movie: {movie_title} ({imdb_id})")
return None
# ---------------------------
# Webhook processors
# ---------------------------
async def process_sonarr_import(webhook_data: SonarrWebhook):
try:
if not webhook_data.series:
_log("WARNING", f"No series data in {webhook_data.eventType} webhook")
return
series_info = webhook_data.series
series_title = series_info.get("title", "Unknown")
imdb_id = series_info.get("imdbId", "")
sonarr_path = series_info.get("path", "")
if imdb_id:
imdb_id = imdb_id.replace("tt", "").strip()
if imdb_id:
imdb_id = f"tt{imdb_id}"
if not imdb_id:
_log("ERROR", f"No IMDb ID found for series: {series_title}")
return
_log("INFO", f"Received webhook for series: {series_title} (IMDb: {imdb_id})")
series_path = find_series_path(series_title, imdb_id, sonarr_path)
if not series_path:
_log("ERROR", f"Could not find series directory for: {series_title} ({imdb_id})")
return
webhook_dict = {
'series': series_info,
'series_path': str(series_path),
'event_type': webhook_data.eventType
}
batcher.add_webhook(imdb_id, webhook_dict, 'tv')
except Exception as e:
_log("ERROR", f"Error processing Sonarr webhook: {e}")
raise
async def process_radarr_import(webhook_data: RadarrWebhook):
try:
if not webhook_data.movie:
_log("WARNING", f"No movie data in {webhook_data.eventType} webhook")
return
movie_info = webhook_data.movie
movie_title = movie_info.get("title", "Unknown")
radarr_path = movie_info.get("folderPath") or movie_info.get("path", "")
imdb_id = (movie_info.get("imdbId") or (webhook_data.remoteMovie or {}).get("imdbId") or "").strip()
if imdb_id:
imdb_id = f"tt{imdb_id.replace('tt','')}"
movie_path = find_movie_path(movie_title, imdb_id, radarr_path)
if not imdb_id and movie_path:
parsed = parse_imdb_from_movie_path(movie_path)
if parsed:
imdb_id = parsed
if not imdb_id:
_log("WARNING", f"No IMDb ID available for movie '{movie_title}'. Proceeding by path only.")
_log("INFO", f"Received webhook for movie: {movie_title} (IMDb: {imdb_id or 'unknown'})")
if webhook_data.renamedMovieFiles:
_log("INFO", f"Renamed {len(webhook_data.renamedMovieFiles)} file(s) for {movie_title}")
if not movie_path:
_log("ERROR", f"Could not find movie directory for: {movie_title} ({imdb_id or 'unknown'})")
return
webhook_dict = {
'movie': movie_info,
'movie_path': str(movie_path),
'event_type': webhook_data.eventType
}
batcher.add_webhook(imdb_id or str(movie_path), webhook_dict, 'movie')
except Exception as e:
_log("ERROR", f"Error processing Radarr webhook: {e}")
raise
# ---------------------------
# Payload reader (json or form)
# ---------------------------
async def _read_payload(request: Request) -> dict:
ctype = (request.headers.get("content-type") or "").lower()
try:
if "application/json" in ctype:
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 {}
# ---------------------------
# Routes
# ---------------------------
@app.post("/webhook/sonarr")
async def sonarr_webhook(request: Request, background_tasks: BackgroundTasks):
_log("INFO", "Hit /webhook/sonarr")
try:
payload = await _read_payload(request)
if not payload:
raise HTTPException(status_code=422, detail="Empty or unreadable Sonarr payload")
_log("DEBUG", f"Raw Sonarr webhook payload: {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"}
await process_sonarr_import(webhook)
return {"status": "accepted", "message": "Sonarr webhook queued for batched processing"}
except HTTPException:
raise
except Exception as e:
_log("ERROR", f"Failed to parse Sonarr webhook: {e}")
raise HTTPException(status_code=422, detail=f"Invalid Sonarr webhook format: {e}")
@app.post("/webhook/radarr")
async def radarr_webhook(request: Request, background_tasks: BackgroundTasks):
_log("INFO", "Hit /webhook/radarr")
try:
payload = await _read_payload(request)
if not payload:
raise HTTPException(status_code=422, detail="Empty or unreadable Radarr payload")
_log("DEBUG", f"Raw Radarr webhook payload: {payload}")
webhook = RadarrWebhook(**payload)
_log("INFO", f"Received Radarr webhook: {webhook.eventType}")
if webhook.eventType not in ["Download", "Upgrade", "Rename", "MovieFileDelete", "MovieDelete", "Test"]:
return {"status": "ignored", "reason": f"Event type {webhook.eventType} not processed"}
if webhook.eventType != "Test":
await process_radarr_import(webhook)
return {"status": "accepted", "message": "Radarr webhook queued for batched processing"}
else:
return {"status": "ok", "message": "Radarr test received"}
except HTTPException:
raise
except Exception as e:
_log("ERROR", f"Failed to parse Radarr webhook: {e}")
raise HTTPException(status_code=422, detail=f"Invalid Radarr webhook format: {e}")
@app.get("/health")
async def health_check() -> HealthResponse:
uptime = datetime.now(timezone.utc) - start_time
try:
conn = db_connect(db_path)
conn.execute("PRAGMA busy_timeout = 5000")
conn.execute("SELECT 1").fetchone()
conn.close()
db_status = "healthy"
except Exception as e:
db_status = f"error: {e}"
return HealthResponse(
status="healthy" if db_status == "healthy" else "degraded",
version="1.7.0",
uptime=str(uptime),
database_status=db_status
)
@app.get("/stats")
async def get_stats():
try:
conn = db_connect(db_path)
conn.execute("PRAGMA busy_timeout = 5000")
ensure_movie_tables(conn)
tv_series_count = conn.execute("SELECT COUNT(*) FROM series").fetchone()[0]
tv_episode_count = conn.execute("SELECT COUNT(*) FROM episode_dates").fetchone()[0]
tv_with_files = conn.execute("SELECT COUNT(*) FROM episode_dates WHERE has_video_file = 1").fetchone()[0]
movie_count = conn.execute("SELECT COUNT(*) FROM movies").fetchone()[0]
movies_with_files = conn.execute("SELECT COUNT(*) FROM movie_dates WHERE has_video_file = 1").fetchone()[0]
tv_sources = conn.execute("""
SELECT source, COUNT(*) as count
FROM episode_dates
WHERE has_video_file = 1
GROUP BY source
ORDER BY count DESC
""").fetchall()
movie_sources = conn.execute("""
SELECT source, COUNT(*) as count
FROM movie_dates
WHERE has_video_file = 1
GROUP BY source
ORDER BY count DESC
""").fetchall()
conn.close()
return {
"tv_series_count": tv_series_count,
"tv_episode_count": tv_episode_count,
"tv_episodes_with_files": tv_with_files,
"movie_count": movie_count,
"movies_with_files": movies_with_files,
"tv_sources": dict(tv_sources),
"movie_sources": dict(movie_sources)
}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/batch/status")
async def get_batch_status():
return batcher.get_status()
# ---------------------------
# Manual scans
# ---------------------------
def run_manual_scan_sync(scan_paths: List[Path], scan_type: str = "both"):
from media_date_cache import iterate_target_path
class MockArgs:
def __init__(self):
self.manage_nfo = config.manage_nfo
self.fix_dir_mtimes = config.fix_dir_mtimes
self.lock_metadata = config.lock_metadata
self.use_cache = True
self.force_refresh = False
self.skip_if_current = True
self.debug = config.debug
self.update_video_mtimes = True
conn = None
try:
conn = db_connect(db_path)
conn.execute("PRAGMA busy_timeout = 30000")
conn.execute("PRAGMA journal_mode = WAL")
ensure_movie_tables(conn)
for scan_path in scan_paths:
_log("INFO", f"Starting manual scan of: {scan_path}")
if scan_type in ["both", "tv"]:
# Only run TV scan on paths that are in tv_paths
if scan_path in config.tv_paths:
tv_targets = iterate_target_path(scan_path)
for series_dir in tv_targets:
try:
apply_series(series_dir, MockArgs(), conn)
except Exception as e:
_log("ERROR", f"Failed processing TV series {series_dir}: {e}")
if scan_type in ["both", "movies"]:
if scan_path in config.movie_paths:
for item in scan_path.iterdir():
if is_movie_dir(item):
try:
apply_movie(item, conn)
except Exception as e:
_log("ERROR", f"Failed processing movie {item}: {e}")
_log("INFO", f"Completed manual scan of: {scan_path}")
finally:
if conn:
try:
conn.close()
except:
pass
@app.post("/manual/scan")
async def manual_scan(background_tasks: BackgroundTasks, path: Optional[str] = None, scan_type: str = "both"):
if scan_type not in ["both", "tv", "movies"]:
raise HTTPException(status_code=400, detail="scan_type must be 'both', 'tv', or 'movies'")
if path:
scan_paths = [Path(path)]
else:
if scan_type == "movies":
scan_paths = list(config.movie_paths)
elif scan_type == "tv":
scan_paths = list(config.tv_paths)
else:
scan_paths = list(config.tv_paths) + list(config.movie_paths)
async def run_scan():
await asyncio.get_event_loop().run_in_executor(None, run_manual_scan_sync, scan_paths, scan_type)
background_tasks.add_task(run_scan)
return {"status": "started", "message": f"Manual {scan_type} scan started for {len(scan_paths)} paths"}
@app.post("/manual/scan/tv")
async def manual_scan_tv(background_tasks: BackgroundTasks, path: Optional[str] = None):
return await manual_scan(background_tasks, path, "tv")
@app.post("/manual/scan/movies")
async def manual_scan_movies(background_tasks: BackgroundTasks, path: Optional[str] = None):
return await manual_scan(background_tasks, path, "movies")
# ---------------------------
# Main
# ---------------------------
if __name__ == "__main__":
if config.debug:
import logging
logging.basicConfig(level=logging.DEBUG)
os.environ["DEBUG"] = "true"
_log("INFO", "Starting NFOGuard Webhook Server")
_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 path: {db_path}")
_log("INFO", f"Manage NFO: {config.manage_nfo}")
_log("INFO", f"Fix dir mtimes: {config.fix_dir_mtimes}")
_log("INFO", f"Lock metadata: {config.lock_metadata}")
_log("INFO", f"Batch delay: {config.batch_delay}s")
_log("INFO", f"Max concurrent items: {config.max_concurrent_series}")
_log("INFO", f"MOVIE_PRIORITY={config.movie_priority}")
_log("INFO", f"MOVIE_POLL_MODE={config.movie_poll_mode}")
_log("INFO", f"MOVIE_DATE_STRATEGY={config.movie_date_strategy}")
_log("INFO", f"MOVIE_DATE_UPDATE_MODE={config.movie_date_update_mode} "
f"(preserve_import={config.movie_preserve_radarr_import}, "
f"preserve_existing={config.movie_preserve_existing})")
_log("INFO", f"MOVIE_DIGITAL_SANITY_CHECK={config.movie_digital_sanity_check} "
f"(ref={config.movie_digital_sanity_ref}, max_lag_days={config.movie_digital_max_lag_days}, "
f"invalid_fallback={config.movie_digital_invalid_fallback})")
if config.radarr_url and config.radarr_api_key:
_log("INFO", f"Radarr enabled @ {config.radarr_url} (timeout={config.radarr_timeout}s, retries={config.radarr_retries})")
if config.tmdb_api_key:
_log("INFO", f"TMDB enabled (country={config.tmdb_country})")
if config.omdb_api_key:
_log("INFO", "OMDb enabled")
if config.jelly_url and config.jelly_api:
_log("INFO", f"Jellyseerr enabled @ {config.jelly_url}")
uvicorn.run(
app, # pass the object, not "app:app"
host="0.0.0.0",
port=int(os.environ.get("PORT", "8080")),
reload=False
)
Reference in New Issue View Git Blame Copy Permalink