Create yml/gostaresh-tasvir.py

yml/gostaresh-tasvir.py

@@ -0,0 +1,168 @@
+import os
+import sys
+import requests
+import base64
+from gradio_client import Client, handle_file
+
+# دریافت متغیرهای سیستمی از گیت‌هاب اکشنز
+raw_prompt = os.environ.get('PROMPT', '')
+run_id = os.environ.get('RUN_ID', '')
+space_url = os.environ.get('SPACE_URL', '')
+github_run_id = os.environ.get('GITHUB_RUN_ID', '')
+
+# تابع گزارش خطای احتمالی جهت اجرای سیستم خودکار تلاش مجدد (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": "gostaresh-tasvir",
+                "client_payload": {
+                    "prompt": raw_prompt,
+                    "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. Decoding configuration from payload...')
+if not raw_prompt.startswith("OUTPAINTCONFIG_"):
+    err_str = "Error: Invalid outpaint configuration payload signature."
+    print(err_str)
+    report_failure(err_str)
+    sys.exit(1)
+
+# استخراج داده‌ها از توکن رمزگذاری شده
+config_str = raw_prompt[len("OUTPAINTCONFIG_"):]
+parts = config_str.split("_")
+config = {}
+i = 0
+while i < len(parts) - 1:
+    key = parts[i]
+    val = parts[i+1]
+    config[key] = val
+    i += 2
+
+user_run_id = config.get("userRunId", run_id)
+img_ext = config.get("imgExt", "png")
+width = int(config.get("width", "1024"))
+height = int(config.get("height", "1024"))
+raw_steps = int(config.get("steps", "8"))
+
+# تعدیل خودکار تعداد گام‌ها جهت هماهنگی با محدودیت ۴ تا ۱۲ سرور میزبان
+outpaint_steps = max(4, min(12, raw_steps))
+
+expand_left = config.get("left") == "true"
+expand_right = config.get("right") == "true"
+expand_top = config.get("top") == "true"
+expand_bottom = config.get("bottom") == "true"
+
+# رمزگشایی پرامپت فارسی
+try:
+    b64_prompt = config.get("prompt", "")
+    b64_prompt += "=" * ((4 - len(b64_prompt) % 4) % 4)
+    persian_prompt = base64.b64decode(b64_prompt).decode('utf-8')
+except Exception as e:
+    persian_prompt = ""
+
+print(f"   -> User Run ID: {user_run_id}")
+print(f"   -> Persian Prompt: {persian_prompt}")
+print(f"   -> Enforced Steps (within 4-12 range): {outpaint_steps}")
+
+# ترجمه مستقیم متن فارسی به انگلیسی بدون وابستگی به کتابخانه‌ها
+english_prompt = ""
+if persian_prompt.strip():
+    try:
+        translate_url = f"https://translate.googleapis.com/translate_a/single?client=gtx&sl=auto&tl=en&dt=t&q={requests.utils.quote(persian_prompt)}"
+        r_trans = requests.get(translate_url, timeout=15)
+        english_prompt = r_trans.json()[0][0][0]
+        print(f"   -> Translated Prompt: {english_prompt}")
+    except Exception as trans_err:
+        print(f"   -> Translation failed, using original: {trans_err}")
+        english_prompt = persian_prompt
+
+# دانلود تصویر مرجع آپلود شده از فضای داکر
+img_url = f"{space_url}/static/images/{user_run_id}_outpaint_img.{img_ext}"
+local_img = f"input.{img_ext}"
+
+print(f"2. Downloading source image from: {img_url}")
+try:
+    req_img = requests.get(img_url, timeout=45)
+    if req_img.status_code != 200:
+        raise Exception(f"Image download failed. Status code: {req_img.status_code}")
+    with open(local_img, 'wb') as f:
+        f.write(req_img.content)
+except Exception as download_err:
+    err_str = f"Error downloading source files: {download_err}"
+    print(err_str)
+    report_failure(err_str)
+    sys.exit(1)
+
+# شبیه‌سازی اتصال مستقیم به موتور هوش مصنوعی گسترش تصویر
+print('3. Connecting to Outpaint Space...')
+try:
+    client = Client("fffiloni/diffusers-image-outpaint")
+    
+    print('4. Generating expanded image...')
+    # تطبیق دقیق آرگومان‌های ارسالی با ساختار ورودی‌های تابع infer در هاگینگ‌فیس
+    result = client.predict(
+        handle_file(local_img), # 0: image (PIL Image)
+        width,                  # 1: width (int)
+        height,                 # 2: height (int)
+        10,                     # 3: overlap_percentage (int)
+        outpaint_steps,         # 4: num_inference_steps (int)
+        "Full",                 # 5: resize_option (str)
+        50,                     # 6: custom_resize_percentage (int)
+        english_prompt,         # 7: prompt_input (str)
+        "Middle",               # 8: alignment (str)
+        expand_left,            # 9: overlap_left (bool)
+        expand_right,           # 10: overlap_right (bool)
+        expand_top,             # 11: overlap_top (bool)
+        expand_bottom,          # 12: overlap_bottom (bool)
+        fn_index=7
+    )
+
+    # استخراج صحیح فایل نهایی تولید شده از ساختار گالری
+    image_path = None
+    if isinstance(result, (list, tuple)):
+        if len(result) > 0:
+            item = result[-1]
+            if isinstance(item, (list, tuple)):
+                image_path = item[0]
+            elif isinstance(item, dict):
+                image_path = item.get('path') or item.get('url')
+            else:
+                image_path = item
+    elif isinstance(result, dict):
+        image_path = result.get('path') or result.get('url')
+    else:
+        image_path = result
+
+    if not image_path or not os.path.exists(str(image_path)):
+        raise Exception("Generated expanded image file not found or invalid.")
+
+    # ارسال فایل نهایی تولید شده به هاست فلاسک با پسوند مناسب
+    print('5. Uploading output back to your Space...')
+    with open(image_path, 'rb') as f:
+        res_upload = requests.post(
+            f'{space_url}/api/webhook/upload', 
+            data={'run_id': run_id, 'github_run_id': github_run_id, 'ext': 'png'}, 
+            files={'file': f}
+        )
+
+    if res_upload.status_code == 200:
+        print('6. SUCCESS! Process complete.')
+    else:
+        raise Exception(f"Webhook upload failed. Status code: {res_upload.status_code}")
+
+except Exception as e:
+    err_str = str(e)
+    print(f"CRITICAL ERROR during generation: {err_str}")
+    report_failure(err_str)
+    sys.exit(1)