alibaba-wan-video

Alibaba Wan 视频

使用此技能选择并操作正确的 Wan 接口。即使版本号看起来相关，也要将托管模型 ID 和可下载检查点视为不同产品。

先选择路线

1. 询问任务、输入、输出时长／分辨率／宽高比、音频需求、部署区域／数据边界、预算、延迟、GPU 容量，以及是否接受托管推理。
2. 选择一条路线：
   • 托管 Model Studio，适用于当前 Wan 2.7 质量、2–15 秒片段、生成／自定义音频、1080P、托管式扩缩容、参考图转视频、续写或托管编辑。
   • 官方开放权重，适用于离线处理、获取权重、固定检查点以确保可复现、适配或自托管。当前官方公开系列为 Wan 2.2（以及旧版 Wan 2.1），而非托管版 Wan 2.7。
3. 绝不能在未告知的情况下替换为托管别名、较旧的 API 协议、社区量化版本或第三方网关。
4. 在发起任何计费请求或大型检查点下载之前，展示确切的模型／修订版本、区域、输入、输出设置、预估费用或存储空间，并取得明确批准。试运行不得产生任何网络变更。

以下事实和价格已于 2026-07-10 核实。执行时请重新检查所链接的第一方页面；可用性、别名、快照、限制和价格均可能变化。

区分不同产品系列

接口	当前生产环境选项	提供的功能	边界
托管版 Wan 2.7	wan2.7-t2v-2026-06-12、wan2.7-i2v-2026-04-25、wan2.7-r2v-2026-06-12、wan2.7-videoedit；也提供动态别名	T2V；首帧／尾帧 I2V；续写；实体／语音参考；编辑；720P／1080P；生成音频或提供音频	根据 Model Studio 条款提供的封闭式托管服务。API 名称不代表存在官方 Wan 2.7 检查点。需要稳定性时，请固定使用带日期的快照。
托管版 Wan 2.6	wan2.6-t2v、wan2.6-i2v、wan2.6-i2v-flash、wan2.6-r2v、wan2.6-r2v-flash；弗吉尼亚区域提供 -us T2V／I2V 变体	更多区域选择、旧版 API 模式、支持音频的生成；flash 版本可能更便宜	使用旧版 size 协议。不要向其发送 Wan 2.7 的 resolution／ratio 请求体。
托管旧版	Wan 2.5 预览版；Wan 2.2／2.1 T2V、I2V、关键帧／VACE、语音／动画变体	兼容性和专用工作流	功能、固定时长、分辨率档位、命名方式（北京区域使用 wanx2.1-*）和价格各不相同。请查阅对应的旧版参考文档。托管的 wan2.2-* ID 并不能证明其与公开检查点逐位一致。
开放版 Wan 2.2	T2V-A14B、I2V-A14B、TI2V-5B、S2V-14B、Animate-14B	离线 T2V／I2V、语音驱动视频、角色动画／替换；Apache-2.0 检查点和代码	不提供托管 SLA、自动安全服务或托管式扩缩容。官方 A14B 示例要求至少 80 GB VRAM；TI2V-5B 文档要求至少 24 GB。
开放版 Wan 2.1	T2V-14B／1.3B、I2V-14B、FLF2V-14B、VACE-14B／1.3B	较早但文档完善的 T2V／I2V、首尾帧及编辑／控制工作流	Apache-2.0，但架构较旧。官方 T2V-1.3B 指南建议使用 480P，并报告需要 8.19 GB VRAM。

Wan 2.2 A14B 是一个双专家 MoE：总参数量约为 27B，每个去噪步骤激活约 14B 参数。TI2V-5B 使用高压缩率 VAE，官方支持以 24 fps 输出 1280x704／704x1280 分辨率的视频。这些事实并不意味着 5B 和 A14B 属于可互换的质量档位。

托管 Model Studio

区域与端点契约

区域、API 密钥、模型可用性、服务部署范围和端点必须匹配。密钥不能跨区域使用。

访问区域	生产环境基地址	Wan 相关的服务范围	备注
新加坡	https://{WorkspaceId}.ap-southeast-1.maas.aliyuncs.com	国际：全球推理节点，中国大陆除外	推荐使用工作空间专属域名。共享的 https://dashscope-intl.aliyuncs.com 仍可供现有集成使用。此处有 Wan 2.7 的相关文档。
中国（北京）	https://{WorkspaceId}.cn-beijing.maas.aliyuncs.com	中国大陆：推理仅限中国	推荐使用工作空间专属域名。共享的 https://dashscope.aliyuncs.com 仍可使用。此处有 Wan 2.7 的相关文档。
德国（法兰克福）	https://{WorkspaceId}.eu-central-1.maas.aliyuncs.com	当前列出的 Wan 2.6 模型可在全球范围使用	法兰克福通常支持 Global 和 EU 工作空间范围，但当前 Wan 定价仅将 T2V/I2V/R2V 2.6 列为 Global。不要假设这些模型可在 EU 范围的工作空间中使用；请在控制台中确认。目前未列出 Wan 2.7。
美国（弗吉尼亚州）	https://dashscope-us.aliyuncs.com	Global，或使用 -us 模型 ID 的 US	当前定价列出了 Wan 2.6 Global 版本以及 wan2.6-t2v-us/wan2.6-i2v-us；目前尚不支持工作空间专属域名。

