Dionysis Media: stories, sound, and a little divine chaos — https://dionysismedia.ca/
' - 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, ) diff --git a/helpers.py b/helpers.py new file mode 100644 index 0000000..d0804eb --- /dev/null +++ b/helpers.py @@ -0,0 +1,502 @@ +""" +helpers.py +========== +Utility and configuration helpers for the Theme Song Booth Flask app. + +These functions are stateless (or use Flask's current_app / session context) +and are imported by app.py. Keeping them here reduces the size of the route file. +""" + +import os +import re +import shutil +import smtplib +import ssl +import time +import base64 +import hmac +import hashlib +import secrets +import json +from email.message import EmailMessage +from pathlib import Path + +from flask import session, current_app, flash +from werkzeug.utils import secure_filename + +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes +from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC + +from mutagen.mp3 import MP3 +from mutagen.easyid3 import EasyID3 + +import requests + + +# --------------------------------------------------------------------------- +# Genre / decade helpers +# --------------------------------------------------------------------------- + +_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) + + +# --------------------------------------------------------------------------- +# Auth / validation helpers +# --------------------------------------------------------------------------- + +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.""" + from flask import redirect, url_for + 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 is_valid_email(email): + """Return True if the given string looks like a valid email address.""" + if not email: + return False + pattern = r"^[^\s@]+@[^\s@]+\.[^\s@]+$" + return re.match(pattern, email) is not None + + +# --------------------------------------------------------------------------- +# File upload helpers +# --------------------------------------------------------------------------- + +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. Applies the configured metadata tags. + """ + 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. + 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) + + +# --------------------------------------------------------------------------- +# Encryption / settings helpers +# --------------------------------------------------------------------------- + +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 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()) + 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') + + +# --------------------------------------------------------------------------- +# Config getters +# --------------------------------------------------------------------------- + +def get_email_config(): + """Return the effective SMTP configuration.""" + 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.""" + 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.""" + 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_callback_expiry_hours(): + """Return the Hermes signed callback token lifetime in hours (default 168 = 7 days).""" + cfg = load_booth_settings() + try: + val = int(cfg.get('callback_expiry_hours', 168)) + except (ValueError, TypeError): + val = 168 + return max(1, val) + + +def get_hermes_api_key(): + """Return the effective Hermes API key.""" + 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 + + +# --------------------------------------------------------------------------- +# Signed callback helpers +# --------------------------------------------------------------------------- + +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. + """ + secret = current_app.config['SECRET_KEY'].encode() + if expires_at is None: + expires_at = int(time.time()) + get_callback_expiry_hours() * 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.""" + 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.""" + from flask import url_for + token = sign_prompt_callback(rid) + return f"{current_app.config['PUBLIC_BASE_URL']}/api/prompt/{rid}?token={token}" + + +# --------------------------------------------------------------------------- +# Email helpers +# --------------------------------------------------------------------------- + +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', 'Dionysis Media: stories, sound, and a little divine chaos — https://dionysismedia.ca/
' + 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.""" + logo_path = Path(current_app.root_path) / 'static' / 'DM-Logo_email.png' + if not logo_path.exists(): + return [] + return [(str(logo_path), 'dm-logo')] + + +# --------------------------------------------------------------------------- +# Booth state +# --------------------------------------------------------------------------- + +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) diff --git a/models.py b/models.py index f56f30d..0baa0b6 100644 --- a/models.py +++ b/models.py @@ -52,7 +52,8 @@ CREATE TABLE IF NOT EXISTS requests ( revision_count INTEGER DEFAULT 0, revision_note TEXT, operator_notes TEXT, - stems_link TEXT + stems_link TEXT, + stems_interest INTEGER DEFAULT 0 ); CREATE INDEX IF NOT EXISTS idx_requests_status ON requests(status); @@ -107,7 +108,7 @@ def init_db(): 'customer_approved', 'approval_notified_at', 'preview_sent_at', 'delivery_sent_at', 'square_payment_ref', 'admin_alert_email', 'player_token', 'revision_count', 'revision_note', 'operator_notes', - 'stems_link' + 'stems_link', 'stems_interest' ], 'revision_history': [ 'id', 'created_at', 'request_id', 'revision_count', 'note', @@ -135,7 +136,7 @@ def now_utc(): return datetime.now(timezone.utc).isoformat() -def create_request(name, email, hobbies, notable_facts, style_genre, extra_requests, vocal_gender=None, pronouns=None): +def create_request(name, email, hobbies, notable_facts, style_genre, extra_requests, vocal_gender=None, pronouns=None, stems_interest=0): """ Insert a new customer request. Returns the auto-generated request id. @@ -143,9 +144,9 @@ def create_request(name, email, hobbies, notable_facts, style_genre, extra_reque db = get_db() cur = db.execute( """INSERT INTO requests - (name, email, hobbies, notable_facts, style_genre, pronouns, extra_requests, vocal_gender, player_token) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", - (name, email, hobbies, notable_facts, style_genre, pronouns, extra_requests, vocal_gender, new_token()) + (name, email, hobbies, notable_facts, style_genre, pronouns, extra_requests, vocal_gender, player_token, stems_interest) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", + (name, email, hobbies, notable_facts, style_genre, pronouns, extra_requests, vocal_gender, new_token(), 1 if stems_interest else 0) ) db.commit() return cur.lastrowid diff --git a/templates/admin/dashboard.html b/templates/admin/dashboard.html index 5dbaf78..c769d4a 100644 --- a/templates/admin/dashboard.html +++ b/templates/admin/dashboard.html @@ -131,7 +131,7 @@Square Payment Reference: {{ req.square_payment_ref }}
{% endif %} + + +