# Minimax Hailuo Video
> Use MiniMax's first-party Hailuo video API safely and reproducibly across global and mainland-China platforms. Covers current text-to-video, image-to-video, first/last-frame, and subject-reference models; prompt and camera craft; region/account routing; exact cost approval; asynchronous jobs and callbacks; artifact provenance; and rights, likeness, disclosure, and data-governance review. Use when designing, implementing, debugging, or evaluating direct MiniMax/Hailuo video-generation API workflows. Do not use for the consumer Hailuo site, Video Agent templates, gateways, or unrelated MiniMax modalities.
- Skill: `calesthio/minimax-hailuo-video` (Agent Skill, multi-file: 2 files)
- Install (CLI): `npx skillmds@latest add calesthio/minimax-hailuo-video`
- Raw SKILL.md: https://api.skillmd.com/api/skills/calesthio/minimax-hailuo-video/raw
- Safety review: pending
- Works with: Claude Code, Claude.ai, OpenAI Codex
- Category: AI & ML
- Author: calesthio (https://skillmd.com/u/calesthio)
- Updated: 2026-09-17
- Page: https://skillmd.com/skills/calesthio/minimax-hailuo-video
---
# MiniMax Hailuo video
Build a reviewed request, obtain exact cost approval, create at most one paid task, then preserve the result and its evidence. Treat documentation gaps as unknowns, not defaults.
## Start with the boundary
Choose one service region and keep its account, key, billing, host, and terms together:
| Region | Console/docs | API host | Billing currency |
|---|---|---|---|
| Global | `platform.minimax.io` | `https://api.minimax.io` | USD |
| Mainland China | `platform.minimaxi.com` | `https://api.minimaxi.com` | CNY |
The two platforms are separate service regions. A key or plan bought on one must not be assumed to work on the other. A `401` can be a wrong-region symptom. Use a pay-as-you-go API key for the examples below; a Token Plan/Subscription Key has separate quotas and model access.
Keep these products out of this workflow:
- `hailuoai.video` is the consumer Hailuo product with separate terms, privacy behavior, features, and commercial-use implications.
- Video Agent/template generation is a different template API. Its deprecated query endpoint and nine-hour link statement do not define general video generation.
- Image, speech, music, and text endpoints are separate modalities. General Hailuo video generation does not document an audio input, audio prompt, or generated-audio control.
- Gateways and resellers have their own schemas, retention, pricing, and support. Do not call them “the MiniMax API.â€
## Route by mode before choosing a model
Current first-party general video generation uses `POST /v1/video_generation` for four modes. The API overview highlights the Hailuo 2.3 family and Hailuo-02; detailed endpoint pages also enumerate older models.
| Mode | Required media field(s) | Documented models | Documented output choices |
|---|---|---|---|
| Text-to-video (T2V) | none; `prompt` required | `MiniMax-Hailuo-2.3`, `MiniMax-Hailuo-02`; T2V endpoint also lists `T2V-01-Director`, `T2V-01` | 2.3/02: 768P at 6 or 10 s; 1080P at 6 s; 24 fps |
| Image-to-video (I2V) | `first_frame_image` | `MiniMax-Hailuo-2.3`, `MiniMax-Hailuo-2.3-Fast`, `MiniMax-Hailuo-02`; endpoint also lists `I2V-01-Director`, `I2V-01-live`, `I2V-01` | 2.3/Fast/02: 768P at 6 or 10 s; 1080P at 6 s. 02 also: 512P at 6 or 10 s; 24 fps |
| First/last frame | `first_frame_image`, `last_frame_image` | `MiniMax-Hailuo-02` | 768P at 6 or 10 s; 1080P at 6 s; 24 fps. The endpoint explicitly says 512P is unsupported |
| Subject reference | `subject_reference` array | `S2V-01` | **Unknown:** its endpoint omits duration, resolution, and current price; the general guide example nevertheless sends 6 s and 1080P |
`MiniMax-Hailuo-2.3-Fast` is I2V-only. Do not select it for T2V. Detailed pages describe older `*-01*` models as 720P/6 s, but current pay-as-you-go pricing omits them; verify availability and price in the correct console before any paid use. Do the same for `S2V-01` rather than borrowing another model's price.
Current pay-as-you-go prices, checked 2026-07-10:
| Combination | Global | Mainland China |
|---|---:|---:|
| 2.3-Fast, 768P, 6 s | $0.19 | ¥1.35 |
| 2.3-Fast, 768P, 10 s | $0.32 | ¥2.25 |
| 2.3-Fast, 1080P, 6 s | $0.33 | ¥2.31 |
| 2.3 or 02, 768P, 6 s | $0.28 | ¥2.00 |
| 2.3 or 02, 768P, 10 s | $0.56 | ¥4.00 |
| 2.3 or 02, 1080P, 6 s | $0.49 | ¥3.50 |
| 02 I2V, 512P, 6 s | $0.10 | ¥0.60 |
| 02 I2V, 512P, 10 s | $0.15 | ¥1.00 |
These are per generated video, not timeless constants. Recheck the region's official pricing page immediately before approval. Packages deduct units and raise RPM; Token Plans use plan quotas. Never turn a package's blended cost or another region's currency into a pay-as-you-go quote.
## Build a request dossier
Before calling create, record:
- region, console/account identifier, key type, model, mode, resolution, duration, variant count, currency, unit price, and pricing-page retrieval time;
- exact prompt, `prompt_optimizer`, `fast_pretreatment`, and every non-default parameter;
- source file SHA-256, MIME type, byte length, dimensions, aspect ratio, provenance, license/consent, and any face/identity present;
- intended audience, jurisdiction, AI disclosure/watermark plan, retention class, and reviewer;
- an attempt ID generated locally before the POST.
For first/last-frame and I2V inputs, official constraints are JPG/JPEG, PNG, or WebP; under 20 MB; short edge over 300 px; aspect ratio from 2:5 through 5:2; public URL or Base64 data URL. The first image determines output aspect/resolution. In first/last mode, a differently sized last image is cropped to match the first. Pre-align composition, subject scale, lighting, and topology to avoid an implausible transition.
The S2V reference documents a different exact shape: `"subject_reference": [{"type": "character", "image": ["https://authorized.example/reference.jpg"]}]`. `type` is `character`; `image` is an array containing the authorized face-reference URL. Do not transpose I2V fields or assume I2V's media transport rules where the S2V page is silent.
A public source URL lets MiniMax fetch that asset and may expose it through URL logs or access controls. Prefer an appropriately scoped URL or bounded Base64 only when policy allows. Never accept arbitrary user URLs into your own downloader without SSRF controls. Hash the original asset; do not log Base64 bodies or signed download URLs.
## Write prompts for this motion system
The prompt limit is 2,000 characters. Hailuo supports explicit camera commands on the documented 2.3, 02, and Director workflows:
`[Truck left]`, `[Truck right]`, `[Pan left]`, `[Pan right]`, `[Push in]`, `[Pull out]`, `[Pedestal up]`, `[Pedestal down]`, `[Tilt up]`, `[Tilt down]`, `[Zoom in]`, `[Zoom out]`, `[Shake]`, `[Tracking shot]`, `[Static shot]`.
Commands in one bracket happen together; official guidance recommends at most three combined commands. Commands placed sequentially express a sequence. They are directional controls, not frame-accurate timecodes.
Use this compact prompt order:
1. Establish subject and initial composition.
2. Describe one physically coherent action arc.
3. State the dominant camera command or simple sequence.
4. Specify a visible end state and stable details that must persist.
For I2V, describe change from the supplied frame instead of contradicting its static appearance. For first/last-frame generation, describe a plausible bridge between the two known states. For a designed shot, set `prompt_optimizer: false`; otherwise the optimizer may rewrite wording or camera intent. `prompt_optimizer` defaults to `true`. `fast_pretreatment` defaults to `false`, requires the optimizer, and is documented for current Hailuo I2V models.
The following are craft heuristics, not API guarantees: one primary action and one dominant camera move usually fit a 6–10 second clip better than dense choreography; explicit end-state language improves reviewability; generated physics, faces, text, hands, occlusion, and identity should still be inspected frame by frame. MiniMax's claims about improved motion, expression, realism, and adherence are provider claims, not independent benchmark findings.
The general endpoint documents no seed, negative prompt, arbitrary aspect-ratio parameter, audio input, audio prompt, or audio-output switch. Do not invent them. Treat the presence and nature of an output audio track as unknown until the actual file is inspected; add separately licensed sound downstream when required.
Across the mode-specific references, `model` is required; T2V requires `prompt`, while I2V and first/last describe the text prompt as optional; prompts may be at most 2,000 characters. `first_frame_image`, `last_frame_image`, and `subject_reference` are required only by their respective modes. `callback_url` is optional. `duration` and `resolution` must follow the model/mode matrix above rather than generic CLI defaults. The current mainland-China T2V page additionally exposes `aigc_watermark` as a boolean defaulting to `false`; the current global T2V page does not.
## Make the paid boundary explicit
Default to dry-run. Show the redacted request, exact one-video cost, variant multiplier, currency, authoritative pricing timestamp/evidence digest, source hashes, local job key, output destination, and exact-host download policy. Canonically hash that complete envelope. Require a human to approve the digest and a finite positive maximum in the same currency; amount-only approval is insufficient because it does not bind the model, prompt, source, region, or custody policy. If the combination has no current published price, stop with `UNKNOWN` and send the operator to the correct console or MiniMax support.
Before POSTing, acquire the job ledger with exclusive-create semantics and persist an `intent-recorded` entry. After a response, persist only an allowlisted result summary—task ID, provider code, status-message/body digests, and timestamps—not raw prompts, Base64, error bodies, or signed URLs. MiniMax documents no idempotency key for video creation. Therefore:
- Retry authenticated GET/poll/download operations with bounded backoff.
- Do not automatically retry a create POST.
- If create times out or the connection breaks before a `task_id` is captured, mark it `create-outcome-unknown`. Reconcile billing/console/support before authorizing a replacement task.
- Treat a 5xx, provider timeout/internal code, malformed envelope, or nominal success with an unusable task ID as equally ambiguous and persist that state before stopping.
- A replay is a new potentially billable video unless the earlier create is conclusively known not to exist.
HTTP success alone is not provider success. Require `base_resp.status_code == 0` and a nonempty `task_id`. Preserve `base_resp.status_code` and `status_msg` for diagnosis. Common codes include 1001 timeout, 1002 rate limit, 1004 authentication, 1008 insufficient balance, 1026 input sensitive, 1027 output sensitive, 2013 invalid parameters, 2049 invalid key, and 2056 usage limit. Never evade a content-safety response.
The published default global video limit is currently 5 requests/minute across the Hailuo 2.3 and 02 series; packages publish higher limits. Treat this as account/plan-specific and recheck the console. Pace polls separately; the guide recommends a 10-second interval.
Check during incidents, but do not infer healthy video generation merely from another listed component's status if video is not shown separately.
## Complete dry-run-first client
This Python 3.11+ client covers currently priced T2V, I2V, and first/last-frame combinations in both regions. It does not execute `S2V-01` or unpriced legacy models. It never retries create, acquires an exclusive attempt ledger before create, resumes known task IDs, checks MiniMax's envelope, polls every 10 seconds, downloads with an exact-host/public-DNS policy and no bearer token, and crash-resumably validates MP4 with `ffprobe` plus a full `ffmpeg` decode. Save it as `hailuo_generate.py` outside the skill directory. Python 3.11+, `ffprobe`, and `ffmpeg` are required.
```python
#!/usr/bin/env python3
import argparse, base64, hashlib, http.client, ipaddress, json, os, pathlib, random, socket, ssl, subprocess, sys, tempfile, threading, time, urllib.error, urllib.parse, urllib.request
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
HOSTS = {"global": "https://api.minimax.io", "cn": "https://api.minimaxi.com"}
PRICES = {
"global": {"currency": "USD", "table": {
("MiniMax-Hailuo-2.3-Fast", "768P", 6): "0.19", ("MiniMax-Hailuo-2.3-Fast", "768P", 10): "0.32",
("MiniMax-Hailuo-2.3-Fast", "1080P", 6): "0.33", ("MiniMax-Hailuo-2.3", "768P", 6): "0.28",
("MiniMax-Hailuo-2.3", "768P", 10): "0.56", ("MiniMax-Hailuo-2.3", "1080P", 6): "0.49",
("MiniMax-Hailuo-02", "512P", 6): "0.10", ("MiniMax-Hailuo-02", "512P", 10): "0.15",
("MiniMax-Hailuo-02", "768P", 6): "0.28", ("MiniMax-Hailuo-02", "768P", 10): "0.56",
("MiniMax-Hailuo-02", "1080P", 6): "0.49"}},
"cn": {"currency": "CNY", "table": {
("MiniMax-Hailuo-2.3-Fast", "768P", 6): "1.35", ("MiniMax-Hailuo-2.3-Fast", "768P", 10): "2.25",
("MiniMax-Hailuo-2.3-Fast", "1080P", 6): "2.31", ("MiniMax-Hailuo-2.3", "768P", 6): "2.00",
("MiniMax-Hailuo-2.3", "768P", 10): "4.00", ("MiniMax-Hailuo-2.3", "1080P", 6): "3.50",
("MiniMax-Hailuo-02", "512P", 6): "0.60", ("MiniMax-Hailuo-02", "512P", 10): "1.00",
("MiniMax-Hailuo-02", "768P", 6): "2.00", ("MiniMax-Hailuo-02", "768P", 10): "4.00",
("MiniMax-Hailuo-02", "1080P", 6): "3.50"}}
}
MODELS = {
"text": {"MiniMax-Hailuo-2.3", "MiniMax-Hailuo-02"},
"image": {"MiniMax-Hailuo-2.3", "MiniMax-Hailuo-2.3-Fast", "MiniMax-Hailuo-02"},
"first-last": {"MiniMax-Hailuo-02"},
}
MIME = {".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp"}
MAX_JSON = 256 * 1024
MAX_VIDEO = 250 * 1024 * 1024
MAX_TOOL_OUTPUT = 512 * 1024
NONTERMINAL = {"Preparing", "Queueing", "Processing"}
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl): return None
NO_REDIRECT = urllib.request.build_opener(NoRedirect)
class APIProblem(Exception):
def __init__(self, kind, http_status=None, body_sha256=None):
super().__init__(kind); self.kind = kind; self.http_status = http_status; self.body_sha256 = body_sha256
def canonical(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
def sha256_bytes(value): return hashlib.sha256(value).hexdigest()
def utc_now(): return datetime.now(timezone.utc).isoformat()
def fsync_directory(path):
if os.name != "nt":
fd = os.open(path, os.O_RDONLY)
try: os.fsync(fd)
finally: os.close(fd)
def image_dimensions(data, suffix):
if suffix == ".png" and data[:8] == b"\x89PNG\r\n\x1a\n" and data[12:16] == b"IHDR":
return int.from_bytes(data[16:20], "big"), int.from_bytes(data[20:24], "big")
if suffix in {".jpg", ".jpeg"} and data[:2] == b"\xff\xd8":
i = 2
while i + 9 < len(data):
if data[i] != 0xff: i += 1; continue
marker = data[i + 1]; i += 2
if marker in {0xd8, 0xd9} or 0xd0 <= marker <= 0xd7: continue
if i + 2 > len(data): break
length = int.from_bytes(data[i:i + 2], "big")
if marker in {0xc0, 0xc1, 0xc2, 0xc3, 0xc5, 0xc6, 0xc7, 0xc9, 0xca, 0xcb, 0xcd, 0xce, 0xcf}:
return int.from_bytes(data[i + 5:i + 7], "big"), int.from_bytes(data[i + 3:i + 5], "big")
if length < 2: break
i += length
if suffix == ".webp" and data[:4] == b"RIFF" and data[8:12] == b"WEBP":
kind = data[12:16]
if kind == b"VP8X" and len(data) >= 30:
return 1 + int.from_bytes(data[24:27], "little"), 1 + int.from_bytes(data[27:30], "little")
if kind == b"VP8L" and len(data) >= 25 and data[20] == 0x2f:
b0, b1, b2, b3 = data[21:25]
return 1 + b0 + ((b1 & 0x3f) << 8), 1 + (b1 >> 6) + (b2 << 2) + ((b3 & 0x0f) << 10)
if kind == b"VP8 " and len(data) >= 30 and data[23:26] == b"\x9d\x01\x2a":
return int.from_bytes(data[26:28], "little") & 0x3fff, int.from_bytes(data[28:30], "little") & 0x3fff
raise SystemExit("Image signature or dimensions are invalid/unsupported")
def atomic_json(path, value):
path = pathlib.Path(path); path.parent.mkdir(parents=True, exist_ok=True)
fd, name = tempfile.mkstemp(prefix=".hailuo-", suffix=".json", dir=path.parent)
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "wb") as out:
out.write(canonical(value) + b"\n"); out.flush(); os.fsync(out.fileno())
os.replace(name, path); fsync_directory(path.parent)
finally:
if os.path.exists(name): os.unlink(name)
def exclusive_json(path, value):
path = pathlib.Path(path); path.parent.mkdir(parents=True, exist_ok=True)
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600)
with os.fdopen(fd, "wb") as out:
out.write(canonical(value) + b"\n"); out.flush(); os.fsync(out.fileno())
fsync_directory(path.parent)
def read_json(path):
raw = pathlib.Path(path).read_bytes()
if len(raw) > MAX_JSON: raise SystemExit("Ledger exceeds JSON cap")
value = json.loads(raw)
if not isinstance(value, dict): raise SystemExit("Ledger is not an object")
return value
def bounded_process(argv, timeout, cap=MAX_TOOL_OUTPUT):
process = subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
chunks, total, overflow, lock = [], [0], threading.Event(), threading.Lock()
def drain(stream):
while True:
data = stream.read(65536)
if not data: return
with lock:
total[0] += len(data)
if total[0] > cap:
overflow.set(); process.kill(); return
chunks.append(data)
threads = [threading.Thread(target=drain, args=(stream,), daemon=True) for stream in (process.stdout, process.stderr)]
for thread in threads: thread.start()
try: code = process.wait(timeout=timeout)
except subprocess.TimeoutExpired:
process.kill(); process.wait(); raise RuntimeError(f"tool timeout: {pathlib.Path(argv[0]).name}")
for thread in threads: thread.join(timeout=2)
output = b"".join(chunks)
if overflow.is_set(): raise RuntimeError(f"tool output cap exceeded: {pathlib.Path(argv[0]).name}")
if code: raise RuntimeError(f"tool failed: {pathlib.Path(argv[0]).name}; diagnostics_sha256={sha256_bytes(output)}")
return output
def media(path_text):
path = pathlib.Path(path_text)
if not path.is_file() or path.suffix.lower() not in MIME or path.stat().st_size >= 20 * 1024 * 1024:
raise SystemExit("Image must be JPG/JPEG, PNG, or WebP and strictly under 20 MB")
data = path.read_bytes()
width, height = image_dimensions(data, path.suffix.lower())
ratio = width / height
if min(width, height) <= 300 or not 0.4 <= ratio <= 2.5:
raise SystemExit("Image short edge must exceed 300 px and aspect ratio must be within 2:5–5:2")
bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 60)
return {
"wire": f"data:{MIME[path.suffix.lower()]};base64,{base64.b64encode(data).decode('ascii')}",
"evidence": {"path": str(path.resolve()), "sha256": hashlib.sha256(data).hexdigest(),
"bytes": len(data), "mime": MIME[path.suffix.lower()],
"width": width, "height": height, "aspect_ratio": ratio}
}
def api_json(method, url, key, body=None, retries=0):
encoded = None if body is None else canonical(body)
for attempt in range(retries + 1):
req = urllib.request.Request(url, data=encoded, method=method,
headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json", "Accept": "application/json"})
try:
with NO_REDIRECT.open(req, timeout=40) as response:
raw = response.read(MAX_JSON + 1)
if len(raw) > MAX_JSON: raise APIProblem("response-too-large")
try: result = json.loads(raw)
except json.JSONDecodeError: raise APIProblem("invalid-json", response.status, sha256_bytes(raw))
if not isinstance(result, dict): raise APIProblem("non-object-json", response.status, sha256_bytes(raw))
return result
except urllib.error.HTTPError as exc:
raw = exc.read(MAX_JSON + 1)
problem = APIProblem("http-error", exc.code, sha256_bytes(raw))
if attempt == retries or exc.code not in {408, 429, 500, 502, 503, 504}: raise problem
except APIProblem: raise
except (urllib.error.URLError, TimeoutError, OSError) as exc:
if attempt == retries: raise APIProblem("transport-" + type(exc).__name__)
time.sleep(min(8, 2 ** attempt) * (1 + random.random() * 0.25))
raise AssertionError("unreachable")
def provider_result(result):
base = result.get("base_resp")
if not isinstance(base, dict) or not isinstance(base.get("status_code"), int):
raise APIProblem("missing-provider-envelope", body_sha256=sha256_bytes(canonical(result)))
return base["status_code"], sha256_bytes(str(base.get("status_msg", "")).encode("utf-8"))
def exact_hosts(raw):
hosts = []
for item in raw.split(","):
host = item.strip().rstrip(".").lower()
if not host: continue
if "*" in host or "/" in host or ":" in host: raise SystemExit("Output hosts must be exact DNS names")
try: ipaddress.ip_address(host)
except ValueError: pass
else: raise SystemExit("Output host policy rejects IP literals")
hosts.append(host)
if not hosts: raise SystemExit("Provide at least one reviewed exact --output-host")
return tuple(sorted(set(hosts)))
def validate_output_url(url, allowed_hosts):
parsed = urllib.parse.urlsplit(url)
if parsed.scheme != "https" or not parsed.hostname or parsed.username or parsed.password or parsed.fragment:
raise RuntimeError("Output URL must be HTTPS with hostname, no userinfo, and no fragment")
if parsed.port not in (None, 443): raise RuntimeError("Output URL must use default HTTPS port")
host = parsed.hostname.rstrip(".").lower()
try: ipaddress.ip_address(host)
except ValueError: pass
else: raise RuntimeError("Output URL rejects IP literals")
if host not in allowed_hosts: raise RuntimeError("Output host is outside the approved exact-host policy")
addresses = []
for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM):
address = ipaddress.ip_address(item[4][0])
if not address.is_global: raise RuntimeError("Output host resolves to a non-public address")
addresses.append(str(address))
if not addresses: raise RuntimeError("Output host did not resolve")
target = parsed.path or "/"
if parsed.query: target += "?" + parsed.query
return host, target, sorted(set(addresses))
class PinnedHTTPSConnection(http.client.HTTPSConnection):
def __init__(self, host, address, timeout):
super().__init__(host, 443, timeout=timeout, context=ssl.create_default_context()); self.address = address
def connect(self):
raw = socket.create_connection((self.address, 443), self.timeout)
try: self.sock = self._context.wrap_socket(raw, server_hostname=self.host)
except Exception: raw.close(); raise
def verify_mp4(path):
path = pathlib.Path(path); size = path.stat().st_size
if size <= 0 or size > MAX_VIDEO: raise RuntimeError("Artifact size is outside policy")
digest = hashlib.sha256()
with path.open("rb") as stream:
header = stream.read(32)
if len(header) < 12 or header[4:8] != b"ftyp": raise RuntimeError("Artifact lacks ISO-BMFF ftyp signature")
stream.seek(0)
for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk)
raw = bounded_process(["ffprobe", "-v", "error", "-show_streams", "-show_format", "-of", "json", str(path)], 30, MAX_JSON)
metadata = json.loads(raw); streams = metadata.get("streams", [])
video = next((item for item in streams if item.get("codec_type") == "video"), None)
if not video: raise RuntimeError("Artifact has no video stream")
bounded_process(["ffmpeg", "-nostdin", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
return {"bytes": size, "sha256": digest.hexdigest(), "qa": {
"container": metadata.get("format", {}).get("format_name"), "video_codec": video.get("codec_name"),
"width": video.get("width"), "height": video.get("height"), "fps": video.get("avg_frame_rate"),
"duration": metadata.get("format", {}).get("duration"),
"audio_present": any(item.get("codec_type") == "audio" for item in streams), "full_decode": "passed"}}
def safe_download(url, destination, allowed_hosts, ledger_path, record):
destination = pathlib.Path(destination); destination.parent.mkdir(parents=True, exist_ok=True)
stage = destination.with_name("." + destination.name + ".hailuo-stage")
expected = record.get("artifact_staged") or record.get("artifact")
if destination.exists():
if not isinstance(expected, dict): raise RuntimeError("Existing artifact has no durable expected digest")
artifact = verify_mp4(destination)
if (artifact["sha256"], artifact["bytes"]) != (expected.get("sha256"), expected.get("bytes")):
raise RuntimeError("Existing artifact does not match durable evidence")
return artifact
if stage.exists() and isinstance(expected, dict):
artifact = verify_mp4(stage)
if (artifact["sha256"], artifact["bytes"]) != (expected.get("sha256"), expected.get("bytes")):
raise RuntimeError("Staged artifact does not match durable evidence")
os.replace(stage, destination); fsync_directory(destination.parent); return artifact
host, target, addresses = validate_output_url(url, allowed_hosts)
url_hash = sha256_bytes(url.encode("utf-8"))
if record.get("download_url_sha256") not in (None, url_hash): raise RuntimeError("Download URL changed during an incomplete transfer")
record.update(state="downloading", download_url_sha256=url_hash, download_host=host, download_started_at=record.get("download_started_at") or utc_now())
atomic_json(ledger_path, record)
if stage.exists(): stage.unlink()
connection = PinnedHTTPSConnection(host, addresses[0], timeout=90)
try:
connection.request("GET", target, headers={"Host": host, "Accept": "video/mp4", "User-Agent": "hailuo-artifact-client/2"})
response = connection.getresponse()
if response.status != 200: raise RuntimeError(f"Output returned HTTP {response.status}")
media_type = response.getheader("Content-Type", "").split(";", 1)[0].lower()
if media_type not in {"video/mp4", "application/octet-stream"}: raise RuntimeError("Unexpected output media type")
announced = response.getheader("Content-Length")
if announced is not None and (not announced.isdigit() or int(announced) > MAX_VIDEO): raise RuntimeError("Invalid or excessive Content-Length")
fd = os.open(stage, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600); total = 0
try:
with os.fdopen(fd, "wb") as out:
while True:
chunk = response.read(1024 * 1024)
if not chunk: break
total += len(chunk)
if total > MAX_VIDEO: raise RuntimeError("Download exceeded byte cap")
out.write(chunk)
out.flush(); os.fsync(out.fileno())
except BaseException:
stage.unlink(missing_ok=True); raise
finally: connection.close()
artifact = verify_mp4(stage)
record.update(state="artifact-staged", artifact_staged=artifact, artifact_validated_at=utc_now())
atomic_json(ledger_path, record)
os.replace(stage, destination); fsync_directory(destination.parent)
return artifact
def main():
p = argparse.ArgumentParser()
p.add_argument("--region", choices=HOSTS, required=True)
p.add_argument("--mode", choices=MODELS, required=True)
p.add_argument("--model", required=True); p.add_argument("--prompt", required=True)
p.add_argument("--resolution", choices=["512P", "768P", "1080P"], required=True)
p.add_argument("--duration", type=int, choices=[6, 10], required=True)
p.add_argument("--first-frame"); p.add_argument("--last-frame")
p.add_argument("--optimizer", action=argparse.BooleanOptionalAction, default=False)
p.add_argument("--fast-pretreatment", action="store_true")
p.add_argument("--execute", action="store_true")
p.add_argument("--job-key", required=True)
p.add_argument("--output-host", required=True, help="Comma-separated reviewed exact DNS hostnames")
p.add_argument("--pricing-checked-at", required=True, help="Timezone-aware ISO-8601 time of the live regional price check")
p.add_argument("--pricing-evidence-sha256", required=True)
p.add_argument("--rights-sha256", required=True); p.add_argument("--moderation-sha256", required=True)
p.add_argument("--qa-plan-sha256", required=True)
p.add_argument("--approval-sha256")
p.add_argument("--max-cost"); p.add_argument("--max-cost-currency", choices=["USD", "CNY"])
p.add_argument("--show-full-plan", action="store_true")
p.add_argument("--ledger", default="hailuo-attempt.json")
p.add_argument("--output", default="hailuo-output.mp4")
a = p.parse_args()
if not a.job_key or any(not (character.isalnum() or character in "-_") for character in a.job_key):
raise SystemExit("job-key accepts letters, digits, hyphen, and underscore")
if a.model not in MODELS[a.mode]: raise SystemExit("Model is not valid for the selected mode")
if len(a.prompt) > 2000: raise SystemExit("Prompt exceeds 2,000 characters")
if a.fast_pretreatment and not a.optimizer: raise SystemExit("fast_pretreatment requires optimizer")
if a.mode in {"image", "first-last"} and not a.first_frame: raise SystemExit("First frame required")
if a.mode == "first-last" and not a.last_frame: raise SystemExit("Last frame required")
key = (a.model, a.resolution, a.duration)
unit = PRICES[a.region]["table"].get(key)
if unit is None: raise SystemExit("UNKNOWN current price/combination; verify in the correct MiniMax console")
if a.model == "MiniMax-Hailuo-02" and a.mode == "text" and a.resolution == "512P":
raise SystemExit("512P pricing/availability is documented for Hailuo-02 I2V, not T2V")
if a.mode == "first-last" and a.resolution == "512P":
raise SystemExit("The first/last-frame endpoint explicitly does not support 512P")
request = {"model": a.model, "prompt": a.prompt, "resolution": a.resolution,
"duration": a.duration, "prompt_optimizer": a.optimizer}
evidence = {}
if a.mode in {"image", "first-last"}:
item = media(a.first_frame); request["first_frame_image"] = item["wire"]; evidence["first_frame"] = item["evidence"]
if a.mode == "first-last":
item = media(a.last_frame); request["last_frame_image"] = item["wire"]; evidence["last_frame"] = item["evidence"]
if a.fast_pretreatment: request["fast_pretreatment"] = True
currency = PRICES[a.region]["currency"]
output_hosts = exact_hosts(a.output_host)
for name, value in (("pricing evidence", a.pricing_evidence_sha256), ("rights", a.rights_sha256),
("moderation", a.moderation_sha256), ("QA plan", a.qa_plan_sha256)):
if len(value) != 64 or any(character not in "0123456789abcdefABCDEF" for character in value):
raise SystemExit(f"{name} must be a SHA-256 digest")
redacted = dict(request)
for field in ("first_frame_image", "last_frame_image"):
if field in redacted: redacted[field] = f""
redacted["prompt"] = "sha256:" + sha256_bytes(a.prompt.encode("utf-8"))
request_hash = sha256_bytes(canonical(request))
output_policy = {"exact_hosts": list(output_hosts), "https": True, "redirects": "reject",
"reject_ip_literals": True, "public_dns": True,
"dns_pinning": "connect-IP-with-TLS-SNI-and-host-certificate",
"max_bytes": MAX_VIDEO, "container": "mp4", "full_decode": True}
envelope = {"schema_version": 1, "job_key": a.job_key, "region": a.region, "api_origin": HOSTS[a.region],
"endpoint": "/v1/video_generation", "mode": a.mode, "model": a.model,
"request_sha256": request_hash, "redacted_request": redacted, "source_evidence": evidence,
"output": str(pathlib.Path(a.output).resolve()), "output_policy": output_policy,
"price": {"amount": unit, "currency": currency, "checked_at": a.pricing_checked_at,
"evidence_sha256": a.pricing_evidence_sha256.lower()},
"governance": {"rights_sha256": a.rights_sha256.lower(),
"moderation_sha256": a.moderation_sha256.lower(),
"qa_plan_sha256": a.qa_plan_sha256.lower()}}
approval = sha256_bytes(canonical(envelope))
print(json.dumps({"approval_sha256": approval, "plan": envelope}, indent=2, ensure_ascii=False))
if a.show_full_plan: print(json.dumps({"protected_payload": request}, indent=2, ensure_ascii=False))
print(f"DRY RUN: one task, exact maximum {currency} {unit}; no network request sent", file=sys.stderr)
if not a.execute: return
if a.approval_sha256 != approval: raise SystemExit("Exact approval digest mismatch")
try: maximum = Decimal(a.max_cost or "")
except InvalidOperation: raise SystemExit("max-cost must be a decimal")
if not maximum.is_finite() or maximum <= 0 or maximum < Decimal(unit):
raise SystemExit("max-cost must be finite, positive, and at least the quoted amount")
if a.max_cost_currency != currency: raise SystemExit("max-cost currency does not match the selected region")
try: checked_at = datetime.fromisoformat(a.pricing_checked_at.replace("Z", "+00:00"))
except ValueError: raise SystemExit("pricing-checked-at must be ISO-8601")
if checked_at.tzinfo is None: raise SystemExit("pricing-checked-at must include a timezone")
age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
if age.total_seconds() < -300 or age.total_seconds() > 86400:
raise SystemExit("Live regional pricing evidence must be no more than 24 hours old")
api_key = os.environ.get("MINIMAX_API_KEY")
if not api_key: raise SystemExit("Set MINIMAX_API_KEY only in the environment")
ledger_path = pathlib.Path(a.ledger)
if ledger_path.exists():
intent = read_json(ledger_path)
if intent.get("request_sha256") != request_hash or intent.get("approval_sha256") != approval:
raise SystemExit("Existing job ledger belongs to a different approved request")
task_id = intent.get("task_id")
if not task_id: raise SystemExit(f"Ledger state {intent.get('state')!r} has no task ID; never replay this job key")
else:
intent = {"schema_version": 1, "state": "intent-recorded", "created_at": utc_now(),
"job_key": a.job_key, "request_sha256": request_hash, "approval_sha256": approval,
"region": a.region, "host": HOSTS[a.region], "mode": a.mode, "model": a.model,
"redacted_request": redacted, "source_evidence": evidence, "output_policy": output_policy,
"price": envelope["price"], "governance": envelope["governance"]}
try: exclusive_json(ledger_path, intent)
except FileExistsError: raise SystemExit("Another process acquired this job ledger")
intent.update(state="create-started", create_started_at=utc_now()); atomic_json(ledger_path, intent)
try: created = api_json("POST", HOSTS[a.region] + "/v1/video_generation", api_key, request, retries=0)
except APIProblem as exc:
intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(), error_kind=exc.kind,
http_status=exc.http_status, error_body_sha256=exc.body_sha256, replay_allowed=False)
atomic_json(ledger_path, intent)
raise SystemExit("Create outcome unknown; do not replay until billing/console/support is reconciled")
try: provider_code, provider_message_sha256 = provider_result(created)
except APIProblem as exc:
intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(), error_kind=exc.kind,
response_sha256=exc.body_sha256, replay_allowed=False); atomic_json(ledger_path, intent)
raise SystemExit("Create success envelope is unusable; do not replay")
if provider_code != 0:
uncertain = provider_code in {1001, 1024, 1033}
intent.update(state="create-outcome-unknown" if uncertain else "create-rejected",
create_outcome_at=utc_now(), provider_status_code=provider_code,
provider_status_message_sha256=provider_message_sha256, replay_allowed=False)
atomic_json(ledger_path, intent)
raise SystemExit("MiniMax rejected or could not confirm create; this job key cannot be replayed")
task_id = created.get("task_id")
if not isinstance(task_id, (str, int)) or not str(task_id).isdigit():
intent.update(state="create-outcome-unknown", create_outcome_at=utc_now(),
error_kind="unusable-success-task-id", response_sha256=sha256_bytes(canonical(created)),
replay_allowed=False); atomic_json(ledger_path, intent)
raise SystemExit("Create returned an unusable task ID; do not replay")
task_id = str(task_id)
intent.update(state="task-created", task_id=task_id, submitted_at=utc_now(),
provider_status_code=0, provider_status_message_sha256=provider_message_sha256)
atomic_json(ledger_path, intent)
deadline = time.monotonic() + 30 * 60
while time.monotonic() < deadline:
query = HOSTS[a.region] + "/v1/query/video_generation?" + urllib.parse.urlencode({"task_id": task_id})
try: status = api_json("GET", query, api_key, retries=3)
except APIProblem as exc:
intent.update(poll_error_kind=exc.kind, poll_http_status=exc.http_status,
poll_error_body_sha256=exc.body_sha256); atomic_json(ledger_path, intent); raise SystemExit("Polling failed; resume this task ID later")
provider_code, provider_message_sha256 = provider_result(status)
if provider_code != 0:
intent.update(poll_provider_status_code=provider_code,
poll_provider_status_message_sha256=provider_message_sha256); atomic_json(ledger_path, intent)
raise SystemExit("MiniMax query returned a provider error; resume after diagnosis")
if status.get("task_id") not in (None, task_id, int(task_id)):
raise SystemExit("Query task ID mismatch")
name = status.get("status")
intent.update(state="polling", last_status=name, last_polled_at=utc_now()); atomic_json(ledger_path, intent)
if name == "Success":
file_id = status.get("file_id")
if not isinstance(file_id, (str, int)) or not str(file_id).isdigit(): raise RuntimeError("Success returned invalid file_id")
file_id = str(file_id)
retrieve = HOSTS[a.region] + "/v1/files/retrieve?" + urllib.parse.urlencode({"file_id": file_id})
retrieved = api_json("GET", retrieve, api_key, retries=3)
retrieve_code, retrieve_message_sha256 = provider_result(retrieved)
if retrieve_code != 0: raise RuntimeError("MiniMax retrieve returned a provider error")
file_record = retrieved.get("file") or {}
if str(file_record.get("file_id")) not in {"None", file_id}: raise RuntimeError("Retrieve file ID mismatch")
url = file_record.get("download_url") or retrieved.get("download_url")
if not url: raise RuntimeError("Retrieve returned no download_url")
artifact = safe_download(url, a.output, set(output_hosts), ledger_path, intent)
expected_dimensions = (status.get("video_width"), status.get("video_height"))
actual_dimensions = (artifact["qa"].get("width"), artifact["qa"].get("height"))
try: actual_duration = Decimal(s
…(truncated)