#!/usr/bin/env python3 """DAOCALM CineSynth integration helper for a customer-configured ComfyUI server. Requires Python 3.10+ and requests. Provide your own working API-format workflow, installed models and nodes, asset permissions, and protected runtime. This is not a DAOCALM SDK. It does not install models, create workflows or guarantee video. """ import argparse import json import os from pathlib import Path import re import sys import time from urllib.parse import quote, urlsplit import uuid def validate_workflow(workflow): if not isinstance(workflow, dict) or not workflow: raise ValueError("Workflow must be a non-empty API-format node map.") for node_id, node in workflow.items(): if (not isinstance(node_id, str) or not isinstance(node, dict) or not isinstance(node.get("class_type"), str) or not isinstance(node.get("inputs"), dict)): raise ValueError("Use ComfyUI File > Export (API), not the editor's visual workflow JSON.") return workflow def artifact_descriptors(history): """Recognize standard file descriptors, including compatible custom nodes.""" seen = set() for node_output in history.get("outputs", {}).values(): if not isinstance(node_output, dict): continue for value in node_output.values(): if not isinstance(value, list): continue for descriptor in value: if not isinstance(descriptor, dict) or not isinstance(descriptor.get("filename"), str): continue name = descriptor["filename"] folder = descriptor.get("subfolder", "") kind = descriptor.get("type", "output") if not name or not isinstance(folder, str) or kind not in ("output", "temp", "input"): continue key = (name, folder, kind) if key not in seen: seen.add(key) yield {"filename": name, "subfolder": folder, "type": kind} class ComfyClient: def __init__(self, base_url, token=None, session=None): parts = urlsplit(base_url) if (parts.scheme not in ("http", "https") or not parts.hostname or parts.username or parts.password or parts.query or parts.fragment): raise ValueError("Use an http(s) base URL without credentials, query or fragment.") if parts.scheme == "http" and parts.hostname not in ("127.0.0.1", "localhost", "::1"): raise ValueError("Remote runtimes require HTTPS. Local HTTP is allowed on loopback only.") self.base_url = base_url.rstrip("/") if session is None: import requests session = requests.Session() self.session = session self.session.headers.update({"Accept": "application/json"}) if token: if "\r" in token or "\n" in token: raise ValueError("Invalid gateway token.") self.session.headers["Authorization"] = "Bearer " + token def submit(self, workflow): validate_workflow(workflow) # No automatic POST retries: a lost response can still mean a queued job. try: response = self.session.post( self.base_url + "/prompt", json={"prompt": workflow, "client_id": str(uuid.uuid4())}, timeout=(5, 30), allow_redirects=False, ) except Exception as exc: raise RuntimeError("Submission outcome is unknown after a connection failure. " "Inspect the runtime queue/history before submitting again.") from exc if response.status_code != 200: raise RuntimeError(f"Submission returned HTTP {response.status_code}. " "Check workflow validation, gateway authentication, and the runtime logs before retrying.") try: data = response.json() except ValueError: raise RuntimeError("Submission returned invalid JSON. Its outcome is unknown; inspect the runtime queue.") from None if not isinstance(data, dict) or not isinstance(data.get("prompt_id"), str) or not data["prompt_id"]: raise RuntimeError("No prompt_id returned. Inspect validation errors and the runtime queue before retrying.") if data.get("node_errors"): print("The runtime accepted this prompt with node validation warnings; inspect its logs.", file=sys.stderr) return data["prompt_id"] def wait(self, prompt_id, timeout=900, poll_interval=2): import requests if timeout <= 0 or poll_interval <= 0: raise ValueError("Timeout and polling interval must be positive.") deadline = time.monotonic() + timeout while time.monotonic() < deadline: remaining = max(0.1, deadline - time.monotonic()) try: response = self.session.get( self.base_url + "/history/" + quote(prompt_id, safe=""), timeout=(min(5, remaining), min(20, remaining)), allow_redirects=False, ) except (requests.exceptions.Timeout, requests.exceptions.ConnectionError): time.sleep(min(poll_interval, max(0, deadline - time.monotonic()))) continue if response.status_code in (429, 502, 503, 504): time.sleep(min(poll_interval, max(0, deadline - time.monotonic()))) continue if response.status_code != 200: raise RuntimeError(f"History returned HTTP {response.status_code}; prompt_id={prompt_id}.") payload = response.json() if not isinstance(payload, dict): raise RuntimeError("History response is not a JSON object.") item = payload.get(prompt_id) if isinstance(item, dict): status = item.get("status", {}) messages = status.get("messages", []) error_event = any(isinstance(event, (list, tuple)) and event and event[0] in ("execution_error", "execution_interrupted") for event in messages) if status.get("status_str") == "error" or error_event: raise RuntimeError(f"Execution failed or was interrupted; prompt_id={prompt_id}. Inspect the runtime logs.") if status.get("completed") is True and status.get("status_str") == "success": return item time.sleep(min(poll_interval, max(0, deadline - time.monotonic()))) raise TimeoutError(f"Polling timed out; prompt_id={prompt_id}. The server job may still be running. " "Resume with --prompt-id; do not submit a duplicate workflow.") def download(self, descriptor, destination, max_bytes=512 * 1024 * 1024): """Stream a /view artifact, with a size cap and no local path traversal.""" response = self.session.get( self.base_url + "/view", params=descriptor, headers={"Accept": "*/*"}, timeout=(5, 30), stream=True, allow_redirects=False, ) try: if response.status_code != 200: raise RuntimeError(f"Artifact download returned HTTP {response.status_code}.") size = response.headers.get("Content-Length") if size and int(size) > max_bytes: raise RuntimeError("Artifact exceeds the configured download size limit.") written = 0 created = False try: with destination.open("xb") as output: created = True for chunk in response.iter_content(chunk_size=65536): if chunk: written += len(chunk) if written > max_bytes: raise RuntimeError("Artifact exceeds the configured download size limit.") output.write(chunk) except BaseException: if created: destination.unlink(missing_ok=True) raise return written finally: response.close() def main(): parser = argparse.ArgumentParser(description=__doc__) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--workflow", type=Path, help="Workflow exported in API format; submitting starts a real runtime job") source.add_argument("--prompt-id", help="Resume polling an existing prompt; does not resubmit it") parser.add_argument("--base-url", default=os.environ.get("COMFY_BASE_URL", "http://127.0.0.1:8188")) parser.add_argument("--output-dir", type=Path, default=Path("cinesynth-output"), help="New output directory; existing directories are not replaced") parser.add_argument("--timeout", type=float, default=900, help="Polling timeout in seconds") parser.add_argument("--max-download-mib", type=int, default=512, help="Per-artifact size cap") parser.add_argument("--max-artifacts", type=int, default=20) args = parser.parse_args() if args.timeout <= 0 or args.max_download_mib <= 0 or args.max_artifacts <= 0: parser.error("Timeout and download limits must be positive.") workflow = None if args.workflow: workflow = validate_workflow(json.loads(args.workflow.read_text(encoding="utf-8"))) try: client = ComfyClient(args.base_url, os.environ.get("COMFY_GATEWAY_BEARER_TOKEN")) except ImportError: parser.error("Install the HTTP client with: python -m pip install requests") args.output_dir.mkdir(parents=True, exist_ok=False) prompt_id = args.prompt_id or client.submit(workflow) with (args.output_dir / "job.json").open("x", encoding="utf-8") as output: json.dump({"prompt_id": prompt_id, "runtime": client.base_url}, output, indent=2) print(f"prompt_id={prompt_id}; recorded in {args.output_dir / 'job.json'}", flush=True) history = client.wait(prompt_id, timeout=args.timeout) with (args.output_dir / "history.json").open("x", encoding="utf-8") as output: json.dump({prompt_id: history}, output, ensure_ascii=False, indent=2) artifacts = list(artifact_descriptors(history)) if len(artifacts) > args.max_artifacts: raise RuntimeError("Artifact count exceeds --max-artifacts. History was saved; review it before downloading.") for index, descriptor in enumerate(artifacts, 1): filename = re.sub(r"[^A-Za-z0-9._-]", "_", Path(descriptor["filename"]).name)[:160] or "artifact" destination = args.output_dir / f"artifact-{index:04d}-{filename}" client.download(descriptor, destination, max_bytes=args.max_download_mib * 1024 * 1024) print(f"Saved {destination}") if not artifacts: print("Execution completed, but no supported file descriptors were found. Inspect history.json; the output node may need an adapter.") else: print("Execution completed. Review generated media and asset permissions before publishing.") if __name__ == "__main__": try: main() except (OSError, RuntimeError, ValueError) as exc: print(f"Error: {exc}", file=sys.stderr) sys.exit(1)