كيفية بناء خط أنابيب فيديو بالذكاء الاصطناعي باستخدام Python و Atlas Cloud

تبدأ معظم الفرق عملها في إنشاء الفيديوهات بالذكاء الاصطناعي عبر إجراء مكالمات API فردية؛ أي إنشاء فيديو واحد، ثم تحميله، ثم الانتقال لغيره. وهذا الأسلوب مناسب للتجربة فقط.

*آخر تحديث: 28 فبراير 2026*

شاهد هذه النماذج أثناء العمل:

بنية خط الأنابيب (Pipeline Architecture)

قبل كتابة الكود، إليك البنية عالية المستوى لما سنقوم ببنائه:

plaintextCopy
1```
2+-------------------+     +--------------------+     +------------------+
3|  إعدادات الأوامر   |     |  Atlas Cloud API   |     |  تخزين المخرجات   |
4|  (JSON/YAML)      |     |                    |     |                  |
5|  - prompts        +---->+  /generateImage    +---->+  /images/        |
6|  - models         |     |  /generateVideo    |     |  /videos/        |
7|  - parameters     |     |  /prediction/get   |     |  /manifest.json  |
8+-------------------+     +--------------------+     +------------------+
9           |                         |                         |
10           v                         v                         v
11+-------------------+     +--------------------+     +------------------+
12|  محرك خط الأنابيب   |     |  الاستعلام وإعادة    |     |  تتبع التكاليف    |
13|                   |     |  المحاولة           |     |                  |
14|  - batch_generate |     |  - التراجع الأسي    |     |  - لكل طلب       |
15|  - concurrency    |     |  - أقصى محاولات    |     |  - تراكمي        |
16|  - model routing  |     |                     |     |  - لكل نموذج     |
17+-------------------+     +--------------------+     +------------------+
18```

يتبع خط الأنابيب مساراً بسيطاً:

1. قراءة إعدادات الأوامر (Prompts) من ملف إدخال منظم.
2. توجيه كل أمر إلى النموذج ونقطة النهاية المناسبة (صور أو فيديو).
3. إرسال جميع الطلبات إلى واجهة برمجة تطبيقات Atlas Cloud مع تحكم في التزامن.
4. الاستعلام عن النتائج مع منطق التراجع الأسي وإعادة المحاولة.
5. تحميل المخرجات المكتملة وحفظها في مجلدات منظمة.
6. تتبع التكاليف وإنشاء ملف ملخص (manifest).

البدء: الوصول إلى الـ API

الخطوة 1: احصل على مفتاح الـ API الخاص بك

سجّل في Atlas Cloud وأنشئ مفتاح API من لوحة التحكم. رصيد USD1 المجاني كافٍ لاختبار خط الأنابيب بالكامل مع العديد من عمليات إنشاء الصور والفيديو.

الخطوة 2: تثبيت المكتبات البرمجية

plaintextCopy
1```bash
2pip install requests pyyaml
3```

لا حاجة لأطر عمل ثقيلة. يستخدم خط الأنابيب فقط مكتبة

1requests

1pyyaml

كود خط الأنابيب الكامل

فيما يلي كود خط الأنابيب العامل بالكامل. يتم شرح كل قسم بعد كتلة الكود.

