#!/usr/bin/env python3 """Small, resumable 1Route WAN3 client. Python 3.10+, requests.""" from __future__ import annotations import argparse, json, os, pathlib, time from urllib.parse import urlparse import requests BASE = "https://video.1route.dev" WAITING = {"queued", "pending", "in_progress", "processing", "submitted"} SUCCESS = {"completed", "success", "succeeded"} FAILURE = {"failed", "failure", "cancelled", "canceled", "expired"} def request(session, method, path, **kwargs): response = session.request(method, BASE + path, timeout=kwargs.pop("timeout", (15, 60)), **kwargs) request_id = response.headers.get("X-Oneapi-Request-Id", "") try: body = response.json() except ValueError: body = {"raw": response.text[:1000]} if response.status_code >= 400: raise RuntimeError(f"HTTP {response.status_code}, request_id={request_id}, body={body}") return body, request_id def result_url(body): candidates = [ (body.get("metadata") or {}).get("url"), body.get("url"), (body.get("data") or {}).get("url") if isinstance(body.get("data"), dict) else None, (body.get("result") or {}).get("url") if isinstance(body.get("result"), dict) else None, ] return next((x for x in candidates if isinstance(x, str) and x.startswith("http")), None) def download(session, url, target): parsed = urlparse(url) if parsed.scheme not in {"http", "https"}: raise RuntimeError("Result URL must be HTTP(S)") # The completed URL is signed. Do not leak the API key to another origin. headers = session.headers if parsed.netloc == urlparse(BASE).netloc else {} with requests.get(url, headers=headers, timeout=(15, 180), stream=True, allow_redirects=True) as response: response.raise_for_status() with target.open("xb") as output: for chunk in response.iter_content(1024 * 1024): if chunk: output.write(chunk) def poll(session, task_id, output, max_wait): deadline = time.monotonic() + max_wait while True: body, request_id = request(session, "GET", f"/v1/videos/{task_id}") status = str(body.get("status", "")).lower() print(f"task={task_id} status={status} progress={body.get('progress', '')} request_id={request_id}") if status in SUCCESS: url = result_url(body) if not url: raise RuntimeError(f"Completed without a result URL: {body}") if output.exists(): raise FileExistsError(f"Refusing to overwrite: {output}") download(session, url, output) output.with_suffix(output.suffix + ".json").write_text( json.dumps(body, ensure_ascii=False, indent=2), encoding="utf-8") print(f"Saved {output} ({output.stat().st_size} bytes)") return if status in FAILURE: raise RuntimeError(f"Terminal task state: {body}") if status not in WAITING: print("Unknown non-terminal status; keeping the existing task.") if time.monotonic() >= deadline: raise TimeoutError(f"Local wait ended. Resume task {task_id}; do not create a replacement.") time.sleep(15) def main(): parser = argparse.ArgumentParser() parser.add_argument("--task-id") parser.add_argument("--model", default="wan3.0-video-480p") parser.add_argument("--prompt", default="A red kite floats above a green field. The camera slowly moves forward.") parser.add_argument("--seconds", type=int, default=2) parser.add_argument("--aspect-ratio", default="16:9") parser.add_argument("--image-url", action="append", default=[]) parser.add_argument("--video", action="append", default=[], metavar="URL,DURATION") parser.add_argument("--audio-url", action="append", default=[]) parser.add_argument("--output", type=pathlib.Path, default=pathlib.Path("wan3-output.mp4")) parser.add_argument("--max-wait", type=int, default=1800) parser.add_argument("--create", action="store_true") parser.add_argument("--confirm-cost", action="store_true") args = parser.parse_args() key = os.environ.get("WAN3_API_KEY", "").strip() if not key: raise SystemExit("Set WAN3_API_KEY in your private environment.") session = requests.Session() session.headers["Authorization"] = "Bearer " + key session.headers["User-Agent"] = "1route-wan3-example/1.0" if args.task_id: poll(session, args.task_id, args.output, args.max_wait) return models, _ = request(session, "GET", "/v1/models") ids = {x.get("id") for x in models.get("data", []) if isinstance(x, dict)} if args.model not in ids: raise SystemExit(f"Model not available for this key: {args.model}") if not 2 <= args.seconds <= 30: raise SystemExit("seconds must be 2..30") reference_videos = [] reference_seconds = 0 for item in args.video: url, sep, duration_text = item.rpartition(",") if not sep: raise SystemExit("--video must be URL,DURATION") duration = int(duration_text) if not 1 <= duration <= 15: raise SystemExit("reference duration must be 1..15 whole seconds") reference_videos.append({"url": url, "duration": duration}) reference_seconds += duration if len(reference_videos) > 5 or reference_seconds > 15 or args.seconds + reference_seconds > 30: raise SystemExit("WAN3 reference/output duration limits exceeded") if "-image" in args.model and reference_videos: raise SystemExit("Image-family models reject reference video") payload = { "model": args.model, "prompt": args.prompt, "seconds": str(args.seconds), "aspect_ratio": args.aspect_ratio, } if args.image_url: payload["reference_images"] = [{"url": url} for url in args.image_url] if reference_videos: payload["reference_videos"] = reference_videos if args.audio_url: payload["reference_audios"] = [{"url": url} for url in args.audio_url] print("Available model verified. Planned request:") print(json.dumps(payload, ensure_ascii=False, indent=2)) print(f"Billable seconds: {args.seconds + reference_seconds}. Check the live rate and currency at {BASE}/pricing/") if not (args.create and args.confirm_cost): print("Preview only. Add --create --confirm-cost after confirming the displayed price.") return # Create exactly once. An ambiguous network failure is deliberately not retried. try: task, request_id = request(session, "POST", "/v1/videos", json=payload, timeout=(15, 360)) except (requests.Timeout, requests.ConnectionError) as exc: raise RuntimeError( "Creation result is ambiguous. Do not retry automatically. " "Check task/usage records with the timestamp before creating again.") from exc task_id = str(task.get("id") or task.get("task_id") or "") if not task_id: raise RuntimeError(f"Creation response did not include an id: {task}") pathlib.Path("wan3-last-task.txt").write_text(task_id + "\n", encoding="utf-8") print(f"Created task={task_id}, request_id={request_id}. Saved wan3-last-task.txt") poll(session, task_id, args.output, args.max_wait) if __name__ == "__main__": main()