كيفية إضافة توليد الفيديو بالذكاء الاصطناعي إلى تطبيق برمجيات كخدمة (SaaS)

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

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

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

خدمة خلفية تستقبل مُحفِّز نصي أو صورة، ترسل مهمة توليد فيديو، تستعلم حتى تكتمل، ثم تعيد رابط الفيديو النهائي. ستتعامل مع أربعة نماذج — 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" هو مُعرّف النموذج لدى CometAPI — راجع صفحة النموذج للتفاصيل الخاصة بالنموذج الأساسي.

• 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

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

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

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). أرسل المهمة في معالج الطلب، أعد معرّف المهمة للعميل، ودع العامل يُخطِر العميل عند جهوزية الفيديو.