| import os |
| import sys |
| import requests |
| import base64 |
| from gradio_client import Client |
|
|
| |
| 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', '') |
|
|
| |
| def report_failure(error_msg): |
| try: |
| requests.post( |
| f"{space_url}/api/webhook/fail", |
| json={ |
| "run_id": run_id, |
| "error": error_msg, |
| "event_type": "effects", |
| "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 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) |
| variant = config.get("variant", "medium") |
| sampler_type = config.get("sampler", "pingpong") |
|
|
| try: |
| duration = float(config.get("duration", "60")) |
| except ValueError: |
| duration = 60.0 |
|
|
| try: |
| steps = int(config.get("steps", "8")) |
| except ValueError: |
| steps = 8 |
|
|
| try: |
| cfg_scale = float(config.get("cfg", "1.0")) |
| except ValueError: |
| cfg_scale = 1.0 |
|
|
| try: |
| seed = int(config.get("seed", "0")) |
| except ValueError: |
| seed = 0 |
|
|
| |
| try: |
| b64_prompt = config.get("prompt", "") |
| b64_prompt += "=" * ((4 - len(b64_prompt) % 4) % 4) |
| prompt = base64.b64decode(b64_prompt).decode('utf-8') |
| except Exception as e: |
| prompt = "" |
|
|
| print(f" -> User Run ID: {user_run_id}") |
| print(f" -> Model Variant: {variant}") |
| print(f" -> Prompt: {prompt}") |
| print(f" -> Duration: {duration}s") |
| print(f" -> Steps: {steps}") |
| print(f" -> CFG Scale: {cfg_scale}") |
| print(f" -> Sampler: {sampler_type}") |
| print(f" -> Seed: {seed}") |
|
|
| print('2. Connecting to Stable Audio 3 Space...') |
| try: |
| hf_token = os.environ.get('HF_TOKEN', '') |
| if hf_token: |
| client = Client("stabilityai/stable-audio-3", token=hf_token) |
| else: |
| client = Client("stabilityai/stable-audio-3") |
|
|
| print('3. Generating audio from Stable Audio 3...') |
| |
| result = client.predict( |
| variant_key=variant, |
| prompt=prompt, |
| duration=duration, |
| steps=steps, |
| cfg_scale=cfg_scale, |
| sampler_type=sampler_type, |
| seed=seed, |
| api_name="/infer" |
| ) |
|
|
| |
| audio_path = None |
| if isinstance(result, (list, tuple)) and len(result) > 0: |
| 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 isinstance(audio_path, dict): |
| audio_path = audio_path.get('path') or audio_path.get('name') or audio_path.get('url') |
|
|
| if not audio_path or not os.path.exists(str(audio_path)): |
| raise Exception("Generated audio file not found or invalid.") |
|
|
| print('4. Uploading generated audio back to Server...') |
| 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('5. 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 audio generation: {err_str}") |
| report_failure(err_str) |
| sys.exit(1) |