工作空间专属域名可将流量隔离到对应工作空间，是文档中建议的生产环境用法。不要仅凭访问 URL 推断数据驻留位置：推理位置由服务部署范围决定，而存储位置由访问区域决定。请同时记录两者。

Wan 2.7 请求格式

所有当前 Wan 2.7 视频模式均使用异步 HTTP：

• 创建：POST {base}/api/v1/services/aigc/video-generation/video-synthesis
• 必需标头：Authorization: Bearer ...、Content-Type: application/json、X-DashScope-Async: enable
• 轮询：GET {base}/api/v1/tasks/{task_id}
• 状态：PENDING、RUNNING、SUCCEEDED、FAILED、CANCELED、UNKNOWN
• 大约每隔 15 秒轮询一次，并使用有上限的退避策略。保存首次返回的 task_id；不要仅仅因为生成缓慢就再次 POST。
• 任务查询和签名后的 video_url 值会在 24 小时后过期。请尽快下载。成功生成的输出为 MP4/H.264。

如果一个工作进程必须支持所有 2.7 模式，请优先使用 HTTP。目前的 SDK 矩阵并不统一：DashScope Python/Java 的 T2V 包装器仅支持 Wan 2.6 及更早版本（文档注明的最低版本分别为 1.25.8/2.22.6），而 2.7 I2V 和 R2V 包装器要求 Python 1.25.16 或 Java 2.22.14。SDK 的“同步”调用只是包装异步服务并阻塞等待；它们不会改变作业时长、计费方式或结果过期时间。固定依赖版本前，请重新核实版本。

核心 T2V 请求体：

{
  "model": "wan2.7-t2v-2026-06-12",
  "input": {
    "prompt": "Generate a single-shot video. Wide shot of a red paper kite rising above a quiet salt flat at dawn; slow dolly forward, soft wind and natural ambience.",
    "negative_prompt": "text overlays, logos, warped kite, abrupt cuts"
  },
  "parameters": {
    "resolution": "720P",
    "ratio": "16:9",
    "duration": 5,
    "prompt_extend": false,
    "watermark": true,
    "seed": 104729
  }
}

Wan 2.7 T2V 接受最多 5,000 个字符的中文或英文提示词，以及最多 500 个字符的负面提示词；超出长度的文本会被截断，因此请在本地进行验证。duration 是 2–15 之间的整数（默认值为 5）；resolution 可以是 720P 或 1080P（默认值为 1080P）；画面比例可以是 16:9、9:16、1:1、4:3 或 3:4。提示词扩展默认启用，水印默认关闭；固定种子有助于提高可复现性，但无法保证完全一致。通过自然语言提示词结构控制单镜头或多镜头；旧版 shot_type 对 2.7 无效。

其他 2.7 模式只需更改 input/model：

// First and last frame I2V. Use only first_frame for first-frame generation.
{
  "model": "wan2.7-i2v-2026-04-25",
  "input": {"prompt": "The lantern drifts downstream as dusk deepens.", "media": [
    {"type": "first_frame", "url": "https://assets.example/first.webp"},
    {"type": "last_frame", "url": "https://assets.example/last.webp"}
  ]},
  "parameters": {"resolution": "720P", "duration": 6, "prompt_extend": false, "watermark": true}
}

// Continuation. first_clip may be combined with last_frame when supported by the exact model docs.
{
  "model": "wan2.7-i2v-2026-04-25",
  "input": {"prompt": "Continue the same camera move and weather.", "media": [
    {"type": "first_clip", "url": "https://assets.example/source.mp4"}
  ]},
  "parameters": {"resolution": "720P", "duration": 10, "prompt_extend": false, "watermark": true}
}

// Reference-to-video. Refer to assets by their one-based Image/Video order in the prompt.
{
  "model": "wan2.7-r2v-2026-06-12",
  "input": {"prompt": "Video 1 enters and gives Image 1 the compass.", "media": [
    {"type": "reference_video", "url": "https://assets.example/person.mp4", "reference_voice": "https://assets.example/voice.wav"},
    {"type": "reference_image", "url": "https://assets.example/robot.webp"}
  ]},
  "parameters": {"resolution": "720P", "ratio": "16:9", "duration": 8, "prompt_extend": false, "watermark": true}
}

对于 I2V，输入图像可以是无 alpha 通道的 JPEG/JPG/PNG、BMP 或 WEBP，文件大小不超过 20 MB，单边像素尺寸为 240–8,000，宽高比为 1:8–8:1。续接用的 first_clip 必须为 MP4/MOV，时长为 2–10 秒，文件大小不超过 100 MB，单边像素尺寸为 240–4,096，宽高比为 1:8–8:1。在续接模式下，duration 表示视频的最终总时长，而非新增秒数；输出宽高比沿用片段的宽高比，计费时长为完整输出视频的时长。

对于 R2V，存在参考视频时，时长为 2–10 秒；不含参考视频时，时长为 2–15 秒。最多可提供一张首帧，以及合计 1–5 张参考图像/个参考视频；角色参考素材必须只包含一个角色。参考图像遵循上述相同的 20 MB 限制。参考视频必须为 MP4/MOV，时长为 1–30 秒，文件大小不超过 100 MB，单边像素尺寸为 240–4,096；reference_voice 必须为 WAV/MP3，时长为 1–10 秒，文件大小不超过 15 MB。提供首帧时，首帧会覆盖 ratio。编辑视频时，请使用 wan2.7-videoedit，提供一个 {"type":"video"}，以及最多四个 reference_image 素材；计费包含输入和输出视频的时长。上传前，请务必重新查阅对应的 API 页面，确认可能变化的限制。

