Create yml/shabihsazi-sada.py

yml/shabihsazi-sada.py

@@ -0,0 +1,176 @@
+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": "shabihsazi-sada",
+                "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("VOICECONFIG_"):
+    err_str = "Error: Invalid voice cloning configuration payload signature."
+    print(err_str)
+    report_failure(err_str)
+    sys.exit(1)
+
+# استخراج داده‌ها از توکن رمزگذاری شده بدون فاصله
+config_str = raw_prompt[len("VOICECONFIG_"):]
+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)
+aud_ext = config.get("audExt", "wav")
+language = config.get("lang", "Auto")
+
+# رمزگشایی متن مرجع (در صورت وجود)
+try:
+    b64_ref = config.get("refText", "")
+    b64_ref += "=" * ((4 - len(b64_ref) % 4) % 4)
+    ref_text = base64.b64decode(b64_ref).decode('utf-8')
+except Exception:
+    ref_text = ""
+
+# رمزگشایی متن نهایی برای شبیه‌سازی
+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"   -> Target Text: {persian_prompt}")
+print(f"   -> Reference Transcription: {ref_text}")
+print(f"   -> Target Language: {language}")
+
+# دانلود فایل صوتی مرجع از هاست فلاسک
+img_url = f"{space_url}/static/images/{user_run_id}_voice_ref.{aud_ext}"
+local_img = f"input.{aud_ext}"
+
+print(f"2. Downloading reference audio from: {img_url}")
+try:
+    req_img = requests.get(img_url, timeout=45)
+    if req_img.status_code != 200:
+        raise Exception(f"Reference audio 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 OmniVoice Space...')
+try:
+    hf_token = os.environ.get('HF_TOKEN', '')
+    if hf_token:
+        client = Client("k2-fsa/OmniVoice", token=hf_token)
+    else:
+        client = Client("k2-fsa/OmniVoice")
+        
+    print('4. Dynamically locating the 12-input clone endpoint...')
+    # اسکن هوشمند متادیتای اندپوینت‌های هاگینگ‌فیس جهت پیدا کردن اتصالات کلون صدا
+    fn_index = None
+    api_name = None
+    
+    dependencies = client.config.get("dependencies", [])
+    for dep in dependencies:
+        inputs = dep.get("inputs", [])
+        if len(inputs) == 12:
+            api_name = dep.get("api_name")
+            fn_index = dep.get("id")
+            break
+            
+    # ساختار دهی نوع فراخوانی بر اساس وجود یا عدم وجود نام اختصاصی
+    endpoint_arg = {}
+    if api_name:
+        endpoint_arg["api_name"] = f"/{api_name}"
+        print(f"   -> Selected endpoint by API Name: /{api_name}")
+    elif fn_index is not None:
+        endpoint_arg["fn_index"] = fn_index
+        print(f"   -> Selected endpoint by Function Index: {fn_index}")
+    else:
+        # حالت پیش‌فرض برای پیشگیری از توقف فرآیند
+        endpoint_arg["fn_index"] = 0
+        print("   -> Warning: Could not locate 12-input endpoint. Falling back to fn_index 0.")
+
+    print('5. Generating synthesized cloned speech...')
+    result = client.predict(
+        persian_prompt,             # 1. Text to synthesize (str)
+        language,                   # 2. Language (str)
+        handle_file(local_img),     # 3. Reference Audio (file)
+        ref_text if ref_text else "", # 4. Reference Text (str)
+        "",                         # 5. Instruct (در حالت کلون ساده خالی رها می‌شود)
+        32,                         # 6. Steps (تعداد مراحل تولید)
+        2.0,                        # 7. Guidance scale (مقیاس هدایت)
+        True,                       # 8. Denoise (کاهش نویز پیش‌فرض)
+        1.0,                        # 9. Speed (سرعت طبیعی)
+        0,                          # 10. Duration (تنظیم خودکار زمان)
+        True,                       # 11. Preprocess prompt (پیش‌پردازش)
+        True,                       # 12. Postprocess output (پس‌پردازش خروجی)
+        **endpoint_arg
+    )
+
+    # استخراج مسیر فایل صوتی تولید شده نهایی
+    audio_path = None
+    if isinstance(result, (list, tuple)):
+        audio_path = result[0]
+    elif isinstance(result, dict):
+        audio_path = result.get('name') or result.get('path') or result.get('url')
+    else:
+        audio_path = result
+
+    if not audio_path or not os.path.exists(str(audio_path)):
+        raise Exception("Generated audio file not found or invalid.")
+
+    # آپلود مجدد فایل صوتی به فضای وب‌هوک هاست فلاسک با پسوند wav
+    print('6. Uploading output back to your Space...')
+    with open(audio_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': 'wav'}, 
+            files={'file': f}
+        )
+
+    if res_upload.status_code == 200:
+        print('7. 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)