Қосымшаңызға видео генерациясын қосу — сурет генерациясын қосумен бірдей емес. API шақыруы бірден жауап береді — бірақ видео әлі дайын емес. Сізге task ID беріледі, және ол дайын болғанға дейін “бітті ме?” деп қайта-қайта сұрауға тура келеді.
Көптеген әзірлеушілер мұны алғаш рет видео API шақырғанда бастан кешіреді: жауаптан видео URL күтеді, ал орнына task ID алады. Бұл нұсқаулық толық ағымды көрсетеді: тапсырманы жіберу, нәтижені поллинг арқылы тексеру, ақауларды өңдеу және URL мерзімі өтіп кетпей тұрып нәтижені сақтау.
Не әзірлейсіз
Мәтіндік промптты немесе суретті қабылдайтын, видео генерациясы тапсырмасын жіберетін, ол аяқталғанша поллинг жасайтын және соңғы видео URL қайтаратын backend қызметі. Төрт үлгімен жұмыс істейсіз — Veo 3 Fast, Sora 2, Kling Video және Runway — барлығы бір ғана API кілті арқылы.
Алдын ала талаптар:
- Python 3.8+ немесе Node.js 18+
- CometAPI кілті
- REST API-лерімен базалық таныстық
Неліктен видео генерациясы өзгеше екенін түсініңіз
Сурет генерациясында сіз сұрау жібересіз де, сол жауапта суретті аласыз. Видео генерациясы async тапсырмалар кезегін пайдаланады:
- Тапсырманы жіберу → жауап ретінде
task_idаласыз - Әр бірнеше секунд сайын мәртебені сұрау (poll)
- Мәртебе терминал күйге жеткенде, видео URL беріледі
- Жүктеп алып, сақтау — URL уақытша ғана
Егер видео генерациясын сурет генерациясы сияқты қарап, алғашқы жауаптың өзінде видеоны күтсеңіз, сұрауыңыз әр жолы таймаутқа ұшырайды.
Өндірістік веб-қызметте бұл поллинг циклы фондық worker-де (Celery, Bull немесе ұқсасы) орындалуы тиіс, request handler-де емес. Төмендегі мысалдар синхронды поллинг қолданады — скрипттер мен прототиптерге жарайды, бірақ көппайдаланушылы жүктемеге арналмаған.
Үлгіні таңдаңыз
| Model | Provider | Макс. ұзақтығы | Бағасы (CometAPI арқылы) | Қайда жақсы қолданылады |
|---|---|---|---|---|
| Veo 3 Fast | 8 сек | $0.05/сек | Жылдам прототиптеу, әлеуметтік роликтер | |
| Sora 2 | OpenAI (CometAPI model ID) | ~10 сек | $0.08/сек | Жоғары сапалы креативті шорттар |
| Kling Video | Kuaishou | 10 сек | $0.13–$2.64/тапсырма | Маркетингтік контент, егжей-тегжейлі басқару |
| Runway Gen-3A Turbo | Runway | 5 немесе 10 сек | $0.32/тапсырма | Суреттен видеоға, коммерциялық контент |
Дереккөз**: CometAPI үлгі беттері, Мамыр 2026. Ескерту: "Sora 2" — CometAPI-дің үлгі идентификаторы — ішкі үлгі туралы мәліметтер үшін олардың үлгі бетіне жүгініңіз.
- Veo 3 Fast мәтіннен-видеоға да, суреттен-видеоға да қолдайды. Секундына ең арзан, бастауға қолайлы.
- Sora 2 видеомен бірге аудионы да нативті түрде генерациялайды — диалог, фондық дыбыс және эффектілер, бөлек TTS қадамынсыз.
- Kling Video сізге
negative_prompt,cfg_scale, камера қозғалысы баптаулары жәнеproрежимін береді. Төртеуінің ішінде басқаруы ең кең. - Runway CometAPI арқылы тек суреттен-видеоға. Оған статикалық сурет пен қозғалыс сипаттамасын беріңіз — ол оны анимациялайды.
Veo тапсырмасын жіберу
Veo multipart/form-data қолданады. Python requests ішінде дұрыс жіберу үшін files= пайдаланыңыз — data=dict application/x-www-form-urlencoded жібереді, бұл басқаша:
import requestsimport osfrom dotenv import load_dotenvload_dotenv()def submit_veo_task(prompt: str, size: str = "16x9") -> str: """Submit a Veo 3 Fast text-to-video task. Returns task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") response = requests.post( "https://api.cometapi.com/v1/videos", headers={"Authorization": f"Bearer {api_key}"}, files={ "prompt": (None, prompt), "model": (None, "veo3-fast"), "size": (None, size) }, timeout=30 ) response.raise_for_status() return response.json()["id"]task_id = submit_veo_task("A paper kite drifting above a wheat field on a windy afternoon")print(f"Task submitted: {task_id}")
Нәтижені сұрау
import timedef poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str: """Poll until Veo task completes. Returns video URL.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") headers = {"Authorization": f"Bearer {api_key}"} url = f"https://api.cometapi.com/v1/videos/{task_id}" elapsed = 0 while elapsed < max_wait: response = requests.get(url, headers=headers, timeout=30) response.raise_for_status() result = response.json() status = result.get("status") if status == "succeeded": return result["output"][0] elif status in ("failed", "cancelled"): raise RuntimeError( f"Task {task_id} failed with status '{status}': " f"{result.get('error', 'no error detail returned')}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Task {task_id} did not complete within {max_wait} seconds")video_url = poll_veo_task(task_id)print(f"Video ready: {video_url}")
Көбірек басқару үшін Kling Video пайдаланыңыз
Kling басқа endpoint құрылымын қолданады және JSON пайдаланады. Назар аударыңыз: Kling-тің терминал мәртебе жолы "succeed" ( "succeeded" емес) — бұл API-дің нақты жауап пішіміне сәйкес келеді:
def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str: """Submit a Kling text-to-video task. Returns task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") response = requests.post( "https://api.cometapi.com/kling/v1/videos/text2video", headers={ "Authorization": f"Bearer {api_key}", "Content-Type": "application/json" }, json={ "model_name": "kling-v1-6", "prompt": prompt, "negative_prompt": "blurry, low quality, watermark", "cfg_scale": 0.5, "mode": mode, # "std" or "pro" "aspect_ratio": "16:9", "duration": duration # "5" or "10" }, timeout=30 ) response.raise_for_status() return response.json()["data"]["task_id"]def poll_kling_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str: """Poll Kling task until complete. Returns video URL.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") headers = {"Authorization": f"Bearer {api_key}"} url = f"https://api.cometapi.com/kling/v1/videos/text2video/{task_id}" elapsed = 0 while elapsed < max_wait: response = requests.get(url, headers=headers, timeout=30) response.raise_for_status() result = response.json() status = result["data"]["task_status"] if status == "succeed": # Kling uses "succeed", not "succeeded" return result["data"]["task_result"]["videos"][0]["url"] elif status == "failed": error_detail = result.get("data", {}).get("task_result", "no detail") raise RuntimeError( f"Kling task {task_id} failed: {error_detail}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Kling task {task_id} timed out after {max_wait}s")
Дереккөз**: CometAPI Kling Video құжаттамасы
Runway арқылы статикалық суретті жандандыру
Runway тек суреттен-видеоға. Ол қосымша тақырыпты да талап етеді (X-Runway-Version):
def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str: """Submit a Runway image-to-video task. Returns task_id.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") response = requests.post( "https://api.cometapi.com/runwayml/v1/image_to_video", headers={ "Authorization": f"Bearer {api_key}", "X-Runway-Version": "2024-11-06", "Content-Type": "application/json" }, json={ "model": "gen3a_turbo", "promptImage": image_url, # must be a stable HTTPS URL "promptText": motion_prompt, "duration": duration, "ratio": "1280:720", "watermark": False }, timeout=30 ) response.raise_for_status() return response.json()["id"]def poll_runway_task(task_id: str, interval: int = 5, max_wait: int = 600) -> str: """Poll Runway task. Returns video URL when done.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY environment variable is not set") headers = { "Authorization": f"Bearer {api_key}", "X-Runway-Version": "2024-11-06" } url = f"https://api.cometapi.com/runwayml/v1/tasks/{task_id}" elapsed = 0 while elapsed < max_wait: response = requests.get(url, headers=headers, timeout=30) response.raise_for_status() result = response.json() status = result.get("status") if status == "task_not_exist": # CometAPI-specific: task is still initializing, retry after a few seconds time.sleep(interval) elapsed += interval continue elif status == "succeeded": return result["output"][0] elif status in ("failed", "cancelled"): raise RuntimeError(f"Runway task {task_id} failed: {result.get('error', 'no detail')}") time.sleep(interval) elapsed += interval raise TimeoutError(f"Runway task {task_id} timed out after {max_wait}s")
Дереккөз**: CometAPI Runway құжаттамасы
URL мерзімі бітпей тұрып видеоны сақтаңыз
Видео генерациясы API-лері уақытша URL береді. Файлды бірден жүктеп алып, өзіңіз басқаратын жерде сақтаңыз:
import requestsimport pathlibdef download_video(url: str, output_path: str) -> None: """Download video from URL to local file using streaming.""" out = pathlib.Path(output_path) if out.parent != pathlib.Path("."): out.parent.mkdir(parents=True, exist_ok=True) with requests.get(url, stream=True, timeout=60) as r: r.raise_for_status() with open(out, "wb") as f: for chunk in r.iter_content(chunk_size=8192): f.write(chunk) print(f"Saved to {output_path}")# Full flowtask_id = submit_veo_task("A timelapse of clouds moving over a city skyline")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")
Өндірісте жергілікті файлға жазуды S3, Cloudflare R2 немесе таңдауыңыздағы басқа қоймаға жүктеуге ауыстырыңыз. Стриминг үлгісі өзгермейді — видеоны толық жадыға жүктемей, байттарды бірден арнаға өткізіңіз.
Ақауларды өңдеу
| Симптом | Мүмкін себеп | Шешім |
|---|---|---|
| Тапсырма queued күйінде 10+ минут тұрып қалды | Сервер жүктемесі жоғары немесе үлгі қолжетімсіз | Басқа үлгімен қайталап көріңіз |
| Алғашқы Runway поллингінде task_not_exist | Тапсырма әлі инициализациялануда | 5 секунд күтіп, қайта көріңіз — CometAPI құжатталған мінез-құлқы |
| Қате хабарламасыз failed | Промпт контент сүзгісін іске қосты | Промптты қайта тұжырымдаңыз |
| Видео URL 403 қайтарады | Жүктеп үлгермей URL мерзімі өтіп кетті | URL алынған бойда бірден жүктеп алыңыз |
| 10 минуттан кейін таймаут | Генерация тым ұзаққа созылды | max_wait мәнін арттырыңыз немесе Veo 3 Fast-қа ауысыңыз |
| Kling "succeed" қайтарады, "succeeded" емес | Kling API стандартты емес статус жолын қолданады | Бұл дұрыс — жоғарыдағы Kling поллинг кодына қараңыз |
Дереккөз: CometAPI видео генерациясы құжаттамасы
Node.js нұсқасы
Node.js 18+ құрамында fetch және FormData нативті бар. Бұл мысалда төрт үлгінің бәрі қамтылған:
// Node.js 18+ — no extra packages neededconst API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY is not set');// --- Veo 3 Fast ---async function submitVeoTask(prompt, size = '16x9') { const form = new FormData(); form.append('prompt', prompt); form.append('model', 'veo3-fast'); form.append('size', size); const res = await fetch('https://api.cometapi.com/v1/videos', { method: 'POST', headers: { 'Authorization': `Bearer ${API_KEY}` }, body: form }); if (!res.ok) throw new Error(`Veo submit failed: ${res.status}`); return (await res.json()).id;}async function pollVeoTask(taskId, intervalMs = 10000, maxWaitMs = 600000) { let elapsed = 0; while (elapsed < maxWaitMs) { const res = await fetch(`https://api.cometapi.com/v1/videos/${taskId}`, { headers: { 'Authorization': `Bearer ${API_KEY}` } }); if (!res.ok) throw new Error(`Poll failed: ${res.status}`); const result = await res.json(); if (result.status === 'succeeded') return result.output[0]; if (['failed', 'cancelled'].includes(result.status)) { throw new Error(`Task ${taskId} failed: ${result.error ?? 'no detail'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Task ${taskId} timed out`);}// --- Kling Video ---async function submitKlingTask(prompt, duration = '5', mode = 'std') { const res = await fetch('https://api.cometapi.com/kling/v1/videos/text2video', { method: 'POST', headers: { 'Authorization': `Bearer ${API_KEY}`, 'Content-Type': 'application/json' }, body: JSON.stringify({ model_name: 'kling-v1-6', prompt, negative_prompt: 'blurry, low quality, watermark', cfg_scale: 0.5, mode, aspect_ratio: '16:9', duration }) }); if (!res.ok) throw new Error(`Kling submit failed: ${res.status}`); return (await res.json()).data.task_id;}async function pollKlingTask(taskId, intervalMs = 10000, maxWaitMs = 600000) { let elapsed = 0; while (elapsed < maxWaitMs) { const res = await fetch( `https://api.cometapi.com/kling/v1/videos/text2video/${taskId}`, { headers: { 'Authorization': `Bearer ${API_KEY}` } } ); if (!res.ok) throw new Error(`Kling poll failed: ${res.status}`); const result = await res.json(); const status = result.data.task_status; if (status === 'succeed') return result.data.task_result.videos[0].url; if (status === 'failed') { throw new Error(`Kling task ${taskId} failed: ${JSON.stringify(result.data.task_result ?? 'no detail')}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Kling task ${taskId} timed out`);}// --- Runway (image-to-video) ---async function submitRunwayTask(imageUrl, motionPrompt, duration = 5) { const res = await fetch('https://api.cometapi.com/runwayml/v1/image_to_video', { method: 'POST', headers: { 'Authorization': `Bearer ${API_KEY}`, 'X-Runway-Version': '2024-11-06', 'Content-Type': 'application/json' }, body: JSON.stringify({ model: 'gen3a_turbo', promptImage: imageUrl, promptText: motionPrompt, duration, ratio: '1280:720', watermark: false }) }); if (!res.ok) throw new Error(`Runway submit failed: ${res.status}`); return (await res.json()).id;}async function pollRunwayTask(taskId, intervalMs = 5000, maxWaitMs = 600000) { let elapsed = 0; while (elapsed < maxWaitMs) { const res = await fetch( `https://api.cometapi.com/runwayml/v1/tasks/${taskId}`, { headers: { 'Authorization': `Bearer ${API_KEY}`, 'X-Runway-Version': '2024-11-06' } } ); if (!res.ok) throw new Error(`Runway poll failed: ${res.status}`); const result = await res.json(); const status = result.status; if (status === 'task_not_exist') { // CometAPI-specific: task still initializing await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; continue; } if (status === 'succeeded') return result.output[0]; if (['failed', 'cancelled'].includes(status)) { throw new Error(`Runway task ${taskId} failed: ${result.error ?? 'no detail'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Runway task ${taskId} timed out`);}// Usage exampleconst taskId = await submitVeoTask('A paper kite drifting above a wheat field');const videoUrl = await pollVeoTask(taskId);console.log('Video ready:', videoUrl);
Келесі қадам
Енді сізде төрт видео үлгісімен жұмыс істейтін код, ақауларды өңдейтін поллинг циклы және генерацияланған контентті жоғалтып алмау үшін жүктеп алу қадамы бар.
Келесіде әзірлеушілер жиі кездесетін мәселе: бір үлгіге қатты байланып қалу, ал арзанырақ немесе жылдамырақ нұсқаға ауысу бірнеше файлды өзгертуге мәжбүрлейді. Келесі мақалада кодты қайта жазбай-ақ сұрауларды әртүрлі үлгілерге қалай маршрутизациялау қарастырылады.
Келесі: Кодыңызды қайта жазбай-ақ AI үлгілері арасында қалай ауысуға болады
Жиі қойылатын сұрақтар
Q: Неліктен API жауабында видеонның орнына task ID аламын?
Видео генерациясы async — Veo, Sora, Kling және Runway сияқты үлгілерге рендерлеуге 2–5 минут қажет. API сұрауыңыз таймаутқа ұшырамас үшін бірден task ID қайтарады. Тапсырма терминал күйге (succeeded, succeed, failed) жеткенше бөлек статус endpoint-ін поллинг жасайсыз.
Q: Генерацияланған видео URL қанша уақытқа жарамды?
Видео генерациясы API-лерінің URL-дері уақытша. URL алынған бойда файлды бірден жүктеп алып, өзіңіздің сақтауыңызға (S3, Cloudflare R2 және т.б.) қойыңыз. URL-ді сақтап, ол бірнеше сағаттан кейін де жұмыс істейді деп күтпеңіз.
Q: Veo 3 Fast пен Kling Video арасындағы айырмашылық қандай?
Veo 3 Fast арзанырақ ($0.05/сек), жылдамырақ және шақыруы қарапайым. Kling Video көбірек басқару береді: negative_prompt, cfg_scale, камера қозғалысы баптаулары және pro сапа режимі. Нәтижені дәл баптау керек болса — Kling, жылдамдық пен төмен шығын қажет болса — Veo 3 Fast таңдаңыз.
Q: Мәтіндік промпттың орнына суреттен видео жасай аламын ба?
Иә. Veo input_reference файлын беру арқылы суреттен-видеоға қолдайды. Kling /kling/v1/videos/image2video endpoint-і арқылы image параметрімен (URL немесе base64) қолдайды. Runway тек суреттен-видеоға — CometAPI арқылы таза мәтінді қабылдамайды.
Q: Неліктен Runway алғашқы поллингте task_not_exist қайтарып береді?
Бұл CometAPI-дің құжатталған мінез-құлқы — тапсырма backend-та әлі инициализациялануда. Бірнеше секунд күтіп, қайта көріңіз. Бұл қате емес. Жоғарыдағы поллинг коды мұны автоматты түрде өңдейді.
Q: Неліктен Kling "succeed" қолданады да, "succeeded" емес?
Бұл Kling API-інің нақты жауап пішімі. Қате емес. Veo және Runway "succeeded" қолданады — Kling "succeed" қолданады. Бірыңғай поллинг орамасын құрып жатсаңыз, екі жолды да өңдеуіңіз керек.
Q: Синхронды поллинг циклын веб-серверде қолдану қауіпсіз бе?
Жоқ. Бұл нұсқаулықтағы поллинг циклы ағынды бірнеше минутқа блоктайды. Нақты көппайдаланушылы веб-қызметте поллингті фондық worker-де (Python үшін Celery, Node.js үшін Bull) орындаңыз. Request handler-де тапсырманы жіберіңіз, клиентке task ID қайтарыңыз, ал видеo дайын болғанда worker клиентті хабардар етсін.