kling-video

使用 Kling AI 制作视频

模型可用性、API 字段、价格、政策和限制均可能变化。以下信息已于 2026-07-10 根据 Kling/Kuaishou 一手资料核实；在付费或发布前，请重新查看链接中的模型页面、计费表、更新日志和条款。

确保使用场景与任务相匹配

询问交付场景（直接 API 或 Kling 网页版/应用）、地理位置、任务、输入媒体、时长、宽高比、分辨率、音频/对话、镜头结构、受众、权利/同意、隐私需求、变体数量和已批准的支出。

使用场景	适用用途	注意区分
国际版直接 API	通过 https://api-singapore.klingai.com 处理服务端工作流；使用 API 密钥进行身份验证；按秒计费的 API	旧版 api.klingai.com、中国境内主机路由、消费级积分或经销商方案
Kling 网页版/应用	交互式创作、Elements 库、会员资格、免费/付费 UI 功能	不要假定 API 中也提供 UI 控件、积分、水印权益或默认设置
kling-v3 / VIDEO 3.0	由提示词驱动的文本/图像/首尾帧生成、原生音频、灵活时长、多镜头	Omni 视频编辑/参考输入
kling-v3-omni / VIDEO 3.0 Omni	参考驱动的生成、视频/参考输入、Elements、编辑、原生音频、多镜头	Kling Image 3.0 Omni 或 Kolors
kling-3.0-turbo	当前速度更快、面向规模化的 3.0 选项，支持原生音频	不要假定所有 3.0 或 Omni 字段都能原样使用
Motion Control	将一段驱动表演转移给参考角色	Avatar、唇形同步、人脸识别或独立音频 API

事实： 当前 API 文档要求所有模型均使用新格式的 API 密钥。旧版 AccessKey/SecretKey JWT 凭据仅对 3.0 及更早版本有效，无法解锁新模型。请使用服务端的 KLING_API_KEY；切勿在浏览器或移动客户端中暴露该密钥。当前国际版主机位于新加坡，而中国境内部署使用不同的北京主机。请勿将主机选择视为数据驻留保证。身份验证

事实： VIDEO 3.0 和 3.0 Omni 支持文生视频、图生视频、原生音频、多镜头叙事，以及灵活生成 3–15 秒的视频。VIDEO 3.0 是由提示词驱动的路径；Omni 则增加了更丰富的图像/视频/Element 参考和编辑工作流。第一方资料中关于一致性、物理效果、文本、唇形同步或质量“完美”“终极”或“工业级”的描述均为供应商声明，并非交付保证。VIDEO 3.0 指南、VIDEO 3.0 Omni 指南、Kuaishou 新闻稿

事实： Kling 团队的技术报告将 Kling-Omni 描述为一个统一的多模态框架，用于生成、编辑和遵循多模态指令。供应商撰写的报告中的基准测试和偏好结果仍属于供应商证据；不要将其转述为独立排名。Kling-Omni 报告

有针对性地选择当前系列

• 当场景可以通过提示词、起始图像或起始/结束帧驱动时，选择 kling-v3。连续性比覆盖范围更重要时，使用单镜头模式；只有在有意剪切时才使用多镜头模式。
• 当请求依赖参考视频、多个图像/元素参考、主体/声音绑定或视频内编辑时，选择 kling-v3-omni。请重新打开确切的 Omni 端点页面，因为参考内容的组合方式变化很快。
• 只有在当前任务页面确认所请求的任务和字段后，才选择 kling-3.0-turbo。官方平台称其速度更快、规模化时更可靠；在有日期的工作负载测试验证这些说法之前，应将其视为提供方的声明。
• 仅在现有兼容性/成本需求下保留 kling-video-o1、kling-v2-6、kling-v2-5-turbo 和更早的 ID。若 3.0 字段被拒绝，不要悄悄降级。
• 表演迁移请使用 Motion Control。它的输入和肖像相似性风险差异很大，因此需要明确获得驱动视频和表演者的授权，而不只是图像生成的批准。

对于 VIDEO 3.0 原生语音，当前产品指南列出中文、英语、日语、韩语和西班牙语，以及一些方言/口音。请针对具体账号确认语言、模型和 API 支持情况。不要承诺声音身份、发音、时序或唇形同步的准确性；请记录发音并检查实际音频。

按确切配置估算价格

