#!/usr/bin/env python3 from __future__ import annotations import datetime import hashlib import json import os import sys import tempfile from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST = ROOT / "script/project-manifest.json" DELETE_KEYS = ("final_master", "footage") def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def resolve_active(manifest: dict, key: str) -> Path: value = (manifest.get("active_paths") or {}).get(key) if not isinstance(value, str) or not value: raise RuntimeError(f"active_paths.{key} is missing") path = Path(value) if not path.is_absolute(): path = ROOT / path resolved = path.resolve() try: resolved.relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"active path escapes project: {key}") from exc return resolved def atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) data = (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8") fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(name, path) finally: if os.path.exists(name): os.unlink(name) def load_json(path: Path, label: str) -> dict: if not path.is_file(): raise RuntimeError(f"{label} is missing: {path}") value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise RuntimeError(f"{label} is not an object") return value def main() -> int: project = load_json(MANIFEST, "project manifest") paths = {key: resolve_active(project, key) for key in ( "final_upload", "final_master", "footage", "transcode_receipt", "postiz_media_receipt", "postiz_schedule_receipt", "cleanup_receipt", )} canon = project.get("canon_sha256") if not isinstance(canon, str) or len(canon) != 64: raise RuntimeError("project canon hash is invalid") upload = paths["final_upload"] if not upload.is_file() or upload.stat().st_size <= 0: raise RuntimeError("verified upload copy is missing") upload_bytes = upload.stat().st_size upload_hash = sha256(upload) if upload_bytes >= 1_000_000_000: raise RuntimeError("upload copy is not under one billion bytes") transcode = load_json(paths["transcode_receipt"], "transcode receipt") media = load_json(paths["postiz_media_receipt"], "Postiz media receipt") schedule = load_json(paths["postiz_schedule_receipt"], "Postiz schedule receipt") for label, receipt in (("transcode", transcode), ("media", media), ("schedule", schedule)): if receipt.get("status") != "completed" or receipt.get("verified") is not True: raise RuntimeError(f"{label} receipt is not terminal verified") if receipt.get("source_canon_sha256") != canon: raise RuntimeError(f"{label} receipt canon mismatch") if transcode.get("artifact_sha256") != upload_hash: raise RuntimeError("upload copy hash differs from transcode receipt") if media.get("video_sha256") != upload_hash or media.get("video_bytes") != upload_bytes: raise RuntimeError("upload copy differs from Postiz media receipt") media_video_path = Path(str(media.get("video_path", ""))).resolve() if media_video_path != upload: raise RuntimeError("Postiz media receipt points to another local video") remote = media.get("video") or {} if not remote.get("id") or not remote.get("path"): raise RuntimeError("Postiz remote media object is incomplete") expected_media_hash = (schedule.get("media_receipt") or {}).get("sha256") if expected_media_hash != sha256(paths["postiz_media_receipt"]): raise RuntimeError("schedule receipt is not linked to current media receipt") posts = schedule.get("posts") or {} integration_ids = schedule.get("integration_ids") or {} expected_ids = {integration_ids.get("facebook"), integration_ids.get("youtube")} if None in expected_ids or set(posts) != expected_ids or len(posts) != 2: raise RuntimeError("schedule receipt does not contain exactly two target integrations") states = {post.get("state") for post in posts.values()} dates = {post.get("publishDate") for post in posts.values()} if not states.issubset({"QUEUE", "PUBLISHED"}) or len(dates) != 1 or None in dates: raise RuntimeError("Postiz schedule readback is not terminal verified") cleanup_path = paths["cleanup_receipt"] resumed_running = None if cleanup_path.exists(): old = load_json(cleanup_path, "cleanup receipt") if old.get("status") == "completed" and old.get("verified") is True: if old.get("upload_copy", {}).get("sha256") != upload_hash: raise RuntimeError("existing cleanup receipt upload hash mismatch") remaining = [record["path"] for record in old.get("delete_targets", []) if Path(record["path"]).exists()] if remaining: raise RuntimeError(f"cleanup receipt is completed but intermediates remain: {remaining}") print(json.dumps({ "status": "completed", "verified": True, "idempotent": True, "receipt": str(cleanup_path), "upload_copy": str(upload), }, ensure_ascii=False)) return 0 if old.get("status") != "running" or old.get("verified") is not False: raise RuntimeError("non-terminal cleanup receipt has unsupported state") if old.get("source_canon_sha256") != canon: raise RuntimeError("running cleanup receipt canon mismatch") if old.get("upload_copy", {}).get("path") != str(upload) or old.get("upload_copy", {}).get("sha256") != upload_hash: raise RuntimeError("running cleanup receipt upload identity mismatch") resumed_running = old if resumed_running is not None: running = resumed_running delete_records = running.get("delete_targets") or [] if not delete_records: raise RuntimeError("running cleanup receipt has no locked deletion set") for record in delete_records: path = Path(str(record.get("path", ""))).resolve() try: path.relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"running cleanup target escapes project: {path}") from exc if path == upload: raise RuntimeError("running cleanup target aliases upload copy") if path.exists() and sha256(path) != record.get("sha256"): raise RuntimeError(f"running cleanup target hash mismatch: {path}") else: delete_candidates = [(key, paths[key], None) for key in DELETE_KEYS] for record in project.get("cleanup_superseded_intermediates") or []: if not isinstance(record, dict) or record.get("verified") is not True: raise RuntimeError("superseded cleanup record is not verified") value = record.get("path") expected_hash = record.get("sha256") receipt_value = record.get("receipt") receipt_hash = record.get("receipt_sha256") if not all(isinstance(item, str) and item for item in (value, expected_hash, receipt_value, receipt_hash)): raise RuntimeError("superseded cleanup record is incomplete") stale_path = (ROOT / value).resolve() receipt_path = (ROOT / receipt_value).resolve() try: stale_path.relative_to(ROOT.resolve()) receipt_path.relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError("superseded cleanup path escapes project") from exc if sha256(receipt_path) != receipt_hash: raise RuntimeError(f"superseded receipt hash mismatch: {receipt_path}") receipt = load_json(receipt_path, "superseded render receipt") if receipt.get("status") != "completed" or receipt.get("verified") is not True: raise RuntimeError(f"superseded render receipt is not verified: {receipt_path}") if receipt.get("source_canon_sha256") != canon or receipt.get("artifact_sha256") != expected_hash: raise RuntimeError(f"superseded artifact lineage mismatch: {stale_path}") if Path(str(receipt.get("output_path", ""))).resolve() != stale_path: raise RuntimeError(f"superseded receipt points to another artifact: {stale_path}") delete_candidates.append((str(record.get("key") or "superseded_intermediate"), stale_path, expected_hash)) superseded = (project.get("poster_v2_promotion") or {}).get("superseded_paths") or {} if superseded: stale_specs = ( ("superseded_final_master", "final_master", project.get("final_render") or {}), ("superseded_final_upload", "final_upload", project.get("transcode_upload") or {}), ) for label, source_key, record in stale_specs: value = superseded.get(source_key) if not isinstance(value, str) or not value: raise RuntimeError(f"superseded path is missing: {source_key}") stale_path = (ROOT / value).resolve() try: stale_path.relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"superseded path escapes project: {source_key}") from exc if record.get("path") != value or not record.get("verified"): raise RuntimeError(f"superseded artifact is not verified in manifest: {source_key}") delete_candidates.append((label, stale_path, record.get("sha256"))) delete_records = [] seen = set() for key, path, expected_hash in delete_candidates: if path == upload: raise RuntimeError(f"cleanup target aliases upload copy: {key}") if path in seen: continue seen.add(path) if not path.is_file(): raise RuntimeError(f"cleanup target is missing before first cleanup: {path}") observed_hash = sha256(path) if expected_hash is not None and observed_hash != expected_hash: raise RuntimeError(f"superseded artifact hash mismatch: {key}") delete_records.append({ "key": key, "path": str(path), "bytes": path.stat().st_size, "sha256": observed_hash, "exists_before": True, }) started_at = datetime.datetime.now(datetime.timezone.utc).isoformat() running = { "status": "running", "verified": False, "source_canon_sha256": canon, "active_version": project.get("active_version"), "postiz_media_receipt": {"path": str(paths["postiz_media_receipt"]), "sha256": sha256(paths["postiz_media_receipt"])}, "postiz_schedule_receipt": {"path": str(paths["postiz_schedule_receipt"]), "sha256": sha256(paths["postiz_schedule_receipt"])}, "upload_copy": {"path": str(upload), "sha256": upload_hash, "bytes": upload_bytes}, "delete_targets": delete_records, "started_at": started_at, } atomic_json(cleanup_path, running) for record in delete_records: Path(record["path"]).unlink(missing_ok=True) upload_hash_after = sha256(upload) checks = { "upload_copy_exists": upload.is_file(), "upload_copy_hash_unchanged": upload_hash_after == upload_hash, "all_targets_absent": all(not Path(record["path"]).exists() for record in delete_records), } verified = all(checks.values()) result = { **running, "status": "completed" if verified else "failed", "verified": verified, "checks": checks, "deleted": [{**record, "exists_after": Path(record["path"]).exists()} for record in delete_records], "bytes_freed": sum(record["bytes"] for record in delete_records), "kept": [ {"key": "final_upload", "path": str(upload), "sha256": upload_hash_after, "bytes": upload.stat().st_size}, {"key": "intro", "path": str(resolve_active(project, "intro"))}, {"key": "tts_audio", "path": str(resolve_active(project, "tts_audio"))}, ], "completed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hash_method": "sha256 streaming 8 MiB", } atomic_json(cleanup_path, result) if not verified: raise RuntimeError(f"cleanup verification failed: {checks}") latest = load_json(MANIFEST, "project manifest") if latest.get("active_paths", {}).get("cleanup_receipt") != str(cleanup_path.relative_to(ROOT)): raise RuntimeError("project manifest changed cleanup receipt during cleanup") latest.setdefault("steps", {})["upload"] = "completed" latest["steps"]["cleanup"] = "completed" latest["cleanup"] = { "status": "completed", "verified": True, "receipt": str(cleanup_path.relative_to(ROOT)), "upload_copy_retained": str(upload.relative_to(ROOT)), "deleted_keys": list(DELETE_KEYS), "bytes_freed": result["bytes_freed"], "completed_at": result["completed_at"], } latest["status"] = "completed" latest["updated_at"] = result["completed_at"] atomic_json(MANIFEST, latest) print(json.dumps({ "status": "completed", "verified": True, "receipt": str(cleanup_path), "deleted": [record["path"] for record in delete_records], "bytes_freed": result["bytes_freed"], "upload_copy_retained": str(upload), "upload_copy_sha256": upload_hash_after, }, ensure_ascii=False)) return 0 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(json.dumps({ "status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}", }, ensure_ascii=False), file=sys.stderr) sys.exit(1)