File size: 5,510 Bytes
bc50ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | """Preview-confirmed retirement of chapter recovery pins, never clip media."""
import json
import os
from pathlib import Path, PurePosixPath
import re
from .checkpoint_manager import (
CheckpointDeleteBlocked, CheckpointGraphManager, _fingerprint,
_strict_run_name, checkpoint_run_lock,
)
class ChapterSnapshotManager:
def __init__(self, output_root):
self.root = Path(output_root).resolve()
def _path(self, address):
if not isinstance(address, str) or not address:
raise ValueError("Select a saved chapter snapshot.")
parts = PurePosixPath(address).parts
if (PurePosixPath(address).is_absolute() or ".." in parts or
"\\" in address or str(PurePosixPath(address)) != address):
raise ValueError("Invalid chapter snapshot address.")
path = self.root
for part in parts:
path /= part
if path.is_symlink():
raise ValueError("Snapshot retirement cannot follow symlinks.")
if not path.resolve().is_relative_to(self.root):
raise ValueError("Chapter snapshot escapes the output directory.")
return path
def _preview(self, run, address):
path = self._path(address)
parts = PurePosixPath(address).parts
if not (len(parts) == 6 and parts[:3] == ("h3_chains", run, "chapters")
and parts[4] == "manifests"
and re.fullmatch(r"[0-9a-f]{32}\.json", parts[5])):
raise ValueError("Select an immutable chapter snapshot inside this run.")
if not path.is_file():
raise FileNotFoundError("The chapter snapshot is no longer available.")
data = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(data, dict):
raise ValueError("Chapter snapshot must be a JSON object.")
chapter = data.get("chapter")
if (data.get("format") != "h3_chain_chapter_manifest_v1"
or data.get("run_name") != run
or not isinstance(chapter, dict)
or not isinstance(chapter.get("number"), int)
or chapter["number"] < 1
or not parts[3].startswith("%02d_" % chapter["number"])
or not isinstance(data.get("segments"), list)
or not data["segments"]
or any(not isinstance(s, dict) or not isinstance(s.get("index"), int)
for s in data["segments"])):
raise ValueError("Invalid chapter snapshot identity.")
identity = {k: v for k, v in data.items()
if k not in ("sealed_at", "chapter_manifest_id", "chapter_manifest_path")}
if (_fingerprint(identity)[:32] != path.stem
or data.get("chapter_manifest_id") != path.stem
or data.get("chapter_manifest_path") != address):
raise ValueError("Chapter snapshot failed its identity check.")
destination = path.parent.parent / "retired_manifests" / path.name
retired_address = destination.relative_to(self.root).as_posix()
self._path(retired_address)
if destination.exists():
raise ValueError("A retired copy already exists; it will not be overwritten.")
active, _ = CheckpointGraphManager(str(self.root)).active_selection(run)
scenes = [{"scene": s["index"], "revision": s.get("revision", ""),
"active": active.get(s["index"]) == s.get("revision")}
for s in data["segments"]]
stat = path.stat()
snapshot = _fingerprint({
"document": data, "active": active,
"file": [stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns],
"path": address, "retired_path": retired_address,
})
return {"ok": True, "allowed": True, "run_name": run,
"path": address, "retired_path": retired_address,
"chapter_number": chapter["number"],
"chapter_title": str(chapter.get("title") or "Chapter %d" % chapter["number"]),
"chapter_manifest_id": path.stem, "scenes": scenes,
"snapshot": snapshot,
"message": "Retiring releases only this snapshot's recovery pins. "
"Its metadata is archived; no clips, active pointers, "
"references, processing takes or exports are deleted."}
def retirement_preview(self, run_name, address):
run = _strict_run_name(run_name)
with checkpoint_run_lock(str(self.root), run):
return self._preview(run, address)
def retire(self, run_name, address, expected_snapshot=""):
run = _strict_run_name(run_name)
with checkpoint_run_lock(str(self.root), run):
preview = self._preview(run, address)
if not expected_snapshot or expected_snapshot != preview["snapshot"]:
raise CheckpointDeleteBlocked(
"Chapter snapshot or active selection changed; preview retirement again.", preview)
path = self._path(preview["path"])
destination = self._path(preview["retired_path"])
destination.parent.mkdir(exist_ok=True)
os.rename(path, destination)
return {**preview, "message": "Chapter snapshot %s retired; recovery metadata "
"kept at %s. No clips were deleted." %
(preview["chapter_manifest_id"][:8], preview["retired_path"])}
|