plaintextCopy
1```python
2import requests
3import time
4import json
5import os
6import logging
7from concurrent.futures import ThreadPoolExecutor, as_completed
8from dataclasses import dataclass, field
9from typing import Optional
10from datetime import datetime
11
12# إعداد السجلات (Logging)
13logging.basicConfig(
14    level=logging.INFO,
15    format="%(asctime)s [%(levelname)s] %(message)s",
16    datefmt="%Y-%m-%d %H:%M:%S"
17)
18logger = logging.getLogger("atlas_pipeline")
19
20@dataclass
21class GenerationResult:
22    """يخزن نتيجة طلب إنشاء واحد."""
23    name: str
24    model: str
25    media_type: str  # "image" أو "video"
26    status: str  # "success", "failed", "error"
27    output_url: Optional[str] = None
28    local_path: Optional[str] = None
29    cost_estimate: float = 0.0
30    duration_seconds: float = 0.0
31    error_message: Optional[str] = None
32
33class AtlasCloudClient:
34    """مغلف العميل لواجهة Atlas Cloud API."""
35
36    BASE_URL = "https://api.atlascloud.ai/api/v1"
37
38    # التسعير لكل نموذج (تقريبي)
39    PRICING = {
40        "black-forest-labs/flux-2-pro/text-to-image": 0.04,        # لكل صورة
41        "google/imagen4-ultra/text-to-image": 0.06,                 # لكل صورة
42        "bytedance/seedance-v1.5-pro/text-to-video": 0.022,       # لكل ثانية
43        "google/veo3.1/text-to-video": 0.03,                       # لكل ثانية
44        "openai/sora-v2/text-to-video": 0.15,                     # لكل ثانية
45    }
46
47    def __init__(self, api_key: str):
48        self.api_key = api_key
49        self.session = requests.Session()
50        self.session.headers.update({
51            "Authorization": f"Bearer {api_key}",
52            "Content-Type": "application/json"
53        })
54
55    def generate_image(
56        self,
57        model: str,
58        prompt: str,
59        width: int = 1024,
60        height: int = 1024
61    ) -> dict:
62        """إرسال طلب إنشاء صورة."""
63        response = self.session.post(
64            f"{self.BASE_URL}/model/generateImage",
65            json={
66                "model": model,
67                "prompt": prompt,
68                "width": width,
69                "height": height
70            }
71        )
72        response.raise_for_status()
73        return response.json()
74
75    def generate_video(
76        self,
77        model: str,
78        prompt: str,
79        duration: int = 5,
80        resolution: str = "1080p"
81    ) -> dict:
82        """إرسال طلب إنشاء فيديو."""
83        response = self.session.post(
84            f"{self.BASE_URL}/model/generateVideo",
85            json={
86                "model": model,
87                "prompt": prompt,
88                "duration": duration,
89                "resolution": resolution
90            }
91        )
92        response.raise_for_status()
93        return response.json()
94
95    def poll_result(
96        self,
97        request_id: str,
98        max_wait: int = 300,
99        initial_interval: int = 5,
100        max_interval: int = 30
101    ) -> Optional[dict]:
102        """الاستعلام عن نتيجة الإنشاء مع التراجع الأسي."""
103        start_time = time.time()
104        interval = initial_interval
105
106        while time.time() - start_time < max_wait:
107            try:
108                response = self.session.get(
109                    f"{self.BASE_URL}/model/prediction/{request_id}/get"
110                )
111                data = response.json()
112
113                if data["status"] == "completed":
114                    return data
115                elif data["status"] == "failed":
116                    logger.error(f"فشل الإنشاء: {data.get('error', 'خطأ غير معروف')}")
117                    return None
118
119                logger.debug(f"الحالة: {data['status']}, بانتظار {interval} ثانية...")
120                time.sleep(interval)
121                interval = min(interval * 1.5, max_interval)
122
123            except requests.RequestException as e:
124                logger.warning(f"فشل طلب الاستعلام: {e}, إعادة المحاولة بعد {interval} ثانية")
125                time.sleep(interval)
126
127        logger.error(f"انتهت المهلة بعد {max_wait} ثانية بانتظار {request_id}")
128        return None
129
130    def estimate_cost(self, model: str, duration: int = 0) -> float:
131        """تقدير تكلفة طلب الإنشاء."""
132        base_price = self.PRICING.get(model, 0.05)
133        if "text-to-video" in model and duration > 0:
134            return base_price * duration
135        return base_price
136
137class VideoPipeline:
138    """تنسيق الإنشاء المجمع للصور والفيديوهات."""
139
140    def __init__(self, api_key: str, output_dir: str = "pipeline_output"):
141        self.client = AtlasCloudClient(api_key)
142        self.output_dir = output_dir
143        self.results: list[GenerationResult] = []
144        self.total_cost = 0.0
145
146        # إنشاء المجلدات
147        os.makedirs(os.path.join(output_dir, "images"), exist_ok=True)
148        os.makedirs(os.path.join(output_dir, "videos"), exist_ok=True)
149
150    def _download_file(self, url: str, filepath: str) -> bool:
151        """تحميل ملف من رابط إلى مسار محلي."""
152        try:
153            response = requests.get(url, timeout=60)
154            response.raise_for_status()
155            with open(filepath, "wb") as f:
156                f.write(response.content)
157            return True
158        except Exception as e:
159            logger.error(f"فشل التحميل لـ {url}: {e}")
160            return False
161
162    def _safe_filename(self, name: str, extension: str) -> str:
163        """تحويل اسم إلى اسم ملف آمن."""
164        safe = name.lower().replace(" ", "_")
165        safe = "".join(c for c in safe if c.isalnum() or c == "_")
166        return f"{safe}.{extension}"
167
168    def _process_image(self, name: str, model: str, prompt: str,
169                       width: int = 1024, height: int = 1024,
170                       retries: int = 2) -> GenerationResult:
171        """إنشاء صورة واحدة مع منطق إعادة المحاولة."""
172        start = time.time()
173        cost = self.client.estimate_cost(model)
174
175        for attempt in range(retries + 1):
176            try:
177                logger.info(f"[صورة] جاري الإنشاء '{name}' (محاولة {attempt + 1})")
178                result = self.client.generate_image(model, prompt, width, height)
179                request_id = result["request_id"]
180
181                data = self.client.poll_result(request_id)
182                if data and data["status"] == "completed":
183                    image_url = data["output"]["image_url"]
184                    filename = self._safe_filename(name, "png")
185                    filepath = os.path.join(self.output_dir, "images", filename)
186                    self._download_file(image_url, filepath)
187
188                    return GenerationResult(
189                        name=name, model=model, media_type="image",
190                        status="success", output_url=image_url,
191                        local_path=filepath, cost_estimate=cost,
192                        duration_seconds=time.time() - start
193                    )
194            except requests.HTTPError as e:
195                if e.response.status_code == 429:
196                    wait = 2 ** (attempt + 2)
197                    logger.warning(f"تم تجاوز الحد، بانتظار {wait} ثانية")
198                    time.sleep(wait)
199                    continue
200                logger.error(f"خطأ HTTP عند إنشاء '{name}': {e}")
201            except Exception as e:
202                logger.error(f"خطأ عند إنشاء '{name}': {e}")
203
204            if attempt < retries:
205                time.sleep(2 ** attempt)
206
207        return GenerationResult(
208            name=name, model=model, media_type="image",
209            status="failed", cost_estimate=0,
210            duration_seconds=time.time() - start,
211            error_message="تم تجاوز الحد الأقصى للمحاولات"
212        )
213
214    def _process_video(self, name: str, model: str, prompt: str,
215                       duration: int = 5, resolution: str = "1080p",
216                       retries: int = 2) -> GenerationResult:
217        """إنشاء فيديو واحد مع منطق إعادة المحاولة."""
218        start = time.time()
219        cost = self.client.estimate_cost(model, duration)
220
221        for attempt in range(retries + 1):
222            try:
223                logger.info(f"[فيديو] جاري الإنشاء '{name}' (محاولة {attempt + 1})")
224                result = self.client.generate_video(model, prompt, duration, resolution)
225                request_id = result["request_id"]
226
227                data = self.client.poll_result(request_id, max_wait=600)
228                if data and data["status"] == "completed":
229                    video_url = data["output"]["video_url"]
230                    filename = self._safe_filename(name, "mp4")
231                    filepath = os.path.join(self.output_dir, "videos", filename)
232                    self._download_file(video_url, filepath)
233
234                    return GenerationResult(
235                        name=name, model=model, media_type="video",
236                        status="success", output_url=video_url,
237                        local_path=filepath, cost_estimate=cost,
238                        duration_seconds=time.time() - start
239                    )
240            except requests.HTTPError as e:
241                if e.response.status_code == 429:
242                    wait = 2 ** (attempt + 2)
243                    logger.warning(f"تم تجاوز الحد، بانتظار {wait} ثانية")
244                    time.sleep(wait)
245                    continue
246                logger.error(f"خطأ HTTP عند إنشاء '{name}': {e}")
247            except Exception as e:
248                logger.error(f"خطأ عند إنشاء '{name}': {e}")
249
250            if attempt < retries:
251                time.sleep(2 ** (attempt + 1))
252
253        return GenerationResult(
254            name=name, model=model, media_type="video",
255            status="failed", cost_estimate=0,
256            duration_seconds=time.time() - start,
257            error_message="تم تجاوز الحد الأقصى للمحاولات"
258        )
259
260    def batch_generate(self, jobs: list[dict], max_workers: int = 3):
261        """تنفيذ مجموعة من وظائف الإنشاء بشكل متزامن."""
262        logger.info(f"بدء مجموعة من {len(jobs)} وظائف مع {max_workers} عمال")
263        start_time = time.time()
264
265        with ThreadPoolExecutor(max_workers=max_workers) as executor:
266            futures = {}
267            for job in jobs:
268                if job["type"] == "image":
269                    future = executor.submit(
270                        self._process_image,
271                        name=job["name"],
272                        model=job["model"],
273                        prompt=job["prompt"],
274                        width=job.get("width", 1024),
275                        height=job.get("height", 1024)
276                    )
277                elif job["type"] == "video":
278                    future = executor.submit(
279                        self._process_video,
280                        name=job["name"],
281                        model=job["model"],
282                        prompt=job["prompt"],
283                        duration=job.get("duration", 5),
284                        resolution=job.get("resolution", "1080p")
285                    )
286                else:
287                    logger.warning(f"نوع وظيفة غير معروف: {job['type']}")
288                    continue
289                futures[future] = job["name"]
290
291            for future in as_completed(futures):
292                result = future.result()
293                self.results.append(result)
294                self.total_cost += result.cost_estimate
295                status_icon = "OK" if result.status == "success" else "FAIL"
296                logger.info(
297                    f"[{status_icon}] {result.name} -- "
298                    f"USD{result.cost_estimate:.3f} -- "
299                    f"{result.duration_seconds:.1f}s"
300                )
301
302        elapsed = time.time() - start_time
303        self._save_manifest()
304        self._print_summary(elapsed)
305
306    def _save_manifest(self):
307        """حفظ بيان النتائج بصيغة JSON."""
308        manifest = {
309            "generated_at": datetime.now().isoformat(),
310            "total_cost": round(self.total_cost, 4),
311            "total_jobs": len(self.results),
312            "successful": sum(1 for r in self.results if r.status == "success"),
313            "failed": sum(1 for r in self.results if r.status != "success"),
314            "results": [
315                {
316                    "name": r.name,
317                    "model": r.model,
318                    "type": r.media_type,
319                    "status": r.status,
320                    "output_url": r.output_url,
321                    "local_path": r.local_path,
322                    "cost": round(r.cost_estimate, 4),
323                    "generation_time": round(r.duration_seconds, 1),
324                    "error": r.error_message
325                }
326                for r in self.results
327            ]
328        }
329        manifest_path = os.path.join(self.output_dir, "manifest.json")
330        with open(manifest_path, "w") as f:
331            json.dump(manifest, f, indent=2)
332        logger.info(f"تم حفظ الملف في {manifest_path}")
333
334    def _print_summary(self, elapsed: float):
335        """طباعة ملخص للمجموعة المنفذة."""
336        success = sum(1 for r in self.results if r.status == "success")
337        failed = len(self.results) - success
338        cost_by_model = {}
339        for r in self.results:
340            cost_by_model[r.model] = cost_by_model.get(r.model, 0) + r.cost_estimate
341
342        print("\n" + "=" * 60)
343        print("ملخص خط الأنابيب")
344        print("=" * 60)
345        print(f"إجمالي الوظائف:    {len(self.results)}")
346        print(f"ناجحة:            {success}")
347        print(f"فاشلة:            {failed}")
348        print(f"إجمالي التكلفة:    USD{self.total_cost:.4f}")
349        print(f"إجمالي الوقت:      {elapsed:.1f}s")
350        print(f"\nالتكلفة حسب النموذج:")
351        for model, cost in sorted(cost_by_model.items()):
352            short_name = model.split("/")[1]
353            print(f"  {short_name}: USD{cost:.4f}")
354        print("=" * 60)
355```

