#!/usr/bin/env python3
"""通過 API易 呼叫 Wan2.7 / HappyHorse 生成影片（文生 / 圖生 / 參考圖生 / 影片編輯）。純標準庫，零依賴。"""
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)
# DashScope 透傳端點。絕不要用 /v1/videos 扁平路徑——它會丟棄 media 欄位
CREATE_URL = "https://api.apiyi.com/wan/api/v1/services/aigc/video-generation/video-synthesis"
TASK_URL = "https://api.apiyi.com/v1/tasks/{}"
# 家族 × 模式 → 模型 ID。編輯版命名不規則（happyhorse 是 1.0 且帶連字元），別手拼
FAMILY_MODELS = {
"wan": {"t2v": "wan2.7-t2v", "i2v": "wan2.7-i2v",
"r2v": "wan2.7-r2v", "edit": "wan2.7-videoedit"},
"happyhorse": {"t2v": "happyhorse-1.1-t2v", "i2v": "happyhorse-1.1-i2v",
"r2v": "happyhorse-1.1-r2v", "edit": "happyhorse-1.0-video-edit"},
}
# r2v 參考素材上限：wan 圖+影片合計 5，happyhorse 僅圖、最多 9
MAX_REFS = {"wan": 5, "happyhorse": 9}
RATIOS = ("16:9", "9:16", "1:1", "4:3", "3:4")
# 出片是非同步任務：720P/5s 實測約 70-140 秒，1080P/長影片可能 5 分鐘以上
POLL_FIRST_DELAY = 15
POLL_INTERVAL = 8
POLL_TIMEOUT = 20 * 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):
"""純檔名 → 存到 <專案根>/wan-output/ 下，確保好找；帶目錄成分則按給定路徑。"""
if os.path.dirname(out):
return os.path.abspath(out)
out_dir = os.path.join(project_root(), "wan-output")
os.makedirs(out_dir, exist_ok=True)
return os.path.join(out_dir, out)
def media_source(src):
"""素材入參：URL / data: 原樣透傳，本地檔案轉 base64 data URI（實測兩系列可用）。"""
if src.startswith(("http://", "https://", "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", "mp4": "video/mp4",
"mov": "video/quicktime"}.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, extra_headers=None):
"""閘道會標 content-encoding: gzip 但實際未壓縮，必須 Accept-Encoding: identity。"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"Accept-Encoding": "identity",
}
if extra_headers:
headers.update(extra_headers)
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):
"""下載結果影片：OSS 簽名直鏈，絕不能帶 Authorization 頭（帶了 403）。"""
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 detect_mode(args):
if args.video:
return "edit"
if args.image:
return "i2v"
if args.ref_image or args.ref_video:
return "r2v"
return "t2v"
def build_media(args, mode):
media = []
if mode == "i2v":
media.append({"type": "first_frame", "url": media_source(args.image)})
elif mode == "r2v":
for src in args.ref_image:
media.append({"type": "reference_image", "url": media_source(src)})
for src in args.ref_video:
media.append({"type": "reference_video", "url": media_source(src)})
elif mode == "edit":
media.append({"type": "video", "url": media_source(args.video)})
for src in args.ref_image:
media.append({"type": "reference_image", "url": media_source(src)})
return media
def main():
api_key = load_api_key()
if not api_key:
sys.exit("未找到 API Key：請在技能目錄的 .env 寫一行 APIYI_API_KEY=sk-xxx"
"（令牌須勾選 Wan&HappyHorse 分組、計費模式為按量）")
parser = argparse.ArgumentParser(description="Wan2.7 / HappyHorse 出影片")
parser.add_argument("prompt", help="提示詞（畫面 + 運鏡 + 氛圍；多素材時用「圖1/影片1」指代）")
parser.add_argument("--model", default="wan", choices=sorted(FAMILY_MODELS),
help="wan=Wan2.7（預設，更便宜）/ happyhorse=HappyHorse-1.1（畫質取向）")
parser.add_argument("--resolution", default="720P", type=str.upper,
choices=("720P", "1080P"), help="解析度，預設 720P（注意無 480P）")
parser.add_argument("--ratio", default=None, choices=RATIOS,
help="寬高比（傳首幀圖時忽略；happyhorse 文件未列此引數，僅顯式傳時傳送）")
parser.add_argument("--duration", type=int, default=5,
help="時長 2-15 整數秒，預設 5；含參考影片時上限 10；編輯模式跟隨源影片")
parser.add_argument("--negative", help="負向提示詞（不想出現的內容，500 字元內）")
parser.add_argument("--no-prompt-extend", action="store_true",
help="關閉提示詞智慧擴寫（預設開啟，短提示詞效果更好）")
parser.add_argument("--seed", type=int, default=None, help="隨機種子，復現用")
parser.add_argument("-i", "--image", help="首幀圖（本地路徑或 URL，圖生影片）")
parser.add_argument("--ref-image", action="append", default=[],
help="參考圖，可重複（wan 與參考影片合計最多 5、happyhorse 最多 9）")
parser.add_argument("--ref-video", action="append", default=[],
help="參考影片 URL，可重複（僅 wan 支援）")
parser.add_argument("--video", help="待編輯影片 URL（影片編輯模式，需配 --ref-image）")
parser.add_argument("-o", "--out", default="output.mp4", help="輸出檔名")
args = parser.parse_args()
if args.image and (args.ref_image or args.ref_video):
sys.exit("首幀模式（-i）與參考模式（--ref-image/--ref-video）互斥，一次只能用一種。")
if args.video and args.image:
sys.exit("影片編輯模式（--video）與首幀模式（-i）互斥。")
if args.video and not args.ref_image:
sys.exit("影片編輯模式需要至少 1 張參考圖（--ref-image）。")
if args.model == "happyhorse" and args.ref_video:
sys.exit("happyhorse 不支援參考影片（--ref-video），僅 wan 支援。")
n_refs = len(args.ref_image) + len(args.ref_video)
if n_refs > MAX_REFS[args.model]:
sys.exit(f"{args.model} 參考素材最多 {MAX_REFS[args.model]} 個，當前 {n_refs} 個。")
if not 2 <= args.duration <= 15:
sys.exit("時長只支援 2-15 整數秒。")
if args.ref_video and args.duration > 10:
sys.exit("含參考影片時時長上限 10 秒。")
mode = detect_mode(args)
model = FAMILY_MODELS[args.model][mode]
input_part = {"prompt": args.prompt}
if args.negative:
input_part["negative_prompt"] = args.negative
media = build_media(args, mode)
if media:
input_part["media"] = media
parameters = {
"resolution": args.resolution,
"duration": args.duration,
"prompt_extend": not args.no_prompt_extend,
}
if args.ratio:
parameters["ratio"] = args.ratio
if args.seed is not None:
parameters["seed"] = args.seed
body = {"model": model, "input": input_part, "parameters": parameters}
try:
resp = api_request(CREATE_URL, api_key, body,
extra_headers={"X-DashScope-Async": "enable"})
except (RuntimeError, OSError) as e:
sys.exit(f"提交失敗：{e}")
task_id = (resp.get("output") or {}).get("task_id") or resp.get("task_id")
if not task_id:
sys.exit(f"提交失敗，響應：{json.dumps(resp, ensure_ascii=False)[:500]}")
print(f"任務已提交 model={model} task_id={task_id}，出片通常需要 2-5 分鐘，開始輪詢…")
t0 = time.time()
time.sleep(POLL_FIRST_DELAY)
while True:
try:
task = api_request(TASK_URL.format(task_id), api_key)
except (RuntimeError, OSError) as e: # 網路抖動不中斷輪詢
print(f" 輪詢異常（繼續）：{e}")
time.sleep(POLL_INTERVAL)
continue
status = str(task.get("status", "unknown")).lower()
progress = task.get("progress", "")
elapsed = round(time.time() - t0)
# progress 常停在 30（上游只有 0/10/30/100 幾檔），不代表卡住
print(f" [{elapsed:>4}s] status={status}" + (f" progress={progress}" if progress != "" else ""))
if status in ("completed", "failed"):
break
if time.time() - t0 > POLL_TIMEOUT:
sys.exit(f"輪詢超時（{POLL_TIMEOUT}s）。任務仍在服務端，可稍後手動查詢：\n"
f" GET {TASK_URL.format(task_id)}")
time.sleep(POLL_INTERVAL)
if status != "completed":
err = task.get("error") or task.get("fail_reason") or task
sys.exit(f"生成失敗（status={status}）：{json.dumps(err, ensure_ascii=False)[:500]}"
"\n（失敗任務不計費）")
result_url = task.get("result_url")
if not result_url:
sys.exit(f"任務完成但未返回影片地址：{json.dumps(task, ensure_ascii=False)[:500]}")
path = resolve_path(args.out)
size = download(result_url, path)
print(f"影片已儲存至 {path}（{size / 1e6:.1f} MB，耗時 {round(time.time() - t0)}s，"
f"模型 {model}）")
if __name__ == "__main__":
main()
