Odpytywanie zadania wideo w Pythonie bez dublowania płatnych generacji

Gdy urlopen rzuca wyjątek po 20 sekundach, wiesz tylko tyle, że jedno żądanie HTTP się nie udało. Film może się w tym czasie spokojnie renderować. Jeśli skrypt w takiej sytuacji wysyła zlecenie od nowa, płacisz za drugą generację. Dlatego rozdzielam w kodzie dwie rzeczy: zlecenie, które wykonuje się raz, i czekanie, które można przerwać i podjąć w dowolnym momencie.

Dwa etapy, dwie funkcje

• Zlecenie. Jedno POST /v1/generate. Zwrócone id od razu zapisuję w pliku albo w bazie aplikacji, zanim zrobię cokolwiek innego.
• Czekanie. Osobna funkcja dostaje samo id i czyta GET /v1/tasks/{id}. Po restarcie programu wołam tylko ją, więc nowe zadanie nie powstaje.
• Dwa różne limity. Timeout pojedynczego żądania (np. 20 s) i limit całego czekania (np. 24 próby co 5 s). Pierwszy chroni przed zawieszonym połączeniem, drugi przed pętlą bez końca.
• Cztery statusy. queued i running znaczą "czekaj dalej", succeeded i failed kończą pętlę. Brak pola output przy running jest normalny, to nie błąd formatu.
• Ponawianie tylko odczytu. Błąd sieci przy GET ponawiam po krótkiej przerwie. Tej zasady nie przenoszę na POST, bo powtórzone zlecenie to nowa płatna generacja.
• Zapis po zakończeniu. Pobieram plik i odkładam obok model oraz wejście. Przy następnych próbach łatwo porównać, co zmieniłem.

Treść zlecenia

To jest ciało POST /v1/generate dla pięciosekundowego klipu. Jeśli potrzebujesz referencji, wstaw adres własnego obrazu dostępnego publicznie.

{
  "model": "kling-3.0",
  "input": {
    "prompt": "A slow camera movement over a forest valley at sunrise, no text",
    "duration": 5
  }
}

Funkcja czekająca na wynik

Wystarczy standardowa biblioteka:

import json, os, time
from urllib.request import Request, urlopen

BASE = "https://kadraris.com/v1/tasks/"
KEY = os.environ["KADRARIS_API_KEY"]

def wait_for_task(task_id):
    # najwyżej 24 odczyty co 5 sekund
    for attempt in range(24):
        req = Request(BASE + task_id, headers={
            "Authorization": "Bearer " + KEY})
        with urlopen(req, timeout=20) as resp:
            task = json.load(resp)
        if task["status"] in ("succeeded", "failed"):
            return task
        time.sleep(5)
    raise TimeoutError("Zapisz id i wroc do zadania pozniej")

Kiedy limit się wyczerpie, funkcja nie zleca niczego nowego. Rzuca wyjątek, a ty wracasz do tego samego id później.

Lista kontrolna przed wdrożeniem

• Po restarcie skrypt czyta zapisane id, a nie wysyła nowego POST.
• Pętla czekania kończy się po określonym czasie.
• Kod odróżnia błąd HTTP (sieć, 5xx) od zadania ze statusem failed.

Koszt

Przykład korzysta z Kling 3.0, aktualna stawka to od 0,546 zł za sekundę. Dla własnych ustawień policz kwotę w cenniku albo w kalkulatorze kosztów. Zadanie ze statusem failed kosztuje 0 zł, saldo nie wygasa i nie ma abonamentu. Poprawnie wygenerowany film jest płatny także wtedy, gdy ostatecznie go nie użyjesz.

Pytania

Czy zamiast pętli mogę użyć parametru wait?

Tak, POST /v1/generate?wait=60 przytrzyma odpowiedź do minuty (maksymalnie 120 s). Film często renderuje się dłużej, więc i tak warto mieć funkcję, która dokończy odczyt po id.

Skąd wezmę cenę przed wysłaniem zlecenia?

Z pola estimated_cost w odpowiedzi API albo z przycisku "Generuj" w Studio, gdzie kwota jest widoczna przed kliknięciem.

Gdzie sprawdzę, jakie pola przyjmuje Kling 3.0?

Na stronie modelu i w dokumentacji generowania. Odczyt zadań opisuje dokumentacja zadań.