#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 百炼 wan3（万相3.0）视频生成调用脚本 ==================================== 仅供 beauty-talk-science-video-script 技能在「用户确认脚本后」调用。 流程： 1. 解析参考素材（人物完整形象照 / 门店内景照 / 插画参考图 / 音色文件）。 - 公网 URL（--media 'https://...:类型'）：默认直接透传，wan3 原生支持公网 URL。 若 config 的 reupload_external_urls=true（wan3 拒收站外 URL 时），先下载再编码 base64 直传。 - 本地文件（生成的人物形象照 / 门店内景 / 插画）：读取后编码为 data:{mime};base64,{b64} 直接放入 media[].url（wan3 原生支持 base64 图像，免去文件上传端点）。 2. 组装 video-synthesis 异步请求（model / prompt / media / parameters），提交任务拿 task_id。 3. 轮询任务状态，直到 SUCCEEDED。 4. 下载生成的视频到本地 output_dir。 端点（与用户所给 LLM base_url 不同，视频服务走同 workspace、同地域、但**不带 llm- 前缀**的主机）： POST https://{WorkspaceId}.{region}.maas.aliyuncs.com/api/v1/services/aigc/video-generation/video-synthesis GET https://{WorkspaceId}.{region}.maas.aliyuncs.com/api/v1/tasks/{task_id} 用户所给 "https://llm-xxxx.cn-beijing.maas.aliyuncs.com/compatible-mode/v1" 是 **LLM 对话网关** （OpenAI 兼容模式），视频生成不使用它；脚本按 workspace_id + region 自动拼视频主机， 也可在 config.api_base 显式覆盖。 凭证： - DASHSCOPE_API_KEY 必须存在于环境变量（不写死在代码/配置里）。 - region / workspace_id / model / resolution / ratio / output_dir 写在同目录 config.json。 - 模型 / Endpoint / Key 必须同地域，跨地域调用会失败。 注意： - 本脚本只做「整段重新生成」，不使用 wan3 的视频编辑 / 延长模式（对应技能第 8 条：不支持二次创作）。 - 单次生成时长上限 30 秒（wan3.0 限制）。超过 30 秒的脚本请在技能侧先拆分/精简，本脚本强制 duration≤30。 """ import os import sys import json import time import base64 import argparse import mimetypes import requests # --------------------------------------------------------------------------- # # 配置与端点 # --------------------------------------------------------------------------- # def load_config(path): with open(path, "r", encoding="utf-8") as f: return json.load(f) def api_base(cfg): """视频服务主机：优先用 config.api_base；否则按 workspace_id + region 拼（不带 llm- 前缀）。""" if cfg.get("api_base"): return cfg["api_base"].rstrip("/") ws = cfg["workspace_id"] region = cfg["region"] return f"https://{ws}.{region}.maas.aliyuncs.com" def auth_headers(key): return { "Authorization": f"Bearer {key}", "Content-Type": "application/json", } def resolve_key(here): """解析百炼 / DashScope API Key。 优先级：环境变量 DASHSCOPE_API_KEY > 脚本同目录 .env（DASHSCOPE_API_KEY=...）。 注意：.env 含密钥，禁止随技能分发；重新打包技能前请删除本地 .env。 """ env = os.environ.get("DASHSCOPE_API_KEY") if env and env.strip(): return env.strip() env_path = os.path.join(here, ".env") if os.path.exists(env_path): try: with open(env_path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue if line.startswith("DASHSCOPE_API_KEY="): v = line.split("=", 1)[1].strip().strip('"').strip("'") if v: return v except Exception: pass sys.exit( "未找到 DASHSCOPE_API_KEY：\n" " (1) 运行前 export DASHSCOPE_API_KEY=sk-真实key（当前 shell 生效）；或\n" " (2) 在脚本同目录创建 .env，写入一行 DASHSCOPE_API_KEY=sk-真实key（详见 .env.example）。\n" "注意：北京地域创建、sk- 开头；密钥严禁提交或分发。" ) # --------------------------------------------------------------------------- # # 素材 -> URL（公网URL直传 / 本地文件 base64 直传） # --------------------------------------------------------------------------- # def local_file_to_data_uri(filepath): """读取本地文件，编码为 data:{mime};base64,{b64}，wan3 原生支持图像 base64 直传。""" mime = mimetypes.guess_type(filepath)[0] or "application/octet-stream" with open(filepath, "rb") as f: b64 = base64.b64encode(f.read()).decode("ascii") return f"data:{mime};base64,{b64}" def download_to_temp(url, suffix=".tmp"): """把公网 URL 下载到临时文件，返回本地路径（用于 reupload_external_urls）。""" import tempfile fd, tmppath = tempfile.mkstemp(suffix=suffix) os.close(fd) try: resp = requests.get(url, timeout=300, stream=True) resp.raise_for_status() with open(tmppath, "wb") as f: for chunk in resp.iter_content(8192): f.write(chunk) except Exception: if os.path.exists(tmppath): os.remove(tmppath) raise return tmppath # --------------------------------------------------------------------------- # # 任务创建与轮询 # --------------------------------------------------------------------------- # def create_task(base, key, payload): url = base + "/api/v1/services/aigc/video-generation/video-synthesis" headers = auth_headers(key) headers["X-DashScope-Async"] = "enable" resp = requests.post(url, headers=headers, json=payload, timeout=120) if resp.status_code not in (200, 202): raise RuntimeError(f"建任务失败：HTTP {resp.status_code} {resp.text[:500]}") return resp.json() def poll_task(base, key, task_id, interval, timeout): url = base + f"/api/v1/tasks/{task_id}" headers = {"Authorization": f"Bearer {key}"} waited = 0 while waited < timeout: try: resp = requests.get(url, headers=headers, timeout=60) if resp.status_code == 200: data = resp.json() status = ( (data.get("output", {}) or {}).get("task_status") or data.get("task_status") ) if status == "SUCCEEDED": return data if status in ("FAILED", "UNKNOWN"): raise RuntimeError(f"任务失败：{json.dumps(data, ensure_ascii=False)[:800]}") except requests.RequestException: pass time.sleep(interval) waited += interval raise RuntimeError("轮询超时（视频生成可能仍在后台，请用 task_id 手动查询）") def extract_video_url(data): out = data.get("output", {}) or {} for k in ("video_url", "videoUrl", "url"): if out.get(k): return out[k] for r in out.get("results") or []: if r.get("url"): return r["url"] raise RuntimeError("任务成功但未找到视频 URL") def download(url, out_path): resp = requests.get(url, timeout=300, stream=True) resp.raise_for_status() with open(out_path, "wb") as f: for chunk in resp.iter_content(8192): f.write(chunk) return out_path # --------------------------------------------------------------------------- # # 入口 # --------------------------------------------------------------------------- # def main(): here = os.path.dirname(os.path.abspath(__file__)) ap = argparse.ArgumentParser(description="百炼 wan3 视频生成（单段，≤30s）") ap.add_argument("--config", default=os.path.join(here, "config.json")) ap.add_argument("--prompt", required=True, help="单段视频提示词（含口播对白与 Image 1 / Audio 1 引用）") ap.add_argument( "--media", nargs="*", default=[], help="参考素材，格式 '路径或URL:类型'。类型：reference_image / reference_audio。" "如 /abs/char.jpg:reference_image 或 https://cdn.x/char.jpg:reference_image；" "公网 URL 默认直接透传；本地文件编码 base64 直传；详见 config.reupload_external_urls。", ) ap.add_argument("--duration", type=int, default=30, help="时长（秒），≤30；建议传技能算出的 Shot 时间分配表之和 T；传 -1 启用智能时长模式（按实际口播内容自动定长，上限 30s）") ap.add_argument("--ratio", default=None, help="画幅，如 9:16") ap.add_argument("--resolution", default=None, help="分辨率，如 720P") ap.add_argument("--model", default=None) ap.add_argument("--output", default=None, help="输出视频本地路径") ap.add_argument("--poll-interval", type=int, default=10) ap.add_argument("--timeout", type=int, default=1200) args = ap.parse_args() if args.duration > 30: sys.exit("wan3 单次生成上限 30 秒，duration 必须 ≤30。请先在技能侧精简口播。") if args.duration != -1 and args.duration < 2: sys.exit("duration 必须 ≥2 秒，或传 -1 启用智能时长模式（模型按实际内容自动定长，上限 30s）。") key = resolve_key(here) cfg = load_config(args.config) base = api_base(cfg) model = args.model or cfg.get("model", "wan3.0-video") resolution = args.resolution or cfg.get("resolution", "720P") ratio = args.ratio or cfg.get("ratio", "9:16") out_dir = os.path.join(here, cfg.get("output_dir", "outputs")) os.makedirs(out_dir, exist_ok=True) # 1) 解析素材 -> URL（公网URL直传 / 本地文件base64直传） reupload = bool(cfg.get("reupload_external_urls")) media = [] for m in args.media: raw = m.strip() if "://" in raw: # 公网 URL 形式：'https://...:reference_image|reference_audio'（类型可选） head, sep, tail = raw.rpartition(":") if sep and tail in ("reference_image", "reference_audio"): raw, mtype = head, tail else: mtype = "reference_image" else: raw, _, mtype = raw.partition(":") mtype = (mtype or "reference_image").strip() raw = raw.strip() if "://" in raw: if reupload: print(f"[下载并base64] {raw} ({mtype}) ...", file=sys.stderr) tmp = download_to_temp(raw) u = local_file_to_data_uri(tmp) try: os.remove(tmp) except OSError: pass else: print(f"[透传URL] {raw} ({mtype})", file=sys.stderr) u = raw media.append({"type": mtype, "url": u}) continue fpath = os.path.expanduser(raw) if not os.path.exists(fpath): sys.exit(f"素材不存在：{fpath}") print(f"[base64直传] {fpath} ({mtype}) ...", file=sys.stderr) u = local_file_to_data_uri(fpath) media.append({"type": mtype, "url": u}) # 2) 组装请求 payload = { "model": model, "input": {"prompt": args.prompt}, "parameters": { "resolution": resolution, "ratio": ratio, "duration": args.duration, "audio": True, "prompt_extend": False, "watermark": False, }, } if media: payload["input"]["media"] = media print("[提交] 创建视频生成任务 ...", file=sys.stderr) created = create_task(base, key, payload) task_id = (created.get("output", {}) or {}).get("task_id") or created.get("task_id") if not task_id: sys.exit(f"未返回 task_id：{json.dumps(created, ensure_ascii=False)[:500]}") print(f"[轮询] task_id={task_id}", file=sys.stderr) # 3) 轮询 result = poll_task(base, key, task_id, args.poll_interval, args.timeout) # 4) 下载 vurl = extract_video_url(result) out_path = args.output or os.path.join(out_dir, f"video_{int(time.time())}.mp4") out_path = os.path.abspath(out_path) _parent = os.path.dirname(out_path) if _parent: # 自动创建输出父目录：支持把成片直接落到「任务文件夹」而无需手动 mkdir os.makedirs(_parent, exist_ok=True) print(f"[下载] {vurl} -> {out_path}", file=sys.stderr) download(vurl, out_path) # 仅供技能解析的结构化结果（走 stdout） print("VIDEO_PATH=" + out_path) print("RESULT_JSON=" + json.dumps( {"task_id": task_id, "video_path": out_path, "video_url": vurl}, ensure_ascii=False, )) if __name__ == "__main__": main()