كيفية إضافة توليد الفيديو بالذكاء الاصطناعي إلى تطبيق SaaS

إضافة توليد الفيديو إلى تطبيقك ليست مماثلة لإضافة توليد الصور. يستجيب نداء API فوراً — لكن الفيديو لا يكون جاهزاً بعد. ستحصل على معرّف مهمة، ويجب أن تواصل السؤال "is it done?" حتى يكتمل.

يصطدم معظم المطوّرين بهذا عند أول مرة يستدعون فيها واجهة فيديو، فينتظرون جسم الاستجابة مع رابط فيديو، لكنهم يتلقّون بدلاً منه معرّف مهمة. يشرح هذا الدليل التدفق الكامل: إرسال مهمة، الاستعلام الدوري عن النتائج، معالجة الإخفاقات، وتخزين الناتج قبل انتهاء صلاحية الرابط.

ما الذي ستبنيه

خدمة خلفية تستقبل مُدخل نصي أو صورة، ترسل مهمة توليد فيديو، تستعلم دورياً حتى الاكتمال، ثم تُرجع رابط الفيديو النهائي. ستعمل مع أربعة نماذج — Veo 3 Fast وSora 2 وKling Video وRunway — عبر مفتاح API واحد.

المتطلبات الأساسية:

• Python 3.8+ أو Node.js 18+
• مفتاح CometAPI
• إلمام أساسي بـ REST APIs

افهم لماذا يختلف توليد الفيديو

في توليد الصور، ترسل طلباً وتتلقى الصورة في الاستجابة نفسها. أما توليد الفيديو فيستخدم طابور مهام غير متزامن:

1. أرسل طلب التوليد → تتلقى task_id
2. استعلم دورياً عن نقطة حالة كل بضع ثوانٍ
3. عندما تصل الحالة إلى وضع نهائي، تحصل على رابط الفيديو
4. نزّل وخزّن الفيديو — الرابط مؤقت

إذا عاملت توليد الفيديو كما لو كان توليد صور وانتظرت أن تحتوي الاستجابة الأولى على الفيديو، فسوف تنتهي مهلة طلبك في كل مرة.

في خدمة ويب للإنتاج، يجب أن يعمل حلقة الاستعلام الدوري في عامل خلفي (Celery أو Bull أو ما شابه)، وليس في معالج الطلب. الأمثلة أدناه تستخدم استعلاماً متزامناً — مناسب للبرامج النصية والنماذج الأولية، لكنه غير مناسب للتعامل مع مستخدمين متزامنين.

اختر نموذجاً

المصدر**: صفحات نماذج CometAPI، مايو 2026. ملاحظة: "Sora 2" هو identifier لنموذج CometAPI — راجع model page للتفاصيل عن النموذج الأساسي.

• Veo 3 Fast يدعم التحويل من النص إلى فيديو ومن الصورة إلى فيديو. الأرخص لكل ثانية، وبداية جيدة.
• Sora 2 يولّد الصوت بشكل أصيل إلى جانب الفيديو — حوار وأصوات محيطة وتأثيرات دون خطوة TTS منفصلة.
• Kling Video يوفّر لك negative_prompt وcfg_scale وإعدادات حركة الكاميرا ووضع pro. الأكثر تحكّماً بين الأربعة.
• Runway عبر CometAPI يدعم من الصورة إلى الفيديو فقط. قدّم صورة ثابتة ووصف حركة، وسيقوم بتحريكها.

إرسال مهمة Veo

يستخدم Veo multipart/form-data. استخدم files= في requests الخاصة بـ Python لإرساله بشكل صحيح — إن data=dict يرسل application/x-www-form-urlencoded، وهو شيء مختلف:

import requestsimport osfrom dotenv import load_dotenv​load_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 time​def 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 بنية نقاط نهاية مختلفة ويستخدم JSON. لاحظ أن سلسلة الحالة النهائية في Kling هي "succeed" (وليس "succeeded") — وهذا مطابق للاستجابة الفعلية للواجهة:

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 docs

حرّك صورة ثابتة باستخدام 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 docs

احفظ الفيديو قبل انتهاء صلاحية عنوان URL

روابط الفيديو من واجهات التوليد مؤقتة. نزّل الملف فوراً وخزّنه في مكان تتحكّم به:

