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

feat: send song requests to Hermes via webhook

Browse source 
- Add Hermes webhook settings form in /admin/settings (URL + HMAC secret)
- Add /admin/request/<id>/send-to-hermes action
- POST request data to Hermes with V2 HMAC signature and callback URL
- Hermes can POST the generated MusicGPT prompt back to /api/prompt/<id>
- Add test button, update README workflow, bump version to 0.9.0
This commit is contained in:
Troll (Hermes Agent) 2026-08-24 23:01:33 +00:00
parent 3c99378044
commit 2f5050b29f
6 changed files with 199 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

72
app.py
View file

@ -77,6 +77,7 @@ from helpers import (
download_album_cover,
process_stems_to_gokapi,
get_ntfy_config, send_ntfy,
get_hermes_webhook_config, set_hermes_webhook_config, send_hermes_webhook,
sign_prompt_callback, verify_prompt_callback, build_prompt_callback_url,
send_email, build_signature_images,
get_booth_open,
@ -1503,6 +1504,38 @@ def admin_request(rid):
)
@app.route('/admin/request/<int:rid>/send-to-hermes', methods=['POST'])
def send_to_hermes(rid):
"""
Operator action: push the current request data to the configured Hermes webhook.
Hermes will generate a MusicGPT prompt and POST it back to the callback URL.
"""
redir = require_admin()
if redir:
return redir
req = get_request_by_id(rid)
if not req:
abort(404)
if req['status'] not in ('pending', 'revisions_requested'):
flash('Request must be pending or awaiting revision to send to Hermes.', 'error')
return redirect(url_for('admin_request', rid=rid))
cfg = get_hermes_webhook_config()
if not cfg['url'] or not cfg['secret']:
flash('Hermes webhook is not configured. Set URL and secret in /admin/settings.', 'error')
return redirect(url_for('admin_request', rid=rid))
callback_url = build_prompt_callback_url(rid)
ok, msg = send_hermes_webhook(rid, req, callback_url=callback_url)
if ok:
flash(f'Sent to Hermes. {msg}', 'success')
else:
flash(f'Failed to send to Hermes: {msg}', 'error')
return redirect(url_for('admin_request', rid=rid))
@app.route('/admin/request/<int:rid>/delete', methods=['POST'])
def admin_delete_request(rid):
"""Delete a single request and remove its uploaded MP3 files."""
@ -1850,6 +1883,42 @@ def admin_settings():
flash('ntfy notification settings saved.', 'success')
return redirect(url_for('admin_settings'))
elif action == 'save_hermes_webhook':
# Update Hermes webhook URL and secret from the settings form.
webhook_url = request.form.get('hermes_webhook_url', '').strip().rstrip('/')
webhook_secret = request.form.get('hermes_webhook_secret', '').strip()
set_hermes_webhook_config(webhook_url, webhook_secret)
flash('Hermes webhook settings saved.', 'success')
return redirect(url_for('admin_settings'))
elif action == 'send_test_hermes_webhook':
# Send a test request to the configured Hermes webhook.
cfg = get_hermes_webhook_config()
if not cfg['url'] or not cfg['secret']:
flash('Configure Hermes webhook URL and secret first.', 'error')
return redirect(url_for('admin_settings'))
ok, msg = send_hermes_webhook(0, {
'email': 'test@example.com',
'name': 'Test Customer',
'pronouns': 'They/Them/Their',
'hobbies': 'testing',
'notable_facts': 'none',
'style_genre': "1980's, Pop, Funk",
'vocal_gender': 'female',
'extra_requests': 'Test webhook',
'stems_interest': 0,
'revision_count': 0,
'revision_note': '',
'suno_title': '',
'suno_style': '',
'suno_lyrics': '',
})
if ok:
flash(f'Test Hermes webhook sent. {msg}', 'success')
else:
flash(f'Test Hermes webhook failed: {msg}', 'error')
return redirect(url_for('admin_settings'))
elif action == 'send_test_ntfy':
# Send a test push notification to the configured ntfy topic.
ntfy = get_ntfy_config()
@ -1911,6 +1980,7 @@ def admin_settings():
musicgpt_api_key_set=bool(get_musicgpt_api_key()),
musicgpt_webhook_url=build_musicgpt_webhook_url(),
musicgpt_autopoll=runtime_settings.get('musicgpt_autopoll', True),
hermes_webhook=get_hermes_webhook_config(),
)
@ -1967,4 +2037,4 @@ with app.app_context():
# If the database path is not yet reachable (e.g. volume not mounted),
# defer to the first request or the explicit init-db command.
import logging
logging.getLogger('app').warning('Startup init_db() failed; database may need manual initialization.', exc_info=True)
logging.getLogger('app').warning('Startup init_db() failed; database may need manual initialization.', exc_info=True)