klingai-job-monitoring

# Kling AI Job Monitoring ## Overview Every Kling AI generation returns a `task_id`. This skill covers polling strategies, batch tracking, timeout handling, and callback-based monitoring for the `/v1/videos/text2video`, `/v1/videos/image2video`, and `/v1/videos/video-extend` endpoints. ## Task Lifecycle | Status | Meaning | Typical Duration | |--------|---------|-----------------| | `submitted` | Queued for processing | 0-30s | | `processing` | Generation in progress | 30-120s (standard), 60-300s (professional) | | `succeed` | Complete, video URL available | Terminal | | `failed` | Generation failed | Terminal | ## Polling a Single Task ```python import jwt, time, os, requests BASE = "https://api.klingai.com/v1" def get_headers(): ak, sk = os.environ["KLING_ACCESS_KEY"], os.environ["KLING_SECRET_KEY"] token = jwt.encode( {"iss": ak, "exp": int(time.time()) + 1800, "nbf": int(time.time()) - 5}, sk, algorithm="HS256", headers={"alg": "HS256", "typ": "JWT"} ) return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} def poll_task(endpoint: str, task_id: str, interval: int = 10, timeout: int = 600): """Poll with adaptive interval and timeout.""" start = time.monotonic() attempts = 0 while time.monotonic() - start < timeout: time.sleep(interval) attempts += 1 r = requests.get(f"{BASE}{endpoint}/{task_id}", headers=get_headers(), timeout=30) data = r.json()["data"] ...