استخدام خط الأنابيب

مع تعريف فئات

1AtlasCloudClient

1VideoPipeline

الاستخدام الأساسي: صور مصغرة + فيديوهات

plaintextCopy
1```python
2API_KEY = "your-atlas-cloud-api-key"
3
4pipeline = VideoPipeline(api_key=API_KEY, output_dir="weekly_content")
5
6jobs = [
7    # إنشاء صور مصغرة باستخدام Flux 2 Pro
8    {
9        "name": "Product Launch Thumbnail",
10        "type": "image",
11        "model": "black-forest-labs/flux-2-pro/text-to-image",
12        "prompt": "صورة مصغرة جذابة لـ YouTube، نص جريء 'إطلاق جديد'، "
13                  "تركيز على المنتج بخلفية متدرجة داكنة، ألوان حيوية، "
14                  "تصميم احترافي، 4K"
15    },
16    {
17        "name": "Tutorial Thumbnail",
18        "type": "image",
19        "model": "black-forest-labs/flux-2-pro/text-to-image",
20        "prompt": "صورة مصغرة لـ YouTube لدرس برمجي، شاشة مقسمة "
21                  "تظهر محرر الأكواد والنتيجة النهائية، جمالية تقنية، "
22                  "تصميم عصري نظيف، نص مقروء وجريء"
23    },
24
25    # إنشاء فيديوهات باستخدام Seedance 2.0 (اقتصادي)
26    {
27        "name": "Product Showcase Seedance",
28        "type": "video",
29        "model": "bytedance/seedance-v1.5-pro/text-to-video",
30        "prompt": "رسوم متحركة أنيقة لعرض المنتج، أداة تقنية حديثة تظهر "
31                  "من ضوء ناعم، دوران ببطء لإظهار كافة الزوايا، "
32                  "خلفية بيضاء بسيطة، إضاءة سينمائية",
33        "duration": 10
34    },
35    {
36        "name": "Brand Intro Seedance",
37        "type": "video",
38        "model": "bytedance/seedance-v1.5-pro/text-to-video",
39        "prompt": "تسلسل مقدمة علامة تجارية ديناميكي، أشكال هندسية مجردة "
40                  "تتجمع لتشكيل شعار، جزيئات ومسارات ضوئية، "
41                  "أسلوب رسوم متحركة احترافي، خلفية داكنة",
42        "duration": 5
43    },
44
45    # إنشاء فيديو سينمائي باستخدام Veo 3.1 (مع صوت)
46    {
47        "name": "Hero Video Veo",
48        "type": "video",
49        "model": "google/veo3.1/text-to-video",
50        "prompt": "لقطة جوية سينمائية لأفق مدينة حديثة في الساعة الذهبية، "
51                  "الكاميرا تتقدم ببطء، توهج عدسي من الشمس الغاربة، "
52                  "أصوات مدينة محيطة، حبيبات فيلم، "
53                  "تصحيح ألوان احترافي",
54        "duration": 8
55    },
56]
57
58pipeline.batch_generate(jobs, max_workers=3)
59```