直接 API 计费页面目前按每个生成秒数定价。kling-v3 在 1080p 下为无声时 $0.112/s，使用原生音频时 $0.168/s；720p 下分别为 $0.084/s 和 $0.126/s。符合条件的 kling-v3/kling-v3-omni 配置支持原生 4K，标价为**$0.42/s**。使用原生音频的 kling-3.0-turbo 在 720p 下为 $0.112/s，在 1080p 下为 $0.14/s。Omni 的价格取决于是否使用视频输入和原生音频。当前 API 计费

每次付费创建前：

1. 重新打开计费表和确切的模型任务页面；根据模型、任务、模式/分辨率、声音、时长、参考素材和变体数量估算费用。
2. 将有日期的每秒费率乘以时长和数量。目前，5 秒的 1080p 无声 kling-v3 创建任务费用为 $0.112 × 5 = $0.56；使用原生音频时为 $0.168 × 5 = $0.84。
3. 将每次重新生成、编辑、延长或生成变体都视为可能产生的新费用。不要假设失败/缩短的任务会退款；Motion Control 明确警告，提取较短的有效动作片段仍可能无法退款。
4. 获取对确切请求摘要和明确费用上限的批准。UI 中的“生成”确认或笼统的“使用 Kling”指示不构成费用批准。
5. 检查余额、套餐有效期、模型权限、并发数、QPS 和 IP 允许列表。代码 1101/1102 与余额/套餐有关；1302/1303 与速率/并发有关。未经授权，不要购买、充值或更改账号设置。

围绕时长和覆盖范围构建提示词

对于单镜头，需指定主体及其持续特征、肢体动作、场景、构图／镜头、摄像机运动、光线／色彩／质感、时间节拍和声音。对于 3–15 秒的视频，动作数量应切合实际。

Single continuous 5-second shot, no cuts. A hand-thrown ceramic cup sits on a dark walnut table.
[0–2s] Macro close-up; a ribbon of steam curls upward in quiet morning light.
[2–5s] Slow 10 cm dolly left reveals a rain-streaked window; keep the cup shape and glaze unchanged.
Natural shallow depth of field, restrained blue-gray palette, believable steam and reflections.
No people, no lettering, no logo. Audio off.

对于图生视频，将图像视为身份／构图锚点，提示词主要描述变化：哪些内容在移动、哪些内容必须保持不变、摄像机行为以及结束条件。对于起始帧／结束帧，要求生成合理的衔接过程，而不是互不相关的事件。

对于多镜头拍摄，让每次剪辑都有明确目的。VIDEO 3.0 提供自动多镜头模式，以及对镜头级时长／构图／视角／动作／摄像机的自定义控制。自定义方案最多可包含六个镜头；各镜头时长之和必须等于总时长。避免在 15 秒内塞入六个互不相关的构思。

{
  "model_name": "kling-v3",
  "multi_shot": true,
  "shot_type": "customize",
  "multi_prompt": [
    {"index": 1, "duration": "3", "prompt": "Wide dawn exterior: a cyclist stops outside a small bakery; gentle locked camera."},
    {"index": 2, "duration": "2", "prompt": "Close-up: the same gloved hand lifts the warm paper bag; steam in backlight."}
  ],
  "duration": "5",
  "mode": "pro",
  "aspect_ratio": "16:9",
  "sound": "off"
}

这是一个示例，并非通用模板。使用前请确认当前的 3.0 任务架构。设置 multi_shot=true 时，自定义镜头提示词会替代单条提示词；请保持索引连续，并确保时长准确。

对于原生音频，在每句引用台词前注明说话者；仅在必要时说明语言／口音，并注明发声意图、环境音、音效和排除项。对话应足够简短，以便自然说出。若必须保证准确性，请将事实性文案、法律声明、指定音乐和品牌文字放在后期制作中处理。

将参考素材和动作作为生产资产进行管理

3.0 Omni 指南目前记载：

• 最多 7 张图片，每张的宽高均至少为 300 px，大小不超过 10 MB，格式为 JPG/JPEG/PNG。
• 1 段参考视频，时长为 3–10 秒，大小不超过 200 MB，分辨率不高于 2K。
• 使用视频时，图片/Elements 总数不得超过 4 个；不使用视频时，最多 7 个。
• 多图 Element 最多可使用 4 个视角。角色语音绑定可使用 5–30 秒的清晰单人语音；视频角色 Element 可使用 3–8 秒的单角色片段。

