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

v0.6.0: form limits, callback expiry, dashboard filter, payment ref, STEMS interest, helper refactor

Browse source
This commit is contained in:
Troll (Hermes Agent) 2026-08-09 05:18:40 +00:00
parent 0bfe8c315c
commit 0a01381d44
9 changed files with 603 additions and 497 deletions
Download patch file Download diff file Expand all files Collapse all files

525
app.py
View file

@ -30,95 +30,36 @@ Admin routes:
# Standard library imports
import os
import re
import shutil
import smtplib
import ssl
import time
from email.message import EmailMessage
from pathlib import Path
import json
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import base64
import hmac
import hashlib
import secrets
# Flask and related imports
from flask import Flask, request, render_template, redirect, url_for, flash, session, send_from_directory, abort, current_app, send_file, jsonify
from flask import Flask, request, render_template, redirect, url_for, flash, session, abort, current_app, send_file, jsonify
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from werkzeug.utils import secure_filename
# Project imports
from config import Config
from models import init_db, close_db, create_request, get_request_by_id, get_request_by_token, list_requests, update_request, now_utc, delete_request, reset_all_requests, get_db
from models import SCHEMA, get_requests_by_email, log_revision, list_revision_history
# Audio metadata import
from mutagen.mp3 import MP3
from mutagen.easyid3 import EasyID3
# HTTP client for ntfy push notifications
import requests
# ---------------------------------------------------------------------------
# Genre / decade helpers
# ---------------------------------------------------------------------------
# Load external lists once at import time. These are shared by the customer
# request form and the admin request editor so the dropdowns stay in sync.
# In production the canonical files live under /mnt/Storage. We also bundle
# copies in the repo so the container still works if that mount is missing.
_GENRES_PATH = Path('/mnt/Storage/Music Genres.txt')
_DECADES_PATH = Path('/mnt/Storage/Decades.txt')
_FALLBACK_GENRES_PATH = Path(__file__).parent / 'lists' / 'music_genres.txt'
_FALLBACK_DECADES_PATH = Path(__file__).parent / 'lists' / 'decades.txt'
def _load_lines(path: Path) -> list[str]:
"""Load a text file and return non-empty stripped lines."""
if not path.exists():
return []
lines = path.read_text(encoding='utf-8').splitlines()
return [line.strip() for line in lines if line.strip()]
def _load_list(primary: Path, fallback: Path) -> list[str]:
"""Load from the primary path, falling back to the bundled copy."""
lines = _load_lines(primary)
if lines:
return lines
return _load_lines(fallback)
MUSIC_GENRES = _load_list(_GENRES_PATH, _FALLBACK_GENRES_PATH)
DECADES = _load_list(_DECADES_PATH, _FALLBACK_DECADES_PATH)
def parse_style_genre(style_genre: str | None) -> dict:
"""
Split a stored combined style string into decade, basic, and additional.
The stored format is 'Decade, Basic, Additional' (additional may be empty).
"""
parts = [p.strip() for p in (style_genre or '').split(',') if p.strip()]
return {
'decade': parts[0] if len(parts) > 0 else '',
'basic_style': parts[1] if len(parts) > 1 else '',
'additional_style': ', '.join(parts[2:]) if len(parts) > 2 else '',
}
def build_style_genre(decade: str, basic_style: str, additional_style: str) -> str:
"""Build the combined style_genre string stored in the database."""
parts = [p.strip() for p in [decade, basic_style, additional_style] if p.strip()]
return ', '.join(parts)
# Helpers (extracted from app.py to keep the route file manageable)
from helpers import (
MUSIC_GENRES, DECADES, parse_style_genre, build_style_genre,
is_admin, require_admin, admin_password_ok, is_valid_email,
allowed_file, upload_path, save_upload,
apply_mp3_tags,
encrypt_value, decrypt_value, decrypt_value_legacy,
settings_file_path, load_booth_settings, save_booth_settings,
get_email_config, get_refresh_seconds, get_kiosk_cycle_seconds, get_kiosk_mode,
get_max_revisions, get_callback_expiry_hours,
get_hermes_api_key, set_hermes_api_key, generate_hermes_api_key, mask_api_key,
get_ntfy_config, send_ntfy,
sign_prompt_callback, verify_prompt_callback, build_prompt_callback_url,
send_email, build_signature_images,
get_booth_open,
)
# ---------------------------------------------------------------------------
@ -147,403 +88,6 @@ STATUS_LABELS = {
'cancelled': 'Cancelled',
}
# ---------------------------------------------------------------------------
# Helper functions
# ---------------------------------------------------------------------------
def is_admin():
"""Return True if the current browser session is logged in as admin."""
return session.get('admin') is True
def require_admin():
"""Redirect to the admin login page if the user is not logged in."""
if not is_admin():
return redirect(url_for('admin_login'))
def admin_password_ok(pw):
"""Check the submitted admin password against the configured one."""
return pw and pw == current_app.config['ADMIN_PASSWORD']
def allowed_file(filename):
"""Return True if the uploaded filename has an allowed extension (mp3)."""
return '.' in filename and filename.rsplit('.', 1)[1].lower() in current_app.config['ALLOWED_EXTENSIONS']
def upload_path(request_id):
"""Return the per-request upload directory path, creating it if necessary."""
p = Path(current_app.config['UPLOAD_FOLDER']) / str(request_id)
p.mkdir(parents=True, exist_ok=True)
return p
def save_upload(request_id, file_obj, version, song_title=None):
"""
Save an uploaded MP3 file for a request, preserving the original filename
with a version prefix (e.g. A - MySong.mp3 / B - MySong.mp3).
Applies the configured metadata tags (title, artist, album, year, comment).
"""
if not file_obj or file_obj.filename == '':
return None
if not allowed_file(file_obj.filename):
flash('Only MP3 files are allowed.', 'error')
return None
original = secure_filename(file_obj.filename)
filename = f"{version.upper()} - {original}"
p = upload_path(request_id)
dest = p / filename
file_obj.save(dest)
apply_mp3_tags(str(dest), song_title)
return str(dest)
def apply_mp3_tags(path, title=None):
"""
Write common ID3 tags on an uploaded MP3 using the runtime metadata defaults.
Writes title, artist, album, and date via EasyID3, plus a comment using both
a COMM frame and a TXXX:Comment frame for broad reader compatibility.
Failures are logged as a warning and do not block the upload.
"""
cfg = load_booth_settings()
try:
audio = MP3(path)
if audio.tags is None:
audio.add_tags()
if not isinstance(audio.tags, EasyID3):
audio.tags = EasyID3()
tags = audio.tags
if title:
tags['title'] = title
if cfg.get('artist'):
tags['artist'] = cfg['artist']
if cfg.get('album'):
tags['album'] = cfg['album']
if cfg.get('year'):
tags['date'] = str(cfg['year'])
audio.save()
if cfg.get('comment'):
from mutagen.id3 import COMM, TXXX
audio2 = MP3(path)
if audio2.tags is None:
audio2.add_tags()
audio2.tags["COMM"] = COMM(encoding=3, lang='eng', desc='Comment', text=cfg['comment'])
audio2.tags["TXXX:Comment"] = TXXX(encoding=3, desc='Comment', text=cfg['comment'])
audio2.save()
except Exception as e:
try:
flash(f'Warning: could not tag MP3: {e}', 'error')
except RuntimeError:
import logging
logging.getLogger('app').warning('Could not tag MP3 %s: %s', path, e)
def _get_fernet():
"""Derive a Fernet key from the Flask SECRET_KEY so stored values are encrypted."""
secret = current_app.config['SECRET_KEY'].encode()
kdf = PBKDF2HMAC(
algorithm=hashes.SHA256(),
length=32,
salt=b'theme-song-booth-v1',
iterations=480000,
)
key = base64.urlsafe_b64encode(kdf.derive(secret))
return Fernet(key)
def encrypt_value(value):
"""Encrypt a string using the Flask SECRET_KEY. Returns base64 ciphertext."""
if not value:
return ''
return _get_fernet().encrypt(value.encode()).decode()
def decrypt_value(ciphertext):
"""Decrypt a string previously encrypted by encrypt_value."""
if not ciphertext:
return ''
try:
return _get_fernet().decrypt(ciphertext.encode()).decode()
except Exception:
return ''
def decrypt_value_legacy(ciphertext):
"""Decrypt or return plaintext. Tolerates unencrypted legacy values."""
if not ciphertext:
return ''
plaintext = decrypt_value(ciphertext)
if plaintext:
return plaintext
# If decryption failed, the value might already be plaintext.
# A Fernet token is base64 and ends with '='; a plain API key does not.
if not ciphertext.endswith('='):
return ciphertext
return ''
def settings_file_path():
"""Return the path to the persistent runtime settings JSON file."""
return Path(current_app.config['DATABASE']).parent / current_app.config['SETTINGS_FILE']
def load_booth_settings():
"""Load persistent runtime settings from JSON file inside the upload parent."""
cfg_path = settings_file_path()
if cfg_path.exists():
try:
cfg = json.loads(cfg_path.read_text())
# Normalize any legacy None metadata/email values to empty strings
# so form fields repopulate correctly after reload.
for key in ('artist', 'album', 'year', 'comment', 'smtp_host', 'smtp_port', 'smtp_user', 'smtp_from'):
if cfg.get(key) is None:
cfg[key] = ''
return cfg
except (json.JSONDecodeError, OSError):
pass
return {}
def save_booth_settings(settings):
"""Persist runtime settings to JSON file."""
cfg_path = settings_file_path()
try:
cfg_path.write_text(json.dumps(settings, indent=2))
except OSError as e:
flash(f'Warning: could not save settings: {e}', 'error')
def get_email_config():
"""
Return the effective SMTP configuration.
Runtime settings in booth_settings.json override environment defaults.
The SMTP password is decrypted from the encrypted value stored on disk.
"""
cfg = load_booth_settings()
return {
'SMTP_HOST': cfg.get('smtp_host', current_app.config['SMTP_HOST']),
'SMTP_PORT': int(cfg.get('smtp_port') or current_app.config['SMTP_PORT']),
'SMTP_USER': cfg.get('smtp_user', current_app.config['SMTP_USER']),
'SMTP_PASS': decrypt_value(cfg.get('smtp_pass', '')) or current_app.config['SMTP_PASS'],
'SMTP_FROM': cfg.get('smtp_from', current_app.config['SMTP_FROM']),
}
def get_refresh_seconds():
"""Return the dashboard auto-refresh interval in seconds (10, 20, or 30)."""
cfg = load_booth_settings()
try:
val = int(cfg.get('refresh_seconds', 10))
except (ValueError, TypeError):
val = 10
return val if val in (10, 20, 30) else 10
def get_kiosk_cycle_seconds():
"""Return the kiosk slide cycle interval in seconds. 0 = static price list; 5+ cycles QR and pricing."""
cfg = load_booth_settings()
try:
val = int(cfg.get('kiosk_cycle_seconds', 10))
except (ValueError, TypeError):
val = 10
if val == 0:
return 0
return max(5, val)
def get_kiosk_mode():
"""
Return 'qr', 'prices', 'queue', or 'cycle' based on kiosk_cycle_seconds setting.
-1 = QR only, 0 = prices only, 1 = queue only, 5+ = cycle through all three.
"""
cfg = load_booth_settings()
try:
val = int(cfg.get('kiosk_cycle_seconds', 10))
except (ValueError, TypeError):
val = 10
if val == -1:
return 'qr'
if val == 0:
return 'prices'
if val == 1:
return 'queue'
return 'cycle'
def get_max_revisions():
"""Return the effective max revisions as an integer."""
cfg = load_booth_settings()
try:
val = int(cfg.get('max_revisions', current_app.config.get('MAX_REVISIONS', 2)))
except (ValueError, TypeError):
val = current_app.config.get('MAX_REVISIONS', 2)
return max(0, val)
def get_hermes_api_key():
"""
Return the effective Hermes API key.
Environment variable HERMES_API_KEY overrides any runtime setting.
"""
env_key = current_app.config.get('HERMES_API_KEY', '')
if env_key:
return env_key
cfg = load_booth_settings()
return decrypt_value_legacy(cfg.get('hermes_api_key', ''))
def set_hermes_api_key(key):
"""Persist a new Hermes API key (encrypted) to runtime settings."""
cfg = load_booth_settings()
cfg['hermes_api_key'] = encrypt_value(key)
save_booth_settings(cfg)
def generate_hermes_api_key():
"""Generate a new random API key for Hermes callback authentication."""
return secrets.token_urlsafe(32)
def mask_api_key(key):
"""Return a masked version of the API key showing only the last 6 characters."""
if not key:
return 'Not set'
if len(key) <= 6:
return '*' * len(key)
return '*' * (len(key) - 6) + key[-6:]
def get_ntfy_config():
"""Return the effective ntfy server URL, topic, and access token from runtime settings."""
cfg = load_booth_settings()
return {
'server': cfg.get('ntfy_server', ''),
'topic': cfg.get('ntfy_topic', ''),
'token': decrypt_value(cfg.get('ntfy_token', '')) or '',
}
def send_ntfy(message, title='Theme Song Booth', priority='default', tags='bell'):
"""Send a push notification to the configured ntfy topic, if configured."""
ntfy = get_ntfy_config()
server = ntfy.get('server', '').rstrip('/')
topic = ntfy.get('topic', '').strip()
if not server or not topic:
return False
url = f"{server}/{topic}"
headers = {
'Title': title,
'Priority': priority,
'Tags': tags,
}
token = ntfy.get('token', '')
if token:
headers['Authorization'] = f'Bearer {token}'
try:
resp = requests.post(url, data=message.encode('utf-8'), headers=headers, timeout=10)
return resp.status_code in (200, 202)
except Exception:
return False
def sign_prompt_callback(rid, expires_at=None):
"""
Create a signed callback token for a specific request ID.
The signature is HMAC-SHA256 over "rid:expires_at" using APP_SECRET_KEY.
Returns a URL-safe token string.
"""
secret = current_app.config['SECRET_KEY'].encode()
if expires_at is None:
# Default expiry: 7 days, so operators have plenty of time to copy the
# callback URL into Hermes and for Hermes to POST back.
expires_at = int(time.time()) + 7 * 24 * 3600
payload = f"{rid}:{expires_at}"
sig = hmac.new(secret, payload.encode(), hashlib.sha256).hexdigest()[:16]
return f"{rid}:{expires_at}:{sig}"
def verify_prompt_callback(token):
"""
Verify a signed callback token. Returns (rid, ok) tuple.
The token must match the HMAC signature and not be expired.
"""
if not token:
return None, False
parts = token.split(':')
if len(parts) != 3:
return None, False
try:
rid = int(parts[0])
expires_at = int(parts[1])
except ValueError:
return None, False
if int(time.time()) > expires_at:
return None, False
expected = sign_prompt_callback(rid, expires_at)
if not hmac.compare_digest(expected, token):
return None, False
return rid, True
def build_prompt_callback_url(rid):
"""Build the full callback URL an operator pastes into Hermes for a request."""
token = sign_prompt_callback(rid)
return f"{current_app.config['PUBLIC_BASE_URL']}/api/prompt/{rid}?token={token}"
def send_email(to, subject, body, attachments=None, inline_images=None):
"""Send an email using the configured or runtime SMTP settings."""
cfg = get_email_config()
if not cfg['SMTP_PASS']:
raise RuntimeError('SMTP password is not configured')
msg = EmailMessage()
msg['From'] = cfg['SMTP_FROM']
msg['To'] = to
msg['Subject'] = subject
msg.set_content(body)
html_body = body.replace('\n', '<br>\n')
if inline_images:
for _, cid in inline_images:
html_body += f'<br><img src="cid:{cid}" alt="Dionysis Media" style="max-width:200px;margin-top:1rem;"/>'
html_body += f'<br><br><hr style="border:none;border-top:1px solid #ddd;"/><p style="font-size:0.9rem;color:#555;">Dionysis Media: stories, sound, and a little divine chaos — <a href="https://dionysismedia.ca/">https://dionysismedia.ca/</a></p>'
msg.add_alternative(html_body, subtype='html')
if inline_images:
for path, cid in inline_images:
with open(path, 'rb') as f:
data = f.read()
ext = Path(path).suffix.lower().lstrip('.')
subtype = ext if ext in ('png', 'jpg', 'jpeg', 'gif', 'webp') else 'png'
msg.get_payload()[1].add_related(data, maintype='image', subtype=subtype, cid=f'<{cid}>')
if attachments:
for path, name in attachments:
with open(path, 'rb') as f:
data = f.read()
msg.add_attachment(data, maintype='audio', subtype='mpeg', filename=name)
with smtplib.SMTP_SSL(cfg['SMTP_HOST'], cfg['SMTP_PORT'], context=ssl.create_default_context()) as server:
server.login(cfg['SMTP_USER'], cfg['SMTP_PASS'])
server.send_message(msg)
def build_signature_images():
"""Return inline image tuple list for static/DM-Logo_email.png (Dionysis Media logo)."""
logo_path = Path(current_app.root_path) / 'static' / 'DM-Logo_email.png'
if not logo_path.exists():
return []
return [(str(logo_path), 'dm-logo')]
def get_booth_open():
"""Return True if the booth is currently marked as open in runtime settings."""
cfg = load_booth_settings()
return cfg.get('booth_open', True)
# ---------------------------------------------------------------------------
# Public customer routes
# ---------------------------------------------------------------------------
@ -585,11 +129,12 @@ def request_form():
'name': request.form.get('name', '').strip(),
'email': request.form.get('email', '').strip().lower(),
'pronouns': pronouns,
'hobbies': request.form.get('hobbies', '').strip(),
'notable_facts': request.form.get('notable_facts', '').strip(),
'hobbies': request.form.get('hobbies', '').strip()[:2000],
'notable_facts': request.form.get('notable_facts', '').strip()[:2000],
'style_genre': build_style_genre(decade, basic_style, additional_style),
'extra_requests': request.form.get('extra_requests', '').strip(),
'extra_requests': request.form.get('extra_requests', '').strip()[:2000],
'vocal_gender': request.form.get('vocal_gender', '').strip(),
'stems_interest': bool(request.form.get('stems_interest')),
}
if not is_valid_email(form_data['email']):
flash('Please enter a valid email address.', 'error')
@ -626,6 +171,7 @@ def request_form():
f"Hobbies: {req['hobbies'] or '-'}",
f"Notable facts: {req['notable_facts'] or '-'}",
f"Extra requests: {req['extra_requests'] or '-'}",
f"Interested in STEMS: {'Yes' if req.get('stems_interest') else 'No'}",
"",
"You'll get another email with a private link to preview two versions of your song when they're ready.",
"",
@ -1174,7 +720,8 @@ def admin_request(rid):
notable_facts=request.form.get('notable_facts', '').strip(),
style_genre=style_genre,
vocal_gender=request.form.get('vocal_gender', '').strip(),
extra_requests=request.form.get('extra_requests', '').strip()
extra_requests=request.form.get('extra_requests', '').strip(),
stems_interest=bool(request.form.get('stems_interest'))
)
flash('Customer info updated.', 'success')
return redirect(url_for('admin_request', rid=rid))
@ -1270,6 +817,13 @@ def admin_request(rid):
except Exception as e:
flash(f'Failed to send preview email: {e}', 'error')
elif action == 'update_payment_ref':
# Update the Square payment reference without sending email or changing status.
payment_ref = request.form.get('square_payment_ref', '').strip()
update_request(rid, square_payment_ref=payment_ref)
flash('Payment reference updated.', 'success')
return redirect(url_for('admin_request', rid=rid))
elif action == 'mark_paid_deliver':
# Finalize: record Square payment ref, attach approved MP3s, email customer.
if req.get('customer_approved', 'none') == 'none':
@ -1464,7 +1018,7 @@ def admin_settings():
'style_genre', 'extra_requests', 'vocal_gender', 'status', 'suno_title', 'suno_style',
'suno_lyrics', 'song_a_path', 'song_b_path', 'customer_approved',
'approval_notified_at', 'preview_sent_at', 'delivery_sent_at',
'square_payment_ref', 'admin_alert_email', 'player_token', 'revision_note', 'revision_count', 'operator_notes', 'stems_link'
'square_payment_ref', 'admin_alert_email', 'player_token', 'revision_note', 'revision_count', 'operator_notes', 'stems_link', 'stems_interest'
}
expected_tables = {'requests', 'revision_history'}
health = {'ok': True, 'missing_columns': [], 'missing_tables': [], 'message': 'Database schema looks good.'}
@ -1670,6 +1224,20 @@ def admin_settings():
flash(f'Database restore failed: {e}', 'error')
return redirect(url_for('admin_settings'))
elif action == 'save_callback_expiry':
# Update the Hermes callback URL expiry lifetime in hours.
try:
val = int(request.form.get('callback_expiry_hours', '168').strip())
if val < 1:
raise ValueError
cfg = load_booth_settings()
cfg['callback_expiry_hours'] = val
save_booth_settings(cfg)
flash(f'Callback link expiry set to {val} hour(s).', 'success')
except ValueError:
flash('Invalid callback expiry. Please enter a positive number of hours.', 'error')
return redirect(url_for('admin_settings'))
elif action == 'regenerate_hermes_key':
# Legacy action: no longer exposed in UI. Key is managed via HERMES_API_KEY env var.
flash('Hermes API key is managed via the HERMES_API_KEY environment variable.', 'info')
@ -1727,6 +1295,7 @@ def admin_settings():
hermes_key_set=hermes_key_set,
hermes_key_just_generated=hermes_key_just_generated,
version=current_app.config['VERSION'],
current_callback_expiry_hours=get_callback_expiry_hours(),
ntfy=runtime_settings,
)