#!/usr/bin/env python3 """Plan/copy known full-path mpv resume mappings; never edit the backup. Original Watch at Ease guide helper, 2026-10-28, MIT license. Runtime verification: mpv 0.41.0 on Linux, ordinary absolute UTF-8 paths. Windows/macOS have not been execution-tested; this release refuses them. This does not invert hashes, infer renamed files, restore playlist redirects, or turn basename-keyed state into full-path-keyed state. """ import argparse import hashlib import json import math import os from pathlib import Path, PurePosixPath import re import stat import sys HASH = re.compile(r"^[0-9A-Fa-f]{32}$") START = re.compile(r"^start=([^\r\n]+)$", re.M) def key(path): return hashlib.md5(path.encode("utf-8")).hexdigest().upper() def clean_absolute(path): if not isinstance(path, str) or not path.startswith("/"): raise ValueError("Expected an absolute Linux media path") if "\x00" in path or "\n" in path or "\r" in path: raise ValueError("Control characters are outside this helper's scope") if str(PurePosixPath(path)) != path or ".." in PurePosixPath(path).parts or path.startswith("//"): raise ValueError("Use a clean absolute path, without ./, ../ or repeated slashes: " + path) return path def read_record(path): if not stat.S_ISREG(path.lstat().st_mode): raise ValueError("State entry is not a regular file: " + str(path)) if path.stat().st_size > 1024 * 1024: raise ValueError("State entry exceeds the 1 MiB scope limit: " + str(path)) raw = path.read_bytes() text = raw.decode("utf-8") values = START.findall(text) if len(values) != 1: return None try: pos = float(values[0]) except ValueError: return None if not math.isfinite(pos) or pos < 0: return None return {"raw": raw, "position": pos, "sha256": hashlib.sha256(raw).hexdigest(), "mtimeNs": path.stat().st_mtime_ns, "name": path.name} def mappings(args): if args.mapping: if args.old_root or args.new_root: raise ValueError("Choose --mapping OR the two root arguments") rows = json.loads(Path(args.mapping).read_text(encoding="utf-8")) if not isinstance(rows, list): raise ValueError("Mapping JSON must be an array of oldPath/newPath objects") return [(clean_absolute(r["oldPath"]), clean_absolute(r["newPath"])) for r in rows] if not args.old_root or not args.new_root: raise ValueError("Provide --old-root and --new-root, or --mapping") old_root = clean_absolute(args.old_root.rstrip("/") or "/") new_root = Path(clean_absolute(args.new_root.rstrip("/") or "/")) if not new_root.is_dir(): raise ValueError("The current media root does not exist") # rglob does not descend through directory symlinks on the tested Python. files = sorted(p for p in new_root.rglob("*") if p.is_file()) return [(str(PurePosixPath(old_root) / p.relative_to(new_root).as_posix()), str(p)) for p in files] def plan(args): backup = Path(args.backup).resolve(strict=True) if not backup.is_dir(): raise ValueError("Backup must be a directory") output = Path(args.output).absolute() if output.resolve().is_relative_to(backup): raise ValueError("Output must be outside the original backup") records = {} for p in backup.iterdir(): if HASH.fullmatch(p.name): k = p.name.upper() if k in records: raise ValueError("Multiple backup filenames represent the same hash") records[k] = p rows, ready, used, destinations = [], [], set(), set() for old, new in mappings(args): old = clean_absolute(old) new = clean_absolute(new) target = Path(new) if not target.is_file(): rows.append({"oldPath": old, "newPath": new, "status": "current-media-missing"}) continue old_key, new_key = key(old), key(new) row = {"oldPath": old, "newPath": new, "oldKey": old_key, "newKey": new_key} if old_key in used or new_key in destinations: raise ValueError("Repeated source or conflicting destination mapping: " + new) used.add(old_key); destinations.add(new_key) record = read_record(records[old_key]) if old_key in records else None if record is None: row["status"] = "no-matching-position-record" else: row.update(status="ready", position=record["position"], sourceSha256=record["sha256"], mtimeGateWouldMatch=record["mtimeNs"] // 1_000_000_000 == target.stat().st_mtime_ns // 1_000_000_000) ready.append((new_key, record)) rows.append(row) return output, ready, {"status": "PLANNED", "backup": str(backup), "output": str(output), "readyCount": len(ready), "rows": rows, "unmatchedBackupHashNames": sorted(k for k in records if k not in used), "limits": "Only explicit clean Linux full-path keys; unmatched names may include playlist/directory redirect markers. A match does not prove unchanged media content. All bytes/options and original state mtime are copied. No live player is launched."} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--backup", required=True) parser.add_argument("--output", required=True) parser.add_argument("--old-root") parser.add_argument("--new-root") parser.add_argument("--mapping", help="UTF-8 JSON array with exact oldPath and newPath values") parser.add_argument("--write", action="store_true", help="Copy ready records to a new, nonexistent output directory") args = parser.parse_args() if not sys.platform.startswith("linux"): raise ValueError("This verified helper release supports Linux only") output, ready, report = plan(args) if args.write: if not ready: raise ValueError("No matching position record; nothing was written") output.mkdir(mode=0o700) # Fails if any output directory already exists. for destination, record in ready: target = output / destination with target.open("xb") as f: f.write(record["raw"]) target.chmod(0o600) os.utime(target, ns=(record["mtimeNs"], record["mtimeNs"])) report["status"] = "APPLIED_TO_NEW_DIRECTORY" (output / "recovery-report.json").write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps(report, ensure_ascii=False, indent=2)) if __name__ == "__main__": try: main() except (OSError, ValueError, KeyError, TypeError, UnicodeError) as e: print(json.dumps({"status": "ERROR", "message": str(e), "backupModified": False, "note": "An I/O failure during --write can leave an incomplete new output directory; use only an APPLIED report."}), file=sys.stderr) sys.exit(2)