الأسلوب المعتمد على ملفات الإعداد (Configuration-Driven)

لخطوط الأنابيب المتكررة، قم بتعريف الوظائف في ملف إعداد YAML:

plaintextCopy
1```yaml
2# pipeline_config.yaml
3output_dir: weekly_content
4max_workers: 3
5
6jobs:
7  - name: Product Hero Image
8    type: image
9    model: google/imagen4-ultra/text-to-image
10    prompt: >
11      تصوير منتج فاخر لسماعات لاسلكية في علبة شحن،
12      سطح عاكس داكن، إضاءة درامية، جمالية تقنية فاخرة،
13      دقة 8K، جودة تجارية
14    width: 2048
15    height: 2048
16
17  - name: Social Media Video
18    type: video
19    model: bytedance/seedance-v1.5-pro/text-to-video
20    prompt: >
21      محتوى تواصل اجتماعي عصري، أيادٍ تفتح علبة منتج تقني فاخر،
22      لحظة كشف مرضية، تفاصيل مقربة، إضاءة طبيعية ساطعة،
23      تنسيق رأسي
24    duration: 10
25    resolution: 1080p
26
27  - name: Cinematic Ad
28    type: video
29    model: google/veo3.1/text-to-video
30    prompt: >
31      إعلان سينمائي لسماعات رأس فاخرة، شخص يرتدي
32      سماعات في مقهى مزدحم، العالم يصبح هادئاً، عمق مجال ضحل،
33      لوحة ألوان دافئة، أصوات مقهى محيطة تتلاشى إلى الصمت
34    duration: 8
35    resolution: 1080p
36```

