#!/usr/bin/env python3
"""通过 API易 调用 Seedance 2.0 生成视频（文生 / 图生 / 参考图生视频）。纯标准库，零依赖。"""
import argparse
import base64
import json
import os
import shutil
import sys
import time
import urllib.error
import urllib.request
# 输出重定向到文件/管道时也逐行落盘，方便 Agent 后台跟踪进度
sys.stdout.reconfigure(line_buffering=True)
TASKS_URL = "https://api.apiyi.com/seedance/api/v3/contents/generations/tasks"
# 短名 → 完整模型 ID
MODELS = {
"mini": "doubao-seedance-2-0-mini-260615",
"fast": "doubao-seedance-2-0-fast-260128",
"std": "doubao-seedance-2-0-260128",
}
# 各型号分辨率上限（mini/fast 传 1080p 上游会 400，客户端直接拦下省一次请求）
MODEL_CAPS = {
"mini": ("480p", "720p"),
"fast": ("480p", "720p"),
"std": ("480p", "720p", "1080p"),
}
RATIOS = ("adaptive", "16:9", "4:3", "1:1", "3:4", "9:16", "21:9")
MAX_REF_IMAGES = 9
# 出片是异步任务：提交后先等一段再查，720p/5s 实测约 90-170 秒
POLL_FIRST_DELAY = 25
POLL_INTERVAL = 15
POLL_TIMEOUT = 15 * 60
def load_api_key():
"""优先读环境变量；否则在脚本所在目录及其父目录找 .env。"""
key = os.environ.get("APIYI_API_KEY")
if key:
return key
here = os.path.dirname(os.path.abspath(__file__))
for d in (here, os.path.dirname(here)):
env_path = os.path.join(d, ".env")
if os.path.exists(env_path):
with open(env_path, encoding="utf-8") as f:
for line in f:
line = line.strip()
if line.startswith("APIYI_API_KEY") and "=" in line:
return line.split("=", 1)[1].strip().strip('"').strip("'")
return None
def project_root():
"""从脚本位置向上找包含 .git 或 .claude 的目录，作为项目根目录；找不到则用当前工作目录。"""
d = os.path.dirname(os.path.abspath(__file__))
while True:
if os.path.isdir(os.path.join(d, ".git")) or os.path.isdir(os.path.join(d, ".claude")):
return d
parent = os.path.dirname(d)
if parent == d:
return os.getcwd()
d = parent
def resolve_path(out):
"""纯文件名 → 存到 <项目根>/seedance-output/ 下，确保好找；带目录成分则按给定路径。"""
if os.path.dirname(out):
return os.path.abspath(out)
out_dir = os.path.join(project_root(), "seedance-output")
os.makedirs(out_dir, exist_ok=True)
return os.path.join(out_dir, out)
def image_source(src):
"""图片入参：URL / asset:// / data: 原样透传，本地文件转 base64 data URI。"""
if src.startswith(("http://", "https://", "asset://", "data:")):
return src
if not os.path.exists(src):
sys.exit(f"图片文件不存在：{src}")
ext = src.lower().rsplit(".", 1)[-1]
mime = {"png": "image/png", "webp": "image/webp"}.get(ext, "image/jpeg")
with open(src, "rb") as f:
return f"data:{mime};base64,{base64.b64encode(f.read()).decode()}"
def api_request(url, api_key, body=None):
"""网关会标 content-encoding: gzip 但实际未压缩，必须 Accept-Encoding: identity。"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept-Encoding": "identity",
}
data = json.dumps(body).encode() if body is not None else None
req = urllib.request.Request(url, data=data, headers=headers,
method="POST" if body is not None else "GET")
try:
with urllib.request.urlopen(req, timeout=60) as r:
return json.loads(r.read())
except urllib.error.HTTPError as e:
raise RuntimeError(f"请求失败 HTTP {e.code}：{e.read().decode(errors='replace')[:800]}")
def download(url, path):
"""下载结果视频：签名直链，不要带 Authorization 头。"""
req = urllib.request.Request(url)
with urllib.request.urlopen(req, timeout=300) as r, open(path, "wb") as f:
shutil.copyfileobj(r, f)
return os.path.getsize(path)
def build_content(args):
content = [{"type": "text", "text": args.prompt}]
if args.image:
content.append({"type": "image_url",
"image_url": {"url": image_source(args.image)},
"role": "first_frame"})
if args.last_frame:
content.append({"type": "image_url",
"image_url": {"url": image_source(args.last_frame)},
"role": "last_frame"})
for src in args.ref_image:
content.append({"type": "image_url",
"image_url": {"url": image_source(src)},
"role": "reference_image"})
return content
def main():
api_key = load_api_key()
if not api_key:
sys.exit("未找到 API Key：请在技能目录的 .env 写一行 APIYI_API_KEY=sk-xxx"
"（令牌须勾选 SeeDance2 / SD2Mini / SD2Fast 分组）")
parser = argparse.ArgumentParser(description="Seedance 2.0 出视频")
parser.add_argument("prompt", help="提示词（画面 + 运镜 + 氛围）")
parser.add_argument("--model", default="mini", choices=sorted(MODELS),
help="mini=最快最便宜（默认）/ fast=极速版 / std=标准版（唯一支持 1080p）")
parser.add_argument("--resolution", default="720p", choices=("480p", "720p", "1080p"),
help="分辨率，默认 720p")
parser.add_argument("--ratio", default="adaptive", choices=RATIOS,
help="宽高比，默认 adaptive（同档位全比例同价）")
parser.add_argument("--duration", type=int, default=5,
help="时长 4-15 整数秒，或 -1 让模型智能决定，默认 5")
parser.add_argument("--no-audio", action="store_true",
help="关闭同步音频（默认带声音）")
parser.add_argument("--seed", type=int, default=None, help="随机种子，复现用")
parser.add_argument("-i", "--image", help="首帧图（本地路径 / URL / asset://），传入即图生视频")
parser.add_argument("--last-frame", help="尾帧图，与 -i 搭配做首尾帧过渡")
parser.add_argument("--ref-image", action="append", default=[],
help=f"参考图（可重复，最多 {MAX_REF_IMAGES} 张），与 -i 互斥")
parser.add_argument("-o", "--out", default="output.mp4", help="输出文件名")
args = parser.parse_args()
if args.image and args.ref_image:
sys.exit("首帧模式（-i）与参考图模式（--ref-image）互斥，一次只能用一种。")
if args.last_frame and not args.image:
sys.exit("--last-frame 必须与 -i（首帧图）搭配使用。")
if len(args.ref_image) > MAX_REF_IMAGES:
sys.exit(f"参考图最多 {MAX_REF_IMAGES} 张。")
if args.duration != -1 and not 4 <= args.duration <= 15:
sys.exit("时长只支持 4-15 整数秒，或 -1 智能时长。")
if args.resolution not in MODEL_CAPS[args.model]:
sys.exit(f"{args.model} 最高支持 {MODEL_CAPS[args.model][-1]}，"
f"1080p 请用 --model std。")
body = {
"model": MODELS[args.model],
"content": build_content(args),
"resolution": args.resolution,
"ratio": args.ratio,
"duration": args.duration,
}
if args.no_audio:
body["generate_audio"] = False
if args.seed is not None:
body["seed"] = args.seed
try:
task = api_request(TASKS_URL, api_key, body)
except (RuntimeError, OSError) as e:
sys.exit(f"提交失败：{e}")
task_id = task.get("id")
if not task_id:
sys.exit(f"提交失败，响应：{json.dumps(task, ensure_ascii=False)[:500]}")
print(f"任务已提交 task_id={task_id}，出片通常需要 2-5 分钟，开始轮询…")
t0 = time.time()
time.sleep(POLL_FIRST_DELAY)
while True:
try:
task = api_request(f"{TASKS_URL}/{task_id}", api_key)
except (RuntimeError, OSError) as e: # 网络抖动不中断轮询
print(f" 轮询异常（继续）：{e}")
time.sleep(POLL_INTERVAL)
continue
status = task.get("status", "unknown")
elapsed = round(time.time() - t0)
print(f" [{elapsed:>4}s] status={status}")
if status in ("succeeded", "failed", "expired"):
break
if time.time() - t0 > POLL_TIMEOUT:
sys.exit(f"轮询超时（{POLL_TIMEOUT}s）。任务仍在服务端，可稍后手动查询：\n"
f" GET {TASKS_URL}/{task_id}")
time.sleep(POLL_INTERVAL)
if status != "succeeded":
err = task.get("error") or task
sys.exit(f"生成失败（status={status}）：{json.dumps(err, ensure_ascii=False)[:500]}")
video_url = (task.get("content") or {}).get("video_url")
if not video_url:
sys.exit(f"任务成功但未返回视频地址：{json.dumps(task, ensure_ascii=False)[:500]}")
path = resolve_path(args.out)
size = download(video_url, path)
tokens = (task.get("usage") or {}).get("completion_tokens", "?")
print(f"视频已保存至 {path}（{size / 1e6:.1f} MB，耗时 {round(time.time() - t0)}s，"
f"计费 {tokens} tokens）")
if __name__ == "__main__":
main()
