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()
填写地址和 Key
切换你的供应商
搭建自动化流程