1120 lines
36 KiB
Python
1120 lines
36 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Apply canonical "date" timestamps (mtime) to media files and cache results.
|
|
|
|
Priority (TV):
|
|
1) Sonarr history/import date (first successful import/release)
|
|
2) Sonarr episode 'airDate' (UTC midnight)
|
|
3) TMDB episode 'air_date'
|
|
4) (optional, with --live-on-miss) keep existing mtime
|
|
|
|
Priority (Movies):
|
|
1) Radarr history/import date (first successful import/release)
|
|
2) Radarr movie 'digitalRelease' or 'physicalRelease'
|
|
3) TMDB movie 'release_date'
|
|
4) (optional, with --live-on-miss) keep existing mtime
|
|
|
|
DB schema (SQLite):
|
|
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'))
|
|
);
|
|
|
|
Usage (examples):
|
|
python media_date_cache.py apply \
|
|
--env /opt/scripts/unmanic/.env \
|
|
--kind tv \
|
|
--db /opt/scripts/unmanic/.cache/media_dates.db \
|
|
--root "/mnt/unionfs/Media/TV" \
|
|
--refresh --live-on-miss
|
|
|
|
python media_date_cache.py apply \
|
|
--env /opt/scripts/unmanic/.env \
|
|
--kind movie \
|
|
--db /opt/scripts/unmanic/.cache/media_dates.db \
|
|
--root "/mnt/unionfs/Media/Movies" \
|
|
--refresh --live-on-miss
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import dataclasses
|
|
import datetime as dt
|
|
import json
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, List, Optional, Tuple
|
|
|
|
import requests
|
|
|
|
# =========================
|
|
# Defaults / ENV variables
|
|
# =========================
|
|
|
|
DEFAULT_SONARR_URL = os.getenv("SONARR_URL", "http://sonarr:8989")
|
|
DEFAULT_SONARR_API_KEY = os.getenv("SONARR_API_KEY", "")
|
|
DEFAULT_RADARR_URL = os.getenv("RADARR_URL", "http://radarr:7878")
|
|
DEFAULT_RADARR_API_KEY = os.getenv("RADARR_API_KEY", "")
|
|
DEFAULT_TMDB_API_KEY = os.getenv("TMDB_API_KEY", "")
|
|
DEFAULT_TMDB_PRIMARY_COUNTRY = os.getenv("TMDB_PRIMARY_COUNTRY", "US")
|
|
DEFAULT_GOTIFY_URL = os.getenv("GOTIFY_URL", "")
|
|
DEFAULT_GOTIFY_TOKEN = os.getenv("GOTIFY_TOKEN", "")
|
|
|
|
# Multi-root support from environment
|
|
TV_ROOTS = os.getenv("TV_ROOTS", "").split(":") if os.getenv("TV_ROOTS") else []
|
|
MOVIE_ROOTS = os.getenv("MOVIE_ROOTS", "").split(":") if os.getenv("MOVIE_ROOTS") else []
|
|
|
|
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)
|
|
|
|
# File parsing patterns
|
|
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})\)")
|
|
|
|
|
|
def log(msg: str) -> None:
|
|
ts = dt.datetime.now().strftime(LOG_TS_FMT)
|
|
print(f"[{ts}] {msg}", flush=True)
|
|
|
|
|
|
def send_gotify_notification(title: str, message: str, priority: int = 5) -> None:
|
|
"""Send notification to Gotify server if configured."""
|
|
if not DEFAULT_GOTIFY_URL or not DEFAULT_GOTIFY_TOKEN:
|
|
return
|
|
|
|
try:
|
|
requests.post(
|
|
f"{DEFAULT_GOTIFY_URL}/message",
|
|
data={
|
|
"title": title,
|
|
"message": message,
|
|
"priority": priority
|
|
},
|
|
headers={"X-Gotify-Key": DEFAULT_GOTIFY_TOKEN},
|
|
timeout=10
|
|
)
|
|
except Exception as e:
|
|
log(f"WARN: Gotify notification failed: {e}")
|
|
|
|
|
|
def load_env_file(path: Optional[str]) -> None:
|
|
"""Load a simple KEY=VALUE .env file."""
|
|
if not path:
|
|
return
|
|
p = Path(path)
|
|
if not p.is_file():
|
|
log(f"WARN: .env file not found: {path}")
|
|
return
|
|
|
|
log(f"Loading environment from: {path}")
|
|
for line in p.read_text().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 optional surrounding quotes
|
|
if (v.startswith("'") and v.endswith("'")) or (v.startswith('"') and v.endswith('"')):
|
|
v = v[1:-1]
|
|
os.environ.setdefault(k, v)
|
|
|
|
|
|
# =================
|
|
# SQLite helpers
|
|
# =================
|
|
|
|
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]]:
|
|
"""Get cached timestamp and source for a path."""
|
|
cur = cx.execute("SELECT chosen_ts, chosen_src FROM media_dates WHERE path = ?", (path,))
|
|
row = cur.fetchone()
|
|
if row:
|
|
return row[0], row[1]
|
|
return 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 helpers
|
|
# =================
|
|
|
|
def _req_json(
|
|
method: str,
|
|
url: str,
|
|
*,
|
|
headers: Optional[Dict[str, str]] = None,
|
|
params: Optional[Dict[str, object]] = None,
|
|
timeout: int = 30,
|
|
retries: int = 3,
|
|
backoff: float = 2.0,
|
|
) -> dict:
|
|
"""Request JSON with retries/backoff; raise on final failure."""
|
|
last_err = None
|
|
for attempt in range(1, retries + 1):
|
|
try:
|
|
r = requests.request(method, url, headers=headers or {}, params=params or {}, timeout=timeout)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except (requests.exceptions.Timeout, requests.exceptions.ReadTimeout) as e:
|
|
last_err = e
|
|
except requests.exceptions.RequestException as e:
|
|
last_err = e
|
|
if attempt < retries:
|
|
time.sleep(backoff ** attempt)
|
|
raise last_err # type: ignore[misc]
|
|
|
|
|
|
# =================
|
|
# Sonarr client
|
|
# =================
|
|
|
|
@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]:
|
|
return _req_json(
|
|
"GET", f"{self.url}/api/v3/series",
|
|
headers=self._h(), timeout=self.timeout, retries=self.retries
|
|
)
|
|
|
|
def episodes_for(self, series_id: int) -> List[dict]:
|
|
return _req_json(
|
|
"GET", f"{self.url}/api/v3/episode",
|
|
headers=self._h(), params={"seriesId": series_id},
|
|
timeout=self.timeout, retries=self.retries
|
|
)
|
|
|
|
def history_for(self, series_id: int) -> List[dict]:
|
|
res = _req_json(
|
|
"GET", f"{self.url}/api/v3/history",
|
|
headers=self._h(), params={
|
|
"page": 1,
|
|
"pageSize": 1000,
|
|
"sortKey": "date",
|
|
"sortDirection": "ascending",
|
|
"seriesId": series_id,
|
|
"eventType": 3 # Downloaded/Imported events only
|
|
},
|
|
timeout=self.timeout, retries=self.retries
|
|
)
|
|
return res.get("records", [])
|
|
|
|
|
|
# =================
|
|
# Radarr client
|
|
# =================
|
|
|
|
@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]:
|
|
return _req_json(
|
|
"GET", f"{self.url}/api/v3/movie",
|
|
headers=self._h(), timeout=self.timeout, retries=self.retries
|
|
)
|
|
|
|
def history_for(self, movie_id: int) -> List[dict]:
|
|
res = _req_json(
|
|
"GET", f"{self.url}/api/v3/history",
|
|
headers=self._h(), params={
|
|
"page": 1,
|
|
"pageSize": 1000,
|
|
"sortKey": "date",
|
|
"sortDirection": "ascending",
|
|
"movieId": movie_id,
|
|
"eventType": 3 # Downloaded/Imported events only
|
|
},
|
|
timeout=self.timeout, retries=self.retries
|
|
)
|
|
return res.get("records", [])
|
|
|
|
|
|
# =================
|
|
# TMDB client
|
|
# =================
|
|
|
|
@dataclasses.dataclass
|
|
class TMDBClient:
|
|
api_key: str
|
|
country: str = DEFAULT_TMDB_PRIMARY_COUNTRY
|
|
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}
|
|
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
|
|
)
|
|
tv = js.get("tv_results") or []
|
|
if tv:
|
|
return tv[0].get("id")
|
|
return 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
|
|
)
|
|
movies = js.get("movie_results") or []
|
|
if movies:
|
|
return movies[0].get("id")
|
|
return None
|
|
|
|
def search_tv(self, name: str, first_air_date_year: Optional[int]) -> Optional[int]:
|
|
js = _req_json(
|
|
"GET", f"{self.base_url}/search/tv",
|
|
params=self._p(query=name, first_air_date_year=first_air_date_year or ""),
|
|
timeout=self.timeout, retries=self.retries
|
|
)
|
|
res = js.get("results") or []
|
|
if res:
|
|
return res[0].get("id")
|
|
return 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") or []
|
|
if res:
|
|
return res[0].get("id")
|
|
return 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")
|
|
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")
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# =================
|
|
# File parsing
|
|
# =================
|
|
|
|
@dataclasses.dataclass
|
|
class EpisodeKey:
|
|
series_title: str
|
|
year: Optional[int]
|
|
imdb_id: Optional[str]
|
|
tmdb_id: Optional[str]
|
|
season: int
|
|
episode: int
|
|
|
|
|
|
@dataclasses.dataclass
|
|
class MovieKey:
|
|
title: str
|
|
year: Optional[int]
|
|
imdb_id: Optional[str]
|
|
tmdb_id: Optional[str]
|
|
|
|
|
|
def parse_episode_from_path(path: Path) -> Optional[EpisodeKey]:
|
|
"""Parse episode info from file path."""
|
|
m = SE_EP_RX.search(path.name)
|
|
if not m:
|
|
return None
|
|
season = int(m.group(1))
|
|
episode = int(m.group(2))
|
|
|
|
series_dir = path.parent
|
|
if series_dir.name.lower().startswith("season "):
|
|
series_dir = series_dir.parent
|
|
|
|
series_title = series_dir.name
|
|
imdb_id = None
|
|
tmdb_id = None
|
|
year = None
|
|
|
|
m2 = IMDB_RX.search(series_title)
|
|
if m2:
|
|
imdb_id = m2.group(1)
|
|
|
|
m3 = TMDB_RX.search(series_title)
|
|
if m3:
|
|
tmdb_id = m3.group(1)
|
|
|
|
m4 = YEAR_RX.search(series_title)
|
|
if m4:
|
|
try:
|
|
year = int(m4.group(1))
|
|
except ValueError:
|
|
year = None
|
|
|
|
clean_title = IMDB_RX.sub("", series_title)
|
|
clean_title = TMDB_RX.sub("", clean_title)
|
|
clean_title = YEAR_RX.sub("", clean_title)
|
|
clean_title = clean_title.strip()
|
|
if not clean_title:
|
|
clean_title = series_title
|
|
|
|
return EpisodeKey(clean_title, year, imdb_id, tmdb_id, season, episode)
|
|
|
|
|
|
def parse_movie_from_path(path: Path) -> Optional[MovieKey]:
|
|
"""Parse movie info from file path."""
|
|
movie_dir = path.parent
|
|
movie_title = movie_dir.name
|
|
|
|
imdb_id = None
|
|
tmdb_id = None
|
|
year = None
|
|
|
|
m1 = IMDB_RX.search(movie_title)
|
|
if m1:
|
|
imdb_id = m1.group(1)
|
|
|
|
m2 = TMDB_RX.search(movie_title)
|
|
if m2:
|
|
tmdb_id = m2.group(1)
|
|
|
|
m3 = YEAR_RX.search(movie_title)
|
|
if m3:
|
|
try:
|
|
year = int(m3.group(1))
|
|
except ValueError:
|
|
year = None
|
|
|
|
clean_title = IMDB_RX.sub("", movie_title)
|
|
clean_title = TMDB_RX.sub("", clean_title)
|
|
clean_title = YEAR_RX.sub("", clean_title)
|
|
clean_title = clean_title.strip()
|
|
if not clean_title:
|
|
clean_title = movie_title
|
|
|
|
return MovieKey(clean_title, year, imdb_id, tmdb_id)
|
|
|
|
|
|
# =================
|
|
# Date choosing
|
|
# =================
|
|
|
|
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_history(sc: SonarrClient, series_id: int) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
"""Return (unix_ts, src, extra) or None."""
|
|
try:
|
|
hist = sc.history_for(series_id)
|
|
except Exception as e:
|
|
log(f"WARN: Sonarr history timeout/err series={series_id}: {e}")
|
|
return None
|
|
|
|
best_ts = None
|
|
best = None
|
|
for rec in hist:
|
|
event_type = rec.get("eventType")
|
|
if event_type not in [1, 3]: # 1=Grabbed, 3=Downloaded
|
|
continue
|
|
|
|
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 best_ts is None or ts < best_ts:
|
|
best_ts = ts
|
|
best = rec
|
|
|
|
if best_ts is not None:
|
|
return best_ts, "sonarr_history", {"history_id": best.get("id"), "event_type": best.get("eventType")}
|
|
return None
|
|
|
|
|
|
def pick_from_sonarr_episode(sc: SonarrClient, series_id: int, season: int, episode: int) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
try:
|
|
eps = sc.episodes_for(series_id)
|
|
except Exception as e:
|
|
log(f"WARN: Sonarr episodes timeout/err series={series_id}: {e}")
|
|
return None
|
|
|
|
for ep in eps:
|
|
if ep.get("seasonNumber") == season and ep.get("episodeNumber") == episode:
|
|
air = ep.get("airDate") or ep.get("airDateUtc")
|
|
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
|
|
|
|
|
|
# Similar functions for Radarr and TMDB...
|
|
def pick_from_tmdb_tv(tmdb: TMDBClient, epk: EpisodeKey) -> Optional[Tuple[int, str, Dict[str, object]]]:
|
|
if not tmdb.api_key:
|
|
return None
|
|
|
|
tv_id: Optional[int] = None
|
|
|
|
if epk.tmdb_id:
|
|
try:
|
|
tv_id = int(epk.tmdb_id)
|
|
except (ValueError, TypeError):
|
|
tv_id = None
|
|
|
|
if tv_id is None and epk.imdb_id:
|
|
try:
|
|
tv_id = tmdb.find_tv_by_imdb(epk.imdb_id)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB find by IMDb failed {epk.imdb_id}: {e}")
|
|
|
|
if tv_id is None:
|
|
try:
|
|
tv_id = tmdb.search_tv(epk.series_title, epk.year)
|
|
except Exception as e:
|
|
log(f"WARN: TMDB search failed title={epk.series_title}: {e}")
|
|
|
|
if tv_id is None:
|
|
return None
|
|
|
|
air = tmdb.episode_air_date(tv_id, epk.season, epk.episode)
|
|
if not air:
|
|
return None
|
|
|
|
ts = parse_date_yyyy_mm_dd(air)
|
|
if ts is None:
|
|
return None
|
|
return ts, "tmdb_airdate", {"tv_id": tv_id, "country": tmdb.country}
|
|
|
|
|
|
# =================
|
|
# NFO Management
|
|
# =================
|
|
|
|
def create_simple_nfo(season: int, episode: int, air_date: Optional[str] = None, date_added: Optional[str] = None) -> str:
|
|
"""Create a simple NFO file content for SxxExx.nfo format."""
|
|
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:
|
|
"""Create a simple NFO file content for movie.nfo format."""
|
|
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 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,
|
|
dry_run: bool = False
|
|
) -> Tuple[int, int]:
|
|
"""Manage NFO files. Returns (created_count, removed_count)."""
|
|
created = 0
|
|
removed = 0
|
|
|
|
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)
|
|
|
|
if not dry_run:
|
|
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:
|
|
simple_nfo_path.write_text(nfo_content, encoding='utf-8')
|
|
created = 1
|
|
except Exception as e:
|
|
log(f"WARN: Failed to create {simple_nfo_path}: {e}")
|
|
else:
|
|
# For dry run, check if we would create/update
|
|
if not simple_nfo_path.exists():
|
|
created = 1
|
|
else:
|
|
try:
|
|
existing_content = simple_nfo_path.read_text(encoding='utf-8', errors='ignore')
|
|
if existing_content.strip() != nfo_content.strip():
|
|
created = 1
|
|
except Exception:
|
|
created = 1
|
|
|
|
# Remove complex NFO files (but keep simple ones and season.nfo)
|
|
for nfo_file in media_dir.glob("*.nfo"):
|
|
nfo_name = nfo_file.name.lower()
|
|
|
|
# Keep these files
|
|
if NFO_SIMPLE_PATTERN.match(nfo_file.name):
|
|
continue # Keep S01E01.nfo format
|
|
if nfo_name == "season.nfo":
|
|
continue # Keep season.nfo
|
|
if nfo_name == "tvshow.nfo":
|
|
continue # Keep tvshow.nfo
|
|
|
|
# This is a complex NFO file - remove it
|
|
if not dry_run:
|
|
try:
|
|
nfo_file.unlink()
|
|
log(f"Removed complex NFO: {nfo_file.name}")
|
|
removed += 1
|
|
except Exception as e:
|
|
log(f"WARN: Failed to remove {nfo_file}: {e}")
|
|
else:
|
|
log(f"Would remove complex NFO: {nfo_file.name}")
|
|
removed += 1
|
|
|
|
return created, removed
|
|
|
|
|
|
# =================
|
|
# Sidecar files
|
|
# =================
|
|
|
|
def find_sidecar_files(media_file: Path) -> List[Path]:
|
|
"""Find sidecar files (subtitles, etc.) that match the media file."""
|
|
sidecars = []
|
|
base_name = media_file.stem
|
|
parent_dir = media_file.parent
|
|
|
|
for ext in SIDECAR_EXTS:
|
|
exact_match = parent_dir / f"{base_name}{ext}"
|
|
if exact_match.exists():
|
|
sidecars.append(exact_match)
|
|
|
|
for lang_file in parent_dir.glob(f"{base_name}.*{ext}"):
|
|
if lang_file not in sidecars:
|
|
sidecars.append(lang_file)
|
|
|
|
return sidecars
|
|
|
|
|
|
def apply_timestamp_to_file(file_path: Path, timestamp: int, dry_run: bool = False) -> bool:
|
|
"""Apply timestamp to a file. Returns True if successful."""
|
|
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
|
|
|
|
|
|
# =================
|
|
# Scanning
|
|
# =================
|
|
|
|
def iter_media_files(roots: List[str]) -> Iterable[Path]:
|
|
for root in roots:
|
|
r = Path(root)
|
|
if not r.exists():
|
|
log(f"WARN: Root path does not exist: {root}")
|
|
continue
|
|
if r.is_file():
|
|
if r.suffix.lower() in VIDEO_EXTS:
|
|
yield r
|
|
continue
|
|
for p in r.rglob("*"):
|
|
if p.is_file() and p.suffix.lower() in VIDEO_EXTS:
|
|
yield p
|
|
|
|
|
|
def choose_series_id_for_path(series_index: List[dict], path: Path) -> Optional[int]:
|
|
"""Try to map a path to a Sonarr seriesId by matching the series' 'path' as a prefix."""
|
|
series_dir = path.parent
|
|
if series_dir.name.lower().startswith("season "):
|
|
series_dir = series_dir.parent
|
|
sdir = str(series_dir)
|
|
|
|
best_len = -1
|
|
best_sid = None
|
|
for s in series_index:
|
|
spath = s.get("path", "")
|
|
if spath and sdir.startswith(spath):
|
|
L = len(spath)
|
|
if L > best_len:
|
|
best_len = L
|
|
best_sid = s.get("id")
|
|
return best_sid
|
|
|
|
|
|
# =================
|
|
# Apply command
|
|
# =================
|
|
|
|
def apply_command(args: argparse.Namespace) -> None:
|
|
"""Main apply command function."""
|
|
log("=== Media Date Sync Starting ===")
|
|
|
|
# Load environment first
|
|
load_env_file(args.env)
|
|
|
|
# NOW get the updated environment variables
|
|
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")
|
|
|
|
# Debug: Show what we loaded
|
|
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'}")
|
|
|
|
# Resolve clients with updated environment
|
|
sc = None
|
|
if 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")
|
|
else:
|
|
log("WARN: Sonarr not configured")
|
|
|
|
tmdb = TMDBClient(tmdb_key, tmdb_country)
|
|
if not tmdb_key:
|
|
log("WARN: TMDB not configured - limited functionality")
|
|
|
|
# Initialize database
|
|
cx = sqlite3.connect(args.db)
|
|
db_init(cx)
|
|
log(f"Database initialized: {args.db}")
|
|
|
|
# Load Sonarr series index
|
|
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}")
|
|
|
|
# Determine root paths
|
|
roots = args.root or []
|
|
if not roots:
|
|
tv_roots = os.getenv("TV_ROOTS", "").split(":") if os.getenv("TV_ROOTS") else []
|
|
movie_roots = os.getenv("MOVIE_ROOTS", "").split(":") if os.getenv("MOVIE_ROOTS") else []
|
|
|
|
if args.kind == "tv" and tv_roots:
|
|
roots = [r for r in tv_roots if r.strip()]
|
|
log(f"Using TV_ROOTS from environment: {', '.join(roots)}")
|
|
elif args.kind == "movie" and movie_roots:
|
|
roots = [r for r in movie_roots if r.strip()]
|
|
log(f"Using MOVIE_ROOTS from environment: {', '.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 = list(iter_media_files(roots))
|
|
log(f"Found {len(files)} media file(s)")
|
|
|
|
if not files:
|
|
log("No media files found. Check your root paths and permissions.")
|
|
return
|
|
|
|
# Processing counters
|
|
checked = 0
|
|
updated = 0
|
|
misses = 0
|
|
cached = 0
|
|
nfo_created = 0
|
|
nfo_removed = 0
|
|
sidecars_updated = 0
|
|
|
|
log(f"Starting processing: refresh={args.refresh}, dry_run={args.dry_run}, manage_nfo={args.manage_nfo}")
|
|
|
|
for fp in files:
|
|
checked += 1
|
|
path = str(fp)
|
|
|
|
if checked <= 5: # Show detailed info for first 5 files
|
|
log(f"Processing file {checked}: {fp.name}")
|
|
|
|
# Check cache first (unless refresh requested)
|
|
if not args.refresh:
|
|
cached_result = db_get_cached(cx, path)
|
|
if cached_result:
|
|
cached_ts, cached_src = cached_result
|
|
if checked <= 5:
|
|
log(f" Using cached result: {cached_src}")
|
|
|
|
# Apply cached timestamp to media file
|
|
if apply_timestamp_to_file(fp, cached_ts, args.dry_run):
|
|
cached += 1
|
|
|
|
# Also update sidecar files
|
|
sidecar_files = find_sidecar_files(fp)
|
|
for sidecar in sidecar_files:
|
|
if apply_timestamp_to_file(sidecar, cached_ts, args.dry_run):
|
|
sidecars_updated += 1
|
|
|
|
continue
|
|
|
|
# Parse media file based on type
|
|
ts_src_extra: Optional[Tuple[int, str, Dict[str, object]]] = None
|
|
air_date_str = None
|
|
date_added_str = None
|
|
|
|
if args.kind == "tv":
|
|
epk = parse_episode_from_path(fp)
|
|
if not epk:
|
|
if checked <= 10: # Only log first 10 parse failures to avoid spam
|
|
log(f"WARN: Could not parse episode info from: {fp.name}")
|
|
misses += 1
|
|
continue
|
|
|
|
if checked <= 5:
|
|
log(f" Parsed: {epk.series_title} S{epk.season:02d}E{epk.episode:02d}")
|
|
|
|
# Try Sonarr first
|
|
series_id = choose_series_id_for_path(series_index, fp) if series_index else None
|
|
if checked <= 5:
|
|
log(f" Sonarr series ID: {series_id}")
|
|
|
|
if sc and series_id:
|
|
if checked <= 5:
|
|
log(f" Checking Sonarr history...")
|
|
# Try history first (import date)
|
|
history_result = pick_from_sonarr_history(sc, series_id)
|
|
if history_result:
|
|
ts_src_extra = history_result
|
|
if checked <= 5:
|
|
log(f" Got history: {history_result[1]}")
|
|
try:
|
|
date_added_dt = dt.datetime.fromtimestamp(history_result[0], tz=dt.timezone.utc)
|
|
date_added_str = date_added_dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
pass
|
|
|
|
if checked <= 5:
|
|
log(f" Checking Sonarr episode data...")
|
|
# Get air date regardless
|
|
episode_result = pick_from_sonarr_episode(sc, series_id, epk.season, epk.episode)
|
|
if episode_result:
|
|
if not ts_src_extra: # Use air date if no history
|
|
ts_src_extra = episode_result
|
|
if not date_added_str:
|
|
try:
|
|
date_added_dt = dt.datetime.fromtimestamp(episode_result[0], tz=dt.timezone.utc)
|
|
date_added_str = date_added_dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
pass
|
|
|
|
# Extract air date for NFO
|
|
try:
|
|
air_date_ts = episode_result[0]
|
|
air_date_dt = dt.datetime.fromtimestamp(air_date_ts, tz=dt.timezone.utc)
|
|
air_date_str = air_date_dt.strftime("%Y-%m-%d")
|
|
except Exception:
|
|
pass
|
|
|
|
# TMDB fallback
|
|
if not ts_src_extra and tmdb_key:
|
|
if checked <= 5:
|
|
log(f" Trying TMDB fallback...")
|
|
tmdb_result = pick_from_tmdb_tv(tmdb, epk)
|
|
if tmdb_result:
|
|
ts_src_extra = tmdb_result
|
|
if checked <= 5:
|
|
log(f" Got TMDB result: {tmdb_result[1]}")
|
|
try:
|
|
air_date_ts = tmdb_result[0]
|
|
air_date_dt = dt.datetime.fromtimestamp(air_date_ts, tz=dt.timezone.utc)
|
|
air_date_str = air_date_dt.strftime("%Y-%m-%d")
|
|
if not date_added_str:
|
|
date_added_str = air_date_dt.strftime("%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
pass
|
|
|
|
else: # movie support would go here
|
|
log("Movie support not implemented yet")
|
|
misses += 1
|
|
continue
|
|
|
|
# Last resort: keep existing mtime if allowed
|
|
if not ts_src_extra and args.live_on_miss:
|
|
try:
|
|
mtime = int(fp.stat().st_mtime)
|
|
ts_src_extra = (mtime, "existing_mtime", {})
|
|
if checked <= 5:
|
|
log(f" Using existing mtime as fallback")
|
|
except Exception:
|
|
ts_src_extra = None
|
|
|
|
if not ts_src_extra:
|
|
if checked <= 10: # Only log first 10 misses to avoid spam
|
|
log(f"WARN: No date found for: {fp.name}")
|
|
misses += 1
|
|
continue
|
|
|
|
ts, src, extra = ts_src_extra
|
|
|
|
if checked <= 5:
|
|
log(f" Final source: {src}, timestamp: {dt.datetime.fromtimestamp(ts)}")
|
|
|
|
# Apply timestamp to media file
|
|
if apply_timestamp_to_file(fp, ts, args.dry_run):
|
|
updated += 1
|
|
|
|
# Apply timestamp to sidecar files
|
|
sidecar_files = find_sidecar_files(fp)
|
|
if sidecar_files and checked <= 5:
|
|
log(f" Found {len(sidecar_files)} sidecar files")
|
|
for sidecar in sidecar_files:
|
|
if apply_timestamp_to_file(sidecar, ts, args.dry_run):
|
|
sidecars_updated += 1
|
|
|
|
# Handle NFO files for TV shows
|
|
if args.kind == "tv" and args.manage_nfo:
|
|
created, removed = manage_nfo_files(
|
|
fp.parent, "tv", epk.season, epk.episode,
|
|
air_date_str, date_added_str, dry_run=args.dry_run
|
|
)
|
|
if (created or removed) and checked <= 5:
|
|
log(f" NFO: created={created}, removed={removed}")
|
|
nfo_created += created
|
|
nfo_removed += removed
|
|
|
|
# Update database
|
|
if not args.dry_run:
|
|
try:
|
|
if args.kind == "tv":
|
|
extra_data = {
|
|
**(extra or {}),
|
|
"series_title": epk.series_title,
|
|
"season": epk.season,
|
|
"episode": epk.episode,
|
|
}
|
|
else:
|
|
extra_data = extra or {}
|
|
|
|
db_upsert(cx, path, args.kind, ts, src, extra_data)
|
|
except sqlite3.OperationalError as e:
|
|
log(f"WARN: DB write failed: {e}")
|
|
|
|
# Progress updates
|
|
if checked % 500 == 0:
|
|
log(f"Progress: {checked}/{len(files)} (updated: {updated}, cached: {cached}, misses: {misses})")
|
|
|
|
# Final results
|
|
completion_msg = f"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"Sidecars updated: {sidecars_updated}\n"
|
|
|
|
if args.manage_nfo:
|
|
completion_msg += f"NFO created: {nfo_created}\n" \
|
|
f"NFO removed: {nfo_removed}\n"
|
|
|
|
completion_msg += f"Dry run: {args.dry_run}"
|
|
|
|
log(f"=== Complete ===")
|
|
log(completion_msg.replace('\n', ', '))
|
|
|
|
# Send notification
|
|
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)
|
|
|
|
# apply
|
|
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 (or use TV_ROOTS/MOVIE_ROOTS in .env)")
|
|
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 simple NFO files with air/release dates and date added (import dates). Remove complex NFOs.")
|
|
|
|
return p
|
|
|
|
|
|
def main() -> None:
|
|
try:
|
|
parser = build_arg_parser()
|
|
args = parser.parse_args()
|
|
|
|
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()
|