التحميل والتشغيل:

plaintextCopy
1```python
2import yaml
3
4with open("pipeline_config.yaml") as f:
5    config = yaml.safe_load(f)
6
7pipeline = VideoPipeline(
8    api_key=API_KEY,
9    output_dir=config["output_dir"]
10)
11pipeline.batch_generate(
12    config["jobs"],
13    max_workers=config.get("max_workers", 3)
14)
15```

تفاصيل التنفيذ الرئيسية

الاستعلام بالتراجع الأسي (Exponential Backoff Polling)

يستغرق إنشاء الفيديو ما بين 30 ثانية إلى 5 دقائق حسب النموذج والمدة. يستخدم خط الأنابيب التراجع الأسي للاستعلام بكفاءة دون إغراق الـ API:

plaintextCopy
1```python
2interval = initial_interval  # يبدأ بـ 5 ثوانٍ
3while time.time() - start_time < max_wait:
4    # ... التحقق من الحالة ...
5    time.sleep(interval)
6    interval = min(interval * 1.5, max_interval)  # ينمو بحد أقصى 30 ثانية
7```

هذا يعني أن عمليات الاستعلام الأولى تحدث كل 5 ثوانٍ، ثم تتباعد تدريجياً لتصل إلى فواصل 30 ثانية للإنشاءات الأطول. هذا يقلل من مكالمات API غير الضرورية بحوالي 60% مقارنة بالاستعلام بفواصل زمنية ثابتة.

