import os, sys, random, requests, time, io from gradio_client import Client, handle_file from PIL import Image, ImageOps prompt = os.environ.get('PROMPT', '').strip() if not prompt: prompt = 'make this image come alive, cinematic motion, smooth animation' else: prompt = f'{prompt}, cinematic motion, smooth animation, high quality' negative_prompt = '色调艳丽, 过曝, 静态, 细节模糊不清, 字幕, 风格, 作品, 画作, 画面, 静止, 整体发灰, 最差质量, 低质量, JPEG压缩残留, 丑陋的, 残缺의, 多余的手指, 画得不好的手部, 画得不好的脸部, 畸形的, 毁容的, 形态畸形的肢体, 手指融合, 静止不动的画面, 杂乱的背景, 三条腿, 背景人很多, 倒着走' image_url = os.environ.get('IMAGE_URL', '') run_id = os.environ.get('RUN_ID') space_url = os.environ.get('SPACE_URL') github_run_id = os.environ.get('GITHUB_RUN_ID') # زمان دیگر قفل نیست و مستقیماً از سمت کاربر (سایت) خوانده می‌شود try: duration = float(os.environ.get('DURATION', 5.0)) except Exception: duration = 5.0 # اضافه کردن تابع گزارش خطا جهت فعال‌سازی سیستم تلاش مجدد در حساب دیگر (Retry) def report_failure(error_msg): try: requests.post( f"{space_url}/api/webhook/fail", json={ "run_id": run_id, "error": error_msg, "event_type": "generate-video", "client_payload": { "prompt": prompt, "duration": duration, "image_url": image_url, "run_id": run_id, "space_url": space_url }, "github_run_id": github_run_id }, timeout=15 ) except Exception as e: print(f"Failed to report failure: {e}") print('1. Downloading and validating input image from Docker Space...') max_download_attempts = 2 download_success = False # حلقه برای تلاش مجدد دانلود عکس در صورت خطای 404 (تاخیر CDN) for dl_attempt in range(max_download_attempts): try: req = requests.get(image_url, timeout=30) # بررسی سلامت لینک دانلود (اطمینان از دریافت نکردن صفحات HTML خطا) if req.status_code == 200: download_success = True break # دانلود موفقیت آمیز بود، خروج از حلقه else: print(f" -> Download attempt {dl_attempt + 1} failed (Status {req.status_code}).") except Exception as e: print(f" -> Download attempt {dl_attempt + 1} connection failed: {e}") # اگر تلاش اول ناموفق بود، 2 ثانیه صبر کن و دوباره تلاش کن if dl_attempt < max_download_attempts - 1: print(" -> Waiting 2 seconds before retrying download...") time.sleep(2.0) # اگر بعد از 2 بار تلاش (و صبر 2 ثانیه ای) باز هم دانلود نشد، به سرور اصلی خطا بفرست if not download_success: err_str = f"Download failed after {max_download_attempts} attempts. URL might be invalid or CDN delayed." print(f"CRITICAL ERROR: {err_str}") report_failure(err_str) sys.exit(1) # بررسی اینکه محتوای دانلود شده واقعاً عکس است و تبدیل واقعی فرمت try: img_data = io.BytesIO(req.content) img = Image.open(img_data) img.verify() # تایید صحت هدرهای تصویر # باز کردن مجدد برای تغییر فرمت واقعی به RGB خالص (حذف کانال آلفا و فرمت‌های متفرقه) img_data.seek(0) clean_img = Image.open(img_data).convert('RGB') clean_img.save('input_raw.png', format='PNG') print(' -> Image successfully downloaded, verified, and strictly converted to PNG format.') except Exception as img_err: err_str = f"Downloaded content is corrupted or not a valid image format: {img_err}" print(f"CRITICAL ERROR: {err_str}") report_failure(err_str) sys.exit(1) print('2. Analyzing image dimensions to prevent stretching...') # تصویر الان قطعا معتبر و دارای ساختار PNG استاندارد است img = Image.open('input_raw.png') orig_w, orig_h = img.size ratio = orig_w / orig_h if ratio > 1.2: target_w, target_h = 832, 480 print(' -> Format: Landscape (16:9)') elif ratio < 0.8: target_w, target_h = 480, 832 print(' -> Format: Portrait (9:16)') else: target_w, target_h = 480, 480 print(' -> Format: Square (1:1)') img_resized = ImageOps.fit(img, (target_w, target_h), Image.Resampling.LANCZOS) img_resized.save('input.png', format='PNG') print(f' -> Image perfectly resized to {target_w}x{target_h}') print('3. Connecting to AI Video Space...') video_path = None space_id = 'zerogpu-aoti/wan2-2-fp8da-aoti-faster' print(f' -> Loading Space: {space_id}') try: client = Client(space_id) except Exception as e: err_str = f"Could not load AI space: {e}" print(f'CRITICAL ERROR: {err_str}') report_failure(err_str) sys.exit(1) max_attempts = 4 for attempt in range(max_attempts): seed = random.randint(1, 2147483647) print(f' -> Attempt {attempt + 1} of {max_attempts} with seed {seed} and duration {duration}s...') try: result = client.predict( handle_file('input.png'), # [0] عکس ورودی prompt, # [1] پرامپت مثبت 6, # [2] مقدار ثابت negative_prompt, # [3] پرامپت منفی duration, # [4] زمان انتخابی کاربر 1, # [5] مقدار ثابت 1, # [6] مقدار ثابت seed, # [7] هسته پردازش True, # [8] رندوم کردن Seed fn_index=0 ) if isinstance(result, list) and len(result) > 0: video_path = result[0].get('video') or result[0].get('path') if isinstance(result[0], dict) else result[0] elif isinstance(result, tuple) and len(result) > 0: video_path = result[0] elif isinstance(result, dict): video_path = result.get('video') or result.get('path') else: video_path = result if video_path and os.path.exists(str(video_path)): print(' -> Success! Video generated.') break except Exception as e: err_msg = str(e) print(f' -> Failed on attempt {attempt + 1}: {err_msg}') # شناسایی خطای سهمیه و خروج فوری جهت پاس دادن به حساب گیت‌هاب بعدی if 'ZeroGPU quota' in err_msg or 'quota' in err_msg.lower(): print(' -> ZeroGPU quota hit. Aborting internal retries to trigger account switch.') report_failure(err_msg) sys.exit(1) time.sleep(5) if not video_path or not os.path.exists(str(video_path)): err_str = "Failed to generate video after all internal attempts." print(f'CRITICAL ERROR: {err_str}') report_failure(err_str) sys.exit(1) print('4. Uploading back to your Docker Space...') try: with open(video_path, 'rb') as f: res = requests.post(f'{space_url}/api/webhook/upload', data={'run_id': run_id, 'github_run_id': github_run_id, 'ext': 'mp4'}, files={'file': f}) if res.status_code == 200: print('5. SUCCESS! Process finished.') else: err_str = f"Upload failed with status {res.status_code}: {res.text}" print(f'CRITICAL ERROR: {err_str}') report_failure(err_str) sys.exit(1) except Exception as e: err_str = f"Upload connection failed: {e}" print(f'CRITICAL ERROR: {err_str}') report_failure(err_str) sys.exit(1)