#!/usr/bin/env python3 """EDL JSON -> validated cuts + concat: the deterministic core of edit-as-code. Reads an edit decision list (schema: assets/edl-schema.json — scenes, clips, time ranges, written rationale), validates it, and produces the final video via per-clip cuts + the concat demuxer. DRY-RUN BY DEFAULT: prints every command it would run and touches nothing until --execute. Re-encode mode (default) is frame-accurate and normalizes codec/resolution/fps across clips so the concat is always safe; --copy is faster but requires keyframe-aligned cut points and identical source parameters. Usage: cut-from-edl.py [--execute] [--copy] [-o OUT] [--workdir DIR] [--json] Input: EDL JSON as positional; clip paths resolve relative to the EDL's directory Output: stdout = planned/executed command list (or --json envelope, schema claude-mods.ffmpeg-ops.edl/v1) Stderr: progress, warnings, errors Exit: 0 ok, 2 usage, 3 EDL or source file missing, 4 EDL invalid, 5 ffmpeg missing (--execute only) Examples: cut-from-edl.py edit.json # dry-run: show the plan cut-from-edl.py edit.json --execute -o final.mp4 cut-from-edl.py edit.json --execute --copy # keyframe-aligned EDLs only cut-from-edl.py edit.json --json | jq '.data.commands' """ import argparse import json import shutil import subprocess import sys from pathlib import Path from typing import NoReturn SCHEMA = "claude-mods.ffmpeg-ops.edl/v1" EXIT_OK, EXIT_USAGE, EXIT_NOT_FOUND, EXIT_VALIDATION, EXIT_MISSING_DEP = 0, 2, 3, 4, 5 def err(json_mode: bool, code: str, message: str, exit_code: int) -> NoReturn: if json_mode: print(json.dumps({"error": {"code": code, "message": message, "details": {}}})) print(f"ERROR: {message}", file=sys.stderr) sys.exit(exit_code) def validate_edl(edl: dict) -> list: """Stdlib structural validation mirroring assets/edl-schema.json.""" problems = [] scenes = edl.get("scenes") if not isinstance(scenes, list) or not scenes: return ["'scenes' must be a non-empty array"] for i, scene in enumerate(scenes): where = f"scenes[{i}]" if not isinstance(scene, dict): problems.append(f"{where} must be an object") continue clips = scene.get("clips") if not isinstance(clips, list) or not clips: problems.append(f"{where}.clips must be a non-empty array") continue for j, clip in enumerate(clips): cw = f"{where}.clips[{j}]" if not isinstance(clip, dict): problems.append(f"{cw} must be an object") continue if not isinstance(clip.get("file"), str) or not clip.get("file"): problems.append(f"{cw}.file must be a non-empty string") start, end = clip.get("start"), clip.get("end") if not isinstance(start, (int, float)) or start < 0: problems.append(f"{cw}.start must be a number >= 0") if not isinstance(end, (int, float)): problems.append(f"{cw}.end must be a number") elif isinstance(start, (int, float)) and end <= start: problems.append(f"{cw}: end ({end}) must be > start ({start})") return problems def video_props(ffprobe: str, path: Path) -> dict: proc = subprocess.run( [ffprobe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,r_frame_rate", "-of", "csv=p=0", str(path)], capture_output=True, text=True) parts = proc.stdout.strip().split(",") if len(parts) == 3: try: num, den = parts[2].split("/") fps = round(int(num) / int(den), 3) if int(den) else 0 return {"width": int(parts[0]), "height": int(parts[1]), "fps": fps} except (ValueError, ZeroDivisionError): pass return {} def main() -> int: ap = argparse.ArgumentParser( description="Cut + concat a final video from an EDL JSON (dry-run by default).", epilog="Examples:\n" " cut-from-edl.py edit.json\n" " cut-from-edl.py edit.json --execute -o final.mp4\n", formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("edl", help="EDL JSON file (see assets/edl-schema.json)") ap.add_argument("--execute", action="store_true", help="actually run the cuts (default: dry-run print only)") ap.add_argument("--copy", action="store_true", help="stream-copy cuts (fast; needs keyframe-aligned points " "and identical source params)") ap.add_argument("-o", "--output", default=None, help="final output path, resolved against the CWD (default: the " "EDL 'output' field resolved against the EDL file, else final.mp4)") ap.add_argument("--workdir", default=None, help="directory for cut segments (default: /edl-cuts)") ap.add_argument("--json", action="store_true", help="emit JSON envelope on stdout") args = ap.parse_args() edl_path = Path(args.edl) if not edl_path.is_file(): err(args.json, "NOT_FOUND", f"EDL not found: {edl_path}", EXIT_NOT_FOUND) try: edl = json.loads(edl_path.read_text(encoding="utf-8")) except json.JSONDecodeError as e: err(args.json, "VALIDATION", f"EDL is not valid JSON: {e}", EXIT_VALIDATION) problems = validate_edl(edl) if problems: err(args.json, "VALIDATION", "EDL failed validation: " + "; ".join(problems[:5]) + (f" (+{len(problems) - 5} more)" if len(problems) > 5 else ""), EXIT_VALIDATION) base = edl_path.resolve().parent workdir = Path(args.workdir) if args.workdir else base / "edl-cuts" # CLI -o resolves against the CWD (normal CLI convention); the EDL's own # 'output' field resolves against the EDL file (schema contract). if args.output: output = Path(args.output).resolve() else: output = Path(edl.get("output") or "final.mp4") if not output.is_absolute(): output = base / output # Resolve and existence-check sources (fatal in execute, warning in dry-run). clips, missing = [], [] for scene in edl["scenes"]: for clip in scene["clips"]: src = Path(clip["file"]) if not src.is_absolute(): src = base / src if not src.is_file(): missing.append(str(src)) clips.append({"scene": scene.get("scene"), "src": src, "start": float(clip["start"]), "end": float(clip["end"])}) if missing: for m in missing: print(f"warning: source missing: {m}", file=sys.stderr) if args.execute: err(args.json, "NOT_FOUND", f"{len(missing)} source file(s) missing (first: {missing[0]})", EXIT_NOT_FOUND) ffmpeg = shutil.which("ffmpeg") ffprobe = shutil.which("ffprobe") if args.execute and not ffmpeg: err(args.json, "MISSING_DEPENDENCY", "ffmpeg not found on PATH", EXIT_MISSING_DEP) # Re-encode mode normalizes every segment to the first clip's geometry/fps, # which is what makes the concat demuxer unconditionally safe. norm_filter = "" if not args.copy and ffprobe and not missing: props = [video_props(ffprobe, c["src"]) for c in clips] props = [p for p in props if p] if props: w, h, fps = props[0]["width"], props[0]["height"], props[0]["fps"] or 30 if any((p["width"], p["height"]) != (w, h) or p["fps"] != props[0]["fps"] for p in props): print(f"note: mixed source params — normalizing all segments to " f"{w}x{h} @ {fps}fps", file=sys.stderr) norm_filter = (f"scale={w}:{h}:force_original_aspect_ratio=decrease," f"pad={w}:{h}:(ow-iw)/2:(oh-ih)/2,fps={fps}") commands, concat_lines = [], [] for n, clip in enumerate(clips, 1): seg = workdir / f"seg{n:03d}.mp4" cmd = ["ffmpeg", "-y", "-ss", f"{clip['start']}", "-to", f"{clip['end']}", "-i", str(clip["src"])] if args.copy: cmd += ["-c", "copy", "-avoid_negative_ts", "make_zero"] else: if norm_filter: cmd += ["-vf", norm_filter] cmd += ["-c:v", "libx264", "-crf", "18", "-preset", "fast", "-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "192k", "-ar", "48000"] cmd.append(str(seg)) commands.append(cmd) concat_lines.append(f"file '{seg.as_posix()}'") concat_txt = workdir / "concat.txt" final_cmd = ["ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_txt), "-c", "copy", "-movflags", "+faststart", str(output)] data = { "edl": str(edl_path), "mode": "copy" if args.copy else "reencode", "executed": bool(args.execute), "workdir": str(workdir), "output": str(output), "segments": len(clips), "missing_sources": missing, "commands": [" ".join(c) for c in commands] + [" ".join(final_cmd)], } if not args.execute: if args.json: print(json.dumps({"data": data, "meta": {"schema": SCHEMA}}, indent=2)) else: print(f"# DRY-RUN — {len(clips)} segment(s) -> {output}") for c in data["commands"][:-1]: print(c) print(f"# concat.txt:\n" + "\n".join(f"# {l}" for l in concat_lines)) print(data["commands"][-1]) print("dry-run only; pass --execute to run", file=sys.stderr) return EXIT_OK workdir.mkdir(parents=True, exist_ok=True) for n, cmd in enumerate(commands, 1): print(f"cutting segment {n}/{len(commands)}...", file=sys.stderr) proc = subprocess.run(cmd, capture_output=True, text=True) if proc.returncode != 0: err(args.json, "VALIDATION", f"segment {n} failed: {(proc.stderr.strip().splitlines() or ['?'])[-1]}", EXIT_VALIDATION) concat_txt.write_text("\n".join(concat_lines) + "\n", encoding="utf-8") # Atomic final write: concat to a temp name, then rename over the # destination. The temp KEEPS the real extension — ffmpeg infers the muxer # from it, and "final.mp4.tmp" would fail with "Invalid argument". tmp_out = output.with_name(output.stem + ".tmp" + output.suffix) final_cmd[-1] = str(tmp_out) # the destination dir must exist BEFORE ffmpeg opens the temp output - # otherwise concat dies with a cryptic "Error opening output files" output.parent.mkdir(parents=True, exist_ok=True) print("concatenating...", file=sys.stderr) proc = subprocess.run(final_cmd, capture_output=True, text=True) if proc.returncode != 0: err(args.json, "VALIDATION", f"concat failed: {(proc.stderr.strip().splitlines() or ['?'])[-1]}", EXIT_VALIDATION) tmp_out.replace(output) if args.json: print(json.dumps({"data": data, "meta": {"schema": SCHEMA}}, indent=2)) else: print(str(output)) print(f"done: {output} ({len(clips)} segments)", file=sys.stderr) print("next: re-transcribe the output and verify no words were clipped " "(see references/edit-as-code.md)", file=sys.stderr) return EXIT_OK if __name__ == "__main__": sys.exit(main())