"""Two-frame Kling example; no model request runs on import. Submission response task-ID path is deliberately NOT guessed. Use a confirmed ID with poll; all network commands require explicit opt-in. """ from __future__ import annotations import argparse import base64 import binascii import io import json import os import sys import time import wave from datetime import datetime, timezone from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit, quote from urllib.request import Request, HTTPRedirectHandler, build_opener from http.client import HTTPException class WorkflowError(Exception): pass class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None def open_url(request, timeout): # Redirects are rejected; credentials never follow a redirect to another host. return build_opener(NoRedirect()).open(request, timeout=timeout) def exclusive(path: Path, data: bytes): path.parent.mkdir(parents=True, exist_ok=True) with path.open("xb") as f: f.write(data) def read_json(path): value = json.loads(Path(path).read_text(encoding="utf-8")) if not isinstance(value, dict): raise WorkflowError("Expected a JSON object") return value def json_bytes(value): return (json.dumps(value, indent=2, ensure_ascii=False) + "\n").encode("utf-8") def require_execute(args): if not args.execute: raise WorkflowError("Network command requires --execute; it may use a billable API") key = os.environ.get("TOKENHOT_API_KEY") if not key: raise WorkflowError("Set TOKENHOT_API_KEY in the environment") return key def api_request(method, url, payload, key, run_dir, stem="response"): # run_dir is reserved by the command BEFORE any request. Do not reuse it. body_path = run_dir / (stem + ".json") meta_path = run_dir / (stem + ".meta.json") if body_path.exists() or meta_path.exists(): raise WorkflowError("Request records already exist; refusing to send again") meta = {"method": method, "endpoint": url.split("?")[0], "started_at": datetime.now(timezone.utc).isoformat(), "http_status": None, "body_saved": False} exclusive(meta_path, json_bytes(meta)) request = Request(url, data=json_bytes(payload) if payload is not None else None, headers={"Authorization": "Bearer " + key, "Content-Type": "application/json"}, method=method) try: try: response = open_url(request, timeout=120) except HTTPError as error: response = error with response: meta["http_status"] = response.code meta["request_id"] = response.headers.get("x-request-id") raw = response.read(64 * 1024 * 1024 + 1) if len(raw) > 64 * 1024 * 1024: raise WorkflowError("Response exceeded the example's local 64 MiB limit") exclusive(body_path, raw) meta["body_saved"] = True except (OSError, URLError, HTTPException, WorkflowError) as error: meta["transport_failure"] = type(error).__name__ meta_path.write_bytes(json_bytes(meta)) raise WorkflowError("Transport/read failure; inspect local metadata. " "A POST outcome can be unknown. No automatic retry.") from None meta_path.write_bytes(json_bytes(meta)) if not (200 <= meta["http_status"] < 300): raise WorkflowError("HTTP rejection; raw body is saved for private inspection, not echoed") try: value = json.loads(raw) except (ValueError, UnicodeError): raise WorkflowError("Invalid JSON; raw response is preserved") from None if not isinstance(value, dict): raise WorkflowError("Non-object JSON; raw response is preserved") return value ENDPOINT = "https://api.tokenhot.ai/v1/video/generations" def url_checked(value, allow_http=False): p = urlsplit(value) if (p.scheme not in (("https", "http") if allow_http else ("https",)) or not p.hostname or p.username or p.password or p.fragment): raise WorkflowError("Expected a public media URL; no embedded credentials or fragments") # Not an SSRF firewall: only use trusted URLs in this local CLI. if p.hostname in ("localhost", "127.0.0.1", "::1") or p.hostname.endswith(".invalid"): raise WorkflowError("Replace local/example URLs with trusted reachable media URLs") return value def make_payload(first, last, prompt): if not isinstance(prompt, str) or not prompt.strip(): raise WorkflowError("Motion prompt must not be empty") return {"model": "kling-v3", "prompt": prompt, "duration": 5, "size": "720P", "file_infos": [{"Type": "Url", "Category": "Image", "Url": url_checked(u), "Usage": role} for u, role in ((first, "FirstFrame"), (last, "LastFrame"))], "audio_generation": False} def query_state(result, task_id=None): if result.get("code") != "success" or not isinstance(result.get("data"), dict): raise WorkflowError("Unexpected query envelope; saved response needs inspection") data = result["data"] if task_id is not None and data.get("task_id") not in (None, task_id): raise WorkflowError("Returned task ID differs from the requested task") status = data.get("status") if status not in ("IN_PROGRESS", "SUCCESS", "FAILURE"): raise WorkflowError("Unrecognized task status; not treated as success") return status def poll_task(task_id, key, run, attempts=12, interval=10): if not isinstance(task_id, str) or not task_id.strip(): raise WorkflowError("A confirmed task ID is required") if not 1 <= attempts <= 120 or not 1 <= interval <= 300: raise WorkflowError("Use attempts 1..120 and interval 1..300 seconds") for number in range(1, attempts + 1): result = api_request("GET", ENDPOINT + "/" + quote(task_id, safe=""), None, key, run, stem=f"query-{number:03d}") status = query_state(result, task_id) print(f"Query {number}: {status}") if status == "SUCCESS": exclusive(run / "result.json", json_bytes(result)) return 0 if status == "FAILURE": raise WorkflowError("Task failed; inspect saved query privately. No resubmission.") if number < attempts: time.sleep(interval) raise WorkflowError("Polling budget ended; reuse the SAME task ID in a new query run") def download_video(result, output, allow_http=False): if query_state(result) != "SUCCESS": raise WorkflowError("Only a SUCCESS result may be downloaded") try: url = result["data"]["data"]["metadata"]["url"] except (KeyError, TypeError): raise WorkflowError("Direct URL absent; inspect provider response. Do not invent a proxy contract") from None if not isinstance(url, str): raise WorkflowError("Direct URL must be a string") url_checked(url, allow_http) if output.exists(): raise WorkflowError("Output exists; choose a fresh MP4 path") # No Authorization header is attached to media GETs. Redirects are rejected. request = Request(url, method="GET") with open_url(request, timeout=120) as response: if response.code != 200: raise WorkflowError("Media response is not HTTP 200") raw = response.read(256 * 1024 * 1024 + 1) if len(raw) > 256 * 1024 * 1024: raise WorkflowError("Media exceeds this example's local 256 MiB download limit") if len(raw) < 12 or raw[4:8] != b"ftyp": raise WorkflowError("Expected MP4 file signature; no file saved") exclusive(output, raw) print("Saved MP4-shaped bytes; play and inspect the video before accepting it") def main(argv=None): p = argparse.ArgumentParser(description=__doc__) sub = p.add_subparsers(dest="command", required=True) q=sub.add_parser("prepare"); q.add_argument("--first", required=True); q.add_argument("--last", required=True) q.add_argument("--prompt", required=True); q.add_argument("--output", required=True) q=sub.add_parser("submit"); q.add_argument("--request", required=True); q.add_argument("--run", required=True) q.add_argument("--execute", action="store_true") q=sub.add_parser("poll"); q.add_argument("--task-id", required=True); q.add_argument("--run", required=True) q.add_argument("--attempts", type=int, default=12); q.add_argument("--interval", type=float, default=10) q.add_argument("--execute", action="store_true") q=sub.add_parser("save"); q.add_argument("--response", required=True); q.add_argument("--output", required=True) q.add_argument("--allow-http", action="store_true"); q.add_argument("--execute", action="store_true") args=p.parse_args(argv) try: if args.command == "prepare": exclusive(Path(args.output), json_bytes(make_payload(args.first,args.last,args.prompt))) print("Prepared JSON locally; no request sent"); return 0 if args.command == "save": if not args.execute: raise WorkflowError("Downloading needs --execute; no API key is used for the media GET") download_video(read_json(args.response),Path(args.output),args.allow_http); return 0 key=require_execute(args) if args.command == "submit": payload=read_json(args.request) if payload.get("model") != "kling-v3": raise WorkflowError("This example only submits kling-v3") run=Path(args.run);run.mkdir(parents=True,exist_ok=False) exclusive(run/"request.json",json_bytes(payload)) api_request("POST",ENDPOINT,payload,key,run) print("Submission response saved. Confirm its actual task ID before using poll.") return 0 if not 1 <= args.attempts <= 120 or not 1 <= args.interval <= 300: raise WorkflowError("Polling settings outside this client's bounds") run=Path(args.run);run.mkdir(parents=True,exist_ok=False) return poll_task(args.task_id,key,run,args.attempts,args.interval) except (WorkflowError,OSError,ValueError,TypeError,HTTPException) as error: text=str(error) if isinstance(error,WorkflowError) else type(error).__name__ print(text,file=sys.stderr);return 2 if __name__ == "__main__": raise SystemExit(main())