# Choose Python OR Node.js, not both for the same intended task.
python h3_client.py --submit
# Resume later without another creation:
python h3_client.py --task-id <TASK_ID>
# Alternative Node.js implementation:
node h3_client.mjs --submit
node h3_client.mjs --task-id <TASK_ID>
Python · 标准库 完整示例
Python · 标准库
"""H3 example: POST once, persist task ID, resume polling without resubmission.
Environment: XCM_API_KEY, optionally XCM_API_ORIGIN (HTTPS origin, no /v1).
Usage: python h3_client.py --submit OR python h3_client.py --task-id TASK_ID
Submitting generates a billable video. Polling success is not a billing guarantee.
"""
import argparse
import json
import os
from pathlib import Path
import time
import urllib.error
import urllib.parse
import urllib.request
class NoRedirect(urllib.request.HTTPRedirectHandler):
def redirect_request(self, req, fp, code, msg, headers, newurl):
return None # Never forward the API key to a redirected host.
def request(method, path, payload=None):
origin = os.environ.get('XCM_API_ORIGIN', 'https://xcmapi.org').rstrip('/')
parts = urllib.parse.urlsplit(origin)
if parts.scheme != 'https' or not parts.netloc or parts.path or parts.query or parts.fragment or parts.username:
raise ValueError('XCM_API_ORIGIN must be an HTTPS origin without /v1')
key = os.environ['XCM_API_KEY']
req = urllib.request.Request(origin + path, method=method,
headers={'Authorization': 'Bearer ' + key, 'Content-Type': 'application/json'},
data=json.dumps(payload).encode() if payload is not None else None)
try:
with urllib.request.build_opener(NoRedirect).open(req, timeout=30) as response:
data = json.load(response)
return response.status, data, None
except urllib.error.HTTPError as error:
retry = error.headers.get('Retry-After', '')
retry_seconds = min(15, max(3, int(retry))) if retry.isdigit() else None
# Do not print a raw body that might contain user content or credentials.
return error.code, {}, retry_seconds
def unwrap(data):
return data['data'] if isinstance(data.get('data'), dict) else data
def poll(task_id, attempts=120):
path = '/v1/videos/' + urllib.parse.quote(task_id, safe='')
for _ in range(attempts):
try:
status, response, retry = request('GET', path)
except (TimeoutError, urllib.error.URLError):
time.sleep(5)
continue # Retrying GET does not create another video.
if status in (429, 502, 503, 504):
time.sleep(retry or 5)
continue
if status != 200:
raise RuntimeError(f'Query HTTP {status}; keep task ID and check the documentation')
data = unwrap(response)
state = data.get('status')
if state in ('failed', 'error', 'cancelled', 'canceled', 'expired'):
raise RuntimeError('Task ended without success; keep task ID for support')
if state in ('done', 'completed', 'succeeded', 'success'):
return data
if state not in ('pending', 'queued', 'in_progress', 'processing', 'running'):
raise RuntimeError('Unrecognized task status; inspect the response privately')
time.sleep(5)
raise TimeoutError('Polling stopped. Resume using --task-id; do not resubmit the POST')
def download(ident, path):
"""Fetch /content, following at most five HTTPS redirects/JSON links."""
import re
from urllib import request as http, error as errors, parse as urls
if not isinstance(ident, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}', ident):
raise ValueError('Invalid task ID')
if not path.is_absolute():
raise ValueError('Choose an explicit absolute output path')
if path.exists():
raise FileExistsError('Output exists; choose a new path')
def checked_url(value, base=None):
if not isinstance(value, str) or not value or any(ord(c) <= 32 or ord(c) == 127 for c in value) or '\\' in value:
raise ValueError('Invalid download URL')
target = urls.urljoin(base, value) if base else value
p = urls.urlsplit(target)
if p.scheme != 'https' or not p.hostname or p.username is not None or p.password is not None or p.fragment or p.port == 0:
raise ValueError('Download URL must be HTTPS without credentials or fragment')
return target, (p.hostname.lower(), p.port or 443)
origin = os.environ.get('XCM_API_ORIGIN', 'https://xcmapi.org').rstrip('/')
origin, auth_origin = checked_url(origin)
parsed = urls.urlsplit(origin)
if parsed.path or parsed.query:
raise ValueError('Use an HTTPS origin without /v1')
key = os.environ.get('XCM_API_KEY', '')
if not key.strip() or '\n' in key or '\r' in key:
raise ValueError('Set XCM_API_KEY in the server environment')
url = origin + '/v1/videos/' + urls.quote(ident, safe='') + '/content'
opener = http.build_opener(NoRedirect)
send_auth = True
hops = 0
limit = 250 * 1024 * 1024
part = path.with_name(path.name + '.partial')
created = False
try:
while True:
url, current_origin = checked_url(url)
send_auth = send_auth and current_origin == auth_origin
headers = {'Accept': 'video/mp4, application/octet-stream, application/json'}
if send_auth:
headers['Authorization'] = 'Bearer ' + key
req = http.Request(url, headers=headers)
try:
response = opener.open(req, timeout=90)
except errors.HTTPError as exc:
response = exc
with response:
kind = response.headers.get('Content-Type', '').split(';')[0].strip().lower()
if response.status in (301, 302, 303, 307, 308):
target = response.headers.get('Location')
elif response.status != 200:
raise RuntimeError(f'Download HTTP {response.status}; retain task ID')
elif kind == 'application/json' or kind.endswith('+json'):
raw = response.read(1024 * 1024 + 1)
if len(raw) > 1024 * 1024:
raise RuntimeError('Download JSON exceeds 1MiB')
try:
data = json.loads(raw)
except (ValueError, UnicodeError):
raise RuntimeError('Invalid download JSON') from None
if isinstance(data, dict) and isinstance(data.get('data'), dict):
data = data['data']
if not isinstance(data, dict):
raise RuntimeError('Expected a download URL object')
video = data.get('video')
results = data.get('result_urls')
target = (data.get('url') or data.get('download_url') or data.get('video_url') or data.get('content_url')
or (video.get('url') if isinstance(video, dict) else None)
or (results[0] if isinstance(results, list) and results else None))
else:
if kind not in ('video/mp4', 'application/octet-stream'):
raise RuntimeError('Not an MP4 download response')
length = response.headers.get('Content-Length')
if length is not None and (not length.isdigit() or not 0 < int(length) <= limit):
raise RuntimeError('Invalid or excessive media length')
first = response.read(4096)
if len(first) < 12 or first[4:8] != b'ftyp':
raise RuntimeError('Response is not an MP4')
size = len(first)
with part.open('xb') as stream:
created = True
stream.write(first)
while True:
chunk = response.read(1024 * 1024)
if not chunk:
break
size += len(chunk)
if size > limit:
raise RuntimeError('Media exceeds the 250MiB client limit')
stream.write(chunk)
if length is not None and size != int(length):
raise RuntimeError('Incomplete media response; retain task ID')
break
if hops >= 5:
raise RuntimeError('Too many download redirects/JSON links (maximum 5)')
url, _ = checked_url(target, url)
hops += 1
# Publish exclusively: a concurrent download must not overwrite this file.
for attempt in range(6):
try:
os.link(part, path)
break
except PermissionError:
if attempt == 5:
raise
time.sleep(.1)
return size
finally:
if created:
part.unlink(missing_ok=True)
def main():
parser = argparse.ArgumentParser(description=__doc__)
action = parser.add_mutually_exclusive_group(required=True)
action.add_argument('--submit', action='store_true')
action.add_argument('--task-id')
parser.add_argument('--output', type=Path, default=Path.cwd() / 'h3-output.mp4')
args = parser.parse_args()
task_id = args.task_id
if args.submit:
status, response, _ = request('POST', '/v1/videos', {
'model': 'h3', 'prompt': 'A cat walking slowly on a sunny beach',
'duration': 6, 'resolution': '480p', 'aspect_ratio': '16:9',
})
if status not in (200, 201, 202):
raise RuntimeError(f'Create HTTP {status}; do not automatically retry POST')
data = unwrap(response)
task_id = data.get('request_id') or data.get('id') or data.get('task_id')
if not isinstance(task_id, str) or not task_id:
raise RuntimeError('No task ID found; do not automatically resubmit')
# Contains only your task ID; use a private application database in production.
Path('h3-last-task.json').write_text(json.dumps({'task_id': task_id}), encoding='utf-8')
import re
if not isinstance(task_id, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}', task_id):
raise ValueError('Invalid task ID; do not resubmit')
print('Task ID:', task_id)
poll(task_id)
size = download(task_id, args.output)
print('Downloaded:', args.output, 'bytes:', size)
if __name__ == '__main__':
main()
Node.js 20+ 完整示例
Node.js 20+
// Node.js 20+. XCM_API_KEY is read only from the environment.
// node h3_client.mjs --submit OR node h3_client.mjs --task-id TASK_ID
import {writeFile} from 'node:fs/promises';
const origin = (process.env.XCM_API_ORIGIN || 'https://xcmapi.org').replace(/\/$/, '');
const endpoint = new URL(origin);
if (endpoint.protocol !== 'https:' || endpoint.pathname !== '/' || endpoint.search || endpoint.hash || endpoint.username || endpoint.password)
throw new Error('XCM_API_ORIGIN must be an HTTPS origin without /v1');
if (!process.env.XCM_API_KEY) throw new Error('Set XCM_API_KEY');
const delay = ms => new Promise(resolve => setTimeout(resolve, ms));
async function request(method, path, body) {
return fetch(origin + path, {method, redirect:'error', signal:AbortSignal.timeout(30000),
headers:{Authorization:`Bearer ${process.env.XCM_API_KEY}`, 'Content-Type':'application/json'},
body:body === undefined ? undefined : JSON.stringify(body)});
}
async function download(taskId, output='h3-output.mp4') {
if(typeof taskId !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}$/.test(taskId))
throw new Error('Invalid task ID');
function checkedURL(value, base) {
if(typeof value !== 'string' || !value || /[\s\\\x00-\x1f\x7f]/.test(value)) throw new Error('Invalid download URL');
const u=new URL(value,base);
if(u.protocol!=='https:' || !u.hostname || u.username || u.password || u.hash || u.port==='0')
throw new Error('Download URL must be HTTPS without credentials or fragment');
return u;
}
async function readBounded(response, limit) {
const chunks=[]; let size=0;
if(!response.body) throw new Error('Missing download body');
for await(const chunk of response.body) {
size+=chunk.byteLength;
if(size>limit) throw new Error('Download body exceeds its limit');
chunks.push(Buffer.from(chunk));
}
return Buffer.concat(chunks,size);
}
let url=checkedURL(origin+'/v1/videos/'+encodeURIComponent(taskId)+'/content');
let sendAuth=true, hops=0;
const key=process.env.XCM_API_KEY;
if(!key || !key.trim() || /[\r\n]/.test(key)) throw new Error('Set XCM_API_KEY');
while(true) {
sendAuth=sendAuth && url.origin===endpoint.origin;
const headers={Accept:'video/mp4, application/octet-stream, application/json'};
if(sendAuth) headers.Authorization='Bearer '+key;
const response=await fetch(url.href,{method:'GET',redirect:'manual',headers,signal:AbortSignal.timeout(90000)});
const kind=(response.headers.get('content-type') || '').split(';')[0].trim().toLowerCase();
let target;
if([301,302,303,307,308].includes(response.status)) {
target=response.headers.get('location');
await response.body?.cancel();
} else if(response.status!==200) {
await response.body?.cancel();
throw new Error('Download HTTP '+response.status+'; retain task ID');
} else if(kind==='application/json' || kind.endsWith('+json')) {
let data=JSON.parse((await readBounded(response,1024*1024)).toString('utf8'));
if(data && typeof data.data==='object' && data.data!==null && !Array.isArray(data.data)) data=data.data;
if(!data || typeof data!=='object' || Array.isArray(data)) throw new Error('Expected a download URL object');
target=data.url || data.download_url || data.video_url || data.content_url || data.video?.url ||
(Array.isArray(data.result_urls)?data.result_urls[0]:undefined);
} else {
if(!['video/mp4','application/octet-stream'].includes(kind)) {
await response.body?.cancel(); throw new Error('Not an MP4 download response');
}
const limit=250*1024*1024, length=response.headers.get('content-length');
if(length!==null && (!/^\d+$/.test(length) || Number(length)<=0 || Number(length)>limit)) {
await response.body?.cancel(); throw new Error('Invalid or excessive media length');
}
const media=await readBounded(response,limit);
if(media.length<12 || media.toString('ascii',4,8)!=='ftyp') throw new Error('Response is not an MP4');
if(length!==null && media.length!==Number(length)) throw new Error('Incomplete media response');
await writeFile(output,media,{flag:'wx',mode:0o600});
return media.length;
}
if(hops>=5) throw new Error('Too many download redirects/JSON links (maximum 5)');
url=checkedURL(target,url);
hops++;
}
}
const args=process.argv.slice(2);
let taskId;
if(args.length === 2 && args[0] === '--task-id') taskId=args[1];
else if(args.length === 1 && args[0] === '--submit') {
// One POST only. Do not retry creation automatically after a network timeout.
const response=await request('POST','/v1/videos',{model:'h3',prompt:'A cat walking slowly on a sunny beach',duration:6,resolution:'480p',aspect_ratio:'16:9'});
if(!response.ok) throw new Error(`Create HTTP ${response.status}; do not auto-retry POST`);
const result=await response.json(), data=result.data ?? result;
taskId=data.request_id ?? data.id ?? data.task_id;
if(typeof taskId !== 'string' || !taskId) throw new Error('Missing task ID; do not resubmit');
await writeFile('h3-last-task.json',JSON.stringify({task_id:taskId}),{mode:0o600});
} else throw new Error('Use --submit OR --task-id TASK_ID');
if(typeof taskId !== 'string' || !/^[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}$/.test(taskId)) throw new Error('Invalid task ID; do not resubmit');
console.log('Task ID:',taskId);
const taskPath='/v1/videos/'+encodeURIComponent(taskId);
let completed=false;
for(let attempt=0;attempt<120;attempt++) {
let response;
try {response=await request('GET',taskPath);} catch {await delay(5000);continue;}
if([429,502,503,504].includes(response.status)){
const retry=Number(response.headers.get('retry-after'));
await delay((Number.isFinite(retry)&&retry>0?Math.min(15,Math.max(3,retry)):5)*1000);continue;
}
if(!response.ok) throw new Error(`Query HTTP ${response.status}; retain task ID`);
const result=await response.json(),data=result.data ?? result,state=data.status;
if(['failed','error','cancelled','canceled','expired'].includes(state))throw new Error('Task failed; retain task ID');
if(['done','completed','succeeded','success'].includes(state)){completed=true;break;}
if(!['pending','queued','in_progress','processing','running'].includes(state))throw new Error('Unknown task status');
await delay(5000);
}
if(!completed)throw new Error('Polling stopped; resume with --task-id, do not resubmit');
const savedBytes=await download(taskId);
console.log('Downloaded: h3-output.mp4 bytes:',savedBytes);
"""Mixed video client. Seedance 2.0 at 5s/720p has one real end-to-end verified case.
Use XCM_API_KEY only in a server environment. POST is never automatically retried.
Submit requires a new explicit receipt path; resume uses its saved task ID.
"""
import argparse,json,os,re,time
from pathlib import Path
from urllib import request,error,parse
MODELS={
'seedance-2.0-900-特惠':(5, '720p'),
'seedance-2.5-900-特惠':(4, '480p'),
'grok-1.5':(3, '720p'),
'minimax-h3-933-2k-支持真人':(15, '2k'),
'sd-2.0-933-720-fast-原生真人':(5, '720p'),
'seedance-2.0-933-特惠':(5, '720p'),
'seedance-2.5-1010-特惠':(5, '720p'),
'seedance-2.5-3010-特惠':(5, '720p'),
'seedance2.0-mini-A':(5, '720p'),
'seedance2.5-pro-G1':(30, '720p'),
}
class NoRedirect(request.HTTPRedirectHandler):
def redirect_request(self,*args):return None
def api(method,path,body=None):
origin=os.environ.get('XCM_API_ORIGIN','https://xcmapi.org').rstrip('/');u=parse.urlsplit(origin)
if u.scheme!='https' or not u.netloc or u.path or u.username or u.password or u.query or u.fragment:raise ValueError('XCM_API_ORIGIN must be an HTTPS origin, without /v1')
key=os.environ.get('XCM_API_KEY','')
if not key or '\n' in key or '\r' in key:raise ValueError('Set XCM_API_KEY in the server environment')
data=json.dumps(body,ensure_ascii=False).encode() if body is not None else None
req=request.Request(origin+path,data=data,method=method,headers={'Authorization':'Bearer '+key,'Content-Type':'application/json'})
try:r=request.build_opener(NoRedirect).open(req,timeout=100 if method=='POST' else 35)
except error.HTTPError as e:r=e
with r:
code=r.status;raw=r.read(2*1024*1024+1);retry=r.headers.get('Retry-After','5');ct=r.headers.get('Content-Type','')
if len(raw)>2*1024*1024 or 'json' not in ct.lower():raise RuntimeError('Not a bounded JSON API response; do not resubmit POST')
try:value=json.loads(raw)
except (ValueError,UnicodeError):raise RuntimeError('Invalid JSON response; do not resubmit POST') from None
return code,value,min(60,max(1,int(retry))) if retry.isdigit() else 5
def unwrap(value):
if not isinstance(value,dict):raise RuntimeError('Expected a task object')
return value['data'] if isinstance(value.get('data'),dict) else value
def extract_id(value):
v=unwrap(value)
if v.get('error'):raise RuntimeError('No accepted task; reconcile before trying a new business order')
ident=v.get('task_id') or v.get('id') or v.get('request_id')
if not isinstance(ident,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}',ident):raise RuntimeError('No valid task ID; never retry this POST automatically')
return ident
def replace_receipt(path,data):
temp=path.with_name(path.name+'.next')
# Contains only task metadata, never KEY, prompt, materials or a signed media URL.
with temp.open('x',encoding='utf8') as f:json.dump(data,f,ensure_ascii=False,indent=2)
for attempt in range(6):
try:temp.replace(path);return
except PermissionError:
if attempt==5:raise
time.sleep(.1)
def submit_once(path,model,prompt,duration=None,resolution=None):
if model not in MODELS:raise ValueError('Unknown complete model ID')
if not path.is_absolute():raise ValueError('Choose an explicit absolute receipt path')
default_seconds,default_resolution=MODELS[model]
receipt={'model':model,'requested_duration':duration if duration is not None else default_seconds,'requested_resolution':resolution or default_resolution,'state':'creation_uncertain','task_id':None}
# Reserve a new business receipt before any network side effect. Existing receipts
# cannot be overwritten to accidentally create another billable task.
with path.open('x',encoding='utf8') as f:json.dump(receipt,f,ensure_ascii=False)
try:
status,response,_=api('POST','/v1/videos',{'model':model,'prompt':prompt,'duration':duration if duration is not None else default_seconds,'resolution':resolution or default_resolution,'ratio':'16:9'})
except (TimeoutError,error.URLError,RuntimeError):
raise RuntimeError('Creation outcome is unknown. Keep this receipt and reconcile; do not resubmit.') from None
if status not in (200,201,202):raise RuntimeError(f'Create HTTP {status}; keep the receipt, inspect the cause, and do not automatically retry POST')
ident=extract_id(response);receipt.update(task_id=ident,state='pending')
try:replace_receipt(path,receipt)
except OSError:raise RuntimeError(f'Accepted task ID: {ident}. Receipt update failed; save this ID and never repeat POST.') from None
return ident
def poll(ident,max_attempts=180,max_elapsed=1800):
if not isinstance(ident,str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}',ident):raise ValueError('Invalid task ID')
deadline=time.monotonic()+max_elapsed
for _ in range(max_attempts):
if time.monotonic()>deadline:break
try:code,response,retry=api('GET','/v1/videos/'+parse.quote(ident,safe=''))
except (TimeoutError,error.URLError):time.sleep(5);continue
if code in (429,502,503,504):time.sleep(retry);continue
if code!=200:raise RuntimeError(f'Query HTTP {code}; retain the task ID')
data=unwrap(response);state=str(data.get('status','')).lower()
if state in ('failed','error','expired','cancelled','canceled'):raise RuntimeError('Task ended without success; reconcile its usage before a new submission')
if state in ('done','completed','succeeded','success'):
return data # /content may provide the media even when polling has no URL.
if state not in ('pending','queued','processing','running','in_progress'):raise RuntimeError('Unknown task state; do not treat it as success')
time.sleep(5)
raise TimeoutError('Polling stopped, not cancelled. Resume with the same receipt; do not create another task')
def download(ident, path):
"""Fetch /content, following at most five HTTPS redirects/JSON links."""
import re
from urllib import request as http, error as errors, parse as urls
if not isinstance(ident, str) or not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.:-]{0,159}', ident):
raise ValueError('Invalid task ID')
if not path.is_absolute():
raise ValueError('Choose an explicit absolute output path')
if path.exists():
raise FileExistsError('Output exists; choose a new path')
def checked_url(value, base=None):
if not isinstance(value, str) or not value or any(ord(c) <= 32 or ord(c) == 127 for c in value) or '\\' in value:
raise ValueError('Invalid download URL')
target = urls.urljoin(base, value) if base else value
p = urls.urlsplit(target)
if p.scheme != 'https' or not p.hostname or p.username is not None or p.password is not None or p.fragment or p.port == 0:
raise ValueError('Download URL must be HTTPS without credentials or fragment')
return target, (p.hostname.lower(), p.port or 443)
origin = os.environ.get('XCM_API_ORIGIN', 'https://xcmapi.org').rstrip('/')
origin, auth_origin = checked_url(origin)
parsed = urls.urlsplit(origin)
if parsed.path or parsed.query:
raise ValueError('Use an HTTPS origin without /v1')
key = os.environ.get('XCM_API_KEY', '')
if not key.strip() or '\n' in key or '\r' in key:
raise ValueError('Set XCM_API_KEY in the server environment')
url = origin + '/v1/videos/' + urls.quote(ident, safe='') + '/content'
opener = http.build_opener(NoRedirect)
send_auth = True
hops = 0
limit = 250 * 1024 * 1024
part = path.with_name(path.name + '.partial')
created = False
try:
while True:
url, current_origin = checked_url(url)
send_auth = send_auth and current_origin == auth_origin
headers = {'Accept': 'video/mp4, application/octet-stream, application/json'}
if send_auth:
headers['Authorization'] = 'Bearer ' + key
req = http.Request(url, headers=headers)
try:
response = opener.open(req, timeout=90)
except errors.HTTPError as exc:
response = exc
with response:
kind = response.headers.get('Content-Type', '').split(';')[0].strip().lower()
if response.status in (301, 302, 303, 307, 308):
target = response.headers.get('Location')
elif response.status != 200:
raise RuntimeError(f'Download HTTP {response.status}; retain task ID')
elif kind == 'application/json' or kind.endswith('+json'):
raw = response.read(1024 * 1024 + 1)
if len(raw) > 1024 * 1024:
raise RuntimeError('Download JSON exceeds 1MiB')
try:
data = json.loads(raw)
except (ValueError, UnicodeError):
raise RuntimeError('Invalid download JSON') from None
if isinstance(data, dict) and isinstance(data.get('data'), dict):
data = data['data']
if not isinstance(data, dict):
raise RuntimeError('Expected a download URL object')
video = data.get('video')
results = data.get('result_urls')
target = (data.get('url') or data.get('download_url') or data.get('video_url') or data.get('content_url')
or (video.get('url') if isinstance(video, dict) else None)
or (results[0] if isinstance(results, list) and results else None))
else:
if kind not in ('video/mp4', 'application/octet-stream'):
raise RuntimeError('Not an MP4 download response')
length = response.headers.get('Content-Length')
if length is not None and (not length.isdigit() or not 0 < int(length) <= limit):
raise RuntimeError('Invalid or excessive media length')
first = response.read(4096)
if len(first) < 12 or first[4:8] != b'ftyp':
raise RuntimeError('Response is not an MP4')
size = len(first)
with part.open('xb') as stream:
created = True
stream.write(first)
while True:
chunk = response.read(1024 * 1024)
if not chunk:
break
size += len(chunk)
if size > limit:
raise RuntimeError('Media exceeds the 250MiB client limit')
stream.write(chunk)
if length is not None and size != int(length):
raise RuntimeError('Incomplete media response; retain task ID')
break
if hops >= 5:
raise RuntimeError('Too many download redirects/JSON links (maximum 5)')
url, _ = checked_url(target, url)
hops += 1
# Publish exclusively: a concurrent download must not overwrite this file.
for attempt in range(6):
try:
os.link(part, path)
break
except PermissionError:
if attempt == 5:
raise
time.sleep(.1)
return size
finally:
if created:
part.unlink(missing_ok=True)
def main():
parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--receipt',type=Path,required=True);mode=parser.add_mutually_exclusive_group(required=True);mode.add_argument('--submit',action='store_true');mode.add_argument('--resume',action='store_true');parser.add_argument('--model',choices=MODELS);parser.add_argument('--prompt',default='A paper boat floats on a calm pond, no people, static camera.');parser.add_argument('--duration',type=int);parser.add_argument('--resolution');parser.add_argument('--output',type=Path);args=parser.parse_args()
if not args.receipt.is_absolute():parser.error('--receipt must be an explicit absolute path')
if args.submit:
if not args.model:parser.error('--submit requires --model')
ident=submit_once(args.receipt,args.model,args.prompt,args.duration,args.resolution)
else:
receipt=json.loads(args.receipt.read_text(encoding='utf8'));ident=receipt.get('task_id')
if not ident:raise SystemExit('Receipt has no confirmed task ID. Reconcile the original submission; do not submit again')
poll(ident);receipt=json.loads(args.receipt.read_text(encoding='utf8'));receipt['state']='done';replace_receipt(args.receipt,receipt)
if args.output:
size=download(ident,args.output);receipt['downloaded_file']=str(args.output);receipt['downloaded_bytes']=size;replace_receipt(args.receipt,receipt)
print('Downloaded:',args.output,'bytes:',size)
print('Completed task:',ident)
print('Download with the same KEY through: /v1/videos/'+parse.quote(ident,safe='')+'/content')
print('Do not send your KEY to any external media URL.')
if __name__=='__main__':main()
填写地址和 Key
切换你的供应商
搭建自动化流程