alibaba-wan-video

Alibaba Wan 動画

このスキルを使って、適切な Wan の利用環境を選択し、操作します。ホスト型モデル ID とダウンロード可能なチェックポイントは、バージョン番号が関連しているように見えても、別の製品として扱ってください。

まず利用環境を選ぶ

1. タスク、入力、出力の長さ／解像度／アスペクト比、音声要件、デプロイリージョン／データ境界、予算、レイテンシ、GPU 容量、マネージド推論を許容できるかを確認します。
2. 次のいずれかを選びます。
   • Hosted Model Studio：最新の Wan 2.7 の品質、2～15 秒のクリップ、生成音声／カスタム音声、1080P、マネージドスケーリング、参照画像からの動画生成、動画の続きの生成、またはホスト型編集に適しています。
   • Official open weights：オフライン処理、重みへのアクセス、再現可能なチェックポイントの固定、適応、またはセルフホストに適しています。現在の公式公開ファミリーは Wan 2.2（および旧版の Wan 2.1）であり、ホスト型 Wan 2.7 ではありません。
3. ホスト型エイリアス、古い API プロトコル、コミュニティ製の量子化モデル、またはサードパーティのゲートウェイに、黙って置き換えないでください。
4. 課金対象のリクエストや大容量チェックポイントのダウンロードを行う前に、正確なモデル／リビジョン、リージョン、入力、出力設定、推定料金または必要ストレージを提示し、明示的な承認を得てください。ドライランではネットワーク上の変更を一切行ってはいけません。

以下の情報と料金は、2026-07-10 時点で確認済みです。提供状況、エイリアス、スナップショット、制限、料金は変動するため、実行時にリンク先のファーストパーティページを再確認してください。

製品ファミリーを分けて扱う

利用環境	現在の本番環境向けの選択肢	提供機能	制約
Hosted 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 チェックポイントの存在を意味しません。安定性が重要な場合は、日付付きのスナップショットを固定してください。
Hosted 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 を含むリクエスト本文を送信しないでください。
Hosted legacy	Wan 2.5 preview、Wan 2.2／2.1 の T2V、I2V、キーフレーム／VACE、音声／アニメーションの各バリアント	互換性と特殊なワークフロー	機能、固定された長さ、解像度の区分、名称（北京では wanx2.1-*）、料金が異なります。正確な旧版リファレンスを確認してください。ホスト型の wan2.2-* ID が、公開チェックポイントとビット単位で同一のものを提供する証拠にはなりません。
Open 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 と記載されています。
Open Wan 2.1	T2V-14B／1.3B、I2V-14B、FLF2V-14B、VACE-14B／1.3B	旧世代ながら、十分に文書化された T2V／I2V、最初と最後のフレームを指定する生成、編集／制御ワークフロー	Apache-2.0 ライセンスですが、旧世代のアーキテクチャです。公式の T2V-1.3B ガイダンスでは 480P が推奨され、VRAM は 8.19 GB とされています。

Wan 2.2 A14B は 2 つのエキスパートを持つ MoE で、総パラメーター数は約 27B、ノイズ除去の各ステップで有効になるパラメーター数は約 14B です。TI2V-5B は高圧縮 VAE を使用し、24 fps で 1280x704／704x1280 の出力を公式にサポートしています。これらの事実は、5B と A14B が品質の段階として互換であることを意味しません。

Hosted Model Studio

リージョンとエンドポイントの要件

リージョン、API キー、モデルの提供状況、サービスのデプロイ範囲、エンドポイントを一致させる必要があります。キーはリージョンをまたいで使用できる認証情報ではありません。

アクセスリージョン	本番環境のベース URL	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 モデルではグローバル	フランクフルトは一般にグローバルおよび EU のワークスペース範囲をサポートしますが、現在の Wan の料金表では、T2V/I2V/R2V 2.6 はグローバルとしてのみ記載されています。これらのモデルが EU 範囲のワークスペースで提供されているとは限りません。コンソールで確認してください。Wan 2.7 は現在記載されていません。
米国（バージニア）	https://dashscope-us.aliyuncs.com	グローバル、または -us モデル ID を使用する米国	現在の料金表には Wan 2.6 のグローバル版と 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 です。

1 つのワーカーですべての 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 では、入力画像はアルファチャンネルなしの 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 枚、参照画像と参照動画は合計で 1〜5 個まで指定できます。キャラクター参照には、キャラクターを 1 人だけ含める必要があります。参照画像には、上記と同じ 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"} を 1 つと、最大 4 つの reference_image アセットを指定してください。料金には入力動画と出力動画の両方の時間が含まれます。アップロード前に、変更される可能性がある制約について、必ず該当する API ページを再確認してください。

請求対象となる、ドライラン優先の T2V クライアント

この完全な例では、シンガポールのスナップショットを固定し、デフォルトで機密情報を伏せた無料プランを表示します。さらに、正確なリクエスト／ワークスペース／出力ポリシーと最新の価格根拠を承認に紐付け、タスクの作成を最大 1 件に制限し、クラッシュ後も再開できるサニタイズ済み台帳を永続化します。また、API キーを転送せず、厳密なホスト名およびパブリック IP のポリシーを適用してダウンロードし、検証後に上書き不可のアトミックな方法で公開します。任意のベース 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 をインストールしてから、まずプランを実行してください。課金対象のタスクを作成できるのは、2 つ目のコマンドだけです。

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 のみです。北京または別のリージョンを使用する場合は、まずリージョンの許可リストと価格表を現在の価格ページの値に置き換え、エンドポイント／モデルの契約を再確認してください。このフェイルクローズ境界を迂回してはなりません。

コストと処理能力の制御

課金対象の秒数あたりの現在の標準単価（USD）：

アクセスリージョン／範囲	モデル	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のコマンドでは、文書化されている最小要件24 GB向けのCPU／オフロードフラグと1280*704を使用します。容量に関する記載はベースラインの目安であり、他のフレーム数、ドライバー、または同一環境で実行されるワークロードに対する保証ではありません。

Wan 2.1および2.2のオープンなコード／チェックポイントには、Apache License 2.0のラベルが付いています。成果物または派生物を再配布する場合は、通知を保持し、Apacheの条件を満たしてください。Wanチームは生成コンテンツに対する権利を主張しないと述べていますが、これは権利侵害がないことや所有権を保証するものではありません。出力が保護対象の素材、人物、ブランド、または他のユーザーの出力に類似する可能性があります。入力の権利、パブリシティ／プライバシー／生体情報に関する同意、音楽／音声のライセンス、およびデプロイ先の法域の法律を確認してください。コミュニティ製のLoRA、量子化、テキストエンコーダー、TTSコンポーネント、コンテナー、データセットには、異なるライセンスが適用される場合があります。

安全性、プライバシー、リリースの承認基準

• 識別可能な人物、声、参考パフォーマンス、ソース動画、ロゴ、著作権で保護された入力ごとに、文書化された許可を取得してください。公開されていることを理由に同意があると推測してはいけません。
• 欺瞞を目的としたなりすまし、同意のない性的コンテンツ、搾取、標的型の嫌がらせ、安全対策を回避するための指示は拒否してください。法律、プラットフォームのポリシー、または状況により必要な場合は、合成メディアであることを明確に表示してください。
• ホステッド Model Studio では、入力と出力は顧客の Member Content として扱われます。現行の製品規約では、Alibaba Cloud は出力の所有権を主張せず、別途同意を得ない限り、Member Content を 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 公式モデルカード