1008 lines
40 KiB
Python
1008 lines
40 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
media_date_cache.py
|
|
|
|
Applies canonical "date added" timestamps to TV and Movie files (and their sidecars),
|
|
using Sonarr/Radarr history and episode/movie metadata, with TMDB fallbacks.
|
|
|
|
Key behavior:
|
|
- Prefer original import date from Sonarr/Radarr; fallback to airdate (TV) or TMDB release date (Movie); fallback to existing mtime with --live-on-miss
|
|
- Writes minimal NFOs (TV: SxxEyy.nfo; Movie: movie.nfo)
|
|
- **NFO mtime always mirrors the video file's actual mtime** after processing (so it matches the file + sidecars)
|
|
- Sidecars (.srt/.ass/.vtt/...) are aligned to the chosen timestamp (or left alone with --only-older)
|
|
- SQLite cache to avoid re-fetching unless --refresh
|
|
- Gotify notifications (optional)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import dataclasses
|
|
from dataclasses import field
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Iterator, List, Optional, Tuple
|
|
import urllib.request
|
|
import urllib.error
|
|
|
|
DEFAULT_GOTIFY_URL = os.getenv("GOTIFY_URL", "")
|
|
DEFAULT_GOTIFY_TOKEN = os.getenv("GOTIFY_TOKEN", "")
|
|
|
|
SONARR_TIMEOUT = int(os.getenv("SONARR_TIMEOUT", "45"))
|
|
SONARR_RETRIES = int(os.getenv("SONARR_RETRIES", "3"))
|
|
RADARR_TIMEOUT = int(os.getenv("RADARR_TIMEOUT", "45"))
|
|
RADARR_RETRIES = int(os.getenv("RADARR_RETRIES", "3"))
|
|
TMDB_TIMEOUT = int(os.getenv("TMDB_TIMEOUT", "45"))
|
|
TMDB_RETRIES = int(os.getenv("TMDB_RETRIES", "3"))
|
|
|
|
LOG_TS_FMT = "%Y-%m-%d %H:%M:%S"
|
|
|
|
VIDEO_EXTS = {".mkv", ".mp4", ".avi", ".m4v", ".mov", ".wmv", ".flv", ".webm"}
|
|
SIDECAR_EXTS = {".srt", ".sub", ".idx", ".ass", ".ssa", ".vtt", ".smi", ".rt", ".txt"}
|
|
NFO_SIMPLE_PATTERN = re.compile(r"^S\d{2}E\d{2}\.nfo$", re.IGNORECASE)
|
|
|
|
SE_EP_RX = re.compile(r"[Ss](\d{1,2})[ ._-]*[Ee](\d{1,3})")
|
|
IMDB_RX = re.compile(r"\[imdb-(tt\d+)\]", re.IGNORECASE)
|
|
TMDB_RX = re.compile(r"\[tmdb-(\d+)\]", re.IGNORECASE)
|
|
YEAR_RX = re.compile(r"\((\d{4})\)")
|
|
|
|
DEBUG_HTTP = False
|
|
DEBUG_TRUNCATE = 3000 # chars
|
|
|
|
def log(msg: str) -> None:
|
|
ts = dt.datetime.now().strftime(LOG_TS_FMT)
|
|
print(f"[{ts}] {msg}", flush=True)
|
|
|
|
def dlog(msg: str) -> None:
|
|
if DEBUG_HTTP:
|
|
log(f"[DEBUG] {msg}")
|
|
|
|
def send_gotify_notification(title: str, message: str, priority: int = 5) -> None:
|
|
if not DEFAULT_GOTIFY_URL or not DEFAULT_GOTIFY_TOKEN:
|
|
return
|
|
try:
|
|
url = f"{DEFAULT_GOTIFY_URL.rstrip('/')}/message"
|
|
data = json.dumps({"title": title, "message": message, "priority": priority}).encode("utf-8")
|
|
req = urllib.request.Request(url, data=data, method="POST")
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("X-Gotify-Key", DEFAULT_GOTIFY_TOKEN)
|
|
with urllib.request.urlopen(req, timeout=15):
|
|
pass
|
|
except Exception as e:
|
|
log(f"WARN: Gotify notification failed: {e}")
|
|
|
|
def load_env_file(env_path: Optional[str]) -> None:
|
|
if not env_path:
|
|
return
|
|
p = Path(env_path)
|
|
if not p.exists():
|
|
log(f"WARN: .env not found: {env_path}")
|
|
return
|
|
for line in p.read_text(encoding="utf-8", errors="ignore").splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
k = k.strip()
|
|
v = v.strip().strip('"').strip("'")
|
|
os.environ[k] = v
|
|
|
|
# ---------------- DB ----------------
|
|
|
|
def db_init(cx: sqlite3.Connection) -> None:
|
|
cx.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS media_dates(
|
|
path TEXT PRIMARY KEY,
|
|
kind TEXT NOT NULL,
|
|
chosen_ts INTEGER NOT NULL,
|
|
chosen_src TEXT NOT NULL,
|
|
extra TEXT,
|
|
last_updated INTEGER DEFAULT (strftime('%s', 'now'))
|
|
)
|
|
"""
|
|
)
|
|
cx.commit()
|
|
|
|
def db_get_cached(cx: sqlite3.Connection, path: str) -> Optional[Tuple[int, str]]:
|
|
cur = cx.execute("SELECT chosen_ts, chosen_src FROM media_dates WHERE path = ?", (path,))
|
|
row = cur.fetchone()
|
|
return (int(row[0]), row[1]) if row else None
|
|
|
|
def db_upsert(cx: sqlite3.Connection, path: str, kind: str, chosen_ts: int, chosen_src: str, extra: Dict[str, object] | None) -> None:
|
|
cx.execute(
|
|
"""
|
|
INSERT INTO media_dates(path, kind, chosen_ts, chosen_src, extra, last_updated)
|
|
VALUES(?,?,?,?,?,?)
|
|
ON CONFLICT(path) DO UPDATE SET
|
|
kind=excluded.kind,
|
|
chosen_ts=excluded.chosen_ts,
|
|
chosen_src=excluded.chosen_src,
|
|
extra=excluded.extra,
|
|
last_updated=excluded.last_updated
|
|
""",
|
|
(path, kind, int(chosen_ts), chosen_src, json.dumps(extra or {}, ensure_ascii=False), int(time.time())),
|
|
)
|
|
cx.commit()
|
|
|
|
# -------------- HTTP helper --------------
|
|
|
|
def _pretty(obj) -> str:
|
|
try:
|
|
return json.dumps(obj, indent=2, ensure_ascii=False)[:DEBUG_TRUNCATE]
|
|
except Exception:
|
|
return str(obj)[:DEBUG_TRUNCATE]
|
|
|
|
def _req_json(method: str, url: str, *, headers: Optional[Dict[str, str]] = None, params: Optional[Dict[str, object]] = None, timeout: int = 30, retries: int = 2) -> dict | list:
|
|
from urllib.parse import urlencode
|
|
full_url = f"{url}?{urlencode(params)}" if params else url
|
|
last_err = None
|
|
for _ in range(max(1, retries)):
|
|
try:
|
|
dlog(f"HTTP {method} {full_url}")
|
|
req = urllib.request.Request(full_url, method=method)
|
|
for k, v in (headers or {}).items():
|
|
req.add_header(k, v)
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
body = resp.read()
|
|
txt = body.decode("utf-8", errors="ignore")
|
|
try:
|
|
js = json.loads(txt)
|
|
dlog(f"← {full_url} OK\n{_pretty(js)}")
|
|
return js
|
|
except Exception as je:
|
|
dlog(f"← {full_url} NON-JSON ({len(txt)} bytes)\n{txt[:DEBUG_TRUNCATE]}")
|
|
raise RuntimeError(f"Non-JSON response from {full_url}") from je
|
|
except urllib.error.HTTPError as e:
|
|
last_err = e
|
|
err_body = e.read().decode("utf-8", errors="ignore")
|
|
dlog(f"← {full_url} HTTP {e.code}\n{err_body[:DEBUG_TRUNCATE]}")
|
|
time.sleep(0.35)
|
|
except Exception as e:
|
|
last_err = e
|
|
dlog(f"← {full_url} ERROR: {e}")
|
|
time.sleep(0.35)
|
|
raise last_err or RuntimeError("request failed")
|
|
|
|
# ---------- API clients + caches ----------
|
|
|
|
@dataclasses.dataclass
|
|
class SonarrClient:
|
|
url: str
|
|
api_key: str
|
|
timeout: int = SONARR_TIMEOUT
|
|
retries: int = SONARR_RETRIES
|
|
def _h(self) -> Dict[str, str]: return {"X-Api-Key": self.api_key}
|
|
def ping(self) -> bool:
|
|
try:
|
|
_req_json("GET", f"{self.url}/api/v3/system/status", headers=self._h(), timeout=self.timeout, retries=self.retries)
|
|
return True
|
|
except Exception as e:
|
|
log(f"WARN: Sonarr ping failed: {e}")
|
|
return False
|
|
def series_index(self) -> List[dict]:
|
|
res = _req_json("GET", f"{self.url}/api/v3/series", headers=self._h(), timeout=self.timeout, retries=self.retries)
|
|
return res if isinstance(res, list) else list(res or [])
|
|
def history_for(self, series_id: int) -> List[dict]:
|
|
res = _req_json("GET", f"{self.url}/api/v3/history/series", headers=self._h(), params={"seriesId": series_id, "pageSize": 1000}, timeout=self.timeout, retries=self.retries)
|
|
if isinstance(res, dict):
|
|
recs = res.get("records")
|
|
return recs if isinstance(recs, list) else []
|
|
if isinstance(res, list):
|
|
return res
|
|
return []
|
|
def episodes_for(self, series_id: int) -> List[dict]:
|
|
res = _req_json("GET", f"{self.url}/api/v3/episode", headers=self._h(), params={"seriesId": series_id}, timeout=self.timeout, retries=self.retries)
|
|
return res if isinstance(res, list) else []
|
|
|
|
@dataclasses.dataclass
|
|
class SonarrCache:
|
|
sc: SonarrClient
|
|
episodes_map: Dict[int, List[dict]] = field(default_factory=dict)
|
|
history_map: Dict[int, List[dict]] = field(default_factory=dict)
|
|
def episodes(self, series_id: int) -> List[dict]:
|
|
if series_id not in self.episodes_map:
|
|
eps = self.sc.episodes_for(series_id)
|
|
self.episodes_map[series_id] = eps
|
|
dlog(f"Cached Sonarr episodes for series {series_id}: {len(eps)}")
|
|
return self.episodes_map[series_id]
|
|
def history(self, series_id: int) -> List[dict]:
|
|
if series_id not in self.history_map:
|
|
hist = self.sc.history_for(series_id)
|
|
self.history_map[series_id] = hist
|
|
dlog(f"Cached Sonarr history for series {series_id}: {len(hist)}")
|
|
return self.history_map[series_id]
|
|
|
|
@dataclasses.dataclass
|
|
class RadarrClient:
|
|
url: str
|
|
api_key: str
|
|
timeout: int = RADARR_TIMEOUT
|
|
retries: int = RADARR_RETRIES
|
|
def _h(self) -> Dict[str, str]: return {"X-Api-Key": self.api_key}
|
|
def ping(self) -> bool:
|
|
try:
|
|
_req_json("GET", f"{self.url}/api/v3/system/status", headers=self._h(), timeout=self.timeout, retries=self.retries)
|
|
return True
|
|
except Exception as e:
|
|
log(f"WARN: Radarr ping failed: {e}")
|
|
return False
|
|
def movies_index(self) -> List[dict]:
|
|
res = _req_json("GET", f"{self.url}/api/v3/movie", headers=self._h(), timeout=self.timeout, retries=self.retries)
|
|
return res if isinstance(res, list) else []
|
|
def history_for(self, movie_id: int) -> List[dict]:
|
|
res = _req_json("GET", f"{self.url}/api/v3/history/movie", headers=self._h(), params={"movieId": movie_id, "pageSize": 1000}, timeout=self.timeout, retries=self.retries)
|
|
if isinstance(res, dict):
|
|
recs = res.get("records")
|
|
return recs if isinstance(recs, list) else []
|
|
if isinstance(res, list):
|
|
return res
|
|
return []
|
|
|
|
@dataclasses.dataclass
|
|
class TMDBClient:
|
|
api_key: str
|
|
primary_country: str = "US"
|
|
base_url: str = "https://api.themoviedb.org/3"
|
|
timeout: int = TMDB_TIMEOUT
|
|
retries: int = TMDB_RETRIES
|
|
def _p(self, **kw) -> Dict[str, object]:
|
|
p = {"api_key": self.api_key, "language": "en-US"}; p.update(kw); return p
|
|
def find_tv_by_imdb(self, imdb_id: str) -> Optional[int]:
|
|
js = _req_json("GET", f"{self.base_url}/find/{imdb_id}", params=self._p(external_source="imdb_id"), timeout=self.timeout, retries=self.retries)
|
|
lst = js.get("tv_results") if isinstance(js, dict) else None
|
|
return lst[0].get("id") if lst else None
|
|
def search_tv(self, name: str, year: Optional[int]) -> Optional[int]:
|
|
js = _req_json("GET", f"{self.base_url}/search/tv", params=self._p(query=name, first_air_date_year=year or ""), timeout=self.timeout, retries=self.retries)
|
|
res = js.get("results") if isinstance(js, dict) else None
|
|
return res[0].get("id") if res else None
|
|
def find_movie_by_imdb(self, imdb_id: str) -> Optional[int]:
|
|
js = _req_json("GET", f"{self.base_url}/find/{imdb_id}", params=self._p(external_source="imdb_id"), timeout=self.timeout, retries=self.retries)
|
|
lst = js.get("movie_results") if isinstance(js, dict) else None
|
|
return lst[0].get("id") if lst else None
|
|
def search_movie(self, name: str, year: Optional[int]) -> Optional[int]:
|
|
js = _req_json("GET", f"{self.base_url}/search/movie", params=self._p(query=name, year=year or "", primary_release_year=year or ""), timeout=self.timeout, retries=self.retries)
|
|
res = js.get("results") if isinstance(js, dict) else None
|
|
return res[0].get("id") if res else None
|
|
def episode_air_date(self, tv_id: int, season: int, episode: int) -> Optional[str]:
|
|
try:
|
|
js = _req_json("GET", f"{self.base_url}/tv/{tv_id}/season/{season}/episode/{episode}", params=self._p(), timeout=self.timeout, retries=self.retries)
|
|
return js.get("air_date") if isinstance(js, dict) else None
|
|
except Exception:
|
|
return None
|
|
def movie_release_date(self, movie_id: int) -> Optional[str]:
|
|
try:
|
|
js = _req_json("GET", f"{self.base_url}/movie/{movie_id}", params=self._p(), timeout=self.timeout, retries=self.retries)
|
|
return js.get("release_date") if isinstance(js, dict) else None
|
|
except Exception:
|
|
return None
|
|
|
|
# -------------- parsers --------------
|
|
|
|
@dataclasses.dataclass
|
|
class EpisodeKey:
|
|
series_title: str
|
|
season: int
|
|
episode: int
|
|
imdb_id: Optional[str]
|
|
tmdb_id: Optional[str]
|
|
year: Optional[int]
|
|
|
|
def parse_episode_from_path(p: Path) -> Optional[EpisodeKey]:
|
|
parent = p.parent
|
|
show_dir = parent.parent.name if parent.name.lower().startswith("season ") else parent.name
|
|
m = SE_EP_RX.search(p.name)
|
|
if not m:
|
|
return None
|
|
season = int(m.group(1)); episode = int(m.group(2))
|
|
imdb_id_match = IMDB_RX.search(p.name) or IMDB_RX.search(show_dir)
|
|
imdb_id = imdb_id_match.group(1) if imdb_id_match else None
|
|
tmdb_id_match = TMDB_RX.search(p.name) or TMDB_RX.search(show_dir)
|
|
tmdb_id = tmdb_id_match.group(1) if tmdb_id_match else None
|
|
year = None
|
|
m3 = YEAR_RX.search(show_dir)
|
|
if m3:
|
|
try:
|
|
year = int(m3.group(1))
|
|
except Exception:
|
|
year = None
|
|
series_title = IMDB_RX.sub("", show_dir)
|
|
series_title = TMDB_RX.sub("", series_title)
|
|
series_title = YEAR_RX.sub("", series_title).strip() or show_dir
|
|
return EpisodeKey(series_title, season, episode, imdb_id, tmdb_id, year)
|
|
|
|
@dataclasses.dataclass
|
|
class MovieKey:
|
|
title: str
|
|
year: Optional[int]
|
|
imdb_id: Optional[str]
|
|
tmdb_id: Optional[str]
|
|
|
|
def parse_movie_from_path(p: Path) -> Optional[MovieKey]:
|
|
movie_dir = p.parent.name
|
|
imdb_id_match = IMDB_RX.search(movie_dir)
|
|
tmdb_id_match = TMDB_RX.search(movie_dir)
|
|
imdb_id = imdb_id_match.group(1) if imdb_id_match else None
|
|
tmdb_id = tmdb_id_match.group(1) if tmdb_id_match else None
|
|
year = None
|
|
m3 = YEAR_RX.search(movie_dir)
|
|
if m3:
|
|
try:
|
|
year = int(m3.group(1))
|
|
except Exception:
|
|
year = None
|
|
clean = IMDB_RX.sub("", movie_dir)
|
|
clean = TMDB_RX.sub("", clean)
|
|
clean = YEAR_RX.sub("", clean).strip() or movie_dir
|
|
return MovieKey(clean, year, imdb_id, tmdb_id)
|
|
|
|
# -------------- date picking --------------
|
|
|
|
def parse_date_yyyy_mm_dd(s: str) -> Optional[int]:
|
|
try:
|
|
y, m, d = map(int, s.split("-"))
|
|
return int(dt.datetime(y, m, d, 0, 0, 0, tzinfo=dt.timezone.utc).timestamp())
|
|
except Exception:
|
|
return None
|
|
|
|
def pick_from_sonarr_import_for_episode_from_history(history: List[dict], episode_id: int) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
import_types = {"episodeFileImported", "downloadFolderImported", "manualImport"}
|
|
grabbed_types = {"grabbed"}
|
|
imports: List[Tuple[int, dict]] = []
|
|
grabs: List[Tuple[int, dict]] = []
|
|
for rec in history or []:
|
|
if not isinstance(rec, dict):
|
|
continue
|
|
if int(rec.get("episodeId", -1)) != int(episode_id):
|
|
continue
|
|
et = rec.get("eventType") or ""
|
|
date_s = rec.get("date")
|
|
if not date_s:
|
|
continue
|
|
try:
|
|
t = dt.datetime.fromisoformat(date_s.replace("Z", "+00:00"))
|
|
ts = int(t.timestamp())
|
|
except Exception:
|
|
continue
|
|
if et in import_types:
|
|
imports.append((ts, rec))
|
|
elif et in grabbed_types:
|
|
grabs.append((ts, rec))
|
|
if imports:
|
|
ts, rec0 = sorted(imports, key=lambda x: x[0])[0]
|
|
return ts, "sonarr_import", {"episodeId": episode_id, "eventType": rec0.get("eventType")}
|
|
if grabs:
|
|
ts, rec0 = sorted(grabs, key=lambda x: x[0])[0]
|
|
return ts, "sonarr_grabbed", {"episodeId": episode_id, "eventType": rec0.get("eventType")}
|
|
return None
|
|
|
|
def pick_from_sonarr_episode_from_list(episodes: List[dict], season: int, episode: int) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
for ep in episodes or []:
|
|
if not isinstance(ep, dict):
|
|
continue
|
|
if ep.get("seasonNumber") == season and ep.get("episodeNumber") == episode:
|
|
air = ep.get("airDateUtc") or ep.get("airDate")
|
|
if not air:
|
|
return None
|
|
if "T" in air:
|
|
try:
|
|
t = dt.datetime.fromisoformat(air.replace("Z", "+00:00"))
|
|
return int(t.timestamp()), "sonarr_airdate", {"episodeId": ep.get("id")}
|
|
except Exception:
|
|
return None
|
|
ts = parse_date_yyyy_mm_dd(air)
|
|
if ts is not None:
|
|
return ts, "sonarr_airdate", {"episodeId": ep.get("id")}
|
|
return None
|
|
return None
|
|
|
|
def pick_from_radarr_history(rc: RadarrClient, movie_id: int) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
try:
|
|
hist = rc.history_for(movie_id)
|
|
except Exception as e:
|
|
log(f"WARN: Radarr history timeout/err movie={movie_id}: {e}")
|
|
return None
|
|
best_ts = None; best = None
|
|
for rec in hist or []:
|
|
if not isinstance(rec, dict):
|
|
continue
|
|
date_s = rec.get("date") or ""
|
|
if not date_s:
|
|
continue
|
|
try:
|
|
t = dt.datetime.fromisoformat(date_s.replace("Z", "+00:00"))
|
|
ts = int(t.timestamp())
|
|
except Exception:
|
|
continue
|
|
if best_ts is None or ts < best_ts:
|
|
best_ts = ts; best = rec
|
|
if best_ts is not None:
|
|
return best_ts, "radarr_history", {"history_id": (best or {}).get("id")}
|
|
return None
|
|
|
|
@dataclasses.dataclass
|
|
class TMDBMovieLike:
|
|
title: str; year: Optional[int]; imdb_id: Optional[str]; tmdb_id: Optional[str]
|
|
|
|
def pick_from_tmdb_movie(tmdb: TMDBClient, mk: TMDBMovieLike) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
if not tmdb.api_key:
|
|
return None
|
|
movie_id = None
|
|
if mk.tmdb_id:
|
|
try:
|
|
movie_id = int(mk.tmdb_id)
|
|
except Exception:
|
|
movie_id = None
|
|
if movie_id is None and mk.imdb_id:
|
|
try:
|
|
movie_id = tmdb.find_movie_by_imdb(mk.imdb_id)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB find movie by IMDb failed {mk.imdb_id}: {e}")
|
|
if movie_id is None:
|
|
try:
|
|
movie_id = tmdb.search_movie(mk.title, mk.year)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB movie search failed title={mk.title}: {e}")
|
|
if movie_id is None:
|
|
return None
|
|
rel = tmdb.movie_release_date(movie_id)
|
|
if rel:
|
|
ts = parse_date_yyyy_mm_dd(rel)
|
|
if ts is not None:
|
|
return ts, "tmdb_movie_release", {"tmdb_movie_id": movie_id, "release_date": rel}
|
|
return None
|
|
|
|
def pick_from_tmdb_tv(tmdb: TMDBClient, series_title: str, season: int, episode: int, imdb_id: Optional[str], tmdb_id: Optional[str], year: Optional[int]) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
if not tmdb.api_key:
|
|
return None
|
|
tv_id = None
|
|
if tmdb_id:
|
|
try:
|
|
tv_id = int(tmdb_id)
|
|
except Exception:
|
|
tv_id = None
|
|
if tv_id is None and imdb_id:
|
|
try:
|
|
tv_id = tmdb.find_tv_by_imdb(imdb_id)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB find by IMDb failed {imdb_id}: {e}")
|
|
if tv_id is None:
|
|
try:
|
|
tv_id = tmdb.search_tv(series_title, year)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB search failed title={series_title}: {e}")
|
|
if tv_id is None:
|
|
return None
|
|
air = tmdb.episode_air_date(tv_id, season, episode)
|
|
if air:
|
|
ts = parse_date_yyyy_mm_dd(air)
|
|
if ts is not None:
|
|
return ts, "tmdb_episode_airdate", {"tmdb_tv_id": tv_id, "air_date": air}
|
|
return None
|
|
|
|
# -------------- NFO helpers --------------
|
|
|
|
def create_simple_nfo(season: int, episode: int, air_date: Optional[str], date_added: Optional[str]) -> str:
|
|
aired_tag = f" <aired>{air_date}</aired>\n" if air_date else ""
|
|
dateadded_tag = f" <dateadded>{date_added}</dateadded>\n" if date_added else ""
|
|
return f"""<?xml version="1.0" encoding="UTF-8" standalone="yes" ?>
|
|
<episodedetails>
|
|
{aired_tag}{dateadded_tag}</episodedetails>"""
|
|
|
|
def create_movie_nfo(title: str, year: Optional[int] = None, release_date: Optional[str] = None, date_added: Optional[str] = None) -> str:
|
|
premiered_tag = f" <premiered>{release_date}</premiered>\n" if release_date else ""
|
|
dateadded_tag = f" <dateadded>{date_added}</dateadded>\n" if date_added else ""
|
|
year_tag = f" <year>{year}</year>\n" if year else ""
|
|
title_tag = f" <title>{title}</title>\n" if title else ""
|
|
return f"""<?xml version="1.0" encoding="UTF-8" standalone="yes" ?>
|
|
<movie>
|
|
{title_tag}{year_tag}{premiered_tag}{dateadded_tag}</movie>"""
|
|
|
|
def _fmt(ts: int) -> str:
|
|
return dt.datetime.fromtimestamp(ts, tz=dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
|
|
def _set_file_ts(path: Path, ts: int, dry_run: bool) -> bool:
|
|
if dry_run:
|
|
return True
|
|
try:
|
|
os.utime(str(path), (ts, ts))
|
|
return True
|
|
except Exception as e:
|
|
log(f"WARN: Failed to set timestamp on {path}: {e}")
|
|
return False
|
|
|
|
def manage_nfo_files(
|
|
media_dir: Path,
|
|
kind: str,
|
|
season: int = 0,
|
|
episode: int = 0,
|
|
air_date: Optional[str] = None,
|
|
date_added: Optional[str] = None,
|
|
title: Optional[str] = None,
|
|
year: Optional[int] = None,
|
|
release_date: Optional[str] = None,
|
|
nfo_timestamp: Optional[int] = None, # final, anchor ts (video mtime)
|
|
dry_run: bool = False
|
|
) -> Tuple[int, int, int]:
|
|
"""
|
|
Create/refresh minimal NFOs AND force the NFO file's mtime to nfo_timestamp.
|
|
Returns (created_count, removed_count, timestamps_synced_count).
|
|
"""
|
|
created = 0
|
|
removed = 0
|
|
synced = 0
|
|
|
|
def _verify_and_set_ts(path: Path, ts: Optional[int]) -> bool:
|
|
if ts is None:
|
|
return False
|
|
if dry_run:
|
|
return True
|
|
before = None
|
|
try:
|
|
before = int(path.stat().st_mtime)
|
|
except Exception:
|
|
before = None
|
|
try:
|
|
os.utime(str(path), (ts, ts))
|
|
cur = int(path.stat().st_mtime)
|
|
if DEBUG_HTTP:
|
|
log(f"[DEBUG] NFO ts set {path.name}: {('unknown' if before is None else _fmt(before))} -> {_fmt(cur)} (target {_fmt(ts)})")
|
|
return abs(cur - int(ts)) <= 2
|
|
except Exception as e:
|
|
log(f"WARN: Failed to set timestamp on {path}: {e}")
|
|
return False
|
|
|
|
if kind == "tv":
|
|
simple_nfo_name = f"S{season:02d}E{episode:02d}.nfo"
|
|
simple_nfo_path = media_dir / simple_nfo_name
|
|
nfo_content = create_simple_nfo(season, episode, air_date, date_added)
|
|
|
|
try:
|
|
should_write = True
|
|
if simple_nfo_path.exists():
|
|
existing_content = simple_nfo_path.read_text(encoding='utf-8', errors='ignore')
|
|
should_write = existing_content.strip() != nfo_content.strip()
|
|
if should_write:
|
|
if not dry_run:
|
|
simple_nfo_path.write_text(nfo_content, encoding='utf-8')
|
|
created += 1
|
|
except Exception as e:
|
|
log(f"WARN: Failed to create/update {simple_nfo_path}: {e}")
|
|
|
|
if simple_nfo_path.exists() and nfo_timestamp is not None:
|
|
if _verify_and_set_ts(simple_nfo_path, nfo_timestamp):
|
|
synced += 1
|
|
|
|
# remove non-simple NFOs
|
|
for nfo_file in media_dir.glob("*.nfo"):
|
|
if nfo_file == simple_nfo_path:
|
|
continue
|
|
n_lower = nfo_file.name.lower()
|
|
if n_lower in ("season.nfo", "tvshow.nfo"):
|
|
continue
|
|
if NFO_SIMPLE_PATTERN.match(nfo_file.name):
|
|
continue
|
|
try:
|
|
if not dry_run:
|
|
nfo_file.unlink()
|
|
removed += 1
|
|
log(f"Removed complex NFO: {nfo_file.name}")
|
|
except Exception as e:
|
|
log(f"WARN: Failed to remove {nfo_file}: {e}")
|
|
|
|
else:
|
|
movie_nfo_path = media_dir / "movie.nfo"
|
|
nfo_content = create_movie_nfo(title or "", year=year, release_date=release_date, date_added=date_added)
|
|
|
|
try:
|
|
should_write = True
|
|
if movie_nfo_path.exists():
|
|
existing_content = movie_nfo_path.read_text(encoding='utf-8', errors='ignore')
|
|
should_write = existing_content.strip() != nfo_content.strip()
|
|
if should_write:
|
|
if not dry_run:
|
|
movie_nfo_path.write_text(nfo_content, encoding='utf-8')
|
|
created += 1
|
|
except Exception as e:
|
|
log(f"WARN: Failed to create/update {movie_nfo_path}: {e}")
|
|
|
|
if movie_nfo_path.exists() and nfo_timestamp is not None:
|
|
if _verify_and_set_ts(movie_nfo_path, nfo_timestamp):
|
|
synced += 1
|
|
|
|
return created, removed, synced
|
|
|
|
# -------------- sidecars --------------
|
|
|
|
def _season_episode_token(name: str) -> Optional[Tuple[int,int]]:
|
|
m = SE_EP_RX.search(name)
|
|
if not m:
|
|
return None
|
|
return int(m.group(1)), int(m.group(2))
|
|
|
|
def find_sidecar_files(media_file: Path) -> List[Path]:
|
|
sidecars: List[Path] = []
|
|
parent_dir = media_file.parent
|
|
base_stem = media_file.stem
|
|
token = _season_episode_token(media_file.name)
|
|
for f in parent_dir.iterdir():
|
|
if not f.is_file():
|
|
continue
|
|
if f == media_file:
|
|
continue
|
|
suf = f.suffix.lower()
|
|
if suf not in SIDECAR_EXTS:
|
|
continue
|
|
if f.stem == base_stem or f.name.startswith(base_stem + "."):
|
|
sidecars.append(f); continue
|
|
if token:
|
|
other = _season_episode_token(f.name)
|
|
if other and other == token:
|
|
sidecars.append(f); continue
|
|
uniq: Dict[str, Path] = {}
|
|
for s in sidecars:
|
|
uniq[s.name] = s
|
|
return list(uniq.values())
|
|
|
|
def apply_timestamp_to_file(file_path: Path, timestamp: int, dry_run: bool = False) -> bool:
|
|
if dry_run:
|
|
return True
|
|
try:
|
|
os.utime(str(file_path), (timestamp, timestamp))
|
|
return True
|
|
except Exception as e:
|
|
log(f"WARN: Failed to set timestamp on {file_path}: {e}")
|
|
return False
|
|
|
|
def iter_media_files(roots: List[str]) -> Iterator[Path]:
|
|
for root in roots:
|
|
base = Path(root)
|
|
if not base.exists():
|
|
continue
|
|
for p in base.rglob("*"):
|
|
if p.is_file() and p.suffix.lower() in VIDEO_EXTS:
|
|
yield p
|
|
|
|
# -------------- matching helpers --------------
|
|
|
|
def choose_series_id_for_path(series_index: List[dict], fp: Path) -> Optional[int]:
|
|
if not series_index:
|
|
return None
|
|
try:
|
|
series_dir = fp.parent.parent.name if fp.parent.name.lower().startswith("season ") else fp.parent.name
|
|
except Exception:
|
|
series_dir = fp.parent.name
|
|
base = series_dir.lower().strip()
|
|
base = YEAR_RX.sub("", base)
|
|
base = IMDB_RX.sub("", base)
|
|
base = TMDB_RX.sub("", base)
|
|
base = base.strip()
|
|
for s in series_index:
|
|
title = (s.get("title") or "").lower().strip()
|
|
if base == title:
|
|
return s.get("id")
|
|
return None
|
|
|
|
def choose_movie_id_for_path(movies_index: List[dict], fp: Path) -> Optional[int]:
|
|
movie_dir = fp.parent.name
|
|
base = movie_dir.lower().strip()
|
|
base = YEAR_RX.sub("", base)
|
|
base = IMDB_RX.sub("", base)
|
|
base = TMDB_RX.sub("", base)
|
|
base = base.strip()
|
|
for m in movies_index:
|
|
title = (m.get("title") or "").lower().strip()
|
|
if base == title:
|
|
return m.get("id")
|
|
return None
|
|
|
|
# -------------- main apply --------------
|
|
|
|
def apply_command(args: argparse.Namespace) -> None:
|
|
global DEBUG_HTTP
|
|
DEBUG_HTTP = bool(args.debug_http)
|
|
|
|
log("=== Media Date Sync Starting ===")
|
|
load_env_file(args.env)
|
|
|
|
sonarr_url = os.getenv("SONARR_URL", "http://sonarr:8989")
|
|
sonarr_key = os.getenv("SONARR_API_KEY", "")
|
|
radarr_url = os.getenv("RADARR_URL", "http://radarr:7878")
|
|
radarr_key = os.getenv("RADARR_API_KEY", "")
|
|
tmdb_key = os.getenv("TMDB_API_KEY", "")
|
|
tmdb_country = os.getenv("TMDB_PRIMARY_COUNTRY", "US")
|
|
|
|
log(f"SONARR_URL: {sonarr_url}")
|
|
log(f"SONARR_API_KEY: {'SET' if sonarr_key else 'MISSING'}")
|
|
log(f"TMDB_API_KEY: {'SET' if tmdb_key else 'MISSING'}")
|
|
|
|
sc = None; scache = None
|
|
if args.kind == "tv" and sonarr_key and sonarr_url:
|
|
sc = SonarrClient(sonarr_url, sonarr_key)
|
|
if not sc.ping():
|
|
log("WARN: Sonarr not reachable")
|
|
sc = None
|
|
else:
|
|
log("Sonarr connection successful")
|
|
scache = SonarrCache(sc)
|
|
elif args.kind == "tv":
|
|
log("WARN: Sonarr not configured")
|
|
|
|
rc = None; movies_index: List[dict] = []
|
|
if args.kind == "movie" and radarr_key and radarr_url:
|
|
rc = RadarrClient(radarr_url, radarr_key)
|
|
if not rc.ping():
|
|
log("WARN: Radarr not reachable")
|
|
rc = None
|
|
else:
|
|
try:
|
|
movies_index = rc.movies_index()
|
|
log(f"Loaded {len(movies_index)} movies from Radarr")
|
|
except Exception as e:
|
|
log(f"WARN: Failed to load Radarr movies: {e}")
|
|
elif args.kind == "movie":
|
|
log("WARN: Radarr not configured")
|
|
|
|
tmdb = TMDBClient(tmdb_key, tmdb_country)
|
|
if not tmdb_key:
|
|
log("WARN: TMDB not configured - limited functionality")
|
|
|
|
cx = sqlite3.connect(args.db)
|
|
db_init(cx)
|
|
log(f"Database initialized: {args.db}")
|
|
|
|
series_index: List[dict] = []
|
|
if args.kind == "tv" and sc:
|
|
try:
|
|
log("Loading Sonarr series index...")
|
|
series_index = sc.series_index()
|
|
log(f"Loaded {len(series_index)} series from Sonarr")
|
|
except Exception as e:
|
|
log(f"WARN: Failed to load Sonarr series: {e}")
|
|
|
|
roots = args.root or []
|
|
only_file = None
|
|
if args.path:
|
|
p = Path(args.path)
|
|
if p.is_file():
|
|
roots = [str(p.parent)]
|
|
only_file = p
|
|
else:
|
|
roots = [str(p)]
|
|
if not roots:
|
|
env_roots = os.getenv("TV_ROOTS" if args.kind == "tv" else "MOVIE_ROOTS", "")
|
|
if env_roots:
|
|
roots = [r for r in env_roots.split(":") if r.strip()]
|
|
log(f"Using {'TV' if args.kind=='tv' else 'MOVIE'}_ROOTS from env: {', '.join(roots)}")
|
|
if not roots:
|
|
log("ERROR: No root paths specified. Use --root or set TV_ROOTS/MOVIE_ROOTS in .env")
|
|
return
|
|
|
|
log(f"Scanning for media files in: {', '.join(roots)}")
|
|
files = [only_file] if (only_file is not None) else list(iter_media_files(roots))
|
|
log(f"Found {len(files)} media file(s)")
|
|
if not files:
|
|
log("No media files found.")
|
|
return
|
|
|
|
checked = updated = misses = cached = 0
|
|
nfo_created = nfo_removed = sidecars_updated = nfo_ts_synced = 0
|
|
source_counts: Dict[str, int] = {}
|
|
|
|
log(f"Starting processing: refresh={args.refresh}, dry_run={args.dry_run}, manage_nfo={args.manage_nfo}")
|
|
|
|
for i, fp in enumerate(files, 1):
|
|
path = str(fp)
|
|
if i % 25 == 1 or i <= 5:
|
|
log(f"Processing file {i}/{len(files)}: {fp.name}")
|
|
checked += 1
|
|
|
|
# pick effective timestamp (canonical "chosen") via cache or fresh
|
|
chosen_ts = None
|
|
chosen_src = None
|
|
extra: Dict[str, object] = {}
|
|
air_date_str = None
|
|
release_date_str = None
|
|
|
|
if not args.refresh:
|
|
cached_result = db_get_cached(cx, path)
|
|
if cached_result:
|
|
cached += 1
|
|
chosen_ts, chosen_src = cached_result
|
|
if i <= 5:
|
|
human = dt.datetime.utcfromtimestamp(chosen_ts).strftime("%Y-%m-%d %H:%M:%S UTC")
|
|
log(f" Using cached result: {chosen_src} @ {human}")
|
|
|
|
if chosen_ts is None:
|
|
if args.kind == "tv":
|
|
epk = parse_episode_from_path(fp)
|
|
if not epk:
|
|
if i <= 10:
|
|
log(f"WARN: Could not parse episode info from: {fp.name}")
|
|
misses += 1
|
|
continue
|
|
series_id = choose_series_id_for_path(series_index, fp) if series_index else None
|
|
if scache and series_id:
|
|
eps = scache.episodes(series_id)
|
|
epres = pick_from_sonarr_episode_from_list(eps, epk.season, epk.episode)
|
|
if epres:
|
|
chosen_ts, chosen_src, extra = epres
|
|
air_date_str = dt.datetime.fromtimestamp(chosen_ts, tz=dt.timezone.utc).strftime("%Y-%m-%d")
|
|
# prefer a real import/grab for THIS episode
|
|
ep_id = extra.get("episodeId")
|
|
if ep_id:
|
|
hist = scache.history(series_id)
|
|
imp = pick_from_sonarr_import_for_episode_from_history(hist, ep_id)
|
|
if imp:
|
|
chosen_ts, chosen_src, extra = imp
|
|
if chosen_ts is None and tmdb_key:
|
|
t = pick_from_tmdb_tv(tmdb, epk.series_title, epk.season, epk.episode, epk.imdb_id, epk.tmdb_id, epk.year)
|
|
if t:
|
|
chosen_ts, chosen_src, extra = t
|
|
air_date_str = dt.datetime.fromtimestamp(chosen_ts, tz=dt.timezone.utc).strftime("%Y-%m-%d")
|
|
else:
|
|
mk = parse_movie_from_path(fp)
|
|
if not mk:
|
|
if i <= 10:
|
|
log(f"WARN: Could not parse movie info from: {fp.name}")
|
|
misses += 1
|
|
continue
|
|
movie_id = choose_movie_id_for_path(movies_index, fp) if movies_index else None
|
|
if rc and movie_id:
|
|
hist_res = pick_from_radarr_history(rc, movie_id)
|
|
if hist_res:
|
|
chosen_ts, chosen_src, extra = hist_res
|
|
if chosen_ts is None and tmdb_key and mk:
|
|
t = pick_from_tmdb_movie(tmdb, TMDBMovieLike(mk.title, mk.year, mk.imdb_id, mk.tmdb_id))
|
|
if t:
|
|
chosen_ts, chosen_src, extra = t
|
|
release_date_str = dt.datetime.fromtimestamp(chosen_ts, tz=dt.timezone.utc).strftime("%Y-%m-%d")
|
|
|
|
if chosen_ts is None and args.live_on_miss:
|
|
try:
|
|
chosen_ts = int(fp.stat().st_mtime)
|
|
chosen_src = "existing_mtime"; extra = {}
|
|
if i <= 5:
|
|
log(" Using existing mtime as fallback")
|
|
except Exception:
|
|
pass
|
|
|
|
if chosen_ts is None:
|
|
if i <= 10:
|
|
log(f"WARN: No date found for: {fp.name}")
|
|
misses += 1
|
|
continue
|
|
|
|
source_counts[chosen_src] = source_counts.get(chosen_src, 0) + 1
|
|
|
|
# Apply to video/sidecars honoring --only-older
|
|
skip_apply = False
|
|
try:
|
|
cur_mtime = int(fp.stat().st_mtime)
|
|
except Exception:
|
|
cur_mtime = None
|
|
|
|
if args.only_older and cur_mtime is not None and chosen_ts >= cur_mtime:
|
|
skip_apply = True
|
|
if i <= 5:
|
|
log(f" Skipping (only-older): chosen >= current")
|
|
|
|
if not skip_apply and apply_timestamp_to_file(fp, chosen_ts, args.dry_run):
|
|
updated += 1
|
|
|
|
# Sidecars follow the chosen timestamp if we applied to video; with only-older skip, leave them as-is
|
|
if not skip_apply:
|
|
for s in find_sidecar_files(fp):
|
|
if apply_timestamp_to_file(s, chosen_ts, args.dry_run):
|
|
sidecars_updated += 1
|
|
|
|
# --------- NFO MIRRORS VIDEO MTIME (anchor) ---------
|
|
# Re-stat the video to get the *actual* anchor mtime (new or unchanged)
|
|
try:
|
|
anchor_ts = int(fp.stat().st_mtime)
|
|
except Exception:
|
|
anchor_ts = chosen_ts # fallback
|
|
|
|
if args.manage_nfo:
|
|
nfo_date_added_str = dt.datetime.fromtimestamp(anchor_ts, tz=dt.timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
|
|
if args.kind == "tv":
|
|
epk2 = parse_episode_from_path(fp)
|
|
c, r, s = manage_nfo_files(
|
|
fp.parent, "tv",
|
|
epk2.season if epk2 else 0,
|
|
epk2.episode if epk2 else 0,
|
|
air_date_str, # informational
|
|
nfo_date_added_str, # <dateadded> mirrors video mtime
|
|
nfo_timestamp=anchor_ts, # NFO mtime mirrors video mtime
|
|
dry_run=args.dry_run
|
|
)
|
|
else:
|
|
mk2 = parse_movie_from_path(fp)
|
|
c, r, s = manage_nfo_files(
|
|
fp.parent, "movie", 0, 0,
|
|
None,
|
|
nfo_date_added_str,
|
|
title=(mk2.title if mk2 else None),
|
|
year=(mk2.year if mk2 else None),
|
|
release_date=release_date_str,
|
|
nfo_timestamp=anchor_ts,
|
|
dry_run=args.dry_run
|
|
)
|
|
nfo_created += c; nfo_removed += r; nfo_ts_synced += s
|
|
|
|
# DB write (respect --only-older for the stored "chosen_ts")
|
|
if not args.dry_run:
|
|
try:
|
|
existing = db_get_cached(cx, path)
|
|
if args.only_older and existing and chosen_ts >= existing[0]:
|
|
pass
|
|
else:
|
|
db_upsert(cx, path, args.kind, chosen_ts, chosen_src, (extra or {}))
|
|
except Exception as e:
|
|
log(f"WARN: DB error: {e}")
|
|
|
|
if i % 50 == 0:
|
|
log(f"Progress: {i}/{len(files)} (updated: {updated}, cached: {cached}, misses: {misses})")
|
|
|
|
completion_msg = (
|
|
"Media date sync completed:\n"
|
|
f"Kind: {args.kind}\n"
|
|
f"Checked: {checked}\n"
|
|
f"Updated: {updated}\n"
|
|
f"Cached: {cached}\n"
|
|
f"Misses: {misses}\n"
|
|
f"NFO created: {nfo_created}, removed: {nfo_removed}\n"
|
|
f"NFO timestamps synced: {nfo_ts_synced}\n"
|
|
f"Sidecars updated: {sidecars_updated}"
|
|
)
|
|
if source_counts:
|
|
src_lines = "\n".join([f" - {k}: {v}" for k, v in sorted(source_counts.items(), key=lambda x: (-x[1], x[0]))])
|
|
completion_msg += f"\nSources Used:\n{src_lines}"
|
|
log(completion_msg)
|
|
send_gotify_notification("Media Date Sync Complete", completion_msg)
|
|
|
|
# -------------- CLI --------------
|
|
|
|
def build_arg_parser() -> argparse.ArgumentParser:
|
|
p = argparse.ArgumentParser(description="Apply canonical mtimes to media files using Sonarr/Radarr/TMDB fallbacks.")
|
|
sub = p.add_subparsers(dest="cmd", required=True)
|
|
ap = sub.add_parser("apply", help="Scan, resolve dates, apply mtimes, and record to DB.")
|
|
ap.add_argument("--env", help="Optional .env file with API keys and URLs")
|
|
ap.add_argument("--kind", choices=["tv", "movie"], required=True, help="Media kind")
|
|
ap.add_argument("--db", required=True, help="SQLite DB file path")
|
|
ap.add_argument("--root", action="append", help="Root folder(s) to scan; repeatable")
|
|
ap.add_argument("--path", help="Process a single file or a single folder (overrides --root)")
|
|
ap.add_argument("--only-older", action="store_true", help="Backfill only: only set earlier dates; never move to a later timestamp")
|
|
ap.add_argument("--refresh", action="store_true", help="Ignore cached results and re-fetch all dates")
|
|
ap.add_argument("--dry-run", action="store_true", help="Do not write mtime/DB; just log what would happen")
|
|
ap.add_argument("--live-on-miss", action="store_true", help="If no date found, keep file's current mtime as chosen date")
|
|
ap.add_argument("--manage-nfo", action="store_true", help="Create/refresh simple NFOs (TV: SxxEyy.nfo; Movies: movie.nfo).")
|
|
ap.add_argument("--debug-http", action="store_true", help="Verbose log of all HTTP requests/responses (Sonarr/Radarr/TMDB) and NFO timestamping")
|
|
return p
|
|
|
|
def main() -> None:
|
|
parser = build_arg_parser()
|
|
args = parser.parse_args()
|
|
try:
|
|
if args.cmd == "apply":
|
|
apply_command(args)
|
|
else:
|
|
parser.error("unknown command")
|
|
except KeyboardInterrupt:
|
|
log("Interrupted."); sys.exit(1)
|
|
except Exception as e:
|
|
log(f"FATAL ERROR: {e}")
|
|
import traceback; traceback.print_exc(); sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|