#!/usr/bin/env python3 """Local fallback runner for Plex Prepper rename plans. The GUI uses only the Python standard library. It validates the complete plan before moving files, builds Plex show/season folders, and safely extracts an optional metadata ZIP created by Plex Prepper. """ from __future__ import annotations import argparse import errno import json import os import queue import re import shutil import stat import sys import threading import uuid import zipfile from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import Callable, Iterable, Optional try: import tkinter as tk from tkinter import filedialog, messagebox, ttk except ImportError: # pragma: no cover - depends on the local Python build tk = None filedialog = messagebox = ttk = None APP_NAME = "Plex Prepper Runner" MAX_PLAN_BYTES = 16 * 1024 * 1024 MAX_FILES = 500 MAX_SCANNED_FILES = 100_000 MAX_ZIP_FILES = 1_000 MAX_ZIP_FILE_BYTES = 50 * 1024 * 1024 MAX_ZIP_TOTAL_BYTES = 500 * 1024 * 1024 METADATA_EXTENSIONS = {".nfo", ".jpg", ".jpeg", ".png", ".webp"} WINDOWS_RESERVED = { "CON", "PRN", "AUX", "NUL", *(f"COM{number}" for number in range(1, 10)), *(f"LPT{number}" for number in range(1, 10)), } INVALID_FILENAME = re.compile(r'[<>:"/\\|?*\x00-\x1f]') YEAR_SUFFIX = re.compile(r"\s*\(\d{4}\)\s*$") SEASON_FOLDER = re.compile(r"^season\s*0*(\d+)$", re.IGNORECASE) class PlanError(Exception): """A plan cannot be applied safely.""" class OperationCanceled(Exception): """The user canceled between operations.""" @dataclass(frozen=True) class PlanItem: row_id: str original_name: str proposed_name: str source_path: str source_size: Optional[int] season: int show_folder: str @property def target_relative(self) -> str: return str( PurePosixPath( self.show_folder, f"Season {self.season:02d}", self.proposed_name, ) ) @dataclass(frozen=True) class RenamePlan: show_name: str show_year: Optional[int] show_folder: str items: tuple[PlanItem, ...] @dataclass(frozen=True) class MoveOperation: item: PlanItem source: Path destination: Path already_ready: bool = False @dataclass(frozen=True) class MetadataOperation: member_name: str destination: Path size: int skip_existing: bool = False ProgressCallback = Callable[[int, int, str, Optional[str], str], None] def safe_path_segment(value: str) -> str: cleaned = INVALID_FILENAME.sub("-", value) cleaned = re.sub(r"-{2,}", "-", cleaned) cleaned = re.sub(r"\s+", " ", cleaned).strip(" .-") if not cleaned: raise PlanError("The show name does not produce a usable folder name.") if len(cleaned) > 240: raise PlanError("The show folder name is longer than 240 characters.") return cleaned def validate_filename(value: str) -> None: if not value or len(value) > 255 or value in {".", ".."}: raise PlanError(f"Invalid filename: {value!r}") if value.endswith((".", " ")) or INVALID_FILENAME.search(value): raise PlanError(f"The filename contains reserved characters: {value}") if value.split(".", 1)[0].upper() in WINDOWS_RESERVED: raise PlanError(f"The filename is reserved by Windows: {value}") def load_plan(path: Path) -> RenamePlan: if not path.is_file(): raise PlanError("Choose a Plex Prepper JSON plan file.") if path.stat().st_size > MAX_PLAN_BYTES: raise PlanError("The plan is larger than the 16 MB safety limit.") try: data = json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError) as error: raise PlanError(f"The plan could not be read: {error}") from error if not isinstance(data, dict): raise PlanError("The plan must contain a JSON object.") show = data.get("show") if not isinstance(show, dict) or not isinstance(show.get("name"), str): raise PlanError("The plan is missing its selected show.") show_name = show["name"].strip() if not show_name: raise PlanError("The selected show name is empty.") show_year = _optional_int(show.get("year")) target = data.get("target") if isinstance(data.get("target"), dict) else {} explicit_folder = target.get("showFolder") show_folder = safe_path_segment( explicit_folder if isinstance(explicit_folder, str) and explicit_folder.strip() else f"{show_name}{f' ({show_year})' if show_year else ''}" ) raw_files = data.get("files") if not isinstance(raw_files, list): raise PlanError("The plan does not contain a file list.") if len(raw_files) > MAX_FILES: raise PlanError(f"A plan can contain at most {MAX_FILES} files.") items = [] for index, raw in enumerate(raw_files): if not isinstance(raw, dict): continue if raw.get("selected") is False or raw.get("status") not in (None, "ready"): continue original = raw.get("originalName") proposed = raw.get("proposedName") if not isinstance(original, str) or not isinstance(proposed, str): continue validate_filename(original) validate_filename(proposed) if Path(original).suffix.casefold() != Path(proposed).suffix.casefold(): raise PlanError(f"The plan tries to change the video type for {original}.") season = _season_from_item(raw) if season is None or season < 0 or season > 999: raise PlanError(f"The season is missing or invalid for {original}.") relative_path = raw.get("relativePath") source_path = relative_path if isinstance(relative_path, str) and relative_path else original source_size = _optional_int(raw.get("sourceSize")) items.append(PlanItem( row_id=str(raw.get("id") or index), original_name=original, proposed_name=proposed, source_path=source_path, source_size=source_size if source_size is None or source_size >= 0 else None, season=season, show_folder=show_folder, )) if not items: raise PlanError("The plan has no selected, ready episodes.") return RenamePlan(show_name, show_year, show_folder, tuple(items)) def _optional_int(value: object) -> Optional[int]: if isinstance(value, bool): return None if isinstance(value, int): return value if isinstance(value, float) and value.is_integer(): return int(value) return None def _season_from_item(raw: dict) -> Optional[int]: direct = _optional_int(raw.get("targetSeason")) if direct is not None: return direct episode = raw.get("episode") if isinstance(episode, dict): season = _optional_int(episode.get("season")) if season is not None: return season parsed = raw.get("parsed") if isinstance(parsed, dict): return _optional_int(parsed.get("season")) return None def choose_show_root(media_root: Path, plan: RenamePlan) -> Path: root_identity = _folder_identity(media_root.name) if root_identity in { _folder_identity(plan.show_name), _folder_identity(plan.show_folder), }: return media_root return media_root / plan.show_folder def _folder_identity(value: str) -> str: without_year = YEAR_SUFFIX.sub("", value) return "".join(character for character in without_year.casefold() if character.isalnum()) def build_move_operations(plan: RenamePlan, media_root: Path) -> tuple[Path, list[MoveOperation]]: media_root = media_root.expanduser() if not media_root.is_dir() or media_root.is_symlink(): raise PlanError("The media location must be a real local folder.") canonical_root = media_root.resolve(strict=True) show_root = choose_show_root(canonical_root, plan) _ensure_inside(canonical_root, show_root, strict=False) operations = [] destinations = set() sources = set() for item in plan.items: source = resolve_source(canonical_root, item) destination = show_root / f"Season {item.season:02d}" / item.proposed_name _ensure_inside(canonical_root, destination, strict=False) destination_key = os.path.normcase(str(destination)).casefold() source_key = os.path.normcase(str(source)).casefold() if destination_key in destinations: raise PlanError(f"More than one episode targets {destination}.") if source_key in sources: raise PlanError(f"The source file is used more than once: {source}") destinations.add(destination_key) sources.add(source_key) existing_destination = _casefold_existing_path(destination) already_ready = False if existing_destination is not None: if not _same_existing_file(source, existing_destination): raise PlanError(f"The destination already exists: {existing_destination}") already_ready = source == destination operations.append(MoveOperation(item, source, destination, already_ready)) return show_root, operations def resolve_source(media_root: Path, item: PlanItem) -> Path: normalized = item.source_path.replace("\\", "/") relative = PurePosixPath(normalized) if relative.is_absolute() or not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): raise PlanError(f"Unsafe source path in plan: {item.source_path}") candidates = [media_root.joinpath(*relative.parts)] if relative.parts[0].casefold() == media_root.name.casefold() and len(relative.parts) > 1: candidates.append(media_root.joinpath(*relative.parts[1:])) for candidate in candidates: if _matches_source(candidate, item): return _validated_source(media_root, candidate) matches = [] scanned = 0 for folder, directories, filenames in os.walk(media_root, followlinks=False): directories[:] = [name for name in directories if not (Path(folder) / name).is_symlink()] for filename in filenames: scanned += 1 if scanned > MAX_SCANNED_FILES: raise PlanError("The media folder contains too many files to search safely.") if filename.casefold() != item.original_name.casefold(): continue candidate = Path(folder) / filename if _matches_source(candidate, item): matches.append(_validated_source(media_root, candidate)) if len(matches) > 1: break if len(matches) > 1: break if not matches: raise PlanError(f"Source file not found: {item.original_name}") if len(matches) > 1: raise PlanError( f"More than one file matches {item.original_name}. Choose a narrower media folder." ) return matches[0] def _matches_source(path: Path, item: PlanItem) -> bool: try: if not path.is_file() or path.is_symlink() or path.name.casefold() != item.original_name.casefold(): return False return item.source_size is None or path.stat().st_size == item.source_size except OSError: return False def _validated_source(media_root: Path, path: Path) -> Path: resolved = path.resolve(strict=True) _ensure_inside(media_root, resolved, strict=True) if resolved.is_symlink() or not resolved.is_file(): raise PlanError(f"The source is not a regular local file: {path}") return resolved def _ensure_inside(root: Path, path: Path, strict: bool) -> None: root_resolved = root.resolve(strict=False) path_resolved = path.resolve(strict=strict) try: common = Path(os.path.commonpath((root_resolved, path_resolved))) except ValueError as error: raise PlanError(f"The path is outside the selected media folder: {path}") from error if common != root_resolved: raise PlanError(f"The path is outside the selected media folder: {path}") def _same_existing_file(left: Path, right: Path) -> bool: if not right.exists(): return False def _casefold_existing_path(path: Path) -> Optional[Path]: if path.exists(): return path if not path.parent.is_dir(): return None try: for child in path.parent.iterdir(): if child.name.casefold() == path.name.casefold(): return child except OSError as error: raise PlanError(f"Could not inspect the destination folder: {error}") from error return None try: return os.path.samefile(left, right) except OSError: return False def inspect_metadata_zip( zip_path: Optional[Path], show_root: Path, overwrite: bool, ) -> list[MetadataOperation]: if zip_path is None: return [] if not zip_path.is_file(): raise PlanError("The metadata ZIP does not exist.") try: archive = zipfile.ZipFile(zip_path) except (OSError, zipfile.BadZipFile) as error: raise PlanError(f"The metadata ZIP could not be opened: {error}") from error with archive: files = [info for info in archive.infolist() if not info.is_dir()] if len(files) > MAX_ZIP_FILES: raise PlanError(f"The metadata ZIP contains more than {MAX_ZIP_FILES} files.") total_size = sum(info.file_size for info in files) if total_size > MAX_ZIP_TOTAL_BYTES: raise PlanError("The metadata ZIP expands beyond the 500 MB safety limit.") parsed = [] for info in files: mode = (info.external_attr >> 16) & 0o170000 if mode == stat.S_IFLNK: raise PlanError(f"Metadata ZIP symlinks are not allowed: {info.filename}") if info.file_size > MAX_ZIP_FILE_BYTES: raise PlanError(f"A metadata file exceeds the 50 MB limit: {info.filename}") parts = _safe_zip_parts(info.filename) parsed.append((info, parts)) strip_root = bool(parsed) and all( len(parts) > 1 and parts[0].casefold() == parsed[0][1][0].casefold() for _, parts in parsed ) operations = [] destinations = set() for info, original_parts in parsed: parts = original_parts[1:] if strip_root else original_parts if not parts: continue parts = _normalize_metadata_parts(parts) if Path(parts[-1]).suffix.casefold() not in METADATA_EXTENSIONS: raise PlanError(f"The ZIP contains a non-metadata file: {info.filename}") for part in parts: validate_filename(part) destination = show_root.joinpath(*parts) _ensure_inside(show_root, destination, strict=False) key = os.path.normcase(str(destination)).casefold() if key in destinations: raise PlanError(f"The ZIP targets the same metadata path twice: {destination}") destinations.add(key) operations.append(MetadataOperation( info.filename, destination, info.file_size, skip_existing=destination.exists() and not overwrite, )) return operations def _safe_zip_parts(value: str) -> tuple[str, ...]: normalized = value.replace("\\", "/") path = PurePosixPath(normalized) if path.is_absolute() or not path.parts or any(part in {"", ".", ".."} for part in path.parts): raise PlanError(f"The metadata ZIP contains an unsafe path: {value}") return path.parts def _normalize_metadata_parts(parts: tuple[str, ...]) -> tuple[str, ...]: normalized = list(parts) if normalized: season_match = SEASON_FOLDER.match(normalized[0]) if season_match: normalized[0] = f"Season {int(season_match.group(1)):02d}" return tuple(normalized) def execute_moves( operations: Iterable[MoveOperation], callback: ProgressCallback, total: int, offset: int = 0, canceled: Optional[threading.Event] = None, ) -> tuple[int, int]: completed: list[tuple[Path, Path]] = [] done = offset moved = 0 try: for operation in operations: if canceled and canceled.is_set(): raise OperationCanceled("The operation was canceled.") if operation.already_ready: done += 1 callback(done, total, f"Already organized: {operation.destination.name}", operation.item.row_id, "Ready") continue operation.destination.parent.mkdir(parents=True, exist_ok=True) if operation.destination.exists(): raise PlanError(f"The destination appeared after validation: {operation.destination}") callback(done, total, f"Renaming {operation.source.name}", operation.item.row_id, "Renaming") _rename_without_copy(operation.source, operation.destination) completed.append((operation.source, operation.destination)) moved += 1 done += 1 callback(done, total, f"Renamed {operation.destination.name}", operation.item.row_id, "Done") except Exception as error: rollback_errors = [] for source, destination in reversed(completed): try: if destination.exists() and not source.exists(): source.parent.mkdir(parents=True, exist_ok=True) os.rename(destination, source) except OSError as rollback_error: rollback_errors.append(f"{destination}: {rollback_error}") if rollback_errors: raise PlanError( f"{error} Rollback also failed: {'; '.join(rollback_errors)}" ) from error raise return moved, done def _rename_without_copy(source: Path, destination: Path) -> None: case_only = ( source.parent == destination.parent and source.name != destination.name and source.name.casefold() == destination.name.casefold() ) temporary = None try: if case_only: temporary = source.parent / f".plex-prepper-{uuid.uuid4().hex}.tmp" os.rename(source, temporary) os.rename(temporary, destination) else: os.rename(source, destination) except OSError as error: if temporary and temporary.exists() and not source.exists(): os.rename(temporary, source) if error.errno == errno.EXDEV: raise PlanError( "The source and destination are on different filesystems. " "Choose the folder that contains the source files so videos are moved without copying." ) from error raise PlanError(f"Could not rename {source.name}: {error}") from error def extract_metadata( zip_path: Path, operations: Iterable[MetadataOperation], overwrite: bool, callback: ProgressCallback, total: int, offset: int, canceled: Optional[threading.Event] = None, ) -> tuple[int, int, int]: written = skipped = 0 done = offset with zipfile.ZipFile(zip_path) as archive: for operation in operations: if canceled and canceled.is_set(): raise OperationCanceled("The operation was canceled.") if operation.skip_existing: skipped += 1 done += 1 callback(done, total, f"Kept existing {operation.destination.name}", None, "Metadata kept") continue operation.destination.parent.mkdir(parents=True, exist_ok=True) if operation.destination.exists() and not overwrite: skipped += 1 done += 1 callback(done, total, f"Kept existing {operation.destination.name}", None, "Metadata kept") continue temporary = operation.destination.parent / f".plex-prepper-{uuid.uuid4().hex}.tmp" callback(done, total, f"Extracting {operation.destination.name}", None, "Writing metadata") try: with archive.open(operation.member_name) as source, temporary.open("xb") as destination: shutil.copyfileobj(source, destination, length=1024 * 1024) destination.flush() os.fsync(destination.fileno()) if operation.destination.exists() and not overwrite: temporary.unlink(missing_ok=True) skipped += 1 else: os.replace(temporary, operation.destination) written += 1 except Exception: temporary.unlink(missing_ok=True) raise done += 1 callback(done, total, f"Wrote {operation.destination.name}", None, "Metadata done") return written, skipped, done def validate_run( plan: RenamePlan, media_root: Path, metadata_zip: Optional[Path], overwrite_metadata: bool, ) -> tuple[Path, list[MoveOperation], list[MetadataOperation]]: show_root, moves = build_move_operations(plan, media_root) metadata = inspect_metadata_zip(metadata_zip, show_root, overwrite_metadata) return show_root, moves, metadata if tk is not None: class RunnerApp(tk.Tk): def __init__(self) -> None: super().__init__() self.title(APP_NAME) self.geometry("980x700") self.minsize(760, 580) self.configure(bg="#edf2ef") self.plan: Optional[RenamePlan] = None self.plan_path: Optional[Path] = None self.events: queue.Queue = queue.Queue() self.cancel_event = threading.Event() self.busy = False self.plan_var = tk.StringVar(value="No plan selected") self.media_var = tk.StringVar(value="Choose the folder containing the source videos") self.zip_var = tk.StringVar(value="Optional") self.show_var = tk.StringVar(value="Load a plan to begin") self.summary_var = tk.StringVar(value="0 episodes") self.status_var = tk.StringVar(value="Waiting for a Plex Prepper plan") self.current_var = tk.StringVar(value="No files have been changed") self.overwrite_var = tk.BooleanVar(value=False) self._configure_style() self._build_ui() self.after(100, self._poll_events) def _configure_style(self) -> None: style = ttk.Style(self) style.theme_use("clam") style.configure("TFrame", background="#ffffff") style.configure("Page.TFrame", background="#edf2ef") style.configure("Header.TFrame", background="#21483b") style.configure("Header.TLabel", background="#21483b", foreground="#ffffff", font=("Helvetica", 23, "bold")) style.configure("HeaderSub.TLabel", background="#21483b", foreground="#c9ddd5", font=("Helvetica", 11)) style.configure("Title.TLabel", background="#ffffff", foreground="#1d2b26", font=("Helvetica", 14, "bold")) style.configure("Body.TLabel", background="#ffffff", foreground="#53625c", font=("Helvetica", 10)) style.configure("Path.TLabel", background="#ffffff", foreground="#2c3934", font=("Menlo", 9)) style.configure("Accent.TButton", background="#2f725b", foreground="#ffffff", padding=(15, 9), font=("Helvetica", 10, "bold")) style.map("Accent.TButton", background=[("active", "#245b49"), ("disabled", "#9aafa7")]) style.configure("TButton", padding=(11, 8), font=("Helvetica", 10)) style.configure("Horizontal.TProgressbar", troughcolor="#dce6e1", background="#3c8b6c") style.configure("Treeview", rowheight=30, font=("Helvetica", 9), background="#ffffff", fieldbackground="#ffffff") style.configure("Treeview.Heading", font=("Helvetica", 9, "bold"), foreground="#31423b") def _build_ui(self) -> None: self.columnconfigure(0, weight=1) self.rowconfigure(2, weight=1) header = ttk.Frame(self, style="Header.TFrame", padding=(25, 19)) header.grid(row=0, column=0, sticky="ew") header.columnconfigure(0, weight=1) ttk.Label(header, text="Plex Prepper Runner", style="Header.TLabel").grid(row=0, column=0, sticky="w") ttk.Label( header, text="Local fallback for organizing episodes and installing Plex metadata", style="HeaderSub.TLabel", ).grid(row=1, column=0, sticky="w", pady=(3, 0)) setup = ttk.Frame(self, padding=18) setup.grid(row=1, column=0, sticky="ew", padx=18, pady=(18, 10)) setup.columnconfigure(1, weight=1) ttk.Label(setup, text="Plan and media", style="Title.TLabel").grid(row=0, column=0, columnspan=3, sticky="w", pady=(0, 12)) self._path_row(setup, 1, "Rename plan", self.plan_var, self._choose_plan, "Open JSON") self._path_row(setup, 2, "Media folder", self.media_var, self._choose_media, "Choose folder") self._path_row(setup, 3, "Metadata ZIP", self.zip_var, self._choose_zip, "Choose ZIP") overwrite = ttk.Checkbutton( setup, text="Replace existing NFO and artwork files", variable=self.overwrite_var, ) overwrite.grid(row=4, column=1, sticky="w", pady=(8, 0)) plan_frame = ttk.Frame(self, padding=18) plan_frame.grid(row=2, column=0, sticky="nsew", padx=18, pady=0) plan_frame.columnconfigure(0, weight=1) plan_frame.rowconfigure(2, weight=1) summary = ttk.Frame(plan_frame) summary.grid(row=0, column=0, sticky="ew", pady=(0, 10)) summary.columnconfigure(0, weight=1) ttk.Label(summary, textvariable=self.show_var, style="Title.TLabel").grid(row=0, column=0, sticky="w") ttk.Label(summary, textvariable=self.summary_var, style="Body.TLabel").grid(row=0, column=1, sticky="e") columns = ("source", "destination", "status") self.tree = ttk.Treeview(plan_frame, columns=columns, show="headings", selectmode="browse") self.tree.heading("source", text="Current file") self.tree.heading("destination", text="Plex destination") self.tree.heading("status", text="Status") self.tree.column("source", width=310, minwidth=150) self.tree.column("destination", width=420, minwidth=180) self.tree.column("status", width=100, minwidth=85, anchor="center") scrollbar = ttk.Scrollbar(plan_frame, orient="vertical", command=self.tree.yview) self.tree.configure(yscrollcommand=scrollbar.set) self.tree.grid(row=2, column=0, sticky="nsew") scrollbar.grid(row=2, column=1, sticky="ns") footer = ttk.Frame(self, padding=18) footer.grid(row=3, column=0, sticky="ew", padx=18, pady=(10, 18)) footer.columnconfigure(0, weight=1) ttk.Label(footer, textvariable=self.status_var, style="Title.TLabel").grid(row=0, column=0, sticky="w") ttk.Label(footer, textvariable=self.current_var, style="Body.TLabel").grid(row=1, column=0, sticky="w", pady=(3, 9)) self.progress = ttk.Progressbar(footer, mode="determinate", maximum=100) self.progress.grid(row=2, column=0, sticky="ew", padx=(0, 16)) actions = ttk.Frame(footer) actions.grid(row=0, column=1, rowspan=3, sticky="e") self.validate_button = ttk.Button(actions, text="Validate plan", command=self._validate) self.validate_button.grid(row=0, column=0, padx=(0, 8)) self.run_button = ttk.Button(actions, text="Prepare library", style="Accent.TButton", command=self._run) self.run_button.grid(row=0, column=1) def _path_row(self, parent, row, label, variable, command, button_text) -> None: ttk.Label(parent, text=label, style="Body.TLabel", width=14).grid(row=row, column=0, sticky="w", pady=4) ttk.Label(parent, textvariable=variable, style="Path.TLabel", anchor="w").grid(row=row, column=1, sticky="ew", padx=(0, 12), pady=4) ttk.Button(parent, text=button_text, command=command).grid(row=row, column=2, sticky="e", pady=4) def _choose_plan(self) -> None: selected = filedialog.askopenfilename( title="Open Plex Prepper plan", filetypes=(("Plex Prepper plan", "*.json"), ("All files", "*.*")), ) if not selected: return try: self.plan_path = Path(selected) self.plan = load_plan(self.plan_path) except PlanError as error: messagebox.showerror(APP_NAME, str(error)) return self.plan_var.set(str(self.plan_path)) year = f" ({self.plan.show_year})" if self.plan.show_year else "" self.show_var.set(f"{self.plan.show_name}{year}") seasons = sorted({item.season for item in self.plan.items}) season_text = ", ".join(f"{season:02d}" for season in seasons) self.summary_var.set(f"{len(self.plan.items)} episodes | Seasons {season_text}") self.status_var.set("Plan loaded") self.current_var.set("Choose the media folder, then validate the complete batch") self.progress["value"] = 0 self._populate_tree() def _choose_media(self) -> None: selected = filedialog.askdirectory(title="Choose the folder containing the source videos") if selected: self.media_var.set(selected) def _choose_zip(self) -> None: selected = filedialog.askopenfilename( title="Open Plex metadata ZIP", filetypes=(("ZIP archive", "*.zip"), ("All files", "*.*")), ) if selected: self.zip_var.set(selected) def _populate_tree(self) -> None: for row in self.tree.get_children(): self.tree.delete(row) if not self.plan: return for index, item in enumerate(self.plan.items): self.tree.insert("", "end", iid=str(index), values=(item.source_path, item.target_relative, "Planned"), tags=(item.row_id,)) def _validate_inputs(self) -> tuple[RenamePlan, Path, Optional[Path]]: if not self.plan: raise PlanError("Open an exported Plex Prepper plan first.") media_value = self.media_var.get() if not media_value or media_value.startswith("Choose "): raise PlanError("Choose the folder that contains the source videos.") zip_value = self.zip_var.get() metadata_zip = None if not zip_value or zip_value == "Optional" else Path(zip_value) return self.plan, Path(media_value), metadata_zip def _validate(self) -> None: self._start_worker(validate_only=True) def _run(self) -> None: try: plan, _, metadata_zip = self._validate_inputs() except PlanError as error: messagebox.showerror(APP_NAME, str(error)) return metadata_note = " and install the metadata package" if metadata_zip else "" if not messagebox.askyesno( "Confirm library preparation", f"Move and rename {len(plan.items)} episode files{metadata_note}?\n\n" "Existing video files will never be replaced. The complete batch is validated first.", ): return self._start_worker(validate_only=False) def _start_worker(self, validate_only: bool) -> None: if self.busy: return try: plan, media_root, metadata_zip = self._validate_inputs() except PlanError as error: messagebox.showerror(APP_NAME, str(error)) return self.busy = True self.cancel_event.clear() self.validate_button.configure(state="disabled") self.run_button.configure(state="disabled") self.status_var.set("Validating every source and destination") self.current_var.set("No files have been changed") self.progress["value"] = 0 overwrite_metadata = self.overwrite_var.get() worker = threading.Thread( target=self._worker, args=(plan, media_root, metadata_zip, overwrite_metadata, validate_only), daemon=True, ) worker.start() def _worker(self, plan, media_root, metadata_zip, overwrite_metadata, validate_only) -> None: moved = 0 try: show_root, moves, metadata = validate_run( plan, media_root, metadata_zip, overwrite_metadata ) total = len(moves) + len(metadata) if validate_only: for operation in moves: status = "Already ready" if operation.already_ready else "Ready" self.events.put(("row", operation.item.row_id, status)) self.events.put(("validated", show_root, len(moves), len(metadata))) return self.events.put(("total", total, show_root)) callback = lambda done, count, message, row_id, status: self.events.put( ("progress", done, count, message, row_id, status) ) moved, done = execute_moves( moves, callback, total, canceled=self.cancel_event ) written = skipped = 0 if metadata_zip: written, skipped, done = extract_metadata( metadata_zip, metadata, overwrite_metadata, callback, total, done, self.cancel_event, ) self.events.put(("done", show_root, moved, written, skipped, done, total)) except Exception as error: message = str(error) if moved: message += ( f"\n\n{moved} episode files were successfully renamed before the metadata " "step failed. Those video changes were kept." ) self.events.put(("error", message)) def _poll_events(self) -> None: try: while True: event = self.events.get_nowait() kind = event[0] if kind == "row": self._set_row_status(event[1], event[2]) elif kind == "validated": _, show_root, moves, metadata = event self._finish_busy() self.progress["value"] = 100 self.status_var.set("Plan is safe to run") self.current_var.set(f"Destination: {show_root}") messagebox.showinfo( APP_NAME, f"Validated {moves} episode files and {metadata} metadata files.\n\nDestination: {show_root}", ) elif kind == "total": _, total, show_root = event self.progress["maximum"] = max(total, 1) self.status_var.set("Preparing Plex library") self.current_var.set(f"Destination: {show_root}") elif kind == "progress": _, done, total, message, row_id, status = event self.progress["maximum"] = max(total, 1) self.progress["value"] = done self.status_var.set(f"{round(done * 100 / max(total, 1))}% complete") self.current_var.set(message) if row_id: self._set_row_status(row_id, status) elif kind == "done": _, show_root, moved, written, skipped, done, total = event self._finish_busy() self.progress["maximum"] = max(total, 1) self.progress["value"] = done self.status_var.set("Library preparation complete") self.current_var.set(f"Destination: {show_root}") messagebox.showinfo( APP_NAME, f"Renamed and moved {moved} episode files.\n" f"Wrote {written} metadata files; kept {skipped} existing files.\n\n" f"Plex folder: {show_root}", ) elif kind == "error": self._finish_busy() self.status_var.set("Nothing further was changed") self.current_var.set(event[1]) messagebox.showerror(APP_NAME, event[1]) except queue.Empty: pass self.after(100, self._poll_events) def _set_row_status(self, row_id: str, status: str) -> None: for row in self.tree.get_children(): if row_id in self.tree.item(row, "tags"): values = list(self.tree.item(row, "values")) values[2] = status self.tree.item(row, values=values) self.tree.see(row) break def _finish_busy(self) -> None: self.busy = False self.validate_button.configure(state="normal") self.run_button.configure(state="normal") def run_cli(args: argparse.Namespace) -> int: plan = load_plan(Path(args.plan)) metadata_zip = Path(args.metadata_zip) if args.metadata_zip else None show_root, moves, metadata = validate_run( plan, Path(args.media_root), metadata_zip, args.overwrite_metadata, ) print(f"Show: {plan.show_name}") print(f"Destination: {show_root}") for operation in moves: marker = "READY" if operation.already_ready else "MOVE" print(f"[{marker}] {operation.source} -> {operation.destination}") print(f"Metadata files: {len(metadata)}") if not args.apply: print("Validation complete. Run again with --apply to make changes.") return 0 total = len(moves) + len(metadata) callback = lambda done, count, message, _row_id, _status: print( f"[{round(done * 100 / max(count, 1)):3d}%] {message}" ) moved, done = execute_moves(moves, callback, total) written = skipped = 0 if metadata_zip: written, skipped, _ = extract_metadata( metadata_zip, metadata, args.overwrite_metadata, callback, total, done, ) print(f"Complete: {moved} videos moved, {written} metadata files written, {skipped} kept.") return 0 def build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser(description="Apply a Plex Prepper rename plan locally.") parser.add_argument("--plan", help="Path to plex-prepper-plan.json") parser.add_argument("--media-root", help="Folder containing the source videos") parser.add_argument("--metadata-zip", help="Optional metadata ZIP from Plex Prepper") parser.add_argument("--overwrite-metadata", action="store_true", help="Replace existing NFO and artwork files") parser.add_argument("--apply", action="store_true", help="Apply the validated plan; omission is a dry run") return parser def main() -> int: parser = build_parser() args = parser.parse_args() if args.plan or args.media_root: if not args.plan or not args.media_root: parser.error("--plan and --media-root must be supplied together") try: return run_cli(args) except (PlanError, OperationCanceled, OSError, zipfile.BadZipFile) as error: print(f"Error: {error}", file=sys.stderr) return 1 if tk is None: print("This Python installation does not include Tkinter. Use the command-line options instead.", file=sys.stderr) return 1 app = RunnerApp() app.mainloop() return 0 if __name__ == "__main__": raise SystemExit(main())