限制按字段区分，而非采用一个全局限制： 当前 Omni 指南指出，生成的 VIDEO 3.0/3.0 Omni 输出时长可以为 3–15 秒，但其常见问题另行规定，上传的参考视频时长上限为 3–10 秒。不要将 15 秒的生成时长公告误解为允许上传 15 秒视频。除非当前确切的 API 任务 schema 明确更改了该输入字段，否则参考视频输入时长上限应保持为 10 秒。任何覆盖此限制的变更都必须保留证据和日期。同样，请确认所请求的参考视频/原生音频/4K 组合受支持，不要将分别发布的功能公告拼凑成组合支持的依据。

上传前对每个源素材进行哈希计算和内容验证；记录字节大小、MIME、权利、同意、用途和远程标识符。使用私有、短时有效的获取 URL 或受支持的 base64，而不要使用公开存储桶。切勿记录签名 URL 或 base64。参考标签/Element 名称必须与实际素材明确对应；标签并不能保证身份、产品颜色、服装、文本或语音得到保留。

Motion Control 会将驱动视频中的身体、面部和手部动作转移到单张角色图片上。匹配全身/半身取景；确保身体/头部始终可见；优先使用连续的 3–30 秒单角色镜头，并采用适度动作，避免剪辑/镜头移动。官方指南警告，多人输入时会选择画面中面积最大的人物；复杂/快速动作可能被截断；生成时长可能较短，但积分仍会消耗。Element 绑定使用的是面部参考信息，而非服装、头发、妆容或道具。Motion Control 指南、技术报告与影响声明

要求参考人物和动作表演者提供书面同意，涵盖动作、面部、语音、对话、分发和实质性变换。拒绝亲密内容滥用、欺骗性冒充、骚扰、剥削未成年人、非法监控，以及未经授权的政治/公众人物劝说。不得弱化 1300/1301 安全响应，也不得将请求转移到网关以规避该响应。

使用故障关闭式直接 API 事务

直接 API 是异步的。当前的视频结果是由提供商托管的交付产物，并非持久存储；第一方 API 资料称，生成的图像/视频会在 30 天后清除，而付费 API 条款称，提示词和响应会记录 30 天。请及时下载到受控存储中。external_task_id 有助于对账，但没有文档说明它是幂等键。如果 POST 结果不明确，且没有持久保存的提供商任务 ID，请停止并进行对账；切勿“以防万一”而重试。

以下完整的示例通过当前国际 API 提交一个由提示词驱动的 kling-v3 文本转视频任务。默认采用离线试运行模式，绑定精确审批、最新定价证据和精确的交付主机允许列表，最多执行一次 POST，仅使用有界 GET 重试，不转发 API 密钥地下载文件，验证 MP4，且绝不覆盖产物。其持久化的暂存产物记录还可弥合验证、原子发布和最终账本状态之间的崩溃窗口。此示例有意不包含 4K、Omni 参考、回调和自动重放创建请求。

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

API_HOST = "api-singapore.klingai.com"
MODEL = "kling-v3"
MAX_JSON = 1024 * 1024
MAX_VIDEO = 250 * 1024 * 1024
MAX_TOOL = 1024 * 1024
RATES = {("std", "off"): Decimal("0.084"), ("std", "on"): Decimal("0.126"),
         ("pro", "off"): Decimal("0.112"), ("pro", "on"): Decimal("0.168")}

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") as f:
            json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
        os.replace(tmp, path); fsync_directory(path.parent)
    finally:
        if os.path.exists(tmp): os.unlink(tmp)

def claim_once(path, value):
    path.parent.mkdir(parents=True, exist_ok=True)
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "w", encoding="utf-8") as f:
        json.dump(value, f, indent=2, sort_keys=True); f.flush(); os.fsync(f.fileno())
    fsync_directory(path.parent)

