import argparse
import json
import os
import re
import time
from contextlib import ExitStack
from pathlib import Path
from openai import OpenAI, APIConnectionError, APIStatusError
parser = argparse.ArgumentParser()
parser.add_argument("action", choices=["create", "status", "wait", "download", "remix"])
parser.add_argument("task_id", nargs="?")
parser.add_argument("--max-polls", type=int, default=60)
parser.add_argument("--image", help="Reference image to upload when creating a task")
args = parser.parse_args()
if args.action != "create" and (not args.task_id or not re.fullmatch(r"[A-Za-z0-9_-]+", args.task_id)):
parser.error("Provide the public gateway task ID (letters, digits, underscores, or hyphens).")
if not 1 <= args.max_polls <= 120:
parser.error("--max-polls must be between 1 and 120.")
if args.image and args.action != "create":
parser.error("--image is only valid with create.")
with OpenAI(base_url=os.environ["API_BASE_URL"].rstrip("/"),
api_key=os.environ["API_KEY"], timeout=120.0, max_retries=0) as client:
try:
if args.action == "create":
# SDK sends multipart; the gateway also accepts JSON for compatible plugins.
with ExitStack() as stack:
extra = {}
if args.image:
extra["input_reference"] = stack.enter_context(Path(args.image).open("rb"))
result = client.videos.create(
model=os.environ["API_MODEL"],
prompt="A paper boat floating on a calm pond, static camera.", **extra,
)
elif args.action == "remix":
result = client.videos.remix(args.task_id, prompt="Change the scene to sunset.")
elif args.action == "download":
task = client.videos.retrieve(args.task_id)
if task.status != "completed":
raise ValueError("Video is not completed.")
# A failed transfer may leave a partial file; remove it before retrying download.
output = Path("video.mp4")
with client.videos.with_streaming_response.download_content(args.task_id) as response:
media_type = response.headers.get("content-type", "").split(";")[0].lower()
if media_type not in ("video/mp4", "application/octet-stream"):
raise ValueError("Expected MP4 bytes for this example.")
with output.open("xb") as target:
size = 0
for chunk in response.iter_bytes():
size += len(chunk)
if size > 256 * 1024 * 1024:
raise ValueError("Example download exceeds 256 MiB.")
target.write(chunk)
if size == 0:
raise ValueError("Empty video.")
print("Saved video.mp4")
raise SystemExit(0)
else:
for attempt in range(args.max_polls if args.action == "wait" else 1):
result = client.videos.retrieve(args.task_id)
if result.status not in ("queued", "in_progress") or args.action == "status":
break
if attempt + 1 == args.max_polls:
raise SystemExit("Polling limit reached; resume with the same task ID. The task was not cancelled.")
time.sleep(5)
if not result.id or result.status not in ("queued", "in_progress", "completed", "failed"):
raise ValueError("Unrecognized video response.")
print(json.dumps({"id": result.id, "status": result.status, "progress": getattr(result, "progress", None)}))
if result.status == "failed":
raise SystemExit("Video task failed. Check the task logs before submitting again.")
except APIStatusError as error:
raise SystemExit(f"Request failed (HTTP {error.status_code}). Keep any task ID; do not automatically create another task.")
except APIConnectionError:
raise SystemExit("Connection failed. Keep any task ID; do not automatically create another task.")
except (ValueError, OSError):
raise SystemExit("Invalid video response or output file. Check status, content type, and any partial download.")