#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ media_date_cache.py One script to: - Set episode 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 , , and - 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 = ['', ""] lines.append(f" {season}") lines.append(f" {episode}") if aired_date: lines.append(f" {ymd(aired_date)}") lines.append(f" {dateadded_iso}") lines.append(f" ") lines.append("\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()