jskala/unmanic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Claude AI for TV shows only

Browse Source
This commit is contained in:
jskala
2025-08-25 16:22:09 -04:00
parent 293eeebbba
commit edcde29f81
394 changed files with 11212 additions and 1095 deletions
Download Patch File Download Diff File Expand all files Collapse all files
media_date_cache.claudev11
Executable
+784
View File
@@ -0,0 +1,784 @@
#!/opt/scripts/unmanic/.venv/bin/python3
# -*- coding: utf-8 -*-
import os, sys, re, json, sqlite3, argparse, time
from pathlib import Path
from datetime import datetime, timezone
import urllib.request, urllib.parse
# =========================
# tiny .env loader + --env anywhere
# =========================
def _iso_now():
return datetime.now(timezone.utc).isoformat(timespec="seconds")
def _log(level, msg):
# Only show DEBUG messages if debug mode is enabled
if level == "DEBUG" and not getattr(_log, 'debug_enabled', False):
return
print(f"[{_iso_now()}] {level}: {msg}")
def _load_env_file(path: str) -> bool:
p = Path(path)
if not p.exists():
_log("WARNING", f".env not found path={path}")
return False
loaded = 0
for raw in p.read_text(encoding="utf-8").splitlines():
line = raw.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
loaded += 1
# normalize variants
if "SONARR_API_KEY" not in os.environ and "SONARR_APIKEY" in os.environ:
os.environ["SONARR_API_KEY"] = os.environ["SONARR_APIKEY"]
if "RADARR_API_KEY" not in os.environ and "RADARR_APIKEY" in os.environ:
os.environ["RADARR_API_KEY"] = os.environ["RADARR_APIKEY"]
_log("INFO", f".env loaded from {path} ({loaded} keys)")
return True
def _preparse_env(argv):
env_path = None
new_argv = []
i = 0
while i < len(argv):
a = argv[i]
if a == "--env" and i + 1 < len(argv):
env_path = argv[i + 1]
i += 2
continue
elif a.startswith("--env="):
env_path = a.split("=", 1)[1]
i += 1
continue
new_argv.append(a)
i += 1
if env_path:
if not _load_env_file(env_path):
_log("WARNING", f".env not loaded path={env_path}")
return new_argv
sys.argv = _preparse_env(sys.argv)
# =========================
# DB schema / connect
# =========================
SCHEMA = """
PRAGMA journal_mode=WAL;
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT
);
CREATE TABLE IF NOT EXISTS series (
imdb_id TEXT PRIMARY KEY,
series_path TEXT NOT NULL,
last_applied TEXT
);
CREATE TABLE IF NOT EXISTS episode_dates (
imdb_id TEXT NOT NULL,
season INTEGER NOT NULL,
episode INTEGER NOT NULL,
aired TEXT,
dateadded TEXT,
source TEXT,
PRIMARY KEY (imdb_id, season, episode)
);
"""
def db_connect(db_path: Path) -> sqlite3.Connection:
db_path.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(str(db_path))
conn.execute("PRAGMA foreign_keys=ON;")
# Check if tables exist and have correct schema
try:
# Test if episode_dates table exists with correct columns
result = conn.execute("PRAGMA table_info(episode_dates);").fetchall()
columns = [row[1] for row in result] # row[1] is column name
required_columns = ['imdb_id', 'season', 'episode', 'aired', 'dateadded', 'source']
if not all(col in columns for col in required_columns):
_log("WARNING", "Database schema outdated, recreating tables...")
conn.execute("DROP TABLE IF EXISTS episode_dates;")
conn.execute("DROP TABLE IF EXISTS series;")
conn.execute("DROP TABLE IF EXISTS meta;")
except sqlite3.OperationalError:
# Tables don't exist, which is fine
pass
# Create/recreate schema
conn.executescript(SCHEMA)
# Verify schema was created correctly
try:
conn.execute("SELECT imdb_id FROM episode_dates LIMIT 0;")
_log("DEBUG", "Database schema verified")
except sqlite3.OperationalError as e:
_log("ERROR", f"Database schema verification failed: {e}")
raise
return conn
# =========================
# HTTP helpers
# =========================
def http_get_json(url, headers=None, timeout=45):
req = urllib.request.Request(url, headers=headers or {})
with urllib.request.urlopen(req, timeout=timeout) as resp:
data = resp.read()
if not data:
return None
try:
return json.loads(data.decode("utf-8"))
except Exception:
return None
def to_utc_iso(s):
# Accepts '2025-08-25T01:27:00Z' or '+00:00' or '2011-12-04'
if not s:
return None
try:
if len(s) == 10 and s[4] == "-" and s[7] == "-":
# date only -> midnight UTC
return datetime.fromisoformat(s).replace(tzinfo=timezone.utc).isoformat(timespec="seconds")
# normalize Z to +00:00
s2 = s.replace("Z", "+00:00")
dt = datetime.fromisoformat(s2)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
else:
dt = dt.astimezone(timezone.utc)
return dt.isoformat(timespec="seconds")
except Exception:
return None
def iso_from_epoch(epoch):
try:
return datetime.fromtimestamp(epoch, tz=timezone.utc).isoformat(timespec="seconds")
except Exception:
return None
def file_mtime_iso(p: Path):
try:
return iso_from_epoch(p.stat().st_mtime)
except Exception:
return None
# =========================
# Source clients
# =========================
class SonarrClient:
def __init__(self):
self.base = os.environ.get("SONARR_URL", "").rstrip("/")
self.key = os.environ.get("SONARR_API_KEY", "")
self.timeout = int(os.environ.get("SONARR_TIMEOUT", "45") or "45")
self.retries = int(os.environ.get("SONARR_RETRIES", "3") or "3")
self.enabled = bool(self.base and self.key)
def _get(self, path, params=None):
if not self.enabled: return None
q = f"{self.base}/api/v3{path}"
if params:
qs = urllib.parse.urlencode(params)
q = f"{q}?{qs}"
headers = {"X-Api-Key": self.key}
for attempt in range(self.retries):
try:
j = http_get_json(q, headers=headers, timeout=self.timeout)
if j is not None:
return j
except Exception as e:
_log("WARNING", f"Sonarr API attempt {attempt+1}/{self.retries} failed: {e}")
time.sleep(0.3)
return None
def series_by_imdb(self, imdb_id):
# Sonarr supports /series/lookup?term=imdbid:tt1234567
j = self._get("/series/lookup", {"term": f"imdbid:{imdb_id}"})
if not j: return None
# pick exact imdb id match
for s in j:
if (s.get("imdbId") or "").lower() == imdb_id.lower():
return s
return None
def episodes_for_series(self, series_id):
return self._get("/episode", {"seriesId": series_id}) or []
def episode_file(self, episode_file_id):
return self._get(f"/episodefile/{episode_file_id}")
def get_episode_import_history(self, episode_id):
"""Get the original import date from history, not just the latest"""
# Get ALL history for this episode by fetching multiple pages
all_records = []
page = 1
page_size = 100
while True:
history = self._get("/history", {
"episodeId": episode_id,
"sortKey": "date",
"sortDir": "asc",
"page": page,
"pageSize": page_size
}) or {}
records = history.get("records", [])
if not records:
break
all_records.extend(records)
# Check if we got all records (if we got less than pageSize, we're done)
if len(records) < page_size:
break
page += 1
if page > 10: # Safety valve - don't go crazy
break
_log("DEBUG", f"Got {len(all_records)} total history records for episode {episode_id}")
# Look for ACTUAL import events (not just renames)
import_events = []
grabbed_events = []
for event in all_records:
event_type = event.get("eventType")
date = event.get("date")
_log("DEBUG", f"History event: {event_type} at {date}")
# Primary import event - this is what we want
if event_type == "downloadFolderImported" and date:
import_events.append({
"date": date,
"sourceTitle": event.get("sourceTitle", ""),
"eventType": event_type
})
# Track grab events as secondary indicators
elif event_type == "grabbed" and date:
grabbed_events.append({
"date": date,
"sourceTitle": event.get("sourceTitle", ""),
"eventType": event_type
})
# Use downloadFolderImported events if we have them
if import_events:
earliest = min(import_events, key=lambda x: x["date"] or "9999-12-31")
_log("INFO", f"Found original import date: {earliest['date']} (downloadFolderImported) for episode {episode_id}")
return earliest["date"]
# If no import events but we have grabs, use the earliest grab + some buffer time
# This handles cases where history is incomplete but we know when it was grabbed
if grabbed_events:
earliest_grab = min(grabbed_events, key=lambda x: x["date"] or "9999-12-31")
_log("WARNING", f"No downloadFolderImported found, using grabbed event: {earliest_grab['date']} for episode {episode_id}")
return earliest_grab["date"]
_log("WARNING", f"No reliable import events found for episode {episode_id}")
return None
class TMDBClient:
def __init__(self):
self.api_key = os.environ.get("TMDB_API_KEY", "")
self.primary_country = os.environ.get("TMDB_PRIMARY_COUNTRY", "US")
self.timeout = int(os.environ.get("TMDB_TIMEOUT", "45") or "45")
self.retries = int(os.environ.get("TMDB_RETRIES", "3") or "3")
self.enabled = bool(self.api_key)
def _get(self, path, params=None):
if not self.enabled: return None
base = "https://api.themoviedb.org/3"
params = params or {}
params["api_key"] = self.api_key
url = f"{base}{path}?{urllib.parse.urlencode(params)}"
for _ in range(self.retries):
j = http_get_json(url, timeout=self.timeout)
if j is not None:
return j
time.sleep(0.3)
return None
def tv_id_from_imdb(self, imdb_id):
j = self._get(f"/find/{imdb_id}", {"external_source": "imdb_id"})
if not j: return None
tv = (j.get("tv_results") or [])
if tv:
return tv[0].get("id")
# sometimes series is treated as multi-type; try "tv_episode" find to resolve parent?
return None
def season_airdates(self, tv_id, season_number):
# returns {episode_number: air_date or None}
j = self._get(f"/tv/{tv_id}/season/{season_number}")
res = {}
if not j: return res
for ep in (j.get("episodes") or []):
num = ep.get("episode_number")
air = ep.get("air_date")
if isinstance(num, int) and air:
res[num] = air
return res
class OMDbClient:
def __init__(self):
self.api_key = os.environ.get("OMDB_API_KEY", "")
self.enabled = bool(self.api_key)
def season_airdates(self, imdb_id, season_number):
# http://www.omdbapi.com/?i=tt2085059&Season=1&apikey=KEY
if not self.enabled: return {}
params = {"i": imdb_id, "Season": str(season_number), "apikey": self.api_key}
url = "http://www.omdbapi.com/?" + urllib.parse.urlencode(params)
j = http_get_json(url, timeout=45)
res = {}
if not j or j.get("Response") != "True":
return res
for ep in j.get("Episodes", []):
try:
num = int(ep.get("Episode"))
except Exception:
continue
rel = ep.get("Released") # '2011-12-04' etc
if rel and rel != "N/A":
res[num] = rel
return res
# =========================
# NFO helpers
# =========================
LONG_NFO_RE = re.compile(r".*-S\d{2}E\d{2}-.*\.nfo$", re.IGNORECASE)
SHORT_NFO_RE = re.compile(r"^S(\d{2})E(\d{2})\.nfo$", re.IGNORECASE)
IMDB_TAG_RE = re.compile(r"\[imdb-(tt\d+)\]", re.IGNORECASE)
def remove_long_nfos(season_dir: Path) -> int:
count = 0
for f in season_dir.glob("*.nfo"):
if LONG_NFO_RE.match(f.name):
try:
f.unlink()
count += 1
except Exception:
pass
return count
def write_episode_nfo(season_dir: Path, season_num: int, ep_num: int, aired: str, dateadded: str, source: str):
# Ensure the season directory exists
try:
season_dir.mkdir(parents=True, exist_ok=True)
except Exception as e:
_log("ERROR", f"Failed creating season directory {season_dir}: {e}")
return
name = f"S{season_num:02d}E{ep_num:02d}.nfo"
dst = season_dir / name
xml = [
'<?xml version="1.0" encoding="UTF-8" standalone="yes"?>',
"<episodedetails>",
f" <season>{season_num}</season>",
f" <episode>{ep_num}</episode>",
f" <aired>{(aired or '').split('T')[0]}</aired>",
]
if dateadded:
xml.append(f" <dateadded>{dateadded}</dateadded>")
xml.append(f" <!-- source: {source} -->")
xml.append("</episodedetails>\n")
content = "\n".join(xml)
try:
dst.write_text(content, encoding="utf-8")
except Exception as e:
_log("ERROR", f"Failed writing {dst}: {e}")
def ensure_season_nfo(season_dir: Path, season_num: int):
# Small, safe season.nfo; don't overwrite if present
p = season_dir / "season.nfo"
if p.exists():
return
xml = f"""<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<season>
<seasonnumber>{season_num}</seasonnumber>
</season>
"""
try:
p.write_text(xml, encoding="utf-8")
except Exception as e:
_log("WARNING", f"Could not write season.nfo in {season_dir}: {e}")
def maybe_tvshow_nfo(series_dir: Path, imdb_id: str):
# Don't clobber; only create minimal file if missing
p = series_dir / "tvshow.nfo"
if p.exists():
return
xml = f"""<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
<tvshow>
<id>{imdb_id}</id>
<uniqueid type="imdb" default="true">{imdb_id}</uniqueid>
</tvshow>
"""
try:
p.write_text(xml, encoding="utf-8")
except Exception as e:
_log("WARNING", f"Could not create tvshow.nfo in {series_dir}: {e}")
def set_dir_mtime(path: Path, iso_timestamp: str):
try:
ts = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")).timestamp()
os.utime(path, (ts, ts), follow_symlinks=False)
except Exception as e:
_log("WARNING", f"Failed to set mtime on {path}: {e}")
def set_media_mtimes(season_dir: Path, iso_timestamp: str):
try:
ts = datetime.fromisoformat(iso_timestamp.replace("Z", "+00:00")).timestamp()
except Exception as e:
_log("WARNING", f"Bad timestamp for media mtimes in {season_dir}: {e}")
return
exts = (".mkv", ".mp4", ".avi", ".mov", ".m4v", ".srt", ".ass", ".vtt")
for f in season_dir.iterdir():
if f.suffix.lower() in exts:
try:
os.utime(f, (ts, ts), follow_symlinks=False)
except Exception:
pass
# =========================
# Core logic
# =========================
def parse_imdb_from_path(p: Path):
m = IMDB_TAG_RE.search(p.name)
return m.group(1).lower() if m else None
def is_series_dir(p: Path) -> bool:
return p.is_dir() and parse_imdb_from_path(p) is not None
def gather_episode_dates(series_dir: Path, imdb_id: str, args, conn: sqlite3.Connection):
sonarr = SonarrClient()
tmdb = TMDBClient()
omdb = OMDbClient()
# results: dict[(season, episode)] = (aired_iso, dateadded_iso, source)
out = {}
# If using cache, first load existing data from DB
if args.use_cache:
_log("INFO", f"Checking cache for existing episode data: {imdb_id}")
cached = conn.execute(
"SELECT season, episode, aired, dateadded, source FROM episode_dates WHERE imdb_id = ?",
(imdb_id,)
).fetchall()
for season, episode, aired, dateadded, source in cached:
out[(season, episode)] = (aired, dateadded, source)
_log("INFO", f"Found {len(cached)} cached episodes for {imdb_id}")
# Find which episodes we have on disk but not in cache
disk_episodes = set()
for season_dir in series_dir.iterdir():
if not (season_dir.is_dir() and season_dir.name.lower().startswith("season ")):
continue
try:
sn = int(season_dir.name.split()[-1])
except Exception:
continue
# Find episode files
media = [p for p in season_dir.iterdir()
if p.is_file() and re.search(r"S(\d{2})E(\d{2})", p.name, re.IGNORECASE)]
for f in media:
m = re.search(r"S(\d{2})E(\d{2})", f.name, re.IGNORECASE)
if m:
sn2, en2 = int(m.group(1)), int(m.group(2))
if sn2 == sn:
disk_episodes.add((sn, en2))
cached_episodes = set(out.keys())
missing_episodes = disk_episodes - cached_episodes
if missing_episodes:
_log("INFO", f"Found {len(missing_episodes)} episodes on disk not in cache: {sorted(missing_episodes)}")
else:
_log("INFO", f"All disk episodes are cached, skipping API queries")
return out
# Only query APIs if not using cache or if we have missing episodes
need_api_query = not args.use_cache or (args.use_cache and missing_episodes)
if need_api_query:
# Prefer Sonarr
if sonarr.enabled:
_log("INFO", f"Querying Sonarr for series: {imdb_id}")
s = sonarr.series_by_imdb(imdb_id)
if s:
series_id = s.get("id")
_log("INFO", f"Found Sonarr series ID: {series_id}")
eps = sonarr.episodes_for_series(series_id)
if isinstance(eps, list):
_log("INFO", f"Found {len(eps)} episodes in Sonarr")
for ep in eps:
season_num = ep.get("seasonNumber")
ep_num = ep.get("episodeNumber")
episode_id = ep.get("id")
if not isinstance(season_num, int) or not isinstance(ep_num, int):
continue
# If using cache, skip episodes we already have unless forced
key = (season_num, ep_num)
if args.use_cache and key in out and not args.force_refresh:
continue
_log("DEBUG", f"Processing S{season_num:02d}E{ep_num:02d} (episode ID: {episode_id})")
air_utc = to_utc_iso(ep.get("airDateUtc"))
date_add = None
src = None
# Try to get original import date from history first
if episode_id:
import_date = sonarr.get_episode_import_history(episode_id)
if import_date:
date_add = to_utc_iso(import_date)
src = "sonarr:history.downloadFolderImported"
_log("INFO", f"S{season_num:02d}E{ep_num:02d}: Using original import date {date_add}")
# Fallback to episode file dateAdded
if not date_add:
ep_file_id = ep.get("episodeFileId")
if ep_file_id:
fobj = sonarr.episode_file(ep_file_id)
if fobj and fobj.get("dateAdded"):
date_add = to_utc_iso(fobj.get("dateAdded"))
src = "sonarr:episodeFile.dateAdded"
_log("WARNING", f"S{season_num:02d}E{ep_num:02d}: No history found, using file dateAdded {date_add}")
# Final fallback to air date
if not date_add and air_utc:
date_add = air_utc
src = "sonarr:episode.airDateUtc"
_log("WARNING", f"S{season_num:02d}E{ep_num:02d}: No dateAdded found, using air date {date_add}")
if air_utc or date_add:
out[key] = (air_utc, date_add, src)
else:
_log("WARNING", f"Series not found in Sonarr for IMDb ID: {imdb_id}")
# Fill gaps with TMDB (only for missing episodes if using cache)
if tmdb.enabled:
tv_id = tmdb.tv_id_from_imdb(imdb_id)
if tv_id:
# find all season folders we have
seasons = sorted({int(p.name.split()[-1]) for p in series_dir.iterdir()
if p.is_dir() and p.name.lower().startswith("season ")
and p.name.split()[-1].isdigit()})
for sn in seasons:
m = tmdb.season_airdates(tv_id, sn)
for ep_num, air in m.items():
key = (sn, ep_num)
if key not in out or not out[key][1]: # no dateadded yet
air_iso = to_utc_iso(air) # date → midnight UTC
out[key] = (air_iso, air_iso, "tmdb:air_date")
# Fill remaining with OMDb (only for missing episodes if using cache)
if omdb.enabled:
# infer seasons as above
seasons = sorted({int(p.name.split()[-1]) for p in series_dir.iterdir()
if p.is_dir() and p.name.lower().startswith("season ")
and p.name.split()[-1].isdigit()})
for sn in seasons:
m = omdb.season_airdates(imdb_id, sn)
for ep_num, air in m.items():
key = (sn, ep_num)
if key not in out or not out[key][1]:
air_iso = to_utc_iso(air)
out[key] = (air_iso, air_iso, "omdb:Released")
# Absolute last resort: file mtime (always check for missing episodes)
for season_dir in series_dir.iterdir():
if not (season_dir.is_dir() and season_dir.name.lower().startswith("season ")):
continue
try:
sn = int(season_dir.name.split()[-1])
except Exception:
continue
# map existing episode files to ep numbers
media = [p for p in season_dir.iterdir()
if p.is_file() and re.search(r"S(\d{2})E(\d{2})", p.name, re.IGNORECASE)]
for f in media:
m = re.search(r"S(\d{2})E(\d{2})", f.name, re.IGNORECASE)
if not m: continue
sn2, en2 = int(m.group(1)), int(m.group(2))
if sn2 != sn: # skip cross-labeling
continue
key = (sn, en2)
if key not in out or not out[key][1]:
mt = file_mtime_iso(f)
out[key] = ( (out.get(key) or (None,))[0] , mt, "file:mtime")
return out
def apply_series(series_dir: Path, args, conn: sqlite3.Connection):
imdb_id = parse_imdb_from_path(series_dir)
if not imdb_id:
_log("ERROR", f"Series path does not contain an IMDb id [imdb-tt...] : {series_dir}")
return
# Store/refresh series row
conn.execute(
"INSERT INTO series(imdb_id, series_path, last_applied) VALUES(?,?,?) "
"ON CONFLICT(imdb_id) DO UPDATE SET series_path=excluded.series_path, last_applied=excluded.last_applied",
(imdb_id, str(series_dir), _iso_now()),
)
conn.commit()
# Clean long-name NFOs and ensure tvshow.nfo/season.nfo if asked
if args.manage_nfo:
for season_dir in series_dir.iterdir():
if season_dir.is_dir() and season_dir.name.lower().startswith("season "):
removed = remove_long_nfos(season_dir)
if removed:
_log("INFO", f"Removed {removed} long-name NFO(s) in {season_dir}")
try:
sn = int(season_dir.name.split()[-1])
ensure_season_nfo(season_dir, sn)
except Exception:
pass
maybe_tvshow_nfo(series_dir, imdb_id)
# Gather dates
dates = gather_episode_dates(series_dir, imdb_id, args, conn)
# Write short NFOs and record DB rows; track per-season latest date for mtime
seasons_latest = {}
for (sn, en), (aired, dateadded, source) in sorted(dates.items()):
# write episode nfo
if args.manage_nfo:
write_episode_nfo(series_dir / f"Season {sn:02d}", sn, en, aired, dateadded, source or "")
# DB upsert
conn.execute(
"INSERT INTO episode_dates(imdb_id, season, episode, aired, dateadded, source) "
"VALUES(?,?,?,?,?,?) "
"ON CONFLICT(imdb_id, season, episode) DO UPDATE SET aired=excluded.aired, dateadded=excluded.dateadded, source=excluded.source",
(imdb_id, sn, en, aired or None, dateadded or None, source or None),
)
# compute latest per season for mtime
if dateadded:
try:
ts = datetime.fromisoformat(dateadded.replace("Z", "+00:00"))
except Exception:
ts = None
if ts:
seasons_latest.setdefault(sn, ts)
if ts > seasons_latest[sn]:
seasons_latest[sn] = ts
conn.commit()
# Fix folder mtimes (season dirs)
if args.fix_dir_mtimes and seasons_latest:
for sn, ts in seasons_latest.items():
season_dir = series_dir / f"Season {sn:02d}"
if season_dir.exists():
iso = ts.astimezone(timezone.utc).isoformat(timespec="seconds")
set_dir_mtime(season_dir, iso)
_log("INFO", f"Updated mtime on {season_dir} -> {iso}")
if args.update_video_mtimes:
set_media_mtimes(season_dir, iso)
def iterate_target_path(path: Path):
# If path is a series dir (has imdb tag), return [path]; if it's a root, return imdb-tag subdirs
if is_series_dir(path):
return [path]
if path.is_dir():
out = []
for p in path.iterdir():
if is_series_dir(p):
out.append(p)
return sorted(out)
return []
# =========================
# CLI
# =========================
def cmd_rebuild_db(args):
_log("INFO", f"Rebuilding DB at {args.db}")
conn = db_connect(Path(args.db))
# wipe content but keep schema
conn.execute("DELETE FROM episode_dates;")
conn.execute("DELETE FROM series;")
conn.commit()
conn.close()
def cmd_apply(args):
# Enable debug logging if requested
if args.debug:
_log.debug_enabled = True
conn = db_connect(Path(args.db))
targets = iterate_target_path(Path(args.path))
if not targets:
_log("ERROR", f"No series found at {args.path} (expects dirs named like 'Show Name [imdb-ttxxxxxxx]')")
return
# Inform about source availability
if not (os.environ.get("SONARR_URL") and (os.environ.get("SONARR_API_KEY") or os.environ.get("SONARR_APIKEY"))):
_log("INFO", "Sonarr not configured (set SONARR_URL and SONARR_API_KEY)")
if not os.environ.get("TMDB_API_KEY"):
_log("INFO", "TMDB not configured (set TMDB_API_KEY)")
if not os.environ.get("OMDB_API_KEY"):
_log("INFO", "OMDb not configured (set OMDB_API_KEY)")
for series_dir in targets:
_log("INFO", f"=== APPLY: {series_dir} ===")
try:
apply_series(series_dir, args, conn)
except KeyboardInterrupt:
raise
except Exception as e:
_log("ERROR", f"Failed applying {series_dir}: {e}")
conn.close()
def build_parser():
p = argparse.ArgumentParser(prog="media_date_cache.py")
p.add_argument("--env", help="Path to .env file (can appear anywhere on the command)", default=None)
sub = p.add_subparsers(dest="cmd", required=True)
p_rebuild = sub.add_parser("rebuild-db", help="Recreate (wipe) episode cache DB")
p_rebuild.add_argument("--db", required=True, help="Path to sqlite DB")
p_rebuild.set_defaults(func=cmd_rebuild_db)
p_apply = sub.add_parser("apply", help="Apply dates → NFO + mtimes for a series dir or a library root")
p_apply.add_argument("--db", required=True, help="Path to sqlite DB")
p_apply.add_argument("--path", required=True, help="Series dir (with [imdb-tt...]) or a root containing them")
p_apply.add_argument("--manage-nfo", action="store_true", help="Create short NFOs; remove long-name NFOs; ensure season.nfo/tvshow.nfo")
p_apply.add_argument("--fix-dir-mtimes", action="store_true", help="Set Season folder mtimes to latest episode 'dateadded'")
p_apply.add_argument("--update-video-mtimes", action="store_true", help="Also set video/subtitle file mtimes to season latest (use with care)")
p_apply.add_argument("--use-cache", action="store_true", help="Use cached episode data when available, only query APIs for missing episodes")
p_apply.add_argument("--force-refresh", action="store_true", help="Force refresh of cached data (use with --use-cache)")
p_apply.add_argument("--debug", action="store_true", help="Enable debug logging")
p_apply.set_defaults(func=cmd_apply)
return p
def main():
parser = build_parser()
args = parser.parse_args()
# (No need to load env here; it was already handled in preparse)
args.func(args)
if __name__ == "__main__":
main()