yhm-ai-video-duoduo/scripts/generate_video.py

#!/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")
    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 < 2:
        sys.exit("duration 必须 ≥2 秒。")

    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)
    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()