712 lines
28 KiB
Python
712 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
media_date_cache.py
|
|
One script to:
|
|
- Set episode <dateadded> using this order of precedence:
|
|
1) Sonarr earliest *history* import (downloadFolderImported/episodeFileImported)
|
|
2) Sonarr airDateUtc / airDate
|
|
3) TMDB air_date
|
|
4) Trakt first_aired
|
|
5) OMDb Released
|
|
6) Existing SxxEyy.nfo (dateadded/ aired)
|
|
7) File mtime (last resort)
|
|
- Write minimal per-episode NFOs (SxxEyy.nfo) with <aired>, <dateadded>, and <!-- source: ... -->
|
|
- Remove long filename episode NFOs (e.g., "Show - SxxEyy - title.nfo")
|
|
- Optionally set season folder mtime to earliest episode date
|
|
- Does NOT modify season.nfo or tvshow.nfo
|
|
|
|
CLI:
|
|
python3 media_date_cache.py rebuild-db \
|
|
--env /opt/scripts/unmanic/.env \
|
|
--db /opt/scripts/unmanic/.cache/media_dates.db
|
|
|
|
python3 media_date_cache.py apply \
|
|
--env /opt/scripts/unmanic/.env \
|
|
--db /opt/scripts/unmanic/.cache/media_dates.db \
|
|
--path "/path/Show Name (Year) [imdb-ttxxxxxxx]" \
|
|
--manage-nfo \
|
|
--fix-dir-mtimes
|
|
"""
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import os
|
|
import re
|
|
import sqlite3
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Optional, Tuple, Dict, List
|
|
|
|
import requests
|
|
import xml.etree.ElementTree as ET
|
|
|
|
# -----------------------------
|
|
# Logging
|
|
# -----------------------------
|
|
def _ts():
|
|
return dt.datetime.now(dt.timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
def _log(level, msg): print(f"[{_ts()}] {level}: {msg}")
|
|
def info(msg): _log("INFO", msg)
|
|
def warn(msg): _log("WARNING", msg)
|
|
def err(msg): _log("ERROR", msg)
|
|
def debug(msg):
|
|
if os.environ.get("MDC_DEBUG") == "1":
|
|
_log("DEBUG", msg)
|
|
|
|
# -----------------------------
|
|
# Minimal .env loader
|
|
# -----------------------------
|
|
def load_env_file(env_path: Optional[str]) -> None:
|
|
if not env_path:
|
|
return
|
|
p = Path(env_path)
|
|
if not p.exists():
|
|
warn(f".env not loaded (file not found) path={env_path}")
|
|
return
|
|
try:
|
|
with p.open("r", encoding="utf-8") as fh:
|
|
for line in fh:
|
|
s = line.strip()
|
|
if not s or s.startswith("#") or "=" not in s:
|
|
continue
|
|
k, v = s.split("=", 1)
|
|
k, v = k.strip(), v.strip()
|
|
if (v.startswith('"') and v.endswith('"')) or (v.startswith("'") and v.endswith("'")):
|
|
v = v[1:-1]
|
|
os.environ.setdefault(k, v)
|
|
info(f".env loaded from {env_path}")
|
|
except Exception as e:
|
|
warn(f".env not loaded (parse error): {e} path={env_path}")
|
|
|
|
# -----------------------------
|
|
# SQLite cache (lightweight)
|
|
# -----------------------------
|
|
SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS episode_dates(
|
|
series_imdb TEXT NOT NULL,
|
|
season INTEGER NOT NULL,
|
|
episode INTEGER NOT NULL,
|
|
dateadded TEXT,
|
|
source TEXT,
|
|
PRIMARY KEY(series_imdb, season, episode)
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_series ON episode_dates(series_imdb);
|
|
"""
|
|
|
|
def db_connect(db_path: str):
|
|
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
|
|
conn = sqlite3.connect(db_path)
|
|
conn.execute("PRAGMA journal_mode=WAL;")
|
|
conn.execute("PRAGMA synchronous=NORMAL;")
|
|
return conn
|
|
|
|
def db_init(conn):
|
|
for stmt in SCHEMA.strip().split(";"):
|
|
s = stmt.strip()
|
|
if s:
|
|
conn.execute(s)
|
|
conn.commit()
|
|
|
|
def db_set_date(conn, series_imdb: str, season: int, episode: int, dateadded: str, source: str):
|
|
conn.execute(
|
|
"INSERT INTO episode_dates(series_imdb, season, episode, dateadded, source) VALUES(?,?,?,?,?) "
|
|
"ON CONFLICT(series_imdb, season, episode) DO UPDATE SET dateadded=excluded.dateadded, source=excluded.source",
|
|
(series_imdb, season, episode, dateadded, source),
|
|
)
|
|
conn.commit()
|
|
|
|
def db_get_date(conn, series_imdb: str, season: int, episode: int) -> Optional[Tuple[str, str]]:
|
|
cur = conn.execute(
|
|
"SELECT dateadded, source FROM episode_dates WHERE series_imdb=? AND season=? AND episode=?",
|
|
(series_imdb, season, episode),
|
|
)
|
|
row = cur.fetchone()
|
|
return (row[0], row[1]) if row else None
|
|
|
|
# -----------------------------
|
|
# Helpers
|
|
# -----------------------------
|
|
IMDB_DIR_RX = re.compile(r"\[imdb-(tt\d+)\]", re.IGNORECASE)
|
|
SE_EP_RX = re.compile(r"[Ss](\d{1,2})[Ee](\d{1,3})")
|
|
|
|
def parse_series_imdb_from_path(series_path: str) -> Optional[str]:
|
|
m = IMDB_DIR_RX.search(Path(series_path).name)
|
|
return m.group(1).lower() if m else None
|
|
|
|
def iso_utc_from_timestamp(ts: float) -> str:
|
|
return dt.datetime.fromtimestamp(ts, tz=dt.timezone.utc).isoformat(timespec="seconds")
|
|
|
|
def iso_utc(date_str_or_iso: str, default_time: str = "00:00:00") -> str:
|
|
"""Normalize YYYY-MM-DD or ISO to ISO UTC (seconds)."""
|
|
s = date_str_or_iso.strip()
|
|
try:
|
|
if "T" in s:
|
|
dtp = dt.datetime.fromisoformat(s.replace("Z", "+00:00"))
|
|
if dtp.tzinfo is None:
|
|
dtp = dtp.replace(tzinfo=dt.timezone.utc)
|
|
return dtp.astimezone(dt.timezone.utc).isoformat(timespec="seconds")
|
|
return f"{dt.date.fromisoformat(s).isoformat()}T{default_time}+00:00"
|
|
except Exception:
|
|
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
|
|
|
|
def iso_utc_now() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds")
|
|
|
|
def ymd(s: Optional[str]) -> Optional[str]:
|
|
if not s: return None
|
|
try:
|
|
return dt.date.fromisoformat(s[:10]).isoformat()
|
|
except Exception:
|
|
return s[:10]
|
|
|
|
def write_episode_nfo(nfo_path: Path, season: int, episode: int, aired_date: Optional[str], dateadded_iso: str, source: str):
|
|
lines = ['<?xml version="1.0" encoding="UTF-8" standalone="yes"?>', "<episodedetails>"]
|
|
lines.append(f" <season>{season}</season>")
|
|
lines.append(f" <episode>{episode}</episode>")
|
|
if aired_date:
|
|
lines.append(f" <aired>{ymd(aired_date)}</aired>")
|
|
lines.append(f" <dateadded>{dateadded_iso}</dateadded>")
|
|
lines.append(f" <!-- source: {source} -->")
|
|
lines.append("</episodedetails>\n")
|
|
nfo_path.write_text("\n".join(lines), encoding="utf-8")
|
|
|
|
def read_existing_short_nfo(folder: Path, season: int, episode: int) -> Tuple[Optional[str], Optional[str]]:
|
|
nfo = folder / f"S{season:02d}E{episode:02d}.nfo"
|
|
if not nfo.exists():
|
|
return None, None
|
|
try:
|
|
tree = ET.parse(nfo)
|
|
root = tree.getroot()
|
|
aired = root.findtext("aired")
|
|
dateadded = root.findtext("dateadded")
|
|
return (aired, dateadded)
|
|
except Exception:
|
|
return None, None
|
|
|
|
def remove_longname_nfos(folder: Path) -> int:
|
|
removed = 0
|
|
for f in folder.glob("*.nfo"):
|
|
name = f.name
|
|
if re.fullmatch(r"[Ss]\d{2}[Ee]\d{2,3}\.nfo", name):
|
|
continue
|
|
if name.lower() in ("season.nfo", "tvshow.nfo"):
|
|
continue
|
|
try:
|
|
f.unlink()
|
|
removed += 1
|
|
except Exception as e:
|
|
warn(f"Failed to remove {f}: {e}")
|
|
return removed
|
|
|
|
def find_episode_video_and_se(folder: Path) -> Dict[Tuple[int, int], Path]:
|
|
videos = {}
|
|
for f in folder.iterdir():
|
|
if not f.is_file(): continue
|
|
if f.suffix.lower() not in (".mkv", ".mp4", ".avi", ".ts", ".m2ts"): continue
|
|
m = SE_EP_RX.search(f.name)
|
|
if not m: continue
|
|
s, e = int(m.group(1)), int(m.group(2))
|
|
videos[(s, e)] = f
|
|
return videos
|
|
|
|
# -----------------------------
|
|
# Sonarr client
|
|
# -----------------------------
|
|
class Sonarr:
|
|
def __init__(self, url: Optional[str], apikey: Optional[str], timeout: int = 45, retries: int = 3):
|
|
self.url = (url or "").rstrip("/")
|
|
self.apikey = apikey or ""
|
|
self.timeout, self.retries = timeout, retries
|
|
self.enabled = bool(self.url and self.apikey)
|
|
if not self.enabled:
|
|
info("Sonarr not configured (set --sonarr-url/--sonarr-apikey or env SONARR_URL/SONARR_API_KEY)")
|
|
|
|
def _get(self, path, params=None):
|
|
if not self.enabled: return None
|
|
p = dict(params or {}); p["apikey"] = self.apikey
|
|
last = None
|
|
for _ in range(self.retries + 1):
|
|
try:
|
|
r = requests.get(self.url + path, params=p, timeout=self.timeout)
|
|
if r.status_code == 404: return None
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
last = e; time.sleep(0.75)
|
|
warn(f"Sonarr GET {path} failed: {last}"); return None
|
|
|
|
def series_by_imdb(self, imdb_id: str):
|
|
return self._get("/api/v3/series/lookup", params={"term": f"imdb:{imdb_id}"})
|
|
|
|
def episode(self, series_id: int, season: int, episode: int):
|
|
j = self._get("/api/v3/episode", params={"seriesId": series_id, "seasonNumber": season, "episodeNumber": episode})
|
|
if not j: return None
|
|
return j[0] if isinstance(j, list) and j else j
|
|
|
|
def episode_file(self, episode_file_id: int):
|
|
return self._get(f"/api/v3/episodefile/{episode_file_id}")
|
|
|
|
# ----- new: history helpers -----
|
|
def _history_series_pages(self, series_id: int, page_size: int = 1000):
|
|
"""Yield history records for a series (handles both paged and list responses)."""
|
|
# Try paged endpoint
|
|
page = 1
|
|
while True:
|
|
j = self._get("/api/v3/history/series",
|
|
params={"seriesId": series_id, "page": page, "pageSize": page_size,
|
|
"sortKey": "date", "sortDirection": "descending"})
|
|
if not j:
|
|
break
|
|
# Some Sonarr setups return a dict with 'records'; others may return a list.
|
|
if isinstance(j, dict) and "records" in j:
|
|
recs = j.get("records") or []
|
|
for r in recs:
|
|
yield r
|
|
total = int(j.get("totalRecords", len(recs)))
|
|
if page * page_size >= total or not recs:
|
|
break
|
|
elif isinstance(j, list):
|
|
if not j: break
|
|
for r in j:
|
|
yield r
|
|
break
|
|
else:
|
|
break
|
|
page += 1
|
|
|
|
def earliest_import_for_episode(self, series_id: int, episode_id: int) -> Optional[str]:
|
|
"""
|
|
Return earliest (oldest) history 'import' date for a given episode.
|
|
Looks for eventType in a broad set that indicates an import.
|
|
"""
|
|
if not self.enabled:
|
|
return None
|
|
# Event types we treat as "import"
|
|
IMPORT_EVENTS = {"downloadFolderImported", "episodeFileImported", "episodeFileAdded"}
|
|
earliest = None
|
|
for rec in self._history_series_pages(series_id):
|
|
try:
|
|
if int(rec.get("episodeId", -1)) != int(episode_id):
|
|
continue
|
|
et = rec.get("eventType") or rec.get("type") or ""
|
|
if et not in IMPORT_EVENTS:
|
|
continue
|
|
d = rec.get("date")
|
|
if not d:
|
|
continue
|
|
# Normalize
|
|
iso = iso_utc(d)
|
|
# Keep the oldest (min)
|
|
dt_ts = dt.datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()
|
|
if earliest is None or dt_ts < earliest[0]:
|
|
earliest = (dt_ts, iso, et)
|
|
except Exception:
|
|
continue
|
|
if earliest:
|
|
return earliest[1]
|
|
return None
|
|
|
|
# -----------------------------
|
|
# TMDB client
|
|
# -----------------------------
|
|
class TMDB:
|
|
def __init__(self, api_key: Optional[str], timeout: int = 45, retries: int = 3, primary_country: Optional[str] = None):
|
|
self.api_key = api_key or ""
|
|
self.timeout, self.retries = timeout, retries
|
|
self.primary_country = primary_country or "US"
|
|
self.enabled = bool(self.api_key)
|
|
if not self.enabled:
|
|
info("TMDB not configured (set --tmdb-apikey or env TMDB_API_KEY)")
|
|
self.base = "https://api.themoviedb.org/3"
|
|
|
|
def _get(self, path, params=None):
|
|
if not self.enabled: return None
|
|
p = dict(params or {}); p["api_key"] = self.api_key
|
|
last = None
|
|
for _ in range(self.retries + 1):
|
|
try:
|
|
r = requests.get(self.base + path, params=p, timeout=self.timeout)
|
|
if r.status_code == 404: return None
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
last = e; time.sleep(0.75)
|
|
warn(f"TMDB GET {path} failed: {last}"); return None
|
|
|
|
def tv_id_from_imdb(self, imdb_id: str) -> Optional[int]:
|
|
j = self._get(f"/find/{imdb_id}", params={"external_source": "imdb_id"})
|
|
if not j:
|
|
warn(f"TMDB: find failed for {imdb_id}"); return None
|
|
tv_results = j.get("tv_results") or []
|
|
if tv_results:
|
|
tv_id = tv_results[0].get("id")
|
|
debug(f"TMDB map IMDb {imdb_id} -> TV {tv_id}")
|
|
return tv_id
|
|
warn(f"TMDB: no tv_results for {imdb_id}"); return None
|
|
|
|
def episode_air_date(self, tv_id: int, season: int, episode: int) -> Optional[str]:
|
|
j = self._get(f"/tv/{tv_id}/season/{season}/episode/{episode}")
|
|
if not j:
|
|
warn(f"TMDB: episode not found tv_id={tv_id} S{season:02d}E{episode:02d}")
|
|
return None
|
|
air_date = j.get("air_date")
|
|
if not air_date:
|
|
warn(f"TMDB: episode no air_date tv_id={tv_id} S{season:02d}E{episode:02d}")
|
|
return air_date
|
|
|
|
# -----------------------------
|
|
# Trakt client (optional)
|
|
# -----------------------------
|
|
class Trakt:
|
|
def __init__(self, client_id: Optional[str], timeout: int = 30, retries: int = 2):
|
|
self.client_id = client_id or ""
|
|
self.enabled = bool(self.client_id)
|
|
if not self.enabled:
|
|
info("Trakt not configured (set TRAKT_CLIENT_ID)")
|
|
self.timeout, self.retries = timeout, retries
|
|
self.base = "https://api.trakt.tv"
|
|
|
|
def _headers(self):
|
|
return {"Content-Type": "application/json", "trakt-api-version": "2", "trakt-api-key": self.client_id}
|
|
|
|
def _get(self, path, params=None):
|
|
if not self.enabled: return None
|
|
last = None
|
|
for _ in range(self.retries + 1):
|
|
try:
|
|
r = requests.get(self.base + path, params=params, headers=self._headers(), timeout=self.timeout)
|
|
if r.status_code == 404: return None
|
|
r.raise_for_status()
|
|
return r.json()
|
|
except Exception as e:
|
|
last = e; time.sleep(0.75)
|
|
warn(f"Trakt GET {path} failed: {last}"); return None
|
|
|
|
def episode_first_aired(self, imdb_id: str, season: int, episode: int) -> Optional[str]:
|
|
j = self._get(f"/shows/{imdb_id}/seasons/{season}/episodes/{episode}?extended=full")
|
|
if not j:
|
|
warn(f"Trakt: episode not found for {imdb_id} S{season:02d}E{episode:02d}")
|
|
return None
|
|
return j.get("first_aired")
|
|
|
|
# -----------------------------
|
|
# OMDb client (optional)
|
|
# -----------------------------
|
|
class OMDb:
|
|
def __init__(self, api_key: Optional[str], timeout: int = 20, retries: int = 2):
|
|
self.api_key = api_key or ""
|
|
self.enabled = bool(self.api_key)
|
|
if not self.enabled:
|
|
info("OMDb not configured (set OMDB_API_KEY)")
|
|
self.timeout, self.retries = timeout, retries
|
|
self.base = "https://www.omdbapi.com/"
|
|
|
|
def _get(self, params):
|
|
if not self.enabled: return None
|
|
p = dict(params or {}); p["apikey"] = self.api_key
|
|
last = None
|
|
for _ in range(self.retries + 1):
|
|
try:
|
|
r = requests.get(self.base, params=p, timeout=self.timeout)
|
|
r.raise_for_status()
|
|
j = r.json()
|
|
if j.get("Response") == "False":
|
|
return None
|
|
return j
|
|
except Exception as e:
|
|
last = e; time.sleep(0.5)
|
|
warn(f"OMDb request failed: {last}"); return None
|
|
|
|
def season_listing(self, imdb_id: str, season: int) -> Optional[List[Dict]]:
|
|
j = self._get({"i": imdb_id, "Season": str(season)})
|
|
if not j:
|
|
warn(f"OMDb: season listing not found for {imdb_id} season {season}")
|
|
return None
|
|
return j.get("Episodes") or []
|
|
|
|
def episode_released(self, imdb_id: str, season: int, episode: int) -> Optional[str]:
|
|
eps = self.season_listing(imdb_id, season)
|
|
if not eps:
|
|
return None
|
|
for e in eps:
|
|
try:
|
|
if int(e.get("Episode", "-1")) == episode:
|
|
rel = e.get("Released")
|
|
try:
|
|
dtp = dt.datetime.strptime(rel, "%d %b %Y")
|
|
return dtp.date().isoformat()
|
|
except Exception:
|
|
return None
|
|
except Exception:
|
|
pass
|
|
warn(f"OMDb: no matching episode {episode} in season {season} for {imdb_id}")
|
|
return None
|
|
|
|
# -----------------------------
|
|
# Source selection
|
|
# -----------------------------
|
|
def choose_dateadded(series_imdb: str,
|
|
season: int,
|
|
episode: int,
|
|
ep_video_file: Path,
|
|
sonarr: Sonarr,
|
|
tmdb: TMDB,
|
|
trakt: Trakt,
|
|
omdb: OMDb,
|
|
folder_for_nfo: Path) -> Tuple[str, str, Optional[str]]:
|
|
"""
|
|
Priority:
|
|
1) Sonarr earliest history import (if SONARR_USE_EARLIEST_IMPORT=1)
|
|
else Sonarr episodeFile.dateAdded
|
|
2) Sonarr airDateUtc / airDate
|
|
3) TMDB air_date
|
|
4) Trakt first_aired
|
|
5) OMDb Released
|
|
6) Existing SxxEyy.nfo
|
|
7) file mtime
|
|
Returns (dateadded_iso, source, aired_date_opt)
|
|
"""
|
|
|
|
use_earliest = os.environ.get("SONARR_USE_EARLIEST_IMPORT", "1") != "0"
|
|
|
|
# 1) Sonarr
|
|
if sonarr.enabled:
|
|
try:
|
|
results = sonarr.series_by_imdb(series_imdb)
|
|
if isinstance(results, list) and results:
|
|
series_obj = None
|
|
for s in results:
|
|
if (s.get("imdbId") or "").lower() == series_imdb.lower():
|
|
series_obj = s; break
|
|
if not series_obj: series_obj = results[0]
|
|
sid = series_obj.get("id")
|
|
if sid:
|
|
ep = sonarr.episode(sid, season, episode)
|
|
if ep:
|
|
# Try earliest import from history first (if enabled)
|
|
if use_earliest and ep.get("id") is not None:
|
|
earliest = sonarr.earliest_import_for_episode(sid, int(ep["id"]))
|
|
if earliest:
|
|
iso = iso_utc(earliest)
|
|
info(f"Using Sonarr earliest history import S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "sonarr:history.earliestImport", ep.get("airDateUtc") or ep.get("airDate") or None
|
|
|
|
# Else: current file's dateAdded
|
|
if ep.get("hasFile") and ep.get("episodeFileId"):
|
|
ef = sonarr.episode_file(int(ep["episodeFileId"]))
|
|
if ef and ef.get("dateAdded"):
|
|
epfile_added_iso = iso_utc(ef["dateAdded"])
|
|
info(f"Using Sonarr episodeFile.dateAdded S{season:02d}E{episode:02d}: {epfile_added_iso}")
|
|
return epfile_added_iso, "sonarr:episodeFile.dateAdded", ep.get("airDateUtc") or ep.get("airDate") or None
|
|
# Fallback: episode.added (rare)
|
|
if ep.get("added"):
|
|
ep_added_iso = iso_utc(ep["added"])
|
|
info(f"Using Sonarr episode.added S{season:02d}E{episode:02d}: {ep_added_iso}")
|
|
return ep_added_iso, "sonarr:episode.added", ep.get("airDateUtc") or ep.get("airDate") or None
|
|
# Next: air date from Sonarr
|
|
air_utc = ep.get("airDateUtc") or ep.get("airDate")
|
|
if air_utc:
|
|
iso = iso_utc(air_utc, "00:00:00")
|
|
info(f"Using Sonarr air date S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "sonarr:airDate", air_utc
|
|
info(f"Sonarr episode found but no usable dates S{season:02d}E{episode:02d}")
|
|
else:
|
|
info(f"Sonarr lookup returned no series for {series_imdb}")
|
|
except Exception as e:
|
|
warn(f"Sonarr error: {e}")
|
|
|
|
# 2) TMDB
|
|
if tmdb.enabled:
|
|
try:
|
|
tv_id = tmdb.tv_id_from_imdb(series_imdb)
|
|
if tv_id:
|
|
air = tmdb.episode_air_date(tv_id, season, episode)
|
|
if air:
|
|
iso = iso_utc(air, "00:00:00")
|
|
info(f"Using TMDB air_date S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "tmdb:air_date", air
|
|
else:
|
|
warn(f"TMDB had no air_date S{season:02d}E{episode:02d}")
|
|
else:
|
|
warn(f"TMDB could not map {series_imdb} to TV id")
|
|
except Exception as e:
|
|
warn(f"TMDB error: {e}")
|
|
else:
|
|
info("TMDB not configured; skipping")
|
|
|
|
# 3) Trakt first_aired
|
|
if trakt.enabled:
|
|
try:
|
|
fa = trakt.episode_first_aired(series_imdb, season, episode)
|
|
if fa:
|
|
iso = iso_utc(fa)
|
|
info(f"Using Trakt first_aired S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "trakt:first_aired", fa
|
|
except Exception as e:
|
|
warn(f"Trakt error: {e}")
|
|
|
|
# 4) OMDb Released
|
|
if omdb.enabled:
|
|
try:
|
|
rel = omdb.episode_released(series_imdb, season, episode)
|
|
if rel:
|
|
iso = iso_utc(rel, "00:00:00")
|
|
info(f"Using OMDb Released S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "omdb:released", rel
|
|
except Exception as e:
|
|
warn(f"OMDb error: {e}")
|
|
|
|
# 5) Existing NFO
|
|
aired0, dateadded0 = read_existing_short_nfo(folder_for_nfo, season, episode)
|
|
if dateadded0:
|
|
iso = iso_utc(dateadded0)
|
|
info(f"Using existing NFO dateadded S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "nfo:dateadded", aired0
|
|
if aired0:
|
|
iso = iso_utc(aired0, "00:00:00")
|
|
info(f"Using existing NFO aired S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "nfo:aired", aired0
|
|
|
|
# 6) File mtime
|
|
try:
|
|
mt = ep_video_file.stat().st_mtime
|
|
iso = iso_utc_from_timestamp(mt)
|
|
except Exception:
|
|
iso = iso_utc_now()
|
|
info(f"Using file mtime S{season:02d}E{episode:02d}: {iso}")
|
|
return iso, "file:mtime", None
|
|
|
|
# -----------------------------
|
|
# Apply to a series
|
|
# -----------------------------
|
|
def apply_to_series(series_path: str, conn, manage_nfo: bool, fix_dir_mtimes: bool,
|
|
sonarr: Sonarr, tmdb: TMDB, trakt: Trakt, omdb: OMDb):
|
|
root = Path(series_path)
|
|
imdb = parse_series_imdb_from_path(series_path)
|
|
if not imdb:
|
|
err(f"Series path does not contain an IMDb id [imdb-tt...] : {series_path}")
|
|
return
|
|
|
|
for folder in sorted([p for p in root.iterdir() if p.is_dir()]):
|
|
# season folder detection (including Specials)
|
|
season_num = None
|
|
nl = folder.name.lower()
|
|
if nl == "specials": season_num = 0
|
|
else:
|
|
m = re.match(r"season\s+(\d+)$", nl)
|
|
if m: season_num = int(m.group(1))
|
|
if season_num is None:
|
|
continue
|
|
|
|
removed = remove_longname_nfos(folder)
|
|
if removed:
|
|
info(f"Removed {removed} long-name NFO(s) in {folder}")
|
|
|
|
videos = find_episode_video_and_se(folder)
|
|
if not videos:
|
|
debug(f"No episode videos in {folder}")
|
|
continue
|
|
|
|
earliest_ts = None
|
|
for (snum, enum), vpath in sorted(videos.items()):
|
|
dateadded_iso, source, aired_opt = choose_dateadded(imdb, snum, enum, vpath, sonarr, tmdb, trakt, omdb, folder)
|
|
# cache
|
|
try:
|
|
db_set_date(conn, imdb, snum, enum, dateadded_iso, source)
|
|
except Exception as e:
|
|
warn(f"DB cache set failed for S{snum:02d}E{enum:02d}: {e}")
|
|
|
|
if manage_nfo:
|
|
nfo = folder / f"S{snum:02d}E{enum:02d}.nfo"
|
|
try:
|
|
write_episode_nfo(nfo, snum, enum, aired_opt or ymd(dateadded_iso), dateadded_iso, source)
|
|
except Exception as e:
|
|
warn(f"Failed to write {nfo}: {e}")
|
|
|
|
try:
|
|
t = dt.datetime.fromisoformat(dateadded_iso.replace("Z", "+00:00")).timestamp()
|
|
earliest_ts = t if earliest_ts is None else min(earliest_ts, t)
|
|
except Exception:
|
|
pass
|
|
|
|
if fix_dir_mtimes and earliest_ts:
|
|
try:
|
|
os.utime(folder, (earliest_ts, earliest_ts))
|
|
info(f"Updated mtime on {folder} -> {iso_utc_from_timestamp(earliest_ts)}")
|
|
except Exception as e:
|
|
warn(f"Failed to utime {folder}: {e}")
|
|
|
|
# -----------------------------
|
|
# CLI
|
|
# -----------------------------
|
|
def cmd_rebuild_db(args):
|
|
load_env_file(args.env)
|
|
info(f"Rebuilding DB at {args.db}")
|
|
conn = db_connect(args.db)
|
|
db_init(conn)
|
|
conn.close()
|
|
|
|
def cmd_apply(args):
|
|
load_env_file(args.env)
|
|
|
|
sonarr = Sonarr(
|
|
url=args.sonarr_url or os.environ.get("SONARR_URL"),
|
|
apikey=args.sonarr_apikey or os.environ.get("SONARR_API_KEY"),
|
|
timeout=int(os.environ.get("SONARR_TIMEOUT", "45")),
|
|
retries=int(os.environ.get("SONARR_RETRIES", "3")),
|
|
)
|
|
tmdb = TMDB(
|
|
api_key=args.tmdb_apikey or os.environ.get("TMDB_API_KEY"),
|
|
timeout=int(os.environ.get("TMDB_TIMEOUT", "45")),
|
|
retries=int(os.environ.get("TMDB_RETRIES", "3")),
|
|
primary_country=os.environ.get("TMDB_PRIMARY_COUNTRY", "US"),
|
|
)
|
|
trakt = Trakt(
|
|
client_id=os.environ.get("TRAKT_CLIENT_ID"),
|
|
timeout=int(os.environ.get("TRAKT_TIMEOUT", "30")),
|
|
retries=int(os.environ.get("TRAKT_RETRIES", "2")),
|
|
)
|
|
omdb = OMDb(
|
|
api_key=os.environ.get("OMDB_API_KEY"),
|
|
timeout=int(os.environ.get("OMDB_TIMEOUT", "20")),
|
|
retries=int(os.environ.get("OMDB_RETRIES", "2")),
|
|
)
|
|
|
|
conn = db_connect(args.db)
|
|
db_init(conn)
|
|
|
|
apply_to_series(args.path, conn, args.manage_nfo, args.fix_dir_mtimes, sonarr, tmdb, trakt, omdb)
|
|
|
|
conn.close()
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser(description="Maintain episode dateadded (Sonarr history earliest import→Sonarr air→TMDB→Trakt→OMDb→NFO→mtime), NFOs, and mtimes.")
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
ap_a = sub.add_parser("rebuild-db", help="Initialize SQLite cache")
|
|
ap_a.add_argument("--env", help="Path to .env file", default=None)
|
|
ap_a.add_argument("--db", help="SQLite DB path", required=True)
|
|
ap_a.set_defaults(func=cmd_rebuild_db)
|
|
|
|
ap_b = sub.add_parser("apply", help="Apply to a single series folder")
|
|
ap_b.add_argument("--env", help="Path to .env file", default=None)
|
|
ap_b.add_argument("--db", help="SQLite DB path", required=True)
|
|
ap_b.add_argument("--path", help="Series folder (must contain [imdb-tt...])", required=True)
|
|
ap_b.add_argument("--manage-nfo", action="store_true", help="Write SxxEyy.nfo files")
|
|
ap_b.add_argument("--fix-dir-mtimes", action="store_true", help="Set season directory mtime to earliest episode date")
|
|
# Optional overrides (usually via .env)
|
|
ap_b.add_argument("--sonarr-url", default=None)
|
|
ap_b.add_argument("--sonarr-apikey", default=None)
|
|
ap_b.add_argument("--tmdb-apikey", default=None)
|
|
ap_b.set_defaults(func=cmd_apply)
|
|
|
|
args = ap.parse_args()
|
|
args.func(args)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
|