التعامل مع حدود الاستخدام (Rate Limit Handling)

عندما تُرجع الـ API حالة 429 (تم تجاوز حد الاستخدام)، يتراجع خط الأنابيب بشكل أسي بدلاً من الفشل فوراً:

plaintextCopy
1```python
2except requests.HTTPError as e:
3    if e.response.status_code == 429:
4        wait = 2 ** (attempt + 2)  # 4 ثوانٍ، 8 ثوانٍ، 16 ثانية
5        logger.warning(f"تم تجاوز الحد، بانتظار {wait} ثانية")
6        time.sleep(wait)
7        continue
8```

هذا ضروري للعمليات المجمعة حيث قد تتجاوز العديد من الطلبات المتزامنة حدود الاستخدام مؤقتاً.

التحكم في التزامن (Concurrency Control)

يحد

1ThreadPoolExecutor

plaintextCopy
1```python
2with ThreadPoolExecutor(max_workers=3) as executor:
3    futures = {executor.submit(process, job): job for job in jobs}
4```

ابدأ بـ

1max_workers=3

تتبع التكاليف

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

plaintextCopy
1```python
2PRICING = {
3    "black-forest-labs/flux-2-pro/text-to-image": 0.04,
4    "bytedance/seedance-v1.5-pro/text-to-video": 0.022,  # لكل ثانية
5    "google/veo3.1/text-to-video": 0.03,                 # لكل ثانية
6}
7```

بالنسبة لنماذج الفيديو، تتناسب التكلفة مع المدة: فيديو لمدة 10 ثوانٍ بـ Seedance 2.0 يكلف USD0.22، بينما فيديو بـ Veo 3.1 يكلف USD0.30. يتتبع ملف البيان التكاليف لكل طلب والتكاليف التراكمية لمراقبة الميزانية.

تقدير التكلفة لتشغيل خط الأنابيب

إليك تكلفة عمليات خط الأنابيب النموذجية:

مقارنة التكلفة بين Seedance 2.0 و Veo 3.1 لنفس الفيديو:

يعد Seedance 2.0 الخيار الأكثر فعالية من حيث التكلفة لإنشاء الفيديو بكميات كبيرة. يوفر Veo 3.1 توازناً جيداً بين الجودة والتكلفة للمقاطع السينمائية القصيرة. تكلفة Sora 2 أعلى بكثير ولكنه يقدم محاكاة فيزيائية لا تضاهى.

ابدأ ببناء خط أنابيب الفيديو الخاص بك -- USD1 رصيد مجاني

نصائح النشر

وظائف Cron للإنشاء المجدول

شغّل خط الأنابيب وفق جدول زمني باستخدام cron:

