#!/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())
