File size: 3,695 Bytes
6b6c45f fcb05eb 6b6c45f fcb05eb 6b6c45f fcb05eb 6b6c45f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 | import os, sys, requests, time
from gradio_client import Client, handle_file
# --- Variables injected by Flask ---
AUDIO_TYPE = '___AUDIO_TYPE___'
NEGATIVE_PROMPT = '''___NEGATIVE_PROMPT___'''
prompt = os.environ.get('PROMPT', '')
try: seed = int(os.environ.get('SEED', '-1'))
except: seed = -1
try: duration = float(os.environ.get('DURATION', '8'))
except: duration = 8.0
file_url = os.environ.get('FILE_URL', '')
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": "generate-audio",
"client_payload": {
"data": {
"prompt": prompt,
"duration": duration,
"seed": seed,
"audio_type": AUDIO_TYPE,
"negative_prompt": NEGATIVE_PROMPT,
"file_url": file_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}")
try:
print('1. Connecting to Audio Space...')
client = Client("hkchengrex/MMAudio")
if AUDIO_TYPE == 'video':
print('2. Downloading video for audio generation...')
req = requests.get(file_url, timeout=30)
ext = file_url.split('.')[-1] if '.' in file_url else 'mp4'
local_vid = f'input.{ext}'
with open(local_vid, 'wb') as f:
f.write(req.content)
print('3. Processing Video-to-Audio...')
result = client.predict(
video=handle_file(local_vid),
prompt=prompt,
negative_prompt=NEGATIVE_PROMPT,
seed=seed,
num_steps=25,
cfg_strength=4.5,
duration=duration,
api_name="/video_to_audio"
)
else:
print('2. Processing Text-to-Audio...')
result = client.predict(
prompt=prompt,
negative_prompt=NEGATIVE_PROMPT,
seed=seed,
num_steps=25,
cfg_strength=4.5,
duration=duration,
api_name="/text_to_audio"
)
print('4. Preparing File Data...')
audio_path = None
if isinstance(result, list) and len(result) > 0:
audio_path = result[0].get('video') or result[0].get('path') or result[0].get('url') if isinstance(result[0], dict) else result[0]
elif isinstance(result, tuple) and len(result) > 0:
audio_path = result[0]
elif isinstance(result, dict):
audio_path = result.get('video') or result.get('path') or result.get('url')
else:
audio_path = result
ext = 'mp4'
if str(audio_path).endswith('.wav'): ext = 'wav'
if str(audio_path).endswith('.mp3'): ext = 'mp3'
print('5. Uploading back to Space...')
with open(audio_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': ext},
files={'file': f}
)
if res.status_code == 200:
print('SUCCESS!')
else:
sys.exit(1)
except Exception as e:
err_str = str(e)
print(f'Generation error: {err_str}')
report_failure(err_str)
sys.exit(1) |