#!/usr/bin/env python3 """Queue a folder of clip prompts through a ComfyUI image-to-video graph. Reads clip-*.txt in order. A header of `handoff-from: clip-NN last frame` extracts the last frame of that clip's video and loads it as the graph's first frame. `handoff-from: none` does not reuse a frame. A fresh clip uses --first-image, or a still beside the prompt (clip-01.png / .jpg / .webp). With neither, first_frame is left unconnected and the clip is text-to-video. Default graph: quickies/comfyui/minimax-fl2v.json Default host: http://127.0.0.1:9000 When the queue finishes, the clips are joined in order into renders/.mp4. The picture and the sound are also written apart, as renders/_video.mp4 and renders/_audio.m4a. Several project folders run one after another. Pass them on the command line, or list them in a --queue file. The next project starts when the one before it has finished. The guide is quickies/comfyui/README.md, beside this program. """ from __future__ import annotations import argparse import json import random import re import shutil import subprocess import sys import time import urllib.error import urllib.parse import urllib.request from pathlib import Path DEFAULT_WORKFLOW = Path(__file__).resolve().parent / "minimax-fl2v.json" DEFAULT_HOST = "http://127.0.0.1:9000" DEFAULT_INPUT = Path("/home/wizdroid/Applications/comfyui/input") CLIP_RE = re.compile(r"^clip-(\d+)", re.IGNORECASE) HANDOFF_RE = re.compile(r"^handoff-from:\s*(.+?)\s*$", re.IGNORECASE | re.MULTILINE) DURATION_RE = re.compile(r"clip-\d+-(\d+)s", re.IGNORECASE) STILL_EXTS = (".png", ".jpg", ".jpeg", ".webp") class BatchError(RuntimeError): pass def prompt_files(folder: Path) -> list[Path]: files = [p for p in folder.glob("clip-*.txt") if p.is_file()] def key(path: Path) -> tuple: match = CLIP_RE.match(path.name) number = int(match.group(1)) if match else 10**9 return (number, path.name) return sorted(files, key=key) def handoff_source(text: str) -> str | None: """Return the clip id that must supply the first frame, or None.""" match = HANDOFF_RE.search(text) if not match: return None value = match.group(1).strip().lower() if value == "none" or value.startswith("new ") or "do not" in value: return None clip = re.search(r"clip-\d+", value, re.IGNORECASE) if not clip: raise BatchError(f"cannot read handoff-from: {match.group(1).strip()}") return clip.group(0).lower() def duration_seconds(path: Path) -> float | None: match = DURATION_RE.search(path.name) if not match: return None return float(match.group(1)) def find_nodes(graph: dict, class_type: str) -> list[tuple[str, dict]]: return [ (node_id, node) for node_id, node in graph.items() if isinstance(node, dict) and node.get("class_type") == class_type ] def bind_graph(graph: dict) -> dict: i2v = find_nodes(graph, "MiniMaxH3ImageToVideo") if len(i2v) != 1: raise BatchError( f"expected one MiniMaxH3ImageToVideo node, found {len(i2v)}" ) prompt_id, prompt_node = i2v[0] if "prompt" not in prompt_node.get("inputs", {}): raise BatchError(f"node {prompt_id} has no prompt input") first = prompt_node["inputs"].get("first_frame") image_id = None if isinstance(first, list) and first: image_id = str(first[0]) image_node = graph.get(image_id, {}) if image_node.get("class_type") != "LoadImage": raise BatchError( f"first_frame comes from {image_id} ({image_node.get('class_type')}), not LoadImage" ) savers = find_nodes(graph, "SaveVideo") if len(savers) != 1: raise BatchError(f"expected one SaveVideo node, found {len(savers)}") duration_id = None for node_id, node in graph.items(): title = (node.get("_meta") or {}).get("title", "") if node.get("class_type") == "PrimitiveFloat" and "duration" in title.lower(): duration_id = node_id break noise = find_nodes(graph, "RandomNoise") return { "prompt_id": prompt_id, "image_id": image_id, "save_id": savers[0][0], "duration_id": duration_id, "noise_id": noise[0][0] if noise else None, } def apply_job(graph: dict, binding: dict, prompt: str, prefix: str, seconds: float | None, image_name: str | None) -> dict: job = json.loads(json.dumps(graph)) job[binding["prompt_id"]]["inputs"]["prompt"] = prompt job[binding["save_id"]]["inputs"]["filename_prefix"] = prefix if binding["duration_id"] and seconds is not None: job[binding["duration_id"]]["inputs"]["value"] = seconds if binding["noise_id"]: job[binding["noise_id"]]["inputs"]["noise_seed"] = random.randint(1, 2**31 - 1) if image_name and binding["image_id"]: job[binding["image_id"]]["inputs"]["image"] = image_name elif binding["image_id"]: job[binding["prompt_id"]]["inputs"].pop("first_frame", None) image_id = binding["image_id"] used = False for node_id, node in job.items(): if node_id == image_id: continue for value in node.get("inputs", {}).values(): if isinstance(value, list) and value and str(value[0]) == image_id: used = True if not used: job.pop(image_id, None) return job def still_beside(prompt_path: Path) -> Path | None: for ext in STILL_EXTS: candidate = prompt_path.with_suffix(ext) if candidate.is_file(): return candidate return None def frame_count(video: Path) -> int: probe = [ "ffprobe", "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=nb_frames", "-of", "default=nokey=1:noprint_wrappers=1", str(video), ] number = subprocess.run(probe, check=False, capture_output=True, text=True).stdout.strip() if number.isdigit() and int(number) > 0: return int(number) probe = [ "ffprobe", "-v", "error", "-count_frames", "-select_streams", "v:0", "-show_entries", "stream=nb_read_frames", "-of", "default=nokey=1:noprint_wrappers=1", str(video), ] number = subprocess.run(probe, check=False, capture_output=True, text=True).stdout.strip() if number.isdigit() and int(number) > 0: return int(number) raise BatchError(f"could not count frames in {video}") def extract_last_frame(video: Path, dest: Path) -> Path: dest.parent.mkdir(parents=True, exist_ok=True) count = frame_count(video) index = count - 1 cmd = [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(video), "-map", "0:v:0", "-vf", f"select=eq(n\\,{index})", "-frames:v", "1", str(dest), ] subprocess.run(cmd, check=True) if not dest.is_file(): raise BatchError(f"ffmpeg wrote no frame: {dest}") return dest def stage_image(src: Path, comfy_input: Path, name: str) -> str: comfy_input.mkdir(parents=True, exist_ok=True) dest = comfy_input / name shutil.copyfile(src, dest) return name def request_json(url: str, payload: dict | None = None, timeout: float = 60) -> dict: data = None headers = {} if payload is not None: data = json.dumps(payload).encode() headers["Content-Type"] = "application/json" req = urllib.request.Request(url, data=data, headers=headers) try: with urllib.request.urlopen(req, timeout=timeout) as response: return json.loads(response.read().decode()) except urllib.error.HTTPError as exc: body = exc.read().decode(errors="replace") raise BatchError(f"{url} returned {exc.code}: {body}") from exc except urllib.error.URLError as exc: raise BatchError(f"cannot reach {url}: {exc.reason}") from exc def queue_prompt(host: str, graph: dict) -> str: result = request_json(f"{host}/prompt", {"prompt": graph}) if result.get("node_errors"): raise BatchError(f"ComfyUI rejected the graph: {result['node_errors']}") prompt_id = result.get("prompt_id") if not prompt_id: raise BatchError(f"ComfyUI returned no prompt_id: {result}") return prompt_id def wait_history(host: str, prompt_id: str, timeout: float) -> dict: deadline = time.time() + timeout url = f"{host}/history/{prompt_id}" while time.time() < deadline: try: history = request_json(url, timeout=30) except BatchError: history = {} entry = history.get(prompt_id) if entry: status = entry.get("status") or {} if status.get("status_str") == "error" or status.get("completed") is False and status.get("status_str"): if status.get("status_str") == "error": raise BatchError(f"ComfyUI failed {prompt_id}: {status}") if status.get("completed") or entry.get("outputs"): return entry time.sleep(3) raise BatchError(f"timed out after {int(timeout)}s waiting for {prompt_id}") def output_videos(entry: dict) -> list[dict]: found = [] for node_out in (entry.get("outputs") or {}).values(): for key in ("videos", "gifs", "images"): for item in node_out.get(key) or []: name = item.get("filename", "") if name.lower().endswith((".mp4", ".webm", ".mkv", ".mov")): found.append(item) return found def download_output(host: str, item: dict, dest: Path) -> Path: query = urllib.parse.urlencode({ "filename": item["filename"], "subfolder": item.get("subfolder", ""), "type": item.get("type", "output"), }) dest.parent.mkdir(parents=True, exist_ok=True) url = f"{host}/view?{query}" try: with urllib.request.urlopen(url, timeout=120) as response: dest.write_bytes(response.read()) except urllib.error.URLError as exc: raise BatchError(f"could not download {url}: {exc.reason}") from exc return dest def clip_key(path: Path) -> str: match = CLIP_RE.match(path.name) if not match: return path.stem return f"clip-{match.group(1)}" def lookup_source(source: str, rendered: dict[str, Path]) -> Path: if source in rendered: return rendered[source] matches = [(key, path) for key, path in rendered.items() if key.startswith(source)] if len(matches) == 1: return matches[0][1] known = ", ".join(sorted(rendered)) or "none yet" raise BatchError( f"{source} needs a finished video for its last frame; have: {known}" ) def fresh_still(prompt_path: Path, first_image: Path | None, used_first: bool) -> tuple[str, Path, bool]: beside = still_beside(prompt_path) if beside: return (f"still {beside.name}", beside, used_first) if first_image and not used_first: return (f"starting still {first_image.name}", first_image, True) return ("no still, first frame bypassed", None, used_first) def stream_kinds(path: Path) -> set[str]: probe = subprocess.run( [ "ffprobe", "-v", "error", "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path), ], check=False, capture_output=True, text=True, ) return {line.strip() for line in probe.stdout.splitlines() if line.strip()} def concat_list(clips: list[Path], dest: Path) -> None: script = dest.with_suffix(".concat.txt") lines = [] for clip in clips: escaped = str(clip.resolve()).replace("'", "'\\''") lines.append(f"file '{escaped}'") script.write_text("\n".join(lines) + "\n") cmd = [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "concat", "-safe", "0", "-i", str(script), "-c", "copy", str(dest), ] try: subprocess.run(cmd, check=True) finally: script.unlink(missing_ok=True) def split_av(sequence: Path, video_path: Path, audio_path: Path) -> Path | None: subprocess.run( [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(sequence), "-map", "0:v:0", "-c", "copy", "-an", str(video_path), ], check=True, ) if "audio" not in stream_kinds(sequence): return None subprocess.run( [ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-i", str(sequence), "-map", "0:a:0", "-c", "copy", "-vn", str(audio_path), ], check=True, ) return audio_path def assemble(folder: Path, files: list[Path], dry_run: bool = False) -> None: renders = folder / "renders" clips = [] missing = [] for path in files: video = renders / f"{path.stem}.mp4" if video.is_file(): clips.append(video) else: missing.append(video.name) sequence = renders / f"{folder.name}.mp4" picture = renders / f"{folder.name}_video.mp4" sound = renders / f"{folder.name}_audio.m4a" if dry_run: print(f"assemble: {' '.join(p.name for p in clips) or '(no renders yet)'}") print(f" {sequence}") print(f" {picture}") print(f" {sound}") return if missing: raise BatchError(f"cannot stitch; missing {', '.join(missing)}") if not clips: raise BatchError("no clips to stitch") renders.mkdir(parents=True, exist_ok=True) concat_list(clips, sequence) audio = split_av(sequence, picture, sound) print(f"stitched {sequence}") print(f"video {picture}") if audio: print(f"audio {audio}") else: print("audio: none in the clips, so no audio file was written") def run(args: argparse.Namespace) -> int: folder = Path(args.directory).expanduser().resolve() if not folder.is_dir(): raise BatchError(f"not a directory: {folder}") files = prompt_files(folder) if not files: raise BatchError(f"no clip-*.txt in {folder}") if args.assemble_only and args.no_assemble: raise BatchError("--assemble-only and --no-assemble cannot be used together") if args.assemble_only: assemble(folder, files, dry_run=args.dry_run) return 0 workflow = Path(args.workflow).expanduser().resolve() graph = json.loads(workflow.read_text()) binding = bind_graph(graph) if binding["image_id"] is None: raise BatchError("graph has no first_frame LoadImage; this runner only feeds that input") renders = folder / "renders" rendered: dict[str, Path] = {} for path in files: existing = renders / f"{path.stem}.mp4" if existing.is_file(): rendered[clip_key(path)] = existing comfy_input = Path(args.input_dir).expanduser() film = folder.name used_first = False first_image = Path(args.first_image).expanduser().resolve() if args.first_image else None if first_image and not first_image.is_file(): raise BatchError(f"starting still not found: {first_image}") for path in files: key = clip_key(path) text = path.read_text() source = handoff_source(text) dest = renders / f"{path.stem}.mp4" if args.skip_existing and dest.is_file(): rendered[key] = dest print(f"skip {path.name} (render exists)") continue seconds = duration_seconds(path) prefix = f"videos/quickies/{film}/{path.stem}" local_image = None video = None if source: video = lookup_source(source, rendered) reason = f"last frame of {source}" else: reason, local_image, used_first = fresh_still(path, first_image, used_first) print(f"{path.name}: {reason}") print(f" prefix {prefix}" + (f" duration {seconds:g}s" if seconds else "")) if args.dry_run: if source and video is not None and not video.is_file(): print(f" would extract the last frame of {video.name} after it renders") elif source and video is not None: print(f" would extract the last frame of {video}") rendered[key] = dest continue if source: if video is None or not video.is_file(): raise BatchError(f"{source} has no video to take a last frame from") frame = renders / f"{video.stem}_last.png" extract_last_frame(video, frame) image_name = stage_image(frame, comfy_input, f"{film}_{video.stem}_last.png") elif local_image is not None: image_name = stage_image(local_image, comfy_input, f"{film}_{path.stem}{local_image.suffix.lower()}") else: image_name = None job = apply_job(graph, binding, text, prefix, seconds, image_name) prompt_id = queue_prompt(args.host.rstrip("/"), job) print(f" queued {prompt_id}") entry = wait_history(args.host.rstrip("/"), prompt_id, args.timeout) videos = output_videos(entry) if not videos: raise BatchError(f"{path.name} finished with no video in the history") download_output(args.host.rstrip("/"), videos[-1], dest) rendered[key] = dest print(f" wrote {dest}") if args.dry_run and not args.no_assemble: assemble(folder, files, dry_run=True) if not args.dry_run and not args.no_assemble: assemble(folder, files) return 0 def self_test() -> int: text_none = "handoff-from: none\nend-frame: hold\n\nintegrated_multimodal_description: hello\n" text_link = "handoff-from: clip-03 last frame\nnext: clip-05\n\nbody\n" assert handoff_source(text_none) is None assert handoff_source(text_link) == "clip-03" assert handoff_source("no header\n") is None graph = json.loads(DEFAULT_WORKFLOW.read_text()) binding = bind_graph(graph) assert binding["image_id"] == "144" assert binding["prompt_id"] == "105:104" job = apply_job(graph, binding, "PROMPT", "videos/quickies/demo/clip-01-15s", 15, "demo.png") assert job["105:104"]["inputs"]["prompt"] == "PROMPT" assert job["144"]["inputs"]["image"] == "demo.png" bare = apply_job(graph, binding, "PROMPT", "videos/quickies/demo/clip-01-15s", 15, None) assert "first_frame" not in bare["105:104"]["inputs"] assert "144" not in bare assert job["92"]["inputs"]["filename_prefix"] == "videos/quickies/demo/clip-01-15s" assert job["105:111"]["inputs"]["value"] == 15 folder = Path("/tmp/batch-clips-self") if folder.exists(): shutil.rmtree(folder) folder.mkdir() (folder / "clip-02-15s.txt").write_text(text_link) (folder / "clip-01-15s.txt").write_text(text_none) (folder / "arc.md").write_text("skip") ordered = [p.name for p in prompt_files(folder)] assert ordered == ["clip-01-15s.txt", "clip-02-15s.txt"], ordered video = folder / "clip.mp4" subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "color=c=black:s=32x32:r=10:d=0.2", "-f", "lavfi", "-i", "color=c=white:s=32x32:r=10:d=0.1", "-filter_complex", "[0][1]concat=n=2:v=1:a=0", "-pix_fmt", "yuv420p", str(video), ], check=True) frame = extract_last_frame(video, folder / "last.png") raw = subprocess.check_output([ "ffmpeg", "-v", "error", "-i", str(frame), "-vf", "scale=1:1,format=rgb24", "-f", "rawvideo", "-", ]) assert list(raw[:3]) == [255, 255, 255], list(raw[:3]) try: lookup_source("clip-09", {}) except BatchError: pass else: raise SystemExit("missing source video should fail") still = folder / "start.png" subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", "color=c=red:s=16x16:d=0.1", "-frames:v", "1", "-update", "1", str(still), ], check=True) (folder / "clip-02-15s.txt").write_text( "handoff-from: clip-01 last frame\n\nbody\n" ) code = main([ "--dry-run", "--first-image", str(still), str(folder), "--workflow", str(DEFAULT_WORKFLOW), ]) if code != 0: raise SystemExit(f"dry-run failed: {code}") code = main([ "--dry-run", "--no-assemble", str(folder), "--workflow", str(DEFAULT_WORKFLOW), ]) if code != 0: raise SystemExit(f"text-only dry-run failed: {code}") renders = folder / "renders" renders.mkdir() for name, color in (("clip-01-15s.mp4", "black"), ("clip-02-15s.mp4", "white")): subprocess.run([ "ffmpeg", "-hide_banner", "-loglevel", "error", "-y", "-f", "lavfi", "-i", f"color=c={color}:s=32x32:r=10:d=0.3", "-f", "lavfi", "-i", "sine=frequency=440:duration=0.3", "-shortest", "-pix_fmt", "yuv420p", str(renders / name), ], check=True) assemble(folder, prompt_files(folder)) sequence = renders / "batch-clips-self.mp4" picture = renders / "batch-clips-self_video.mp4" sound = renders / "batch-clips-self_audio.m4a" assert sequence.is_file() and picture.is_file() and sound.is_file() assert "audio" not in stream_kinds(picture) assert "video" not in stream_kinds(sound) assert "video" in stream_kinds(sequence) and "audio" in stream_kinds(sequence) queue_file = folder / "queue.txt" first = folder / "proj-a" second = folder / "proj-b" for project in (first, second): project.mkdir() (project / "clip-01-15s.txt").write_text(text_none) queue_file.write_text( f"# overnight\nproj-a {still.name}\n\nproj-b {still.name}\n" ) code = main([ "--dry-run", "--no-assemble", "--queue", str(queue_file), "--workflow", str(DEFAULT_WORKFLOW), ]) if code != 0: raise SystemExit(f"queue dry-run failed: {code}") code = main([ "--dry-run", "--no-assemble", "--first-image", str(still), str(first), str(second), "--workflow", str(DEFAULT_WORKFLOW), ]) if code != 0: raise SystemExit(f"multi-dir dry-run failed: {code}") shutil.rmtree(folder) print("self-test ok") return 0 def load_queue(path: Path) -> list[tuple[str, str | None]]: """Each line is FOLDER [FIRST_IMAGE]. Paths are relative to the queue file.""" if not path.is_file(): raise BatchError(f"queue file not found: {path}") base = path.parent jobs = [] for lineno, raw in enumerate(path.read_text().splitlines(), 1): line = raw.split("#", 1)[0].strip() if not line: continue parts = line.split() if len(parts) > 2: raise BatchError(f"{path}:{lineno}: expected FOLDER [FIRST_IMAGE]") folder = Path(parts[0]) if not folder.is_absolute(): folder = base / folder image = None if len(parts) == 2: image = Path(parts[1]) if not image.is_absolute(): image = base / image image = str(image) jobs.append((str(folder), image)) if not jobs: raise BatchError(f"no projects in {path}") return jobs def project_list(args: argparse.Namespace) -> list[tuple[str, str | None]]: jobs = [] if args.queue: jobs.extend(load_queue(Path(args.queue).expanduser())) for directory in args.directories: jobs.append((directory, None)) return jobs def parse_args(argv: list[str]) -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("directories", nargs="*", help="prompt folders, run in the order given") parser.add_argument("--queue", help="text file of projects: FOLDER [FIRST_IMAGE] per line") parser.add_argument("--workflow", default=str(DEFAULT_WORKFLOW), help="ComfyUI API graph") parser.add_argument("--host", default=DEFAULT_HOST, help="ComfyUI base URL") parser.add_argument("--input-dir", default=str(DEFAULT_INPUT), help="ComfyUI input folder") parser.add_argument("--first-image", help="still for the first clip that has no handoff") parser.add_argument("--timeout", type=float, default=2700, help="seconds to wait for one clip") parser.add_argument("--skip-existing", action="store_true", help="keep renders/.mp4") parser.add_argument("--no-assemble", action="store_true", help="leave the clips separate") parser.add_argument("--assemble-only", action="store_true", help="stitch renders already on disk") parser.add_argument("--dry-run", action="store_true", help="print the queue and do not generate") parser.add_argument("--self-test", action="store_true", help="check parsing, the graph, and last-frame extract") return parser.parse_args(argv) def main(argv: list[str]) -> int: args = parse_args(argv) if args.self_test: return self_test() try: jobs = project_list(args) except BatchError as exc: print(f"batch-clips: {exc}", file=sys.stderr) return 1 if not jobs: parse_args(["--help"]) return 2 failures = [] for index, (directory, image) in enumerate(jobs, 1): print(f"\n=== project {index}/{len(jobs)}: {directory} ===") one = argparse.Namespace(**vars(args)) one.directory = directory if image is not None: one.first_image = image try: run(one) except BatchError as exc: print(f"batch-clips: {directory}: {exc}", file=sys.stderr) failures.append(str(directory)) if failures: print( f"batch-clips: {len(failures)} of {len(jobs)} project(s) failed", file=sys.stderr, ) return 1 return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))