#!/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()