import requestsimport pathlib​def 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 أو مخزن تختاره. يظل نمط البث كما هو — مرّر البايتات مباشرة بدلاً من تحميل الفيديو كله في الذاكرة.

معالجة حالات الفشل

المصدر: CometAPI video generation docs

نسخة Node.js

يتضمّن Node.js 18+ دعماً أصيلاً لـ fetch وFormData. يغطي هذا المثال النماذج الأربعة كلها:

// Node.js 18+ — no extra packages needed​const 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);

ما التالي

لديك الآن شيفرة عاملة لأربعة نماذج فيديو، وحلقة استعلام تتعامل مع الإخفاقات، وخطوة تنزيل تمنعك من فقدان المحتوى المُولَّد.

المشكلة التالية التي يواجهها معظم المطوّرين: ثبّتوا نموذجاً واحداً، والتبديل إلى خيار أرخص أو أسرع يتطلّب تعديل عدة ملفات. المقال التالي يغطّي كيفية توجيه الطلبات عبر النماذج دون إعادة كتابة شيفرتك.

التالي: كيفية التبديل بين نماذج الذكاء الاصطناعي دون إعادة كتابة الشفرة

الأسئلة الشائعة

س: لماذا أحصل على معرّف مهمة بدلاً من فيديو في استجابة API؟

توليد الفيديو غير متزامن — النماذج مثل Veo وSora وKling وRunway تستغرق 2–5 دقائق للمعالجة. تُرجع الواجهة معرّف مهمة فوراً حتى لا تنتهي مهلة طلبك. تستعلم دورياً عن نقطة حالة منفصلة حتى تصل المهمة إلى حالة نهائية (succeeded، succeed، failed).

س: كم تدوم صلاحية رابط الفيديو المُولَّد؟

روابط الفيديو من واجهات التوليد مؤقتة. نزّل الملف فور حصولك على الرابط وخزّنه في مساحة التخزين الخاصة بك (S3، Cloudflare R2، إلخ). لا تخزّن الرابط وتتوقع عمله بعد ساعات.

س: ما الفرق بين Veo 3 Fast وKling Video؟

Veo 3 Fast أرخص ($0.05/sec) وأسرع وأسهل في النداء. يتيح لك Kling Video تحكّماً أكبر: negative_prompt وcfg_scale وإعدادات حركة الكاميرا ووضع جودة pro. إن كنت تحتاج إلى ضبط دقيق للمخرجات فاختر Kling. إن كنت تحتاج السرعة وتكلفة منخفضة فاختر Veo 3 Fast.

س: هل يمكنني توليد فيديو من صورة بدلاً من مطالبة نصية؟

نعم. يدعم Veo من الصورة إلى الفيديو بتمرير ملف input_reference. يدعمه Kling عبر نقطة /kling/v1/videos/image2video مع معامل image (رابط URL أو base64). Runway يدعم من الصورة إلى الفيديو فقط — لا يقبل مطالبات نصية فقط عبر CometAPI.

س: لماذا يعيد Runway task_not_exist في أول عملية استعلام؟

هذا سلوك موثّق لدى CometAPI — ما تزال المهمة في طور التهيئة على الواجهة الخلفية. انتظر بضع ثوانٍ وأعد المحاولة. هذا ليس خطأ. يتعامل كود الاستعلام أعلاه مع ذلك تلقائياً.

س: لماذا يستخدم Kling "succeed" بدلاً من "succeeded"؟

هذه هي صيغة الاستجابة الفعلية لواجهة Kling. ليس خطأً مطبعياً. تستخدم Veo وRunway "succeeded" — بينما يستخدم Kling "succeed". إذا كنت تبني غلاف استعلام موحّداً، ستحتاج إلى التعامل مع السلسلتين.

س: هل حلقة الاستعلام المتزامنة آمنة للاستخدام في خادم ويب؟

لا. حلقة الاستعلام في هذا الدليل تحجب الخيط لدقائق. في خدمة ويب حقيقية مع مستخدمين متزامنين، شغّل الاستعلام في عامل خلفي (Celery لـ Python، Bull لـ Node.js). أرسل المهمة في معالج الطلب، أعد معرّف المهمة إلى العميل، ودع العامل يُخطر العميل عندما يصبح الفيديو جاهزاً.