yrtria/booth-musicgpt
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Make MusicGPT webhook robust: store cover/cost, download when URLs available

Browse source
This commit is contained in:
Troll (Hermes Agent) 2026-08-10 23:17:56 +00:00
parent 2c11738d2e
commit 57bdf69942
1 changed files with 11 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

23
app.py
View file

@ -324,22 +324,21 @@ def musicgpt_webhook():
if data.get('is_flagged'): if data.get('is_flagged'):
update_fields['musicgpt_error'] = data.get('reason') or 'Flagged by MusicGPT' update_fields['musicgpt_error'] = data.get('reason') or 'Flagged by MusicGPT'
if new_status in ('COMPLETED', 'FINISHED'): if new_status in ('COMPLETED', 'FINISHED'):
# Real webhooks may omit audio URLs; fetch full conversion details if needed. # Webhooks often fire before audio URLs are ready. Store whatever the
if not (data.get('conversion_path') or data.get('conversion_path_1') or data.get('conversion_path_2')): # webhook gives us, then rely on the dashboard manual refresh to
poll_result = musicgpt_poll_status(task_id) # download files once the API reports them fully available.
conversion = poll_result.get('conversion') or {} if data.get('album_cover_path'):
if conversion and isinstance(conversion, dict): update_fields['album_cover_url'] = data.get('album_cover_path')
data.update(conversion)
download_musicgpt_outputs(req, data, version=version)
# Re-fetch to get updated paths.
row = db.execute('SELECT * FROM requests WHERE id = ?', (req['id'],)).fetchone()
req = dict(row)
if req.get('song_a_path') and req.get('song_b_path'):
update_fields['status'] = 'songs_uploaded'
try: try:
update_fields['musicgpt_cost'] = float(data.get('conversion_cost') or 0) update_fields['musicgpt_cost'] = float(data.get('conversion_cost') or 0)
except (ValueError, TypeError): except (ValueError, TypeError):
pass pass
# Still attempt download in case the webhook is the rare complete one.
download_musicgpt_outputs(req, data, version=version)
row = db.execute('SELECT * FROM requests WHERE id = ?', (req['id'],)).fetchone()
req = dict(row)
if req.get('song_a_path') and req.get('song_b_path'):
update_fields['status'] = 'songs_uploaded'
elif new_status in ('FAILED', 'ERROR'): elif new_status in ('FAILED', 'ERROR'):
update_fields['musicgpt_error'] = data.get('reason') or data.get('error') or 'MusicGPT reported failure' update_fields['musicgpt_error'] = data.get('reason') or data.get('error') or 'MusicGPT reported failure'
update_request(req['id'], **update_fields) update_request(req['id'], **update_fields)