可计费、优先试运行的 T2V 客户端

这个完整示例固定使用新加坡快照，默认输出经过脱敏的免费计划，将精确请求、工作区和输出策略以及最新定价证据绑定到审批中，最多创建一个任务，持久化可从崩溃中恢复的脱敏记录，通过精确主机名和公共 IP 策略下载，且不会转发 API 密钥，并在原子化且不覆盖现有文件的发布前进行验证。它刻意不接受任意基础 URL 或输出 URL。

#!/usr/bin/env python3
import argparse, hashlib, http.client, ipaddress, json, os, random, socket, ssl
import pathlib, subprocess, sys, tempfile, threading, time, uuid
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from urllib.parse import urlsplit
import requests

MODEL = "wan2.7-t2v-2026-06-12"
PRICE_USD_PER_SECOND = {"720P": Decimal("0.10"), "1080P": Decimal("0.15")}
REGION_SUFFIX = {"sg": "ap-southeast-1.maas.aliyuncs.com"}
TERMINAL = {"SUCCEEDED", "FAILED", "CANCELED", "UNKNOWN"}
MAX_JSON = 1024 * 1024
MAX_VIDEO = 1024 * 1024 * 1024
MAX_TOOL = 1024 * 1024
EXPECTED = {
    ("720P", "16:9"): (1280, 720), ("720P", "9:16"): (720, 1280),
    ("720P", "1:1"): (960, 960), ("720P", "4:3"): (1104, 832),
    ("720P", "3:4"): (832, 1104), ("1080P", "16:9"): (1920, 1080),
    ("1080P", "9:16"): (1080, 1920), ("1080P", "1:1"): (1440, 1440),
    ("1080P", "4:3"): (1648, 1248), ("1080P", "3:4"): (1248, 1648)}

class APIError(RuntimeError):
    def __init__(self, message, status=None, retry_after=None, body_sha256=None):
        super().__init__(message); self.status = status; self.retry_after = retry_after; self.body_sha256 = body_sha256

def canonical(value):
    return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()

def fsync_directory(path):
    if os.name == "nt": return
    fd = os.open(path, os.O_RDONLY)
    try: os.fsync(fd)
    finally: os.close(fd)

def atomic_json(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp = tempfile.mkstemp(prefix=path.name + ".", dir=path.parent)
    try:
        with os.fdopen(fd, "w", encoding="utf-8", newline="\n") as f:
            json.dump(value, f, ensure_ascii=False, indent=2, sort_keys=True)
            f.write("\n"); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); fsync_directory(path.parent)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)

def read_record(path):
    if path.stat().st_size > MAX_JSON: raise RuntimeError("attempt record exceeds cap")
    value = json.loads(path.read_bytes())
    if not isinstance(value, dict): raise RuntimeError("attempt record must be an object")
    return value

