视频完整代码与下载
下面的客户端覆盖“提交一次、保存 ID、轮询、下载”，适用于本手册的 7 个视频型号。选择型号通过请求 JSON 文件完成，程序不替你把不同型号的参数强行转换成同一套。
下载文件
| 文件 | 用途 |
|---|---|
| mxb-video.py | Python 3.10+ 标准库，无需 pip 依赖 |
| mxb-video.mjs | Node.js 22+，无需 npm 依赖 |
| mxb-upload.py | 本地参考图片、视频、音频上传 |
| grok-6s-text.json | Grok 固定 6 秒 |
| grok-6s-v15-text.json | Grok 1.5 固定 6 秒 |
| grok-variable-text.json | grok-video-1.5 |
| seedance-mini-text.json | Seedance Mini，480p / 4 秒 |
| seedance-fast-text.json | Seedance Fast，720p / 5 秒 |
| sd-text.json | SD-2.0-MZ1，720p / 5 秒 |
| h3-text.json | MiniMax-H3，768P / 5 秒 |
旧 mxb-seedance.py 地址保留，现只提供当前 Mini/Fast 默认型号，仍可通过 --request 指定 JSON。新接入建议使用通用客户端。
准备 Key 和文件
将脚本与选中的 JSON 下载到同一目录。先修改 JSON 的 prompt，再按鉴权与地址设置自己的 Key。用 GET /v1/models 核对权限，不通过重复生成检查权限。
Bash 安全输入示例：
bash
read -r -s -p 'API Key: ' MXB_API_KEY
export MXB_API_KEY
printf '\n'
PowerShell：
powershell
$secureKey = Read-Host 'API Key' -AsSecureString
$env:MXB_API_KEY = [System.Net.NetworkCredential]::new('', $secureKey).Password
不要将真实 Key 写进下载的源文件、请求 JSON、命令历史或截图。不同任务使用不同的 --state 路径。
Python：新建与续查
bash
python mxb-video.py --request grok-6s-text.json --state my-video.json --output my-video.mp4
新建时必须提供 --request。程序在 POST 前创建状态文件，拿到 ID 后更新它。再次运行同一命令会读取已保存的 ID，只查询和下载，不再次生成：
bash
python mxb-video.py --state my-video.json --output my-video.mp4
已有真实任务 ID 时：
bash
python mxb-video.py --task "$MXB_VIDEO_TASK_ID" --output my-video.mp4
总等待时间可加 --wait 1800，单位为秒。切换备用本站入口可加 --base https://openpix.cc；已有状态文件必须继续使用原入口。
Node.js：相同用法
bash
node mxb-video.mjs --request h3-text.json --state h3-task.json --output h3.mp4
node mxb-video.mjs --state h3-task.json --output h3.mp4
node mxb-video.mjs --task "$MXB_VIDEO_TASK_ID" --output h3.mp4
第二行、第三行是恢复方式，按实际情况选用。不要为同一业务同时启动两个新建命令。Node 与 Python 使用兼容的状态文件结构，但不要并发操作同一个状态文件或输出文件。
Windows PowerShell 运行
Python / Node 命令本身也可直接在 PowerShell 中运行。需要使用环境变量时写 $env:MXB_VIDEO_TASK_ID。
只需要原始请求时，用 curl.exe 加 JSON 文件，避免内联中文和引号转义：
powershell
curl.exe --fail-with-body --max-time 120 "https://openpix.cc/v1/videos" -H "Authorization: Bearer $env:MXB_API_KEY" -H "Content-Type: application/json" --data-binary "@sd-text.json"
此命令只创建任务，不会自动轮询。手动保存返回 ID 后按任务章节查询；生产接入优先使用上面的持久化脚本。
上传本地素材后生成
bash
python mxb-upload.py --kind image --file reference.png
python mxb-upload.py --kind video --model dreamina-seedance-2-0-mini-260615 --file reference.mp4
python mxb-upload.py --kind audio --model SD-2.0-MZ1 --file reference.wav
按需要执行其中一条，复制返回 url 到该型号正确的参考字段，保存新的 JSON，再执行一次生成。上传失败就停止，不静默去掉参考素材。完整限制见素材上传。
状态文件与恢复原则
| 文件/情况 | 处理 |
|---|---|
| 状态文件存在且有 task_id | 继续查询此 ID，修改请求 JSON 不会创建新任务 |
| status=submission_unknown 且没有 ID | 提交结果不明，先查控制台；不能删除文件后盲目重跑 |
| 本地等待到期 | 任务可能仍在执行，保留状态文件续查 |
| 下载失败 | 使用原 ID 重试下载，不能新建任务 |
| 确定要创建新任务 | 明确完成核查后，使用新的 --state 文件名 |
| .part 文件 | 临时下载，失败时清理；不能当完整视频交付 |
示例仅把 API Key 放在内存，不写入状态文件。错误退出不会打印原始响应、完整请求头或带签名的媒体链接。为方便排障，业务系统应安全记录任务 ID 和公开错误摘要。
示例的验证范围与生产补充
已对各型号 JSON 和文档中的完整生成请求做当前契约校验，并用离线响应验证提交去重、续查、失败和下载错误处理。本次文档发布没有调用收费生成接口。
客户端会检查任务身份、状态和 MP4 文件头，拒绝携带 Key 自动重定向，并设置示例下载上限 512 MiB。文件头检查不等于完整解码；生产系统还应使用媒体工具验证整个文件，加入任务队列、数据库事务、监控、随机退避和账单核对。示例状态文件用于单机单任务，不替代分布式并发控制。
Python 完整源码
python
"""Python 3.10+; submit once, resume a saved task, download from this site's API."""
import argparse
import json
import os
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None # Never forward the API key to another URL.
class ApiError(Exception):
def __init__(self, status):
self.status = status
super().__init__(f"HTTP {status}; check the task and usage logs")
class VideoClient:
def __init__(self, key, base="https://openpix.cc"):
if base.rstrip("/") not in ("https://openpix.cc", "https://openpix.cc"):
raise ValueError("Use one of the documented site origins")
if not key or "\n" in key or "\r" in key:
raise ValueError("Set MXB_API_KEY in your server environment")
self.base, self.key = base.rstrip("/"), key
self.opener = urllib.request.build_opener(NoRedirect())
def open(self, path, payload=None, timeout=30):
data = None if payload is None else json.dumps(payload).encode("utf-8")
headers = {"Authorization": "Bearer " + self.key}
if data is not None:
headers["Content-Type"] = "application/json"
request = urllib.request.Request(self.base + path, data=data, headers=headers)
try:
return self.opener.open(request, timeout=timeout)
except urllib.error.HTTPError as error:
error.close()
raise ApiError(error.code) from None
def json(self, path, payload=None):
with self.open(path, payload, timeout=120 if payload is not None else 30) as response:
result = json.load(response)
if not isinstance(result, dict):
raise ValueError("Unexpected API response; check logs before submitting again")
return result
@staticmethod
def task_path(task):
if not isinstance(task, str) or not re.fullmatch(r"[A-Za-z0-9_-]{1,200}", task):
raise ValueError("Invalid task ID")
return "/v1/videos/" + urllib.parse.quote(task, safe="")
def wait(self, task, seconds):
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
try:
result = self.json(self.task_path(task))
except (urllib.error.URLError, TimeoutError):
time.sleep(5)
continue
except ApiError as error:
if error.status not in (429, 500, 502, 503, 504):
raise
time.sleep(5)
continue
if result.get("id") != task:
raise ValueError("Response task ID mismatch; saved task remains available")
status = str(result.get("status", "")).lower()
if status in ("completed", "succeeded", "success", "done"):
return result
if status in ("failed", "failure", "error", "cancelled", "canceled", "expired"):
raise ValueError("Task failed; inspect its error and refund in the site logs")
if status not in ("queued", "pending", "submitted", "in_progress", "running", "processing"):
raise ValueError("Unknown task state; keep the ID and check site logs")
time.sleep(5)
raise TimeoutError("Waiting stopped; run again with the saved state or --task to resume")
def download(self, task, output):
temporary = output.with_name(output.name + ".part")
try:
with self.open(self.task_path(task) + "/content", timeout=180) as response:
if response.status != 200:
raise ValueError("Unexpected download response")
first = response.read(32)
if b"ftyp" not in first:
raise ValueError("The content response is not an MP4 file")
with temporary.open("wb") as stream:
stream.write(first)
size = len(first)
while chunk := response.read(1024 * 1024):
size += len(chunk)
if size > 512 * 1024 * 1024:
raise ValueError("Download exceeds the example's 512 MiB limit")
stream.write(chunk)
temporary.replace(output)
except BaseException:
temporary.unlink(missing_ok=True)
raise
def get_task(client, state, payload):
if state.exists():
saved = json.loads(state.read_text(encoding="utf-8"))
if saved.get("base") != client.base:
raise ValueError("Saved origin differs; use the same --base as the original task")
if not saved.get("task_id"):
raise ValueError("Submission result unknown: check site logs before creating another task")
client.task_path(saved["task_id"])
return saved["task_id"]
# Reserve the state file BEFORE sending POST. A crash must not trigger a duplicate.
with state.open("x", encoding="utf-8") as stream:
json.dump({"base": client.base, "status": "submission_unknown"}, stream)
stream.flush()
os.fsync(stream.fileno())
result = client.json("/v1/videos", payload)
task = result.get("id")
client.task_path(task)
print("Accepted task:", task, flush=True)
temporary = state.with_name(state.name + ".tmp")
temporary.write_text(json.dumps({"base": client.base, "task_id": task}), encoding="utf-8")
temporary.replace(state)
return task
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", default="https://openpix.cc")
parser.add_argument("--request", type=Path, help="UTF-8 JSON request file")
parser.add_argument("--state", type=Path, default=Path("video-task.json"))
parser.add_argument("--task", help="Existing task ID; never submits a generation")
parser.add_argument("--output", type=Path, default=Path("video-result.mp4"))
parser.add_argument("--wait", type=int, default=1800, help="Polling deadline in seconds")
args = parser.parse_args()
if args.wait <= 0:
raise ValueError("--wait must be positive")
client = VideoClient(os.environ.get("MXB_API_KEY", ""), args.base)
payload = None
if not args.task and not args.state.exists():
if not args.request:
raise ValueError("New tasks require --request with a model-specific JSON file")
payload = json.loads(args.request.read_text(encoding="utf-8-sig"))
if not isinstance(payload, dict):
raise ValueError("Request file must contain a JSON object")
task = args.task or get_task(client, args.state, payload)
client.task_path(task)
print("Querying task:", task, flush=True)
client.wait(task, args.wait)
client.download(task, args.output)
print("Saved video:", args.output)
if __name__ == "__main__":
try:
main()
except (OSError, ValueError, ApiError) as error:
# Do not dump request headers, raw API responses or signed media addresses.
print(f"Stopped ({type(error).__name__}). Keep the task state and check site logs; do not auto-resubmit.", file=sys.stderr)
sys.exit(1)
Node.js 完整源码
javascript
// Node.js 22+. No dependency, no automatic POST retry.
// Keep this state file; an unknown submission must be checked before retrying.
import { readFile, writeFile, rename, open, unlink, mkdir } from 'node:fs/promises';
import { dirname, resolve } from 'node:path';
import { pathToFileURL } from 'node:url';
import { parseArgs } from 'node:util';
export class ApiError extends Error {
constructor(status) { super('HTTP ' + status + '; check task and usage logs'); this.status = status; }
}
const sleep = ms => new Promise(resolve => setTimeout(resolve, ms));
export class VideoClient {
constructor(key, base = 'https://openpix.cc', transport = fetch) {
base = base.replace(/\/$/, '');
if (!['https://openpix.cc', 'https://openpix.cc'].includes(base)) throw new Error('Invalid API origin');
if (!key || /[\r\n]/.test(key)) throw new Error('Set MXB_API_KEY');
this.key = key; this.base = base; this.transport = transport;
}
taskPath(task) {
if (typeof task !== 'string' || !/^[A-Za-z0-9_-]{1,200}$/.test(task)) throw new Error('Invalid task ID');
return '/v1/videos/' + encodeURIComponent(task);
}
async request(path, payload, timeout = 30000) {
const response = await this.transport(this.base + path, {
method: payload === undefined ? 'GET' : 'POST',
headers: { Authorization: 'Bearer ' + this.key, ...(payload === undefined ? {} : { 'Content-Type': 'application/json' }) },
body: payload === undefined ? undefined : JSON.stringify(payload),
redirect: 'error', signal: AbortSignal.timeout(timeout),
});
if (!response.ok) { await response.body?.cancel(); throw new ApiError(response.status); }
return response;
}
async json(path, payload) {
const response = await this.request(path, payload, payload === undefined ? 30000 : 120000);
const value = await response.json();
if (!value || typeof value !== 'object' || Array.isArray(value)) throw new Error('Unexpected response');
return value;
}
async wait(task, seconds, pause = sleep, now = () => performance.now()) {
const deadline = now() + seconds * 1000;
let delay = 5000;
while (now() < deadline) {
let value;
try { value = await this.json(this.taskPath(task)); }
catch (error) {
const transient = error instanceof ApiError ? [429,500,502,503,504].includes(error.status)
: error instanceof TypeError || ['TimeoutError','AbortError'].includes(error.name);
if (!transient) throw error;
await pause(Math.min(delay, Math.max(0, deadline - now())));
delay = Math.min(delay * 2, 30000); continue;
}
if (value.id !== task) throw new Error('Response task ID mismatch');
const status = String(value.status || '').toLowerCase();
if (['completed','success','succeeded','done'].includes(status)) return value;
if (['failed','failure','error','canceled','cancelled','expired'].includes(status)) throw new Error('Task failed; check error and bill');
if (!['queued','pending','submitted','in_progress','running','processing'].includes(status)) throw new Error('Unknown state; keep task ID');
delay = 5000;
await pause(Math.min(delay, Math.max(0, deadline - now())));
}
throw new Error('Waiting stopped; resume the saved task');
}
async download(task, output) {
const temporary = output + '.part';
let file;
try {
const response = await this.request(this.taskPath(task) + '/content', undefined, 180000);
if (response.status !== 200 || !response.body) { await response.body?.cancel(); throw new Error('Unexpected download'); }
await mkdir(dirname(resolve(output)), { recursive: true });
file = await open(temporary, 'w');
let size = 0, first = Buffer.alloc(0);
for await (const chunk of response.body) {
size += chunk.length;
if (size > 512 * 1024 * 1024) throw new Error('Example download limit: 512 MiB');
if (first.length < 32) first = Buffer.concat([first, Buffer.from(chunk).subarray(0, 32 - first.length)]);
await file.writeFile(chunk);
}
if (!first.includes(Buffer.from('ftyp'))) throw new Error('Content is not MP4');
await file.sync(); await file.close(); file = null;
await rename(temporary, output);
} catch (error) {
await file?.close().catch(() => {});
await unlink(temporary).catch(() => {});
throw error;
}
}
}
export async function getTask(client, state, payload) {
let saved;
try { saved = JSON.parse(await readFile(state, 'utf8')); }
catch (error) { if (error.code !== 'ENOENT') throw error; }
if (saved !== undefined) {
if (saved?.base !== client.base) throw new Error('Saved origin differs');
if (!saved.task_id) throw new Error('Submission unknown; check site records before another POST');
client.taskPath(saved.task_id); return saved.task_id;
}
if (!payload || typeof payload !== 'object' || Array.isArray(payload)) throw new Error('Provide a request JSON object');
const handle = await open(state, 'wx');
try {
await handle.writeFile(JSON.stringify({ base: client.base, status: 'submission_unknown' }));
await handle.sync();
} finally { await handle.close(); }
const result = await client.json('/v1/videos', payload);
const task = result.id; client.taskPath(task);
console.log('Accepted task:', task);
await writeFile(state + '.tmp', JSON.stringify({ base: client.base, task_id: task }));
await rename(state + '.tmp', state);
return task;
}
async function main() {
const { values } = parseArgs({ options: {
base: { type: 'string', default: 'https://openpix.cc' },
request: { type: 'string' }, task: { type: 'string' },
state: { type: 'string', default: 'video-task.json' },
output: { type: 'string', default: 'video-result.mp4' },
wait: { type: 'string', default: '1800' },
} });
const seconds = Number(values.wait);
if (!Number.isFinite(seconds) || seconds <= 0) throw new Error('--wait must be positive');
const client = new VideoClient(process.env.MXB_API_KEY, values.base);
let payload;
if (!values.task && values.request) payload = JSON.parse((await readFile(values.request, 'utf8')).replace(/^\uFEFF/, ''));
const task = values.task || await getTask(client, values.state, payload);
client.taskPath(task);
console.log('Querying task:', task);
await client.wait(task, seconds);
await client.download(task, values.output);
console.log('Saved video:', values.output);
}
if (process.argv[1] && pathToFileURL(resolve(process.argv[1])).href === import.meta.url) {
main().catch(error => {
console.error('Stopped (' + error.constructor.name + '). Keep the task state and check site logs; do not auto-resubmit.');
process.exitCode = 1;
});
}
上传工具完整源码
python
"""Python 3.10+: upload reference media; this script does not generate a video."""
import argparse
import base64
import json
import os
import sys
import urllib.parse
import urllib.request
from pathlib import Path
ORIGINS = ("https://openpix.cc", "https://openpix.cc")
MODELS = ("dreamina-seedance-2-0-mini-260615", "dreamina-seedance-2-0-fast-260128", "MiniMax-H3", "SD-2.0-MZ1")
MIMES = {
".png": "image/png", ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".webp": "image/webp",
".mp4": "video/mp4", ".mov": "video/quicktime",
".mp3": "audio/mpeg", ".wav": "audio/wav", ".m4a": "audio/mp4", ".aac": "audio/aac",
}
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None
def prepare(kind, model, path):
mime = MIMES.get(path.suffix.lower(), "")
if not mime.startswith(kind + "/"):
raise ValueError("File extension does not match the selected kind")
if kind != "image" and model not in MODELS:
raise ValueError("Video/audio upload requires a documented model")
if model == "SD-2.0-MZ1" and kind == "video":
raise ValueError("This model does not accept reference video")
if kind == "audio" and model != "MiniMax-H3" and path.suffix.lower() not in (".mp3", ".wav"):
raise ValueError("Use MP3 or PCM WAV for this model")
limit = 20 * 1024 * 1024 if kind == "image" else (15_000_000 if kind == "audio" else (50_000_000 if model == "MiniMax-H3" else 200_000_000))
if not 0 < path.stat().st_size <= limit:
raise ValueError("Empty file or exceeds this upload example's size limit")
with path.open("rb") as stream:
data = stream.read(limit + 1)
if len(data) > limit:
raise ValueError("File grew beyond the size limit")
if kind == "image":
data_url = "data:" + mime + ";base64," + base64.b64encode(data).decode("ascii")
return "/v1/video-reference-images", "application/json", json.dumps({"image": data_url}).encode()
return "/v1/video-reference-media/" + urllib.parse.quote(model, safe="") + "/" + kind, mime, data
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--base", choices=ORIGINS, default=ORIGINS[0])
parser.add_argument("--kind", choices=("image", "video", "audio"), required=True)
parser.add_argument("--model", choices=MODELS)
parser.add_argument("--file", type=Path, required=True)
args = parser.parse_args()
key = os.environ.get("MXB_API_KEY", "")
if not key or "\r" in key or "\n" in key:
raise ValueError("Set MXB_API_KEY in your server environment")
path, mime, data = prepare(args.kind, args.model, args.file)
request = urllib.request.Request(args.base + path, data=data, headers={
"Authorization": "Bearer " + key, "Content-Type": mime,
})
with urllib.request.build_opener(NoRedirect()).open(request, timeout=120) as response:
result = json.load(response)
if not isinstance(result, dict) or not isinstance(result.get("url"), str):
raise ValueError("Unexpected upload result")
parsed = urllib.parse.urlsplit(result["url"])
if parsed.username or parsed.password or parsed.scheme + "://" + parsed.netloc not in ORIGINS:
raise ValueError("Unexpected upload URL origin")
# URL is a capability: keep this output private, never publish logs containing it.
print(json.dumps(result, ensure_ascii=False, indent=2))
if __name__ == "__main__":
try:
main()
except (OSError, ValueError) as error:
code = getattr(error, "code", None)
print("Upload stopped" + (" (HTTP " + str(code) + ")" if code else "") + ". Check the file and model rules; no generation was submitted.", file=sys.stderr)
sys.exit(1)