plaintextCopy
1# إنشاء محتوى أسبوعي كل اثنين الساعة 6 صباحاً
20 6 * * 1 cd /path/to/project && python run_pipeline.py --config weekly.yaml
3```

أنشئ سكريبت بسيط لنقطة الدخول:

plaintextCopy
1```python
2# run_pipeline.py
3import os
4import argparse
5import yaml
6from pipeline import VideoPipeline
7
8parser = argparse.ArgumentParser()
9parser.add_argument("--config", required=True)
10args = parser.parse_args()
11
12with open(args.config) as f:
13    config = yaml.safe_load(f)
14
15API_KEY = os.environ["ATLAS_CLOUD_API_KEY"]
16pipeline = VideoPipeline(api_key=API_KEY, output_dir=config["output_dir"])
17pipeline.batch_generate(config["jobs"], max_workers=config.get("max_workers", 3))
18```

البنية المعتمدة على الطوابير (Queue-Based Architecture)

بالنسبة لعمليات النشر الأكبر، استخدم طابور مهام مثل Celery أو Redis Queue لفصل إرسال الوظائف عن المعالجة:

plaintextCopy
1```python
2# tasks.py (مثال Celery)
3import os
4from celery import Celery
5from pipeline import AtlasCloudClient
6
7app = Celery("video_tasks", broker="redis://localhost:6379")
8client = AtlasCloudClient(os.environ["ATLAS_CLOUD_API_KEY"])
9
10@app.task(bind=True, max_retries=3)
11def generate_video_task(self, prompt, model, duration):
12    try:
13        result = client.generate_video(model, prompt, duration)
14        data = client.poll_result(result["request_id"])
15        if data and data["status"] == "completed":
16            return {"url": data["output"]["video_url"], "status": "success"}
17        return {"status": "failed"}
18    except Exception as e:
19        self.retry(countdown=60, exc=e)
20```

هذه البنية مناسبة للأنظمة الإنتاجية حيث تأتي طلبات إنشاء الفيديو من تطبيق ويب أو API، وتحتاج النتائج إلى التسليم بشكل غير متزامن عبر webhooks أو الاستعلام.

إدارة متغيرات البيئة

لا تقم أبداً بتضمين مفاتيح الـ API في الكود. استخدم متغيرات البيئة:

plaintextCopy
1```python
2import os
3
4API_KEY = os.environ.get("ATLAS_CLOUD_API_KEY")
5if not API_KEY:
6    raise ValueError("لم يتم ضبط متغير البيئة ATLAS_CLOUD_API_KEY")
7```

للتطوير المحلي، استخدم ملف

1.env

1python-dotenv

plaintextCopy
1# .env
2ATLAS_CLOUD_API_KEY=your-key-here
3```

plaintextCopy
1```python
2from dotenv import load_dotenv
3load_dotenv()
4```

توسيع خط الأنابيب

إضافة إنشاء الفيديو من صورة (Image-to-Video)

تدعم بعض النماذج استخدام صورة تم إنشاؤها كمدخل لإنشاء فيديو. قم بتوسيع خط الأنابيب لربط إنشاء الصورة بالفيديو:

plaintextCopy
1```python
2def generate_image_then_video(self, name, image_prompt, video_prompt,
3                              image_model, video_model, duration=5):
4    """إنشاء صورة، ثم استخدامها كمدخل لإنشاء فيديو."""
5    # الخطوة 1: إنشاء الصورة الأساسية
6    image_result = self._process_image(
7        f"{name}_base", image_model, image_prompt
8    )
9    if image_result.status != "success":
10        return image_result
11
12    # الخطوة 2: استخدام رابط الصورة كمدخل لإنشاء الفيديو
13    response = self.client.session.post(
14        f"{self.client.BASE_URL}/model/generateVideo",
15        json={
16            "model": video_model,
17            "prompt": video_prompt,
18            "image_url": image_result.output_url,
19            "duration": duration
20        }
21    )
22    # ... استعلم وحمّل كالمعتاد
23```

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

كم عدد الطلبات المتزامنة التي يمكنني إجراؤها؟

تدعم Atlas Cloud طلبات متزامنة متعددة لكل مفتاح API. ابدأ بـ 3 عمال وزد إلى 5-8 بناءً على فئة حسابك. يتعامل خط الأنابيب مع حدود الاستخدام تلقائياً.

هل يمكنني خلط وظائف الصور والفيديو في نفس المجموعة؟

نعم. يقوم خط الأنابيب بتوجيه كل وظيفة إلى نقطة النهاية الصحيحة بناءً على حقل

1type

كم من الوقت تستغرق طلبات إنشاء الفيديو؟

تختلف المدة حسب النموذج: عادةً ما يكتمل Seedance 2.0 في 30-90 ثانية، و Veo 3.1 في 60-120 ثانية، و Sora 2 في 60-180 ثانية.

ماذا يحدث إذا فشل الإنشاء أثناء المجموعة؟

يتم تسجيل الوظائف الفاشلة وإدراجها في البيان مع رسائل الخطأ. يواصل خط الأنابيب معالجة الوظائف المتبقية.

الحكم النهائي

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

إن الجمع بين Flux 2 Pro لإنشاء صور سريع، و Seedance 2.0 للفيديو الاقتصادي، و Veo 3.1 للمقاطع السينمائية يوفر تغطية كاملة لاحتياجات إنتاج المحتوى.