Қолданбаңызға видео генерациясын қосу — сурет генерациясын қосу сияқты емес. API шақыруы бірден жауап береді — бірақ видео әлі дайын емес. Сіз task ID аласыз, және ол дайын болғанша «бітті ме?» деп қайта-қайта сұрауыңыз керек.
Көптеген әзірлеушілер мұнымен видео API-ді алғаш рет шақырғанда бетпе-бет келеді: жауаптан видео URL күтеді, ал орнына task ID алады. Бұл нұсқаулық толық үрдісті көрсетеді: тапсырманы жіберу, нәтижені поллинг арқылы күту, сәтсіздіктерді өңдеу және URL мерзімі өтіп кетпей тұрып нәтижені сақтау.
Нені құрасыз
Мәтіндік промптты немесе суретті қабылдайтын, видео генерация тапсырмасын жіберетін, аяқталғанша поллинг жасайтын және соңғы видео URL-ын қайтаратын бэкенд қызмет. Сіз төрт модельмен жұмыс жасайсыз — Veo 3 Fast, Sora 2, Kling Video және Runway — бәрі бір API кілті арқылы.
Алдын ала талаптар:
- Python 3.8+ немесе Node.js 18+
- CometAPI кілті
- REST API-лерімен базалық таныстық
Видео генерациясының неге өзгеше екенін түсініңіз
Сурет генерациясында сіз сұрау жіберіп, сол жауапта суретті қайтарасыз. Видео генерациясы асинхронды тапсырмалар кезегін пайдаланады:
- Жіберу генерация сұрауы → жауапта
task_idкеледі - Әр бірнеше секунд сайын күй эндпоинтін поллинг жасау
- Күй терминалды мәнге жеткенде, видео URL береді
- Жүктеп алып, сақтау — URL уақытша
Егер видео генерациясын сурет генерациясы сияқты қабылдап, алғашқы жауапта видео болады деп күтсеңіз, сұрауыңыз әр жолы тайм-аутқа ұшырайды.
Өндірістік веб-қызметте бұл поллинг циклі сұрауды өңдеушіде емес, фондық жұмысшыда (Celery, Bull немесе ұқсасы) орындалуы тиіс. Төмендегі мысалдар синхронды поллингті қолданады — сценарийлер мен прототиптер үшін жарайды, бірақ бір уақытта бірнеше пайдаланушыны өңдеу үшін жарамсыз.
Модельді таңдаңыз
| Model | Provider | Max duration | Price (via CometAPI) | Best for |
|---|---|---|---|---|
| Veo 3 Fast | 8 sec | $0.05/sec | Жылдам прототиптеу, әлеуметтік қысқа бейнелер | |
| Sora 2 | OpenAI (via CometAPI model ID) | ~10 sec | $0.08/sec | Жоғары сапалы креативті қысқалар |
| Kling Video | Kuaishou | 10 sec | $0.13–$2.64/task | Маркетинг контенті, егжей-тегжейлі басқару |
| Runway Gen-3A Turbo | Runway | 5 or 10 sec | $0.32/task | Суреттен видеоға, коммерциялық контент |
Source**: CometAPI model pages, May 2026. Note: "Sora 2" is CometAPI's model identifier — refer to their model page for the underlying model details.
- 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: """Veo 3 Fast мәтіннен-видеоға тапсырмасын жіберу. task_id қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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("Желді түстен кейін бидай алқабының үстінде қалықтап жүрген қағаз батпырауық")print(f"Тапсырма жіберілді: {task_id}")
Нәтижені поллинг арқылы күту
import timedef poll_veo_task(task_id: str, interval: int = 10, max_wait: int = 600) -> str: """Veo тапсырмасы аяқталғанша поллинг жасау. Видео URL қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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_id} '{status}' күйімен сәтсіз аяқталды: " f"{result.get('error', 'қате туралы қосымша мәлімет жоқ')}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Тапсырма {task_id} {max_wait} секунд ішінде аяқталмады")video_url = poll_veo_task(task_id)print(f"Видео дайын: {video_url}")
Көбірек басқару үшін Kling Video қолданыңыз
Kling басқа эндпоинт құрылымын және JSON пайдаланады. Назар аударыңыз: Kling-тің соңғы күй жолы "succeed" ( "succeeded" емес ) — бұл API-ның нақты жауап форматына сәйкес келеді:
def submit_kling_task(prompt: str, duration: str = "5", mode: str = "std") -> str: """Kling мәтіннен-видеоға тапсырмасын жіберу. task_id қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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": "бұлдыр, төмен сапа, сутаңба", "cfg_scale": 0.5, "mode": mode, # "std" немесе "pro" "aspect_ratio": "16:9", "duration": duration # "5" немесе "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: """Kling тапсырмасын аяқталғанша поллинг жасау. Видео URL қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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 "succeed" қолданады, "succeeded" емес return result["data"]["task_result"]["videos"][0]["url"] elif status == "failed": error_detail = result.get("data", {}).get("task_result", "мәлімет жоқ") raise RuntimeError( f"Kling тапсырмасы {task_id} сәтсіз аяқталды: {error_detail}" ) time.sleep(interval) elapsed += interval raise TimeoutError(f"Kling тапсырмасы {task_id} {max_wait} с кейін тайм-аутқа ұшырады")
Source**: CometAPI Kling Video docs
Runway көмегімен статикалық суретті анимациялау
Runway тек суреттен-видеоға. Сонымен бірге қосымша тақырып (X-Runway-Version) талап етеді:
def submit_runway_task(image_url: str, motion_prompt: str, duration: int = 5) -> str: """Runway суреттен-видеоға тапсырмасын жіберу. task_id қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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, # тұрақты 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: """Runway тапсырмасын поллинг жасау. Дайын болғанда видео URL қайтарады.""" api_key = os.getenv("COMETAPI_KEY") if not api_key: raise ValueError("COMETAPI_KEY орта айнымалысы орнатылмаған") 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-ге тән: тапсырма әлі инициализациялануда, бірнеше секундтан соң қайталап көріңіз time.sleep(interval) elapsed += interval continue elif status == "succeeded": return result["output"][0] elif status in ("failed", "cancelled"): raise RuntimeError(f"Runway тапсырмасы {task_id} сәтсіз аяқталды: {result.get('error', 'мәлімет жоқ')}") time.sleep(interval) elapsed += interval raise TimeoutError(f"Runway тапсырмасы {task_id} {max_wait}s кейін тайм-аутқа ұшырады")
Source**: CometAPI Runway docs
URL мерзімі өтіп кетпей тұрып видеоны сақтаңыз
Генерация API-лерінен келген видео URL-дары уақытша. Файлды бірден жүктеп алып, өзіңіз басқаратын жерде сақтаңыз:
import requestsimport pathlibdef download_video(url: str, output_path: str) -> None: """Ағын арқылы URL-дан видеоны жергілікті файлға жүктеу.""" 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"{output_path} жолына сақталды")# Толық үрдістask_id = submit_veo_task("Қаланың көкжиегі үстімен қозғалып жатқан бұлттардың таймлапсы")video_url = poll_veo_task(task_id)download_video(video_url, "output/city_timelapse.mp4")
Өндірісте жергілікті файлға жазуды S3, Cloudflare R2 немесе таңдауыңыздағы қоймаға жүктеумен алмастырыңыз. Ағындық үлгі өзгермейді — бүкіл видеоны жадқа жүктемей, байттарды тікелей өткізіңіз.
Сәтсіздіктерді өңдеу
| Symptom | Likely cause | Fix |
|---|---|---|
Тапсырма queued күйінде 10+ мин тұрып қалды | Сервер жүктемесі немесе модель қолжетімсіз | Басқа модельмен қайталап көріңіз |
Алғашқы Runway поллингінде task_not_exist | Тапсырма әлі инициализациялануда | 5 сек күтіп, қайта көріңіз — CometAPI құжатталған мінез-құлқы |
| Қате хабарламасыз сәтсіз аяқталды | Промпт контент сүзгісін іске қосты | Промптты қайта тұжырымдаңыз |
| Видео URL 403 қайтарады | Жүктеп үлгермей тұрып URL мерзімі өтіп кетті | URL алған бойда дереу жүктеп алыңыз |
| 10 минуттан кейін тайм-аут | Генерация тым ұзаққа созылды | max_wait-ты арттырыңыз немесе Veo 3 Fast-қа ауысыңыз |
| Kling «succeed» қайтарады, «succeeded» емес | Kling API-ы стандартты емес күй жолын қолданады | Бұл дұрыс — жоғарыдағы Kling поллинг кодын қараңыз |
Source: CometAPI video generation docs
Node.js нұсқасы
Node.js 18+ құрамында fetch және FormData нативті түрде бар. Бұл мысал төрт модельдің бәрін қамтиды:
// Node.js 18+ — қосымша пакеттер қажет емесconst API_KEY = process.env.COMETAPI_KEY;if (!API_KEY) throw new Error('COMETAPI_KEY орнатылмаған');// --- 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 жіберу сәтсіз болды: ${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(`Поллинг сәтсіз: ${res.status}`); const result = await res.json(); if (result.status === 'succeeded') return result.output[0]; if (['failed', 'cancelled'].includes(result.status)) { throw new Error(`Тапсырма ${taskId} сәтсіз аяқталды: ${result.error ?? 'мәлімет жоқ'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Тапсырма ${taskId} тайм-аутқа ұшырады`);}// --- 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: 'бұлдыр, төмен сапа, сутаңба', cfg_scale: 0.5, mode, aspect_ratio: '16:9', duration }) }); if (!res.ok) throw new Error(`Kling жіберу сәтсіз: ${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 поллинг сәтсіз: ${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 тапсырмасы ${taskId} сәтсіз: ${JSON.stringify(result.data.task_result ?? 'мәлімет жоқ')}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Kling тапсырмасы ${taskId} тайм-аутқа ұшырады`);}// --- 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 жіберу сәтсіз: ${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 поллинг сәтсіз: ${res.status}`); const result = await res.json(); const status = result.status; if (status === 'task_not_exist') { // CometAPI-ге тән: тапсырма әлі инициализациялануда 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 тапсырмасы ${taskId} сәтсіз: ${result.error ?? 'мәлімет жоқ'}`); } await new Promise(r => setTimeout(r, intervalMs)); elapsed += intervalMs; } throw new Error(`Runway тапсырмасы ${taskId} тайм-аутқа ұшырады`);}// Пайдалану мысалыconst taskId = await submitVeoTask('Бидай алқабының үстінде қалықтап жүрген қағаз батпырауық');const videoUrl = await pollVeoTask(taskId);console.log('Видео дайын:', videoUrl);
Келесі не
Енді сізде төрт видео моделі үшін жұмыс істейтін код, сәтсіздіктерді өңдейтін поллинг циклі және генерацияланған контентті жоғалтпайтындай жүктеп алу қадамы бар.
Келесіде әзірлеушілер кездесетін мәселе: бір модельді хардкодтап қояды да, арзанырақ немесе жылдамырақ нұсқаға ауысу бірнеше файлды өзгертуді талап етеді. Келесі мақала кодты қайта жазбай-ақ сұрауларды әртүрлі модельдерге қалай бағыттауды түсіндіреді.
Келесі: How to Switch Between AI Models Without Rewriting Your Code
FAQ
Q: Неліктен API жауаптан видеоны емес, task ID аламын?
Видео генерациясы асинхронды — Veo, Sora, Kling және Runway сияқты модельдерге рендерлеу үшін 2–5 минут керек. Сұрауыңыз тайм-аутқа ұшырамасын деп API бірден task ID қайтарады. Тапсырма терминалды күйге (succeeded, succeed, failed) жеткенше бөлек күй эндпоинтін поллинг жасайсыз.
Q: Генерацияланған видео URL қанша уақытқа жарамды?
Видео генерация API-лерінің URL-дары уақытша. URL алған бойда бірден файлды жүктеп алып, өзіңіздегі сақтауға (S3, Cloudflare R2, т.б.) шығарыңыз. URL-ды сақтап, бірнеше сағаттан кейін де жұмыс істейді деп күтпеңіз.
Q: Veo 3 Fast пен Kling Video арасында қандай айырмашылық бар?
Veo 3 Fast арзан ($0.05/sec), жылдам және шақыруы жеңіл. Kling Video көбірек басқару береді: negative_prompt, cfg_scale, камера қозғалыстары және pro сапа режимі. Нәтижені майда баптау керек болса — Kling; жылдамдық пен төмен құн керек болса — Veo 3 Fast таңдаңыз.
Q: Мәтін орнына суреттен видео жасай аламын ба?
Иә. Veo суреттен-видеоға input_reference файлын беру арқылы қолдайды. Kling бұны /kling/v1/videos/image2video эндпоинті арқылы image параметрімен (URL немесе base64) қолдайды. Runway тек суреттен-видеоға — CometAPI арқылы таза мәтіндік промпт қабылдамайды.
Q: Неліктен Runway алғашқы поллингте task_not_exist қайтарады?
Бұл CometAPI құжатталған мінез-құлқы — тапсырма бэкендте әлі инициализациялануда. Бірнеше секунд күтіп, қайта көріңіз. Бұл қате емес. Жоғарыдағы поллинг коды мұны автоматты түрде өңдейді.
Q: Неліктен Kling "succeed" пайдаланады, "succeeded" емес?
Бұл Kling API-ының нақты жауап форматы. Қате емес. Veo және Runway "succeeded" қолданады — Kling "succeed". Біріктірілген поллинг орамын жасасаңыз, екі жолды да өңдеуіңіз керек.
Q: Синхронды поллинг циклі веб-серверде қолдануға қауіпсіз бе?
Жоқ. Бұл нұсқаулықтағы поллинг циклі ағынды минуттар бойы блоктайды. Нақты веб-қызметте, поллингті фондық жұмысшыда (Python үшін Celery, Node.js үшін Bull) орындаңыз. Тапсырманы сұрау өңдеушісінен жіберіп, client-ке task ID қайтарыңыз да, видео дайын болғанда жұмысшы клиентке хабарлассын.