def run_bounded(argv, timeout):
    process = subprocess.Popen(argv, shell=False, stdin=subprocess.DEVNULL,
        stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    out, err, total, lock, overflow = bytearray(), bytearray(), [0], threading.Lock(), threading.Event()
    def drain(stream, bucket):
        while True:
            chunk = stream.read(65536)
            if not chunk: break
            with lock:
                room = max(0, MAX_TOOL - total[0]); bucket.extend(chunk[:room]); total[0] += len(chunk)
                if total[0] > MAX_TOOL: overflow.set()
            if overflow.is_set():
                try: process.kill()
                except OSError: pass
    threads = [threading.Thread(target=drain, args=(process.stdout, out), daemon=True),
               threading.Thread(target=drain, args=(process.stderr, err), daemon=True)]
    for thread in threads: thread.start()
    timed_out = False
    try: code = process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        timed_out = True; process.kill(); code = process.wait()
    for thread in threads: thread.join()
    return code, bytes(out), bytes(err), timed_out, overflow.is_set()

def api_json(session, method, url, payload=None, timeout=(10, 60)):
    with session.request(method, url, json=payload, timeout=timeout, allow_redirects=False, stream=True) as response:
        raw = bytearray()
        for chunk in response.iter_content(65536):
            raw.extend(chunk)
            if len(raw) > MAX_JSON: raise RuntimeError("API response exceeds cap")
        digest = hashlib.sha256(raw).hexdigest()
        if not 200 <= response.status_code < 300:
            retry = response.headers.get("Retry-After")
            try: retry = min(60.0, max(0.0, float(retry)))
            except (TypeError, ValueError): retry = None
            raise APIError(f"API HTTP {response.status_code}", response.status_code, retry, digest)
        try: value = json.loads(raw)
        except json.JSONDecodeError as exc: raise APIError("API returned invalid JSON", response.status_code, body_sha256=digest) from exc
        if not isinstance(value, dict): raise RuntimeError("API response is not an object")
        return value

def safe_get(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except APIError as exc:
            if exc.status != 429 and (exc.status is None or exc.status < 500): raise
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = exc.retry_after or min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        except requests.RequestException:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = min(20, 2 ** attempt) * (1 + random.random() * 0.25)
        if time.monotonic() + delay >= deadline: raise TimeoutError("poll retry would cross deadline")
        time.sleep(delay)

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address):
        super().__init__(host, 443, timeout=120, context=ssl.create_default_context()); self.address = address
    def connect(self):
        sock = socket.create_connection((self.address, 443), self.timeout)
        try: self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
        except BaseException:
            sock.close(); raise

def download_video(url, stage_path, allowed_hosts):
    parsed = urlsplit(url); host = (parsed.hostname or "").lower()
    if (parsed.scheme != "https" or parsed.username or parsed.password or parsed.fragment
            or parsed.port not in (None, 443) or not parsed.path or host not in allowed_hosts):
        raise RuntimeError("result URL is not an exact approved HTTPS host")
    try: ipaddress.ip_address(host); raise RuntimeError("IP-literal result host rejected")
    except ValueError: pass
    addresses = sorted({item[4][0] for item in socket.getaddrinfo(host, 443, type=socket.SOCK_STREAM)})
    if not addresses or any(not ipaddress.ip_address(address).is_global for address in addresses):
        raise RuntimeError("result host did not resolve only to public addresses")
    connection = PinnedHTTPSConnection(host, addresses[0]); created = False
    try:
        target = parsed.path + (("?" + parsed.query) if parsed.query else "")
        connection.request("GET", target, headers={"Accept": "video/mp4,application/octet-stream"})
        response = connection.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("result redirect rejected")
        if response.status != 200: raise RuntimeError(f"result HTTP {response.status}")
        mime = response.getheader("Content-Type", "").split(";", 1)[0].lower()
        if mime not in {"video/mp4", "application/octet-stream"}: raise RuntimeError("result MIME rejected")
        length = response.getheader("Content-Length")
        if length and (not length.isdigit() or int(length) > MAX_VIDEO): raise RuntimeError("declared result size rejected")
        fd = os.open(stage_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600); created = True; total = 0
        with os.fdopen(fd, "wb") as f:
            while True:
                chunk = response.read(1024 * 1024)
                if not chunk: break
                total += len(chunk)
                if total > MAX_VIDEO: raise RuntimeError("result crossed byte cap")
                f.write(chunk)
            f.flush(); os.fsync(f.fileno())
        return stage_path
    except BaseException:
        if created and stage_path.exists(): stage_path.unlink()
        raise
    finally: connection.close()

def inspect_mp4(path, output, resolution, ratio, duration):
    with path.open("rb") as f: head = f.read(12)
    if len(head) < 12 or head[4:8] != b"ftyp": raise RuntimeError("MP4 signature rejected")
    args = ["ffprobe", "-v", "error", "-show_entries",
            "format=duration,size:stream=codec_type,codec_name,width,height,r_frame_rate", "-of", "json", str(path)]
    code, out, _, timed_out, overflow = run_bounded(args, 30)
    if code or timed_out or overflow: raise RuntimeError("bounded ffprobe failed")
    probe = json.loads(out); streams = probe.get("streams", []); videos = [s for s in streams if s.get("codec_type") == "video"]
    if len(videos) != 1 or videos[0].get("codec_name") != "h264": raise RuntimeError("expected one H.264 stream")
    width, height = videos[0].get("width"), videos[0].get("height")
    if (width, height) != EXPECTED[(resolution, ratio)]: raise RuntimeError("dimensions do not match approval")
    try: actual = Decimal(str(probe.get("format", {}).get("duration", "")))
    except InvalidOperation as exc: raise RuntimeError("invalid media duration") from exc
    if not actual.is_finite() or abs(actual - Decimal(duration)) > Decimal("0.75"):
        raise RuntimeError("media duration outside tolerance")
    code, _, _, timed_out, overflow = run_bounded(["ffmpeg", "-v", "error", "-i", str(path), "-f", "null", "-"], 300)
    if code or timed_out or overflow: raise RuntimeError("bounded full decode failed")
    digest = hashlib.sha256()
    with path.open("rb") as f:
        for chunk in iter(lambda: f.read(1024 * 1024), b""): digest.update(chunk)
    return {"path": str(output), "bytes": path.stat().st_size, "sha256": digest.hexdigest(),
            "media": {"duration": str(actual), "width": width, "height": height,
                      "videoCodec": "h264", "audioStreams": sum(s.get("codec_type") == "audio" for s in streams)}}

def main():
    p = argparse.ArgumentParser()
    p.add_argument("--prompt", required=True)
    p.add_argument("--negative-prompt", default="text overlays, logos, deformed objects")
    p.add_argument("--region", choices=sorted(REGION_SUFFIX), default="sg")
    p.add_argument("--resolution", choices=sorted(PRICE_USD_PER_SECOND), default="720P")
    p.add_argument("--ratio", choices=["16:9", "9:16", "1:1", "4:3", "3:4"], default="16:9")
    p.add_argument("--duration", type=int, default=5)
    p.add_argument("--seed", type=int, default=104729)
    p.add_argument("--workspace-id", required=True, help="non-secret Singapore Model Studio workspace ID")
    p.add_argument("--out-dir", type=pathlib.Path, default=pathlib.Path("wan-run"))
    p.add_argument("--attempt-id", required=True)
    p.add_argument("--max-usd", required=True)
    p.add_argument("--pricing-checked-at", required=True)
    p.add_argument("--pricing-evidence-sha256", required=True)
    p.add_argument("--delivery-host", action="append", required=True,
                   help="exact approved result host; repeat for each host")
    p.add_argument("--approval-sha256")
    p.add_argument("--execute", action="store_true"); p.add_argument("--resume", action="store_true")
    p.add_argument("--show-full-plan", action="store_true")
    a = p.parse_args()

    if not 1 <= len(a.prompt) <= 5000 or len(a.negative_prompt) > 500:
        p.error("prompt must be 1..5000 chars; negative prompt <=500")
    if not 2 <= a.duration <= 15 or not 0 <= a.seed <= 2147483647:
        p.error("duration must be 2..15 and seed 0..2147483647")
    try: attempt = uuid.UUID(a.attempt_id); ceiling = Decimal(a.max_usd)
    except (ValueError, InvalidOperation) as exc: p.error(f"attempt ID or maximum invalid: {exc}")
    if attempt.version != 4 or str(attempt) != a.attempt_id.lower(): p.error("attempt ID must be canonical UUIDv4")
    estimate = PRICE_USD_PER_SECOND[a.resolution] * a.duration
    if not ceiling.is_finite() or ceiling <= 0 or ceiling < estimate:
        p.error("maximum must be finite, positive, and cover the estimate")
    if (len(a.pricing_evidence_sha256) != 64
            or any(c not in "0123456789abcdef" for c in a.pricing_evidence_sha256)):
        p.error("pricing evidence must be a lowercase SHA-256")
    try: checked_at = datetime.fromisoformat(a.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError as exc: p.error(f"pricing timestamp must be ISO-8601: {exc}")
    if checked_at.tzinfo is None: p.error("pricing timestamp must include a timezone")
    allowed_hosts = sorted(set(host.lower().rstrip(".") for host in a.delivery_host))
    if (not allowed_hosts or len(allowed_hosts) > 8
            or any(not host or len(host) > 253
                   or any(c not in "abcdefghijklmnopqrstuvwxyz0123456789-." for c in host) for host in allowed_hosts)):
        p.error("delivery hosts must be 1..8 exact DNS names")
    for host in allowed_hosts:
        try: ipaddress.ip_address(host); p.error("delivery hosts cannot be IP literals")
        except ValueError: pass
    out_dir = a.out_dir.resolve(); record_path = out_dir / "attempt.json"
    video_path = out_dir / "output.mp4"; stage_path = out_dir / ".output.mp4.wan-stage"
    body = {"model": MODEL, "input": {"prompt": a.prompt,
            "negative_prompt": a.negative_prompt}, "parameters": {
            "resolution": a.resolution, "ratio": a.ratio, "duration": a.duration,
            "prompt_extend": False, "watermark": True, "seed": a.seed}}
    workspace = a.workspace_id
    if (not workspace or len(workspace) > 100
            or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-" for c in workspace)):
        p.error("workspace ID contains unsupported characters")
    host = f"{workspace}.{REGION_SUFFIX[a.region]}"
    request_hash = hashlib.sha256(canonical(body)).hexdigest()
    envelope = {"backend": "alibaba-model-studio-hosted", "region": a.region,
        "serviceDeploymentScope": "International", "host": host,
        "endpoint": "/api/v1/services/aigc/video-generation/video-synthesis",
        "request": body, "attemptId": str(attempt), "attemptRecord": str(record_path),
        "outputPolicy": {"path": str(video_path), "stagePath": str(stage_path),
                         "exactDeliveryHosts": allowed_hosts, "maxBytes": MAX_VIDEO,
                         "mime": ["video/mp4", "application/octet-stream"], "noOverwrite": True,
                         "requiredCodec": "h264", "fullDecode": True},
        "cost": {"pricingCheckedAt": checked_at.isoformat(), "pricingEvidenceSha256": a.pricing_evidence_sha256,
                 "rateUsdPerOutputSecond": str(PRICE_USD_PER_SECOND[a.resolution]),
                 "seconds": a.duration, "creates": 1, "estimatedMaxUsd": str(estimate),
                 "approvedMaxUsd": str(ceiling)}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest()
    safe_body = {**body, "input": {"prompt": "sha256:" + hashlib.sha256(a.prompt.encode()).hexdigest(),
                                    "negative_prompt": "sha256:" + hashlib.sha256(a.negative_prompt.encode()).hexdigest()}}
    print(json.dumps({"dryRun": not a.execute, "approvalSha256": approval,
        "plan": {**envelope, "request": safe_body}}, ensure_ascii=False, indent=2))
    if a.show_full_plan: print(json.dumps({"protectedFullPlan": envelope}, ensure_ascii=False, indent=2))
    if not a.execute:
        print("PLAN ONLY: no API request was made", file=sys.stderr)
        return 0
    if a.approval_sha256 != approval: raise SystemExit("exact approval digest mismatch")
    age = datetime.now(timezone.utc) - checked_at.astimezone(timezone.utc)
    if age.total_seconds() < -300 or age.total_seconds() > 86400:
        raise SystemExit("pricing evidence must be rechecked within 24 hours of execution")
    base = f"https://{workspace}.{REGION_SUFFIX[a.region]}"
    if a.resume:
        if not record_path.exists(): raise SystemExit("resume record missing")
        record = read_record(record_path)
        if record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash:
            raise SystemExit("resume does not match exact approved request")
        task_id = record.get("task_id")
        if not isinstance(task_id, str) or not task_id: raise SystemExit("no known task ID; never replay create")
        expected_artifact = record.get("artifact_staged") or record.get("artifact")
        candidate = video_path if video_path.exists() else stage_path if stage_path.exists() else None
        if candidate is not None:
            if not isinstance(expected_artifact, dict):
                if record.get("status") == "downloading" and candidate == stage_path:
                    stage_path.unlink(); fsync_directory(stage_path.parent)
                else: raise SystemExit("unclaimed output/stage exists; refusing overwrite or adoption")
            else:
                artifact = inspect_mp4(candidate, video_path, a.resolution, a.ratio, a.duration)
                if (artifact.get("sha256") != expected_artifact.get("sha256")
                        or artifact.get("bytes") != expected_artifact.get("bytes")):
                    raise SystemExit("recovered artifact does not match durable staged evidence")
                if candidate == stage_path:
                    if video_path.exists(): raise SystemExit("output appeared during recovery")
                    os.replace(stage_path, video_path); fsync_directory(video_path.parent)
                record.update(status="artifact_saved", artifact=artifact, updated_unix=int(time.time()))
                record.pop("artifact_staged", None); atomic_json(record_path, record)
                print(json.dumps(record, indent=2)); return 0
        key = os.environ.get("DASHSCOPE_API_KEY")
        if not key: raise SystemExit("Set DASHSCOPE_API_KEY to resume polling")
        session = requests.Session(); session.headers.update({"Authorization": f"Bearer {key}",
            "Content-Type": "application/json", "X-DashScope-Async": "enable"})
    else:
        key = os.environ.get("DASHSCOPE_API_KEY")
        if not key: raise SystemExit("Set DASHSCOPE_API_KEY")
        session = requests.Session(); session.headers.update({"Authorization": f"Bearer {key}",
            "Content-Type": "application/json", "X-DashScope-Async": "enable"})
        out_dir.parent.mkdir(parents=True, exist_ok=True)
        try: out_dir.mkdir(exist_ok=False)
        except FileExistsError as exc: raise SystemExit("attempt directory already claimed; use resume") from exc
        record = {"attempt_id": str(attempt), "approval_sha256": approval, "request_sha256": request_hash,
                  "status": "posting", "task_id": None, "created_unix": int(time.time())}
        atomic_json(record_path, record)
        try: created = api_json(session, "POST", base + envelope["endpoint"], body, (10, 120))
        except Exception as exc:
            known_rejection = (isinstance(exc, APIError) and exc.status is not None and 400 <= exc.status < 500
                               and exc.status not in {408, 409, 425, 429})
            record.update(status="create_rejected" if known_rejection else "create_outcome_unknown",
                          error_type=type(exc).__name__, http_status=getattr(exc, "status", None),
                          error_body_sha256=getattr(exc, "body_sha256", None))
            atomic_json(record_path, record)
            if known_rejection: raise RuntimeError("provider rejected create before acceptance") from exc
            raise RuntimeError("create outcome unknown; reconcile billing/logs, never replay") from exc
        task_id = created.get("output", {}).get("task_id")
        if (not isinstance(task_id, str) or not 1 <= len(task_id) <= 200
                or any(c not in "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789-_." for c in task_id)):
            record.update(status="create_outcome_unknown", error_type="MissingOrInvalidTaskId",
                          create_response_sha256=hashlib.sha256(canonical(created)).hexdigest())
            atomic_json(record_path, record)
            raise RuntimeError("create returned no safe task ID; reconcile, never replay")
        record.update(status="task_id_saved", task_id=task_id, request_id=created.get("request_id"),
                      create_response_sha256=hashlib.sha256(canonical(created)).hexdigest(), updated_unix=int(time.time()))
        atomic_json(record_path, record)

    deadline = time.monotonic() + 12 * 60
    result = None
    while time.monotonic() < deadline:
        result = safe_get(lambda: api_json(session, "GET", base + f"/api/v1/tasks/{task_id}", timeout=(10, 30)), deadline)
        status = result.get("output", {}).get("task_status")
        print(f"task={task_id} status={status}", file=sys.stderr)
        if status in TERMINAL:
            break
        if status not in {"PENDING", "RUNNING"}: raise RuntimeError("unrecognized task status")
        if time.monotonic() + 15 >= deadline: break
        time.sleep(15)
    if result is None or result.get("output", {}).get("task_status") in {"PENDING", "RUNNING"}:
        record.update(status="polling_paused", updated_unix=int(time.time())); atomic_json(record_path, record)
        raise TimeoutError(f"Polling timed out; resume GET with task_id {task_id}, do not POST")
    output = result.get("output", {})
    if output.get("task_status") != "SUCCEEDED":
        message = str(output.get("message", ""))
        record.update(status="generation_" + str(output.get("task_status", "unknown")).lower(),
                      service_code=output.get("code"), message_sha256=hashlib.sha256(message.encode()).hexdigest())
        atomic_json(record_path, record); raise RuntimeError("task ended without a successful artifact")
    url = output.get("video_url", "")
    if not isinstance(url, str) or not url: raise RuntimeError("successful task omitted result URL")
    usage = result.get("usage") or {}; actual_cost = None
    try:
        seconds = Decimal(str(usage.get("output_video_duration")))
        if seconds.is_finite() and seconds >= 0: actual_cost = PRICE_USD_PER_SECOND[a.resolution] * seconds
    except InvalidOperation: pass
    record.update(status="downloading", result_response_sha256=hashlib.sha256(canonical(result)).hexdigest(),
                  result_url_sha256=hashlib.sha256(url.encode()).hexdigest(), usage=usage,
                  actual_cost_usd=None if actual_cost is None else str(actual_cost),
                  actual_cost_within_approval=None if actual_cost is None else actual_cost <= ceiling,
                  stage_path=str(stage_path), updated_unix=int(time.time()))
    atomic_json(record_path, record)
    try:
        download_video(url, stage_path, set(allowed_hosts))
        artifact = inspect_mp4(stage_path, video_path, a.resolution, a.ratio, a.duration)
    except BaseException as exc:
        record.update(status="artifact_validation_failed", error_type=type(exc).__name__)
        atomic_json(record_path, record); raise
    record.update(status="artifact_staged", artifact_staged=artifact, updated_unix=int(time.time()))
    atomic_json(record_path, record)
    if video_path.exists(): raise RuntimeError("output appeared before publication")
    os.replace(stage_path, video_path); fsync_directory(video_path.parent)
    record.update(status="artifact_saved", artifact=artifact, rights_review="required",
                  moderation_review="required", creative_qa="pending", ai_disclosure="required",
                  provenance_credentials="not_documented", updated_unix=int(time.time()))
    record.pop("artifact_staged", None); atomic_json(record_path, record)
    print(json.dumps(record, ensure_ascii=False, indent=2)); return 0

if __name__ == "__main__":
    main()

安装 requests，然后先运行计划。只有第二条命令可以创建付费任务：

python wan_hosted.py --workspace-id "$WAN_WORKSPACE_ID" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.50 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$WAN_RESULT_HOST" --prompt "Generate a single-shot video. A red kite rises over a salt flat at dawn."
DASHSCOPE_API_KEY=... python wan_hosted.py --execute --approval-sha256 "<exact digest>" --workspace-id "$WAN_WORKSPACE_ID" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.50 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$WAN_RESULT_HOST" --prompt "Generate a single-shot video. A red kite rises over a salt flat at dawn."

BILLING_SHA256 是实际审阅过的带日期第一方定价证据的摘要。WAN_RESULT_HOST 是当前 Model Studio/OSS 交付协议允许的精确主机名，绝不能根据返回的 URL 推断出通配符。仅当需要额外的精确获批主机名时，才重复使用 --delivery-host。执行时会拒绝超过 24 小时的定价证据。仅当持久化尝试记录中已包含任务 ID 时，才使用 --execute --resume 和完全不变的参数；posting 或 create_outcome_unknown 状态需要与账单、日志或支持团队核对，绝不能重放。

该可执行程序刻意只接受 --region sg。对于北京或其他区域，请先用当前定价页面中的值替换区域允许列表和价格表，并重新审查端点/模型协议；不要绕过此默认失败边界。

成本和容量控制

当前标准费率，以美元计，按可计费秒数计算：

访问区域 / 范围	模型	720P	1080P
新加坡 / 国际	Wan 2.7 T2V、I2V、R2V、视频编辑	$0.10	$0.15
北京 / 中国大陆	Wan 2.7 T2V、I2V、R2V、视频编辑	$0.086012	$0.143353
法兰克福 / 全球	Wan 2.6 T2V、I2V、R2V	$0.086012	$0.143353
弗吉尼亚州 / 全球	Wan 2.6 T2V、I2V、R2V	$0.086012	$0.143353
弗吉尼亚州 / 美国	wan2.6-t2v-us、wan2.6-i2v-us	$0.10	$0.15

不同模式的计费依据仍然不同。I2V 和 T2V 输入免费，因此其可计费时长为成功生成的视频时长。对于 Wan 2.7 R2V，若有 N 个参考视频，每个参考视频按 min(actual duration, 5/N seconds) 计费，并将这些输入时长计入输出时长；图片和首帧不计入 N。视频编辑按输入视频时长加输出视频时长计费。失败的请求不计费，也不消耗免费额度。Flash、旧版和无声输出的价格不同；请查阅实时价格表，不要自行推算。不要将促销/免费额度视为预算控制手段。

新加坡和北京对列出的 Wan 2.7 T2V/I2V/R2V/编辑模型实施速率限制：每个 Alibaba Cloud 账号每秒最多提交 5 个任务，同时最多运行 5 个任务。限额会合并计算 RAM 用户、工作区和密钥的用量。速率限制不等于累计支出限额：请使用账号预算提醒、本地作业队列、并发信号量和单个作业的成本上限。请重新核实这些值，因为它们可能会变。

官方开放权重部署

除非用户明确接受社区移植版本或量化版本，否则只能使用官方 Wan-Video GitHub 和 Wan-AI Hugging Face/ModelScope 资源。记录以下信息：

• 仓库 URL 和不可变的 Git 提交；
• 检查点仓库和不可变的修订版本/提交；
• 每个文件的哈希值、软件包锁定文件、CUDA/PyTorch/驱动程序版本、GPU 类型、提示词、负面提示词、随机种子、尺寸、帧数、采样设置和输出哈希值；
• 每个固定资源随附的许可证文本，以及所有依赖项的许可证。

不要把下载权重当作顺带操作。首先运行 git ls-remote 或检查注册表元数据，说明预期的存储需求和许可证，获得批准后，再固定已审核的哈希值。截至核验时，官方 Wan 2.2 仓库没有 GitHub 发布资源；main 不是生产版本。

操作人员批准并固定两个修订版本后，可使用以下示例：

git clone https://github.com/Wan-Video/Wan2.2.git
cd Wan2.2
git checkout --detach "$REVIEWED_WAN_GIT_COMMIT"
python -m venv .venv
. .venv/bin/activate
python -m pip install --require-hashes -r requirements.lock
huggingface-cli download Wan-AI/Wan2.2-TI2V-5B \
  --revision "$REVIEWED_HF_COMMIT" --local-dir ./Wan2.2-TI2V-5B
sha256sum -c checkpoint.sha256
python generate.py --task ti2v-5B --size '1280*704' \
  --ckpt_dir ./Wan2.2-TI2V-5B --offload_model True --convert_model_dtype \
  --t5_cpu --base_seed 104729 --save_file ./wan-ti2v-5b.mp4 \
  --prompt "Wide shot of a red paper kite rising above a quiet salt flat at dawn."
ffprobe -v error -show_streams -show_format ./wan-ti2v-5b.mp4
ffmpeg -v error -i ./wan-ti2v-5b.mp4 -f null -

requirements.lock 和 checkpoint.sha256 是组织生成的审核文件，并非 Wan 仓库提供的文件。请在受控的依赖解析阶段创建这些文件；切勿编造其内容。官方仓库要求 PyTorch >=2.4.0。其 A14B T2V/I2V 基准命令使用 1280*720；TI2V-5B 命令使用 1280*704，并带有 CPU/卸载标志，以满足文档所述的 24 GB 最低显存要求。容量说明仅为基准参考，不能保证适用于其他帧数、驱动程序或同机运行的工作负载。

Wan 2.1 和 2.2 的开源代码/检查点标注为 Apache License 2.0。重新分发作品或衍生作品时，请保留声明并遵守 Apache 许可条款。Wan 团队表示不主张对生成内容享有权利，但这并不构成不侵权或所有权保证：输出可能与受保护的素材、人物、品牌或其他用户的输出相似。请审核输入内容的权利、公开权/隐私权/生物识别信息相关同意、音乐/语音许可，以及部署所在地的法律。社区 LoRA、量化版本、文本编码器、TTS 组件、容器和数据集可能受不同许可证约束。

安全、隐私与发布审核

• 为每一位可识别的人物、声音、参考表演、源视频、徽标和受版权保护的输入获取书面授权。切勿根据内容公开可用就推断已获得同意。
• 拒绝欺骗性冒充、未经同意的私密内容、剥削、针对性骚扰，以及旨在规避安全防护的指令。在法律、平台政策或具体情境要求时，明确标注合成媒体。
• 托管的 Model Studio 将输入和输出视为客户成员内容。其当前产品条款规定，Alibaba Cloud 不主张拥有输出的所有权；未经另行同意，也不会使用成员内容开发或改进 Model Studio 模型。客户仍须对权利、同意、数据保护和输出用途负责。这并不保证 API 载荷完全不会被保留；对于敏感工作负载，请获取适用的数据处理协议（DPA）和数据保留条款。
• watermark API 选项会在右下角添加可见的 AI Generated 标记。在政策或法律要求时，切勿禁用、裁剪或移除溯源信息。在发布工作流中添加持久的溯源信息/元数据，而不要仅依赖可见标记。
• 开放权重部署会将内容审核、访问控制、滥用监控、密钥/资产隔离、日志记录、事件响应和删除责任转移给运营方。发布前对输入和输出进行审核。
• 切勿记录 API 密钥或带签名的输出 URL。使用密钥管理器、遵循最小权限原则的工作区密钥、短期有效的源 URL、独立的暂存存储，以及生命周期删除。签名结果链接会过期；仅归档已批准的输出。

仅在完成以下检查后发布：任务/模型/版本凭证；确切时长、尺寸、帧率、编解码器、音频存在情况/同步；完整解码；黑帧/冻结帧/重复帧；提示词和参考内容的一致性；身份偏移；时序瑕疵；文本/徽标缺陷；安全/内容审核；溯源信息；权利审批；以及实际计费用量是否在获批上限内。仅在政策允许时保留被拒绝的输出和诊断元数据。

第一方来源

• Wan 2.7 文生视频 API
• Wan 2.7 图生视频 API
• Wan 参考视频生成 API
• Wan 2.7 视频编辑 API
• 旧版 Wan T2V API
• 区域访问和部署范围
• 模型定价
• 速率限制
• Model Studio 产品条款
• Model Studio 安全与隐私
• Wan 2.2 官方代码仓库和 Apache-2.0 许可证
• Wan 2.1 官方代码仓库和技术报告
• Wan 2.2 TI2V-5B 官方模型卡