"""scripts/generate-video.py -- CI-friendly video generation."""
import argparse
import jwt
import time
import os
import requests
import sys
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 main():
parser = argparse.ArgumentParser()
parser.add_argument("--prompt", required=True)
parser.add_argument("--model", default="kling-v2-master")
parser.add_argument("--duration", default="5")
parser.add_argument("--mode", default="standard")
parser.add_argument("--output", default="output/")
parser.add_argument("--timeout", type=int, default=600)
args = parser.parse_args()
os.makedirs(args.output, exist_ok=True)
r = requests.post(f"{BASE}/videos/text2video", headers=get_headers(), json={
"model_name": args.model,
"prompt": args.prompt,
"duration": args.duration,
"mode": args.mode,
})
r.raise_for_status()
task_id = r.json()["data"]["task_id"]
print(f"Task submitted: {task_id}")
start = time.monotonic()
while time.monotonic() - start < args.timeout:
time.sleep(15)
result = requests.get(
f"{BASE}/videos/text2video/{task_id}", headers=get_headers()
).json()
status = result["data"]["task_status"]
elapsed = int(time.monotonic() - start)
print(f"[{elapsed}s] Status: {status}")
if status == "succeed":
video_url = result["data"]["task_result"]["videos"][0]["url"]
filepath = os.path.join(args.output, f"{task_id}.mp4")
with open(filepath, "wb") as f:
f.write(requests.get(video_url).content)
print(f"Saved: {filepath}")
return
if status == "failed":
print(f"FAILED: {result['data'].get('task_status_msg')}", file=sys.stderr)
sys.exit(1)
print("TIMEOUT: generation did not complete", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
import yaml
with open("video-prompts.yml") as f:
config = yaml.safe_load(f)
for video in config["videos"]:
task_id = submit_async(video["prompt"], model=video["model"])
print(f"{video['name']}: {task_id}")
generate-video:
image: python:3.11-slim
stage: build
script:
- pip install PyJWT requests
- python3 scripts/generate-video.py --prompt "$VIDEO_PROMPT" --output output/
artifacts:
paths:
- output/*.mp4
expire_in: 7 days
variables:
KLING_ACCESS_KEY: $KLING_ACCESS_KEY
KLING_SECRET_KEY: $KLING_SECRET_KEY
