#!/usr/bin/env python3 """Daily movie duplicate audit + safe auto-resolve (see media_stack/dedupe.py). Radarr's upgrade-delete misses a file the pipeline already renamed, leaving it beside the keeper. SAFE: Radarr tracks the chosen keeper, so the untracked leftover is recycled automatically. RISKY: Radarr tracks a NON-keeper (or nothing); re-pointing it unattended could trigger a cutoff-unmet re-grab, so these are only REPORTED. `--force` resolves them (move the tracked file, RescanMovie, verify). Removals go to ``$MEDIA_ROOT/.dupe-recycle/``. Default is a DRY RUN; the cron runs ``--apply --notify``. Exit 1 if RISKY items need manual review and were not resolved. """ from __future__ import annotations import argparse import datetime import json import os import shutil import sys import time from contextlib import ExitStack from pathlib import Path from media_stack import paths from media_stack.clients.arr import ArrClient from media_stack.clients.telegram import send as telegram_send from media_stack.dedupe import choose_keeper, is_video from media_stack.locking import acquire_file_lock, lock_path_for from media_stack.probe import already_processed, probe def log(msg: str) -> None: print(f"{datetime.datetime.now().isoformat(timespec='seconds')} {msg}", flush=True) def _radarr_key() -> str: if "RADARR_API_KEY" in os.environ: return os.environ["RADARR_API_KEY"] return paths.load_env_file(paths.MEDIA_STACK_ROOT / ".env").get("RADARR_API_KEY", "") def _videos_in(folder: Path) -> list[str]: return sorted(f.name for f in folder.iterdir() if f.is_file() and is_video(f.name)) def _video_meta(folder: Path, name: str) -> dict: p = folder / name try: size = p.stat().st_size except OSError: size = 0 processed = already_processed(probe(p) or {}) return {"name": name, "size": size, "processed": processed} def _notify(summary: str) -> None: """Best-effort; the token comes from the plugin .env, never the stack .env.""" env_path = Path(os.environ.get( "TELEGRAM_ENV", str(Path.home() / ".claude" / "channels" / "telegram" / ".env"))) vals = paths.load_env_file(env_path) token = os.environ.get("TELEGRAM_BOT_TOKEN") or vals.get("TELEGRAM_BOT_TOKEN", "") chat = os.environ.get("TELEGRAM_CHAT_ID") or vals.get("TELEGRAM_CHAT_ID", "") if not token or not chat: log("notify skipped: no Telegram credentials") return try: ok = telegram_send(token, chat, summary) log("notify sent" if ok else "notify failed (non-fatal)") except Exception as e: log(f"notify error (non-fatal): {type(e).__name__}") def main(argv: list[str] | None = None) -> int: ap = argparse.ArgumentParser(description="Resolve duplicate movie files (recoverable).") ap.add_argument("--apply", action="store_true", help="perform moves (default: dry run)") ap.add_argument("--force", action="store_true", help="also resolve RISKY cases (re-point Radarr via rescan)") ap.add_argument("--notify", action="store_true", help="send a Telegram summary") ap.add_argument("--movies-dir", default=str(paths.MEDIA_ROOT / "movies")) args = ap.parse_args(argv) movies_dir = Path(args.movies_dir) # same pool as movies_dir (instant rename), outside the library Jellyfin scans recycle = movies_dir.parent / ".dupe-recycle" mode = "APPLY" if args.apply else "DRY-RUN" key = _radarr_key() arr = ArrClient(paths.RADARR_URL, key) if key else None by_folder: dict[str, dict] = {} if arr: movies = arr.movies() if movies is None: log("WARNING: could not list Radarr movies — classification degraded to RISKY") else: by_folder = {os.path.basename(m.get("path", "")): m for m in movies} if not movies_dir.is_dir(): log(f"movies dir not found: {movies_dir}") return 1 resolved: list[str] = [] # SAFE auto-resolved forced: list[str] = [] # RISKY resolved under --force flagged: list[str] = [] # RISKY left for manual review skipped_locked: list[str] = [] # a pipeline holds a lock — retried next pass errors: list[str] = [] reclaimed = 0 manifest: list[dict] = [] for folder in sorted(p for p in movies_dir.iterdir() if p.is_dir()): vids = _videos_in(folder) if len(vids) <= 1: continue metas = [_video_meta(folder, v) for v in vids] keeper, extras = choose_keeper(metas) keeper_name = keeper["name"] extra_names = [e["name"] for e in extras] movie = by_folder.get(folder.name) tracked = None if movie and (movie.get("movieFile") or {}).get("relativePath"): tracked = os.path.basename(movie["movieFile"]["relativePath"]) safe = tracked == keeper_name tag = "SAFE" if safe else "RISKY" log(f"[{tag}] {folder.name}: keep={keeper_name!r} " f"move={extra_names!r} radarr_tracks={tracked!r}") if not safe and not args.force: flagged.append(folder.name) continue if not args.apply: (resolved if safe else forced).append(folder.name) continue # Move under the per-file locks consolidate-subs.py / normalize-audio.py take (a mid-pipeline # move would corrupt state); any extra locked skips the folder. try: with ExitStack() as locks: if not all(locks.enter_context(acquire_file_lock(folder / e["name"])) for e in extras): log(f" {folder.name}: a pipeline holds a lock — skipping this pass") skipped_locked.append(folder.name) continue dest = recycle / folder.name dest.mkdir(parents=True, exist_ok=True) moved = [] for e in extras: shutil.move(str(folder / e["name"]), str(dest / e["name"])) reclaimed += e["size"] moved.append(e["name"]) manifest.append({"folder": folder.name, "keeper": keeper_name, "tracked": tracked, "moved": moved, "safe": safe}) if not safe and arr and movie: # RISKY + --force: Radarr tracked a moved file; rescan, then verify arr.rescan_movie(movie["id"]) time.sleep(4) remaining = _videos_in(folder) mv = arr.movies() or [] cur = next((m for m in mv if m.get("id") == movie["id"]), {}) cur_tracked = os.path.basename((cur.get("movieFile") or {}).get("relativePath", "")) or None if len(remaining) == 1 and cur.get("hasFile") and cur_tracked == remaining[0]: forced.append(folder.name) else: errors.append(f"{folder.name} (rescan left tracks={cur_tracked} " f"remaining={len(remaining)})") else: resolved.append(folder.name) # locks released: these lock files are dead (the "never unlink" rule guards a file # reappearing at the path) for name in moved: try: os.unlink(lock_path_for(folder / name)) except OSError: pass except Exception as e: errors.append(f"{folder.name}: {type(e).__name__}: {e}") if args.apply and manifest: try: recycle.mkdir(parents=True, exist_ok=True) stamp = datetime.datetime.now().isoformat(timespec="seconds").replace(":", "") (recycle / f"manifest-{stamp}.json").write_text( json.dumps({"when": stamp, "moves": manifest}, indent=2)) except OSError as e: log(f"manifest write failed (non-fatal): {e}") gb = reclaimed / (1024 ** 3) summary = (f"movie-dedupe {mode}: resolved={len(resolved)} " f"flagged(manual)={len(flagged)} forced={len(forced)} " f"locked-skipped={len(skipped_locked)} errors={len(errors)} " f"reclaimed={gb:.1f}GB") log(summary) if flagged: log("manual review (Radarr tracks a non-keeper; re-run with --force " f"after checking): {flagged}") if errors: log(f"errors: {errors}") if args.notify and (resolved or forced or flagged or errors): body = summary if flagged: body += "\n\nmanual review needed: " + ", ".join(flagged) if errors: body += "\n\nerrors: " + "; ".join(errors) _notify(body) # locked-skipped is retried next pass and does not fail the run return 1 if flagged or errors else 0 if __name__ == "__main__": sys.exit(main())