def read_small_json(path):
    if path.stat().st_size > MAX_JSON: raise RuntimeError("record exceeds cap")
    with path.open("rb") as f: raw = f.read(MAX_JSON + 1)
    value = json.loads(raw)
    if not isinstance(value, dict): raise RuntimeError("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(method, path, key, payload=None, timeout=60):
    body = canonical(payload) if payload is not None else None
    headers = {"Authorization": "Bearer " + key, "Accept": "application/json"}
    if body is not None: headers["Content-Type"] = "application/json"
    connection = http.client.HTTPSConnection(API_HOST, 443, timeout=timeout, context=ssl.create_default_context())
    try:
        connection.request(method, path, body=body, headers=headers); response = connection.getresponse()
        raw = response.read(MAX_JSON + 1)
        if len(raw) > MAX_JSON: raise RuntimeError("API response exceeds cap")
        if response.status < 200 or response.status >= 300:
            retry = response.getheader("Retry-After")
            try: retry = min(60.0, max(0.0, float(retry)))
            except (TypeError, ValueError): retry = None
            raise APIError(f"API HTTP {response.status}", response.status, retry, hashlib.sha256(raw).hexdigest())
        try: value = json.loads(raw)
        except json.JSONDecodeError as exc:
            raise APIError("API returned invalid JSON", response.status, body_sha256=hashlib.sha256(raw).hexdigest()) from exc
        if not isinstance(value, dict): raise RuntimeError("API response is not an object")
        return value
    finally: connection.close()

def safe_get(call, deadline, attempts=5):
    for attempt in range(attempts):
        try: return call()
        except Exception as exc:
            if attempt + 1 == attempts or time.monotonic() >= deadline: raise
            delay = getattr(exc, "retry_after", None) or min(20, 2 ** attempt) * (1 + random.random() * 0.25)
            if time.monotonic() + delay >= deadline: raise
            time.sleep(delay)

class PinnedHTTPSConnection(http.client.HTTPSConnection):
    def __init__(self, host, address):
        super().__init__(host, 443, timeout=60, 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("delivery URL host is not an exact approved host")
    try: ipaddress.ip_address(host); raise RuntimeError("IP-literal delivery URL 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(item).is_global for item in addresses):
        raise RuntimeError("delivery host did not resolve publicly")
    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"}); response = connection.getresponse()
        if 300 <= response.status < 400: raise RuntimeError("delivery redirect rejected")
        if response.status != 200: raise RuntimeError(f"delivery HTTP {response.status}")
        if response.getheader("Content-Type", "").split(";", 1)[0].lower() != "video/mp4":
            raise RuntimeError("delivery MIME rejected")
        length = response.getheader("Content-Length")
        if length and (not length.isdigit() or int(length) > MAX_VIDEO): raise RuntimeError("declared 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("download 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, mode, aspect, duration, sound):
    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("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: raise RuntimeError("expected one video stream")
    if sound == "on" and not any(s.get("codec_type") == "audio" for s in streams):
        raise RuntimeError("native audio was requested but absent")
    width, height = videos[0].get("width"), videos[0].get("height"); expected = 720 if mode == "std" else 1080
    if not isinstance(width, int) or not isinstance(height, int) or min(width, height) != expected:
        raise RuntimeError("resolution does not match approved mode")
    expected_ratio = {"16:9": Decimal(16) / 9, "9:16": Decimal(9) / 16, "1:1": Decimal(1)}[aspect]
    if abs(Decimal(width) / Decimal(height) - expected_ratio) > Decimal("0.02"):
        raise RuntimeError("aspect ratio does 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("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("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": videos[0].get("codec_name"),
                      "audioStreams": sum(s.get("codec_type") == "audio" for s in streams)}}

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--prompt", required=True); parser.add_argument("--negative-prompt", default="")
    parser.add_argument("--duration", type=int, choices=range(3, 16), default=5)
    parser.add_argument("--aspect", choices=("16:9", "9:16", "1:1"), default="16:9")
    parser.add_argument("--mode", choices=("std", "pro"), default="pro")
    parser.add_argument("--sound", choices=("off", "on"), default="off")
    parser.add_argument("--attempt-id", required=True); parser.add_argument("--attempt-record", type=Path, default=Path("kling-attempt.json"))
    parser.add_argument("--output", type=Path, default=Path("kling-output.mp4"))
    parser.add_argument("--max-usd", required=True); parser.add_argument("--approval-sha256")
    parser.add_argument("--pricing-checked-at", required=True)
    parser.add_argument("--pricing-evidence-sha256", required=True)
    parser.add_argument("--delivery-host", action="append", required=True,
                        help="exact approved Kling delivery host; repeat for each host")
    parser.add_argument("--execute", action="store_true"); parser.add_argument("--resume", action="store_true")
    parser.add_argument("--show-full-plan", action="store_true")
    args = parser.parse_args()
    if not args.prompt.strip() or len(args.prompt) > 2500 or len(args.prompt.encode()) > 10000:
        raise SystemExit("prompt must be 1..2500 characters and at most 10000 UTF-8 bytes")
    if len(args.negative_prompt) > 2500 or len(args.negative_prompt.encode()) > 10000:
        raise SystemExit("negative prompt exceeds local cap")
    try: attempt = uuid.UUID(args.attempt_id); maximum = Decimal(args.max_usd)
    except (ValueError, InvalidOperation) as exc: raise SystemExit("attempt ID or maximum invalid") from exc
    if attempt.version != 4 or str(attempt) != args.attempt_id.lower(): raise SystemExit("attempt ID must be canonical UUIDv4")
    if (len(args.pricing_evidence_sha256) != 64
            or any(c not in "0123456789abcdef" for c in args.pricing_evidence_sha256)):
        raise SystemExit("pricing evidence must be a lowercase SHA-256")
    try: checked_at = datetime.fromisoformat(args.pricing_checked_at.replace("Z", "+00:00"))
    except ValueError as exc: raise SystemExit("pricing timestamp must be ISO-8601") from exc
    if checked_at.tzinfo is None: raise SystemExit("pricing timestamp must include a timezone")
    allowed_hosts = sorted(set(host.lower().rstrip(".") for host in args.delivery_host))
    if (not allowed_hosts or len(allowed_hosts) > 8
            or any(not host or len(host) > 253 or "*" in host or "/" in host or "@" in host for host in allowed_hosts)):
        raise SystemExit("delivery hosts must be 1..8 exact DNS names")
    for host in allowed_hosts:
        try: ipaddress.ip_address(host); raise SystemExit("delivery hosts cannot be IP literals")
        except ValueError: pass
    estimate = RATES[(args.mode, args.sound)] * args.duration
    if not maximum.is_finite() or maximum <= 0 or maximum < estimate: raise SystemExit("maximum must be finite and cover estimate")
    output, record_path = args.output.resolve(), args.attempt_record.resolve()
    stage_path = output.with_name("." + output.name + ".kling-stage")
    if output.suffix.lower() != ".mp4" or output == record_path or stage_path in {output, record_path}:
        raise SystemExit("output path rejected")
    request = {"model_name": MODEL, "prompt": args.prompt, "negative_prompt": args.negative_prompt,
               "duration": str(args.duration), "mode": args.mode, "aspect_ratio": args.aspect,
               "sound": args.sound, "external_task_id": str(attempt)}
    request_hash = hashlib.sha256(canonical(request)).hexdigest()
    envelope = {"backend": "kling-direct-international", "endpoint": "/v1/videos/text2video", "host": API_HOST,
        "request": request, "attemptId": str(attempt), "attemptRecord": str(record_path),
        "outputPolicy": {"path": str(output), "stagePath": str(stage_path), "exactDeliveryHosts": allowed_hosts,
                         "maxBytes": MAX_VIDEO, "mime": "video/mp4", "fullDecode": True, "noOverwrite": True},
        "cost": {"pricingCheckedAt": checked_at.isoformat(), "pricingEvidenceSha256": args.pricing_evidence_sha256,
                 "rateUsdPerSecond": str(RATES[(args.mode, args.sound)]),
                 "seconds": args.duration, "creates": 1, "estimatedMaxUsd": str(estimate), "approvedMaxUsd": str(maximum)}}
    approval = hashlib.sha256(canonical(envelope)).hexdigest()
    safe_request = {**request, "prompt": "sha256:" + hashlib.sha256(args.prompt.encode()).hexdigest(),
                    "negative_prompt": "sha256:" + hashlib.sha256(args.negative_prompt.encode()).hexdigest()}
    print(json.dumps({"dryRun": not args.execute, "approvalSha256": approval,
        "plan": {**envelope, "request": safe_request}}, indent=2))
    if args.show_full_plan: print(json.dumps({"protectedFullPlan": envelope}, indent=2))
    if not args.execute: return 0
    if args.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")
    key = os.getenv("KLING_API_KEY")
    if not key: raise SystemExit("set server-side KLING_API_KEY")
    if args.resume:
        if not record_path.exists(): raise SystemExit("resume record missing")
        record = read_small_json(record_path)
        if record.get("approval_sha256") != approval or record.get("request_sha256") != request_hash:
            raise SystemExit("resume does not match exact request")
        task_id = record.get("task_id")
        if not isinstance(task_id, str) or not task_id: raise SystemExit("no known task ID; do not replay create")
        expected_artifact = record.get("artifact_staged") or record.get("artifact")
        candidate = output if output.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, output, args.mode, args.aspect, args.duration, args.sound)
                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 output.exists(): raise SystemExit("output appeared during recovery")
                    os.replace(stage_path, output); fsync_directory(output.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
    else:
        if output.exists() or stage_path.exists(): raise SystemExit("output or stage already exists")
        record = {"attempt_id": str(attempt), "approval_sha256": approval, "request_sha256": request_hash,
                  "status": "posting", "task_id": None, "created_unix": int(time.time())}
        try: claim_once(record_path, record)
        except FileExistsError as exc: raise SystemExit("attempt already claimed; use resume") from exc
        try: response = api_json("POST", "/v1/videos/text2video", key, request, 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 account, never replay") from exc
        if response.get("code") != 0:
            record.update(status="create_rejected", service_code=response.get("code")); atomic_json(record_path, record)
            raise RuntimeError("provider rejected create")
        data = response.get("data") or {}; task_id = data.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"); atomic_json(record_path, record)
            raise RuntimeError("create returned no safe task ID; reconcile")
        record.update(status="task_id_saved", task_id=task_id, request_id=response.get("request_id"), updated_unix=int(time.time()))
        atomic_json(record_path, record)
    deadline = time.monotonic() + 1200
    while True:
        response = safe_get(lambda: api_json("GET", "/v1/videos/text2video/" + task_id, key, timeout=45), deadline)
        if response.get("code") != 0: raise RuntimeError("task query failed")
        data = response.get("data") or {}; status = data.get("task_status")
        if status == "succeed": break
        if status == "failed":
            message = str(data.get("task_status_msg", ""))
            record.update(status="generation_failed", task_status_msg_sha256=hashlib.sha256(message.encode()).hexdigest())
            atomic_json(record_path, record)
            raise RuntimeError("generation failed")
        if status not in {"submitted", "processing"}: raise RuntimeError("unknown task status")
        if time.monotonic() >= deadline: raise TimeoutError("generation exceeded 20 minutes")
        time.sleep(8 + random.random() * 2)
    videos = ((data.get("task_result") or {}).get("videos") or [])
    if len(videos) != 1 or not isinstance(videos[0], dict) or not isinstance(videos[0].get("url"), str):
        raise RuntimeError("expected exactly one video URL")
    url = videos[0]["url"]; output.parent.mkdir(parents=True, exist_ok=True)
    record.update(status="downloading", delivery_url_sha256=hashlib.sha256(url.encode()).hexdigest(), stage_path=str(stage_path))
    atomic_json(record_path, record)
    try:
        download_video(url, stage_path, set(allowed_hosts))
        artifact = inspect_mp4(stage_path, output, args.mode, args.aspect, args.duration, args.sound)
    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, final_unit_deduction=data.get("final_unit_deduction"),
        remote_purge_expected_within_days=30, rights_review="required", creative_qa="pending",
        ai_disclosure="required", provenance_credentials="not_documented", updated_unix=int(time.time()))
    atomic_json(record_path, record)
    if output.exists(): raise RuntimeError("output appeared before publication")
    os.replace(stage_path, output); fsync_directory(output.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

if __name__ == "__main__": raise SystemExit(main())

试运行并批准打印出的摘要：

python kling_generate.py --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.56 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$KLING_DELIVERY_HOST" --prompt "Single continuous five-second macro dolly shot of steam rising from a ceramic cup; rain outside; no people, text, or logo."

仅执行一次未经更改的请求：

python kling_generate.py --execute --approval-sha256 "<exact digest>" --attempt-id 7f8f06b7-9e68-4f0a-bf16-3eaf3ee9d6c4 --max-usd 0.56 --pricing-checked-at 2026-07-10T12:00:00Z --pricing-evidence-sha256 "$BILLING_SHA256" --delivery-host "$KLING_DELIVERY_HOST" --prompt "Single continuous five-second macro dolly shot of steam rising from a ceramic cup; rain outside; no people, text, or logo."

BILLING_SHA256 是已审核的注明日期的第一方计费证据的 SHA-256；KLING_DELIVERY_HOST 是当前直接 API 交付约定授权的精确主机名，而不是通配符，也不是根据不可信 URL 推断出的主机。仅在需要添加其他精确获批主机时，才重复使用 --delivery-host。定价证据超过 24 小时时，执行将被拒绝。

仅当记录中已有任务 ID 时，才使用相同参数并加上 --resume --execute。请勿使用 --resume 恢复 posting 或 create_outcome_unknown 状态；应先核对 external_task_id、账户用量、请求 ID、日志或联系支持。恢复操作只能采用暂存/已发布的产物，且其字节内容和 SHA-256 必须与持久化账本证据相符。由于文档界面正在积极重构，投入生产前请在实时的 kling-v3 文本转视频页面上验证端点和字段类型。

管理数据、权利与发布

事实： 付费 API 条款规定，Kling 作为数据处理方，不会使用付费 API 的提示词／文件／响应来改进模型，并会为服务／安全目的将提示词／响应记录保留 30 天。API 隐私政策称，数据存储在新加坡的服务器上，运营团队／服务提供商可能会在国际范围内访问／传输数据。出于法律、合规、争议或正当目的，数据保留期限可能延长。这并不意味着零保留或数据存储于欧盟境内。API 付费条款，API 隐私政策

事实： 付费 API 条款允许商业用途，但要求运营方拥有输入内容的权利／授权，并提醒生成内容仍可能侵权。所有权／版权资格取决于适用法律。Kling 通用条款要求，除非另有书面许可，分发输出内容时必须显著标明 Kling AI。遵循具有约束力的账户／API 协议和当地合成媒体法律中更严格的规定；不要据此推断“无水印”就意味着无需披露。API 付费条款，Kling 条款

没有任何经审查的一手来源承诺，Kling 直接生成的视频会附带 C2PA 或持久的加密来源凭证。将来源凭证状态视为未知。保留请求／来源哈希和原始文件；添加清晰的 AI 披露信息及任何必需的品牌标识，但不要声称这些内容由服务提供商签名。

发布前，应逐帧检查并从头到尾聆听。确认尺寸、时长、帧率、音频流、完整解码、响度／峰值、对白／口型同步／说话人分配、时间逻辑、人脸／手部／解剖结构、身份／服装／产品连续性、徽标／文字／颜色、不安全内容、事实／广告声明、权利、披露信息和平台规格。技术上有效的 MP4 并不等同于获批的创意母版。

按阶段处理故障

• POST 前： 修复架构、权利、身份验证、余额、模型访问权限或确切的审批；尚未接受生成请求。
• POST 返回任务 ID： 保存该 ID，然后查询对应任务。处理缓慢、429、500、503 或 504 都不代表可以再次创建任务。
• POST 结果未知： 停止操作。external_task_id 是用于核对的标识，并非有文档保证的幂等性机制。
• 安全审核拒绝： 不要通过改写来规避管控，也不要转由经销商处理。重新评估其正当性和同意情况。
• 任务失败或动作缩短： 记录状态和实际扣费；不要承诺退款。
• 交付 URL 或媒体无效： 隔离／删除临时文件，保留经过清理的证据，不要发布。新的付费生成需要新的尝试 ID 和审批。
• 预览／更新不一致： 重新查看官方身份验证、模型、计费、更新和政策页面。不要猜测替代 ID，也不要将网关协议与直接 API 混用。

如实标注证据

• 事实： 当前一手 API 字段、模型能力、价格、条款或限制。
• 服务商声明： Kling/Kuaishou 关于质量、基准测试、“原生”特性、一致性、物理效果或效率的声明，未经独立验证。
• 观察结果： 可复现任务的带日期结果，包括界面、模型、请求哈希、输入、账户区域和输出检查。
• 启发式建议： 生产建议，例如限制动作、匹配运动参考的构图，或每次编辑只做一项更改。
• 未知： 未记录的硬性延迟、创建操作的确切幂等性、直接 API C2PA、保证的身份/文本/音频准确性，或当前任务页面上不可见的任何字段。

重新核验一手来源：API 概览、身份验证/错误、计费、3.0 任务索引、3.0 指南、3.0 Omni 指南、Motion Control 指南、API 付费条款、API 隐私政策、Kling-Omni 报告和 Motion Control 报告。