Skip to content

Instantly share code, notes, and snippets.

@knowlet

knowlet/main.py

Created May 19, 2026 05:36
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save knowlet/b5ac83ae48c54a25e5dd3e91b5a7af1b to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save knowlet/b5ac83ae48c54a25e5dd3e91b5a7af1b to your computer and use it in GitHub Desktop.
Download ZIP
openrouter video-generation cli python script
Raw
main.py
#!/usr/bin/env python3
from __future__ import annotations
import argparse
import base64
import json
import mimetypes
import os
import time
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional
import requests
# Alpha video API (not /v1/chat/completions); see:
# https://openrouter.notion.site/video-generation-testing
ALPHA_VIDEOS = "https://openrouter.ai/api/alpha/videos"
DEFAULT_PROMPT = (
"A golden retriever runs through a sunlit meadow with wildflowers, "
"shot in slow motion with warm natural lighting and shallow depth of field."
)
class OpenRouterError(Exception):
pass
def _image_file_to_data_url(path: Path) -> str:
data = path.read_bytes()
mime, _ = mimetypes.guess_type(path.name)
if not mime or not mime.startswith("image/"):
mime = "image/jpeg"
b64 = base64.standard_b64encode(data).decode("ascii")
return f"data:{mime};base64,{b64}"
def build_input_references(
references_json: Optional[Path],
image_urls: Optional[List[str]],
image_files: Optional[List[Path]],
) -> Optional[List[Dict[str, Any]]]:
"""Image-to-video: same shape as chat image parts."""
if references_json is not None:
raw = references_json.read_text(encoding="utf-8")
try:
parsed = json.loads(raw)
except json.JSONDecodeError as e:
raise OpenRouterError(
f"Invalid JSON in --references-json: {e}"
) from e
if not isinstance(parsed, list):
raise OpenRouterError("--references-json must contain a JSON array.")
return parsed
urls = [u for u in (image_urls or []) if u]
files = list(image_files or [])
if not urls and not files:
return None
out: List[Dict[str, Any]] = []
for url in urls:
out.append({"type": "image_url", "image_url": {"url": url}})
for image_file in files:
path = image_file.expanduser().resolve()
if not path.is_file():
raise OpenRouterError(f"Image file not found: {path}")
data_url = _image_file_to_data_url(path)
out.append({"type": "image_url", "image_url": {"url": data_url}})
return out
def resolve_api_key(explicit: Optional[str]) -> str:
key = explicit or os.getenv("OPENROUTER_API_KEY")
if not key:
raise OpenRouterError("請設定 OPENROUTER_API_KEY，或使用 --api-key。")
return key
def get_headers(api_key: str) -> Dict[str, str]:
return {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"X-Title": "openrouter-alpha-video",
}
def submit_video_generation(
headers: Dict[str, str],
model: str,
prompt: str,
duration: int,
resolution: str,
aspect_ratio: Optional[str],
input_references: Optional[List[Dict[str, Any]]],
) -> Dict[str, Any]:
payload: Dict[str, Any] = {
"model": model,
"prompt": prompt,
"duration": duration,
"resolution": resolution,
}
if aspect_ratio:
payload["aspect_ratio"] = aspect_ratio
if input_references:
payload["input_references"] = input_references
resp = requests.post(ALPHA_VIDEOS, headers=headers, json=payload, timeout=120)
if resp.status_code >= 400:
raise OpenRouterError(
f"Submit failed: HTTP {resp.status_code}\n{resp.text}"
)
return resp.json()
def extract_job_id(response_json: Dict[str, Any]) -> Optional[str]:
return next(
(
str(response_json[key])
for key in ("id", "job_id", "jobId")
if key in response_json and response_json[key]
),
None,
)
def poll_video_job(
headers: Dict[str, str],
job_id: str,
poll_interval: int,
timeout: int,
) -> Dict[str, Any]:
url = f"{ALPHA_VIDEOS}/{job_id}"
deadline = time.time() + timeout
last_status: Dict[str, Any] = {}
while time.time() < deadline:
resp = requests.get(url, headers=headers, timeout=60)
if resp.status_code >= 400:
raise OpenRouterError(
f"Polling failed: HTTP {resp.status_code}\n{resp.text}"
)
status_json = resp.json()
last_status = status_json
status = status_json.get("status")
if isinstance(status, str):
status = status.lower()
if status == "completed":
return status_json
if status in ("failed", "cancelled", "expired"):
err = status_json.get("error")
detail = json.dumps(status_json, indent=2, ensure_ascii=False)
if err is not None:
detail = f"{err}\n{detail}"
raise OpenRouterError(f"Generation ended with status {status!r}:\n{detail}")
time.sleep(poll_interval)
raise TimeoutError(
"Timed out waiting for generation.\n"
+ json.dumps(last_status, indent=2, ensure_ascii=False)
)
def extract_video_url(obj: Any) -> Optional[str]:
urls = obj.get("unsigned_urls") if isinstance(obj, dict) else None
if isinstance(urls, list) and urls:
first = urls[0]
if isinstance(first, str):
return first
if isinstance(first, dict):
u = first.get("url")
if isinstance(u, str):
return u
return None
def download_file(url: str, output_path: Path) -> None:
with requests.get(url, stream=True, timeout=300) as resp:
if resp.status_code >= 400:
raise OpenRouterError(
f"Download failed: HTTP {resp.status_code}\n{resp.text}"
)
with output_path.open("wb") as f:
for chunk in resp.iter_content(chunk_size=1024 * 1024):
if chunk:
f.write(chunk)
def download_job_content(
headers: Dict[str, str], job_id: str, output_path: Path
) -> None:
url = f"{ALPHA_VIDEOS}/{job_id}/content"
with requests.get(url, headers=headers, stream=True, timeout=300) as resp:
if resp.status_code >= 400:
raise OpenRouterError(
f"Download failed: HTTP {resp.status_code}\n{resp.text}"
)
with output_path.open("wb") as f:
for chunk in resp.iter_content(chunk_size=1024 * 1024):
if chunk:
f.write(chunk)
def parse_args() -> argparse.Namespace:
default_output = Path(
f"output_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4"
)
p = argparse.ArgumentParser(
description="OpenRouter alpha video：文字轉影片 / 圖片轉影片",
)
p.add_argument(
"prompt",
nargs="?",
default=DEFAULT_PROMPT,
help="影片描述（預設為內建範例 prompt）",
)
p.add_argument(
"-m",
"--model",
default="alibaba/wan-2.6",
help="模型 ID（預設：%(default)s）",
)
p.add_argument(
"-d",
"--duration",
type=int,
default=5,
metavar="SEC",
help="秒數（預設：%(default)s）",
)
p.add_argument(
"-r",
"--resolution",
default="720p",
help="解析度（預設：%(default)s）",
)
p.add_argument(
"--poll-interval",
type=int,
default=10,
metavar="SEC",
help="輪詢間隔秒數（預設：%(default)s）",
)
p.add_argument(
"--poll-timeout",
type=int,
default=600,
metavar="SEC",
help="輪詢逾時秒數（預設：%(default)s）",
)
p.add_argument(
"--aspect-ratio",
default=None,
metavar="RATIO",
help="例如 16:9、9:16（選填）",
)
p.add_argument(
"-o",
"--output",
type=Path,
default=default_output,
help="輸出檔案（預設：%(default)s）",
)
p.add_argument(
"--api-key",
default=None,
metavar="KEY",
help="API key；若省略則使用環境變數 OPENROUTER_API_KEY",
)
p.add_argument(
"--image-url",
action="append",
default=None,
dest="image_urls",
metavar="URL",
help="圖片轉影片：參考圖 HTTPS URL（可重複；input_references 順序為先全部 URL，再全部本機檔）",
)
p.add_argument(
"--image-file",
action="append",
default=None,
dest="image_files",
type=Path,
metavar="PATH",
help="圖片轉影片：本機參考圖（可重複；會轉成 data URI；順序見 --image-url）",
)
p.add_argument(
"--references-json",
type=Path,
default=None,
metavar="PATH",
help="圖片轉影片：JSON 檔，內容為 input_references 陣列",
)
return p.parse_args()
def main() -> int:
args = parse_args()
api_key = resolve_api_key(args.api_key)
headers = get_headers(api_key)
input_references = build_input_references(
args.references_json,
args.image_urls,
args.image_files,
)
print(f"Model: {args.model}")
print(f"Prompt: {args.prompt}")
print(f"Duration: {args.duration}s")
print(f"Resolution: {args.resolution}")
if input_references:
n = len(input_references)
print(f"Mode: image-to-video ({n} reference image{'s' if n != 1 else ''})")
else:
print("Mode: text-to-video")
submit_response = submit_video_generation(
headers,
args.model,
args.prompt,
args.duration,
args.resolution,
args.aspect_ratio,
input_references,
)
print("Submit response:")
print(json.dumps(submit_response, indent=2, ensure_ascii=False))
job_id = extract_job_id(submit_response)
if not job_id:
print("No job ID found in response.")
return 2
print(f"Job ID: {job_id}")
status_json = poll_video_job(
headers,
job_id,
args.poll_interval,
args.poll_timeout,
)
print("Final status:")
print(json.dumps(status_json, indent=2, ensure_ascii=False))
try:
download_job_content(headers, job_id, args.output)
except OpenRouterError:
if video_url := extract_video_url(status_json):
download_file(video_url, args.output)
else:
raise
print(f"Saved to: {args.output.resolve()}")
return 0
if __name__ == "__main__":
raise SystemExit(main())
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment