1348 lines
57 KiB
Python
1348 lines
57 KiB
Python
"""
|
|
app.py
|
|
======
|
|
Main Flask application for the Theme Song Booth.
|
|
|
|
This module defines all HTTP routes, helper functions, the email layer,
|
|
runtime settings persistence, MP3 metadata tagging, rate limiting, and
|
|
database health/maintenance helpers.
|
|
|
|
It is meant to be served by gunicorn inside a Docker container (see Dockerfile).
|
|
|
|
Public routes (customers):
|
|
- / -> redirects to /request
|
|
- /request -> customer submits their info
|
|
- /thanks/<id> -> confirmation page after submission
|
|
- /play/<token> -> private player page with Version A and B
|
|
- /play/<token>/approve -> customer picks a version
|
|
- /play/<token>/revise -> customer asks for changes
|
|
- /audio/<token>/<v>.mp3 -> serves the uploaded MP3 files
|
|
|
|
Admin routes:
|
|
- /admin/login -> password login
|
|
- /admin/logout -> clears session
|
|
- /admin -> dashboard queue with status filters and auto-refresh
|
|
- /admin/settings -> runtime settings, health check, DB stats, backup/restore, reset
|
|
- /admin/request/<id> -> detail/edit page for a single request
|
|
- /admin/request/<id>/delete -> deletes one request and its uploaded files
|
|
- /admin/reset -> deletes ALL requests and ALL files
|
|
"""
|
|
|
|
# Standard library imports
|
|
import os
|
|
import hmac
|
|
import shutil
|
|
import time
|
|
from pathlib import Path
|
|
|
|
# Flask and related imports
|
|
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
|
|
|
|
# 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
|
|
|
|
# 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,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App setup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# Create the Flask app and load configuration from Config class.
|
|
app = Flask(__name__)
|
|
app.config.from_object(Config)
|
|
|
|
# Global request rate limiting by remote IP (default 60/min). The public form is further limited to 5/min.
|
|
limiter = Limiter(get_remote_address, app=app, default_limits=["60 per minute"])
|
|
|
|
# Ensure the SQLite connection is closed at the end of each request.
|
|
app.teardown_appcontext(close_db)
|
|
|
|
# Human-readable labels for each status value stored in the database.
|
|
STATUS_LABELS = {
|
|
'pending': 'Pending',
|
|
'prompt_ready': 'Prompt Ready',
|
|
'songs_uploaded': 'Songs Uploaded — Awaiting Approval',
|
|
'revisions_requested': 'Revisions Requested',
|
|
'awaiting_payment': 'Awaiting Payment',
|
|
'paid': 'Paid',
|
|
'delivered': 'Delivered',
|
|
'cancelled': 'Cancelled',
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Public customer routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route('/')
|
|
def index():
|
|
"""Root route: redirect customers straight to the request form."""
|
|
return redirect(url_for('request_form'))
|
|
|
|
|
|
@app.route('/request', methods=['GET', 'POST'])
|
|
@limiter.limit("5 per minute")
|
|
def request_form():
|
|
"""
|
|
Public request form.
|
|
GET -> shows the form with the banner image, or a closed message if the booth is closed.
|
|
POST -> validates the email, creates a database record, sends a
|
|
confirmation email, and redirects to the thanks page.
|
|
Rate limited to 5 submissions per minute per IP.
|
|
"""
|
|
if not get_booth_open():
|
|
return render_template('closed.html')
|
|
|
|
if request.method == 'POST':
|
|
decade = request.form.get('decade', '').strip()
|
|
basic_style = request.form.get('basic_style', '').strip()
|
|
additional_style = request.form.get('additional_style', '').strip()
|
|
pronouns = request.form.get('pronouns', '').strip()
|
|
if not pronouns:
|
|
flash('Please select your pronouns.', 'error')
|
|
return render_template('request.html', form=request.form, decades=DECADES, genres=MUSIC_GENRES), 400
|
|
if not decade:
|
|
flash('Please select a decade.', 'error')
|
|
return render_template('request.html', form=request.form, decades=DECADES, genres=MUSIC_GENRES), 400
|
|
if not basic_style:
|
|
flash('Please select a basic style.', 'error')
|
|
return render_template('request.html', form=request.form, decades=DECADES, genres=MUSIC_GENRES), 400
|
|
form_data = {
|
|
'name': request.form.get('name', '').strip(),
|
|
'email': request.form.get('email', '').strip().lower(),
|
|
'pronouns': pronouns,
|
|
'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()[: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')
|
|
return render_template('request.html', form=form_data, decades=DECADES, genres=MUSIC_GENRES), 400
|
|
rid = create_request(**form_data)
|
|
|
|
# Notify operator via ntfy when a new request comes in.
|
|
try:
|
|
send_ntfy(
|
|
f"New request #{rid} from {form_data['name']} ({form_data['email']})\nStyle: {form_data['style_genre'] or '-'}",
|
|
title='New Theme Song Request',
|
|
priority='high',
|
|
tags='musical_note'
|
|
)
|
|
except Exception:
|
|
# Push notification failure should never break the customer form.
|
|
pass
|
|
|
|
# Send confirmation email with a summary of what the customer asked for.
|
|
req = get_request_by_id(rid)
|
|
if req:
|
|
try:
|
|
body_lines = [
|
|
f"Hi {req['name']},",
|
|
"",
|
|
"Thanks for stopping by the Trollgorithm Theme Song Booth! We've received your request and will start crafting your custom song soon.",
|
|
"",
|
|
"Here's what we have on file:",
|
|
f"Name: {req['name']}",
|
|
f"Email: {req['email']}",
|
|
f"Pronouns: {req['pronouns'] or '-'}",
|
|
f"Style / genre: {req['style_genre'] or '-'}",
|
|
f"Preferred singer voice / gender: {req['vocal_gender'] or 'No preference'}",
|
|
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.",
|
|
"",
|
|
"— Trollgorithm / Dionysis Media"
|
|
]
|
|
send_email(req['email'], 'Your theme song request is received', '\n'.join(body_lines), inline_images=build_signature_images())
|
|
except Exception as e:
|
|
flash(f'Your request was saved, but we could not send a confirmation email: {e}', 'error')
|
|
|
|
flash('Your request has been submitted! Check your email soon.', 'success')
|
|
return redirect(url_for('thanks', rid=rid))
|
|
return render_template('request.html', form=None, decades=DECADES, genres=MUSIC_GENRES)
|
|
|
|
|
|
@app.route('/api/ping', methods=['GET'])
|
|
@app.route('/api/key-test', methods=['GET'])
|
|
@limiter.limit('4 per minute')
|
|
def api_key_test():
|
|
"""
|
|
Diagnostic endpoint for verifying the Hermes API key configuration.
|
|
|
|
Accepts a Bearer token in the Authorization header and compares it against
|
|
the configured HERMES_API_KEY. Returns plain JSON so callers can distinguish
|
|
key mismatch from networking / signed-token issues.
|
|
|
|
Rate limited to 4 per minute to prevent brute-force guessing.
|
|
"""
|
|
expected_key = get_hermes_api_key()
|
|
if not expected_key:
|
|
return jsonify({'ok': False, 'reason': 'not_configured'}), 500
|
|
|
|
auth_header = request.headers.get('Authorization', '').strip()
|
|
if not auth_header.startswith('Bearer '):
|
|
return jsonify({'ok': False, 'reason': 'missing_bearer'}), 401
|
|
|
|
provided_key = auth_header[7:].strip()
|
|
if not hmac.compare_digest(expected_key, provided_key):
|
|
return jsonify({'ok': False, 'reason': 'key_mismatch'}), 401
|
|
|
|
return jsonify({'ok': True, 'reason': 'valid'}), 200
|
|
|
|
|
|
@app.route('/api/prompt/<int:rid>', methods=['POST'])
|
|
def api_update_prompt(rid):
|
|
"""
|
|
Hermes callback endpoint.
|
|
|
|
Accepts a JSON POST with generated Suno prompt fields and updates the
|
|
matching request. Two layers of auth:
|
|
1) A per-request signed callback token in the query string.
|
|
2) A Hermes API key in the Authorization header (Bearer).
|
|
|
|
Only records in 'pending' status can be updated. On success, status is set
|
|
to 'prompt_ready'.
|
|
"""
|
|
# Layer 1: verify the signed callback URL token.
|
|
callback_token = request.args.get('token', '').strip()
|
|
token_rid, token_ok = verify_prompt_callback(callback_token)
|
|
if not token_ok or token_rid != rid:
|
|
abort(401)
|
|
|
|
# Layer 2: verify the Hermes API key from the Authorization header.
|
|
expected_key = get_hermes_api_key()
|
|
if not expected_key:
|
|
abort(500, description='Hermes API key is not configured')
|
|
auth_header = request.headers.get('Authorization', '').strip()
|
|
if not auth_header.startswith('Bearer '):
|
|
abort(401)
|
|
provided_key = auth_header[7:].strip()
|
|
if not hmac.compare_digest(expected_key, provided_key):
|
|
abort(401)
|
|
|
|
req = get_request_by_id(rid)
|
|
if not req:
|
|
abort(404)
|
|
# Allow updates when the request is pending or when a revision has been requested.
|
|
if req['status'] not in ('pending', 'revisions_requested'):
|
|
abort(409, description='Request is no longer pending or awaiting revision')
|
|
|
|
data = request.get_json(silent=True) or {}
|
|
|
|
# Optional email verification to make sure clipboard matches record.
|
|
provided_email = data.get('email', '').strip().lower()
|
|
if provided_email and provided_email != req['email'].lower():
|
|
abort(400, description='Email mismatch')
|
|
|
|
# When this is a revision, keep the revision note and status as revisions_requested
|
|
# so the operator still sees it as "needs new songs", and preserve customer_approved.
|
|
new_status = 'revisions_requested' if req['status'] == 'revisions_requested' else 'prompt_ready'
|
|
update_request(rid,
|
|
suno_title=data.get('suno_title', '').strip(),
|
|
suno_style=data.get('suno_style', '').strip(),
|
|
suno_lyrics=data.get('suno_lyrics', '').strip(),
|
|
status=new_status
|
|
)
|
|
|
|
return jsonify({'ok': True, 'request_id': rid, 'status': new_status}), 200
|
|
|
|
|
|
@app.route('/thanks/<int:rid>')
|
|
def thanks(rid):
|
|
"""Confirmation page shown after a customer submits a request."""
|
|
req = get_request_by_id(rid)
|
|
if not req:
|
|
abort(404)
|
|
return render_template('thanks.html', req=req)
|
|
|
|
|
|
@app.route('/status', methods=['GET', 'POST'])
|
|
def status_lookup():
|
|
"""
|
|
Public order status lookup page.
|
|
Customers enter their email to see all their requests and their current
|
|
statuses, plus the private player link once songs have been uploaded.
|
|
"""
|
|
requests_list = []
|
|
email = ''
|
|
searched = False
|
|
if request.method == 'POST':
|
|
email = request.form.get('email', '').strip().lower()
|
|
if not is_valid_email(email):
|
|
flash('Please enter a valid email address.', 'error')
|
|
else:
|
|
requests_list = get_requests_by_email(email)
|
|
searched = True
|
|
return render_template('status.html', email=email, requests=requests_list, searched=searched, statuses=STATUS_LABELS)
|
|
|
|
|
|
@app.route('/faq')
|
|
def faq():
|
|
"""Customer-facing frequently asked questions page."""
|
|
return render_template('faq.html')
|
|
|
|
|
|
@app.route('/play/<token>')
|
|
def play(token):
|
|
"""
|
|
Private player page for a customer.
|
|
The token is a cryptographically random URL-safe string generated at request time.
|
|
Shows A/B audio players, approval buttons, or a revision note depending on status.
|
|
"""
|
|
req = get_request_by_token(token)
|
|
if not req:
|
|
abort(404)
|
|
|
|
# Load runtime max revisions setting.
|
|
max_revisions = get_max_revisions()
|
|
revisions_left = max(0, max_revisions - int(req.get('revision_count') or 0))
|
|
|
|
return render_template('player.html', req=req, revisions_left=revisions_left)
|
|
|
|
|
|
@app.route('/play/<token>/approve', methods=['POST'])
|
|
def approve(token):
|
|
"""
|
|
Customer has chosen Version A, Version B, or both.
|
|
Updates the request status to 'awaiting_payment' so the operator can collect payment.
|
|
Operator email alerts are intentionally disabled; the dashboard is the single queue.
|
|
"""
|
|
req = get_request_by_token(token)
|
|
if not req:
|
|
abort(404)
|
|
choice = request.form.get('choice')
|
|
if choice not in ('a', 'b', 'both'):
|
|
flash('Invalid selection.', 'error')
|
|
return redirect(url_for('play', token=token))
|
|
|
|
update_request(req['id'], customer_approved=choice, status='awaiting_payment', approval_notified_at=now_utc())
|
|
|
|
flash('Thanks! Please return to the booth to finalize payment.', 'success')
|
|
return redirect(url_for('play', token=token))
|
|
|
|
|
|
@app.route('/play/<token>/revise', methods=['POST'])
|
|
def revise(token):
|
|
"""
|
|
Customer asked for changes.
|
|
Enforces the runtime max revisions limit, archives the current A/B MP3 files,
|
|
stores the revision note, and resets status to 'revisions_requested'.
|
|
"""
|
|
note = request.form.get('revision_note', '').strip()
|
|
req = get_request_by_token(token)
|
|
if not req:
|
|
abort(404)
|
|
|
|
# Enforce max revisions limit for customer-submitted revisions.
|
|
max_revisions = get_max_revisions()
|
|
current_count = int(req.get('revision_count') or 0)
|
|
if current_count >= max_revisions:
|
|
flash('Revision limit reached. Please speak to the booth operator if you need further changes.', 'error')
|
|
return redirect(url_for('play', token=token))
|
|
|
|
# Increment revision counter and archive current files before new versions are uploaded.
|
|
new_count = current_count + 1
|
|
upload_dir = Path(current_app.config['UPLOAD_FOLDER']) / str(req['id'])
|
|
old_a, old_b = req.get('song_a_path'), req.get('song_b_path')
|
|
new_a, new_b = old_a, old_b
|
|
if upload_dir.exists():
|
|
for field, version in (('song_a_path', 'A'), ('song_b_path', 'B')):
|
|
path = req.get(field)
|
|
if path and Path(path).exists():
|
|
old = Path(path)
|
|
archived = upload_dir / f"Rev{new_count}-{old.name}"
|
|
try:
|
|
old.rename(archived)
|
|
if field == 'song_a_path':
|
|
new_a = str(archived)
|
|
else:
|
|
new_b = str(archived)
|
|
req[field] = str(archived)
|
|
except OSError:
|
|
pass
|
|
log_revision(req['id'], new_count, note, old_a=old_a, old_b=old_b, new_a=new_a, new_b=new_b)
|
|
update_request(req['id'], revision_note=note, status='revisions_requested',
|
|
song_a_path=req.get('song_a_path'), song_b_path=req.get('song_b_path'),
|
|
customer_approved='none', revision_count=int(new_count))
|
|
|
|
# NOTE: No operator email is sent; the dashboard is the single queue.
|
|
flash('Your feedback has been saved. We will regenerate and update you.', 'success')
|
|
return redirect(url_for('play', token=token))
|
|
|
|
|
|
@app.route('/api/stream/<token>/<version>.mp3')
|
|
def stream_audio(token, version):
|
|
"""
|
|
Stream an uploaded MP3 through a backend proxy endpoint.
|
|
|
|
This hides the real file path from the customer. The endpoint checks the
|
|
player token and serves bytes with Range request support so the HTML audio
|
|
player can seek. The URL is still interceptable in-browser, but it is not a
|
|
direct file path and can be gated or expired later.
|
|
"""
|
|
req = get_request_by_token(token)
|
|
if not req:
|
|
abort(404)
|
|
if version not in ('a', 'b'):
|
|
abort(404)
|
|
|
|
# Both versions must exist before any streaming happens.
|
|
a_path = req.get('song_a_path')
|
|
b_path = req.get('song_b_path')
|
|
if not a_path or not b_path:
|
|
abort(404)
|
|
for p in (a_path, b_path):
|
|
if not Path(p).exists():
|
|
abort(404)
|
|
|
|
path = a_path if version == 'a' else b_path
|
|
file_path = Path(path)
|
|
file_size = file_path.stat().st_size
|
|
|
|
range_header = request.headers.get('Range', '')
|
|
start = 0
|
|
end = file_size - 1
|
|
status_code = 200
|
|
|
|
if range_header and range_header.startswith('bytes='):
|
|
try:
|
|
range_value = range_header[len('bytes='):].strip()
|
|
if '-' in range_value:
|
|
parts = range_value.split('-')
|
|
if parts[0]:
|
|
start = int(parts[0])
|
|
if parts[1]:
|
|
end = min(int(parts[1]), file_size - 1)
|
|
if start >= file_size or start > end:
|
|
abort(416)
|
|
status_code = 206
|
|
except ValueError:
|
|
start = 0
|
|
end = file_size - 1
|
|
status_code = 200
|
|
|
|
def generate():
|
|
with open(file_path, 'rb') as f:
|
|
f.seek(start)
|
|
remaining = end - start + 1
|
|
chunk_size = 64 * 1024
|
|
while remaining > 0:
|
|
to_read = min(chunk_size, remaining)
|
|
data = f.read(to_read)
|
|
if not data:
|
|
break
|
|
yield data
|
|
remaining -= len(data)
|
|
|
|
response = current_app.response_class(generate(), mimetype='audio/mpeg')
|
|
response.status_code = status_code
|
|
response.headers['Accept-Ranges'] = 'bytes'
|
|
response.headers['Content-Disposition'] = 'inline'
|
|
response.headers['Content-Length'] = str(end - start + 1)
|
|
if status_code == 206:
|
|
response.headers['Content-Range'] = f'bytes {start}-{end}/{file_size}'
|
|
return response
|
|
|
|
|
|
@app.route('/audio/<token>/<version>.mp3')
|
|
def audio(token, version):
|
|
"""
|
|
Legacy audio endpoint. Replaced by /api/stream/<token>/<version>.mp3.
|
|
Returns 404 so old direct links do not work.
|
|
"""
|
|
abort(404)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Admin routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.route('/admin/login', methods=['GET', 'POST'])
|
|
def admin_login():
|
|
"""Simple session-based admin login. Password is set via ADMIN_PASSWORD env var."""
|
|
if is_admin():
|
|
return redirect(url_for('admin_dashboard'))
|
|
if request.method == 'POST':
|
|
if admin_password_ok(request.form.get('password', '')):
|
|
session['admin'] = True
|
|
return redirect(url_for('admin_dashboard'))
|
|
flash('Invalid password.', 'error')
|
|
return render_template('admin/login.html')
|
|
|
|
|
|
@app.route('/admin/logout')
|
|
def admin_logout():
|
|
"""Clear the admin session."""
|
|
session.pop('admin', None)
|
|
return redirect(url_for('admin_login'))
|
|
|
|
|
|
@app.route('/admin')
|
|
def admin_dashboard():
|
|
"""
|
|
Main operator queue.
|
|
Optional ?status= filter lets operators focus on one state at a time.
|
|
Auto-refresh interval is controlled from /admin/settings.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
status_filter = request.args.get('status')
|
|
requests = list_requests(status_filter)
|
|
return render_template('admin/dashboard.html', requests=requests, statuses=STATUS_LABELS, current_status=status_filter, refresh_seconds=get_refresh_seconds())
|
|
|
|
|
|
@app.route('/admin/sales')
|
|
def admin_sales():
|
|
"""
|
|
Sales report: delivered requests only.
|
|
Shows customer email, name, chosen version, and Square payment reference.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
sales = list_requests(status='delivered')
|
|
return render_template('admin/sales.html', sales=sales, statuses=STATUS_LABELS)
|
|
|
|
|
|
@app.route('/admin/pricing', methods=['GET', 'POST'])
|
|
def admin_pricing():
|
|
"""
|
|
Pricing configuration page.
|
|
Fixed items: one_song, both_songs, wav_per_song, stems_per_song.
|
|
Plus up to 5 custom name/price pairs.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
|
|
fixed_keys = ['one_song', 'both_songs', 'wav_per_song', 'stems_per_song']
|
|
custom_count = 5
|
|
|
|
cfg = load_booth_settings()
|
|
if 'pricing' not in cfg:
|
|
cfg['pricing'] = {}
|
|
|
|
if request.method == 'POST':
|
|
pricing = {}
|
|
for key in fixed_keys:
|
|
pricing[key] = request.form.get(key, '').strip()
|
|
for i in range(1, custom_count + 1):
|
|
name = request.form.get(f'custom_name_{i}', '').strip()
|
|
price = request.form.get(f'custom_price_{i}', '').strip()
|
|
if name:
|
|
pricing[f'custom_{i}'] = {'name': name, 'price': price}
|
|
else:
|
|
pricing[f'custom_{i}'] = None
|
|
cfg['pricing'] = pricing
|
|
save_booth_settings(cfg)
|
|
flash('Pricing saved.', 'success')
|
|
return redirect(url_for('admin_pricing'))
|
|
|
|
pricing = cfg.get('pricing', {})
|
|
fixed = {key: pricing.get(key, '') for key in fixed_keys}
|
|
customs = []
|
|
for i in range(1, custom_count + 1):
|
|
entry = pricing.get(f'custom_{i}')
|
|
customs.append({
|
|
'name': entry.get('name', '') if isinstance(entry, dict) else '',
|
|
'price': entry.get('price', '') if isinstance(entry, dict) else ''
|
|
})
|
|
return render_template('admin/pricing.html', fixed=fixed, customs=customs)
|
|
|
|
|
|
@app.route('/kiosk')
|
|
def kiosk():
|
|
"""
|
|
Public kiosk display page for the booth.
|
|
Shows open/closed banner, pricing, and optionally cycles with a QR code.
|
|
Auto-refreshes so pricing updates are picked up quickly.
|
|
"""
|
|
cfg = load_booth_settings()
|
|
pricing = cfg.get('pricing', {})
|
|
|
|
fixed_keys = {
|
|
'one_song': 'One Song',
|
|
'both_songs': 'Both Songs',
|
|
'wav_per_song': 'WAV files / song',
|
|
'stems_per_song': 'STEM files / song'
|
|
}
|
|
price_items = []
|
|
for key, label in fixed_keys.items():
|
|
val = pricing.get(key, '').strip()
|
|
if val:
|
|
price_items.append({'label': label, 'price': val})
|
|
for i in range(1, 6):
|
|
entry = pricing.get(f'custom_{i}')
|
|
if isinstance(entry, dict):
|
|
name = entry.get('name', '').strip()
|
|
price = entry.get('price', '').strip()
|
|
if name and price:
|
|
price_items.append({'label': name, 'price': price})
|
|
|
|
cycle_seconds = get_kiosk_cycle_seconds()
|
|
mode = get_kiosk_mode()
|
|
booth_open = get_booth_open()
|
|
|
|
# Build the public queue: only active statuses, mapped to friendly names,
|
|
# using the local-part of the email as the customer name.
|
|
KIOSK_STATUS_MAP = {
|
|
'pending': 'Received',
|
|
'prompt_ready': 'Trollgorithm Recording in Studio',
|
|
'songs_uploaded': 'Trollgorithm Recording in Studio',
|
|
'revisions_requested': 'Waiting for Customer Response',
|
|
'awaiting_payment': 'Payment Due',
|
|
}
|
|
active_statuses = set(KIOSK_STATUS_MAP.keys())
|
|
raw_queue = list_requests()
|
|
queue = []
|
|
for r in raw_queue:
|
|
if r.get('status') in active_statuses:
|
|
email = r.get('email') or ''
|
|
name = email.split('@')[0] if '@' in email else email
|
|
queue.append({
|
|
'id': r['id'],
|
|
'name': name or 'Guest',
|
|
'status': KIOSK_STATUS_MAP[r['status']],
|
|
'raw_status': r['status'],
|
|
})
|
|
|
|
return render_template(
|
|
'kiosk.html',
|
|
booth_open=booth_open,
|
|
price_items=price_items,
|
|
cycle_seconds=cycle_seconds,
|
|
mode=mode,
|
|
queue=queue,
|
|
refresh_seconds=30
|
|
)
|
|
|
|
|
|
@app.route('/admin/request/<int:rid>', methods=['GET', 'POST'])
|
|
def admin_request(rid):
|
|
"""
|
|
Detail/edit page for a single request.
|
|
GET -> render customer info (email editable), prompt, upload status,
|
|
email status, operator notes, and delivery forms.
|
|
POST -> handle one of six actions:
|
|
update_customer_email, save_operator_notes, save_prompt, upload_songs,
|
|
notify_customer, mark_paid_deliver
|
|
Uploaded MP3s are tagged with metadata defaults from /admin/settings.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
req = get_request_by_id(rid)
|
|
if not req:
|
|
abort(404)
|
|
|
|
# Small helpers exposed to the template for status badges.
|
|
def file_exists(path):
|
|
return bool(path and Path(path).exists())
|
|
|
|
def basename(path):
|
|
return Path(path).name if path else ''
|
|
|
|
# Collect any extra MP3 files in the request folder (archived revisions).
|
|
upload_dir = Path(current_app.config['UPLOAD_FOLDER']) / str(rid)
|
|
current_paths = {req['song_a_path'], req['song_b_path']}
|
|
extra_files = []
|
|
if upload_dir.exists():
|
|
for f in upload_dir.iterdir():
|
|
if f.is_file() and f.suffix.lower() == '.mp3' and str(f) not in current_paths:
|
|
extra_files.append(str(f))
|
|
extra_files.sort()
|
|
|
|
if request.method == 'POST':
|
|
action = request.form.get('action')
|
|
|
|
if action == 'update_customer_email':
|
|
new_email = request.form.get('email', '').strip().lower()
|
|
if not is_valid_email(new_email):
|
|
flash('Please enter a valid email address.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
update_request(rid, email=new_email)
|
|
flash('Customer email updated.', 'success')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
elif action == 'update_customer_info':
|
|
new_email = request.form.get('email', '').strip().lower()
|
|
pronouns = request.form.get('pronouns', '').strip()
|
|
if not is_valid_email(new_email):
|
|
flash('Please enter a valid email address.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
if not pronouns:
|
|
flash('Pronouns are required.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
style_genre = build_style_genre(
|
|
request.form.get('decade', '').strip(),
|
|
request.form.get('basic_style', '').strip(),
|
|
request.form.get('additional_style', '').strip()
|
|
)
|
|
update_request(rid,
|
|
email=new_email,
|
|
name=request.form.get('name', '').strip(),
|
|
pronouns=pronouns,
|
|
hobbies=request.form.get('hobbies', '').strip(),
|
|
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(),
|
|
stems_interest=bool(request.form.get('stems_interest'))
|
|
)
|
|
flash('Customer info updated.', 'success')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
elif action == 'save_operator_notes':
|
|
update_request(rid, operator_notes=request.form.get('operator_notes', '').strip())
|
|
flash('Operator notes saved.', 'success')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
elif action == 'save_stems_link':
|
|
update_request(rid, stems_link=request.form.get('stems_link', '').strip())
|
|
flash('Stems share link saved.', 'success')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
elif action == 'cancel_request':
|
|
update_request(rid, status='cancelled')
|
|
flash('Request marked as cancelled.', 'success')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
elif action == 'save_prompt':
|
|
# Store the generated title/style/lyrics and mark prompt ready.
|
|
update_request(rid,
|
|
suno_title=request.form.get('suno_title', '').strip(),
|
|
suno_style=request.form.get('suno_style', '').strip(),
|
|
suno_lyrics=request.form.get('suno_lyrics', '').strip(),
|
|
status='prompt_ready'
|
|
)
|
|
flash('Prompt saved.', 'success')
|
|
|
|
elif action == 'delete_songs':
|
|
# Remove selected uploaded MP3s and reset status so new songs can be uploaded.
|
|
to_delete = request.form.getlist('delete_song')
|
|
if not to_delete:
|
|
flash('Select at least one song to delete.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
update_fields = {}
|
|
for version in to_delete:
|
|
if version == 'a':
|
|
path = req.get('song_a_path')
|
|
if path and Path(path).exists():
|
|
try:
|
|
Path(path).unlink()
|
|
except OSError:
|
|
pass
|
|
update_fields['song_a_path'] = None
|
|
elif version == 'b':
|
|
path = req.get('song_b_path')
|
|
if path and Path(path).exists():
|
|
try:
|
|
Path(path).unlink()
|
|
except OSError:
|
|
pass
|
|
update_fields['song_b_path'] = None
|
|
|
|
if update_fields:
|
|
# Wipe any prior customer approval because the old files are gone.
|
|
update_fields['customer_approved'] = 'none'
|
|
# Reset to prompt_ready since songs need to be uploaded again.
|
|
update_fields['status'] = 'prompt_ready'
|
|
update_request(rid, **update_fields)
|
|
flash('Selected songs deleted. Request reset to Prompt Ready.', 'success')
|
|
else:
|
|
flash('No uploaded songs selected for deletion.', 'error')
|
|
|
|
elif action == 'upload_songs':
|
|
# Save uploaded MP3 files for Version A and/or Version B.
|
|
# Use the saved Suno title as the MP3 title tag if available.
|
|
song_title = req.get('suno_title') or None
|
|
a_path = save_upload(rid, request.files.get('song_a'), 'a', song_title)
|
|
b_path = save_upload(rid, request.files.get('song_b'), 'b', song_title)
|
|
fields = {}
|
|
if a_path:
|
|
fields['song_a_path'] = a_path
|
|
if b_path:
|
|
fields['song_b_path'] = b_path
|
|
if fields:
|
|
fields['status'] = 'songs_uploaded'
|
|
update_request(rid, **fields)
|
|
flash('Songs uploaded.', 'success')
|
|
|
|
elif action == 'notify_customer':
|
|
# Email the customer a private player link. Both songs must be uploaded first.
|
|
if not (req['song_a_path'] and req['song_b_path']):
|
|
flash('Both songs must be uploaded first.', 'error')
|
|
else:
|
|
player_link = f"{current_app.config['PUBLIC_BASE_URL']}/play/{req['player_token']}"
|
|
body = f"Hi {req['name']},\n\nYour custom theme song has been created. Listen to both versions and let us know which one you want:\n\n{player_link}\n\n- Version A\n- Version B\n- Or both versions\n\nOnce you make your choice, we'll send you to the booth to finalize payment and deliver your files.\n\nThanks for stopping by!\n\n— {current_app.config['BOOTH_NAME']}"
|
|
try:
|
|
send_email(req['email'], 'Your custom theme song is ready — listen and pick your version', body, inline_images=build_signature_images())
|
|
update_request(rid, preview_sent_at=now_utc(), status='songs_uploaded')
|
|
flash('Preview email sent.', 'success')
|
|
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':
|
|
flash('Customer must approve a version before you can mark paid or deliver.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
payment_ref = request.form.get('square_payment_ref', '').strip()
|
|
if not payment_ref:
|
|
flash('Square payment reference is required.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
# Build list of selected files from checkboxes.
|
|
selected = request.form.getlist('deliver_file')
|
|
if not selected:
|
|
flash('Select at least one file to deliver.', 'error')
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
# Record payment immediately so the reference is preserved even if email fails.
|
|
update_request(rid, square_payment_ref=payment_ref, status='delivered')
|
|
|
|
attachments = []
|
|
for path in selected:
|
|
p = Path(path)
|
|
if p.exists():
|
|
attachments.append((str(p), p.name))
|
|
|
|
# Compute 3-month expiry date for any stems share link.
|
|
from datetime import datetime, timedelta
|
|
expiry_date = (datetime.utcnow() + timedelta(days=90)).strftime('%B %d, %Y')
|
|
|
|
player_link = f"{current_app.config['PUBLIC_BASE_URL']}/play/{req['player_token']}"
|
|
|
|
# Format lyrics the same way they appear on the customer player page:
|
|
# skip section headers like [Chorus] and blank lines, keep stanza breaks.
|
|
lyrics_section = ''
|
|
if req.get('suno_lyrics'):
|
|
lyric_lines = []
|
|
for line in req['suno_lyrics'].splitlines():
|
|
stripped = line.strip()
|
|
if stripped and not (stripped.startswith('[') and stripped.endswith(']')):
|
|
lyric_lines.append(stripped)
|
|
elif not stripped and lyric_lines and lyric_lines[-1] != '':
|
|
lyric_lines.append('')
|
|
if lyric_lines:
|
|
lyrics_section = '\n'.join(['', '---', 'Song Lyrics', '---', ''] + lyric_lines + [''])
|
|
|
|
body_lines = [
|
|
f"Hi {req['name']},",
|
|
"",
|
|
"Thanks for your payment! Your selected song(s) are attached to this email.",
|
|
f"You can also keep streaming them here: {player_link}",
|
|
]
|
|
if lyrics_section:
|
|
body_lines.append(lyrics_section)
|
|
body_lines += [
|
|
"",
|
|
"Enjoy!",
|
|
"",
|
|
f"— {current_app.config['BOOTH_NAME']}"
|
|
]
|
|
if req.get('stems_link'):
|
|
body_lines.insert(4, f"Your stems / extras are available here: {req['stems_link']}")
|
|
body_lines.insert(5, f"This share link expires on {expiry_date} (3 months from today). Please download before then.")
|
|
body_lines.insert(6, "")
|
|
body = '\n'.join(body_lines)
|
|
try:
|
|
send_email(req['email'], 'Your theme song files are here!', body, attachments=attachments, inline_images=build_signature_images())
|
|
update_request(rid, delivery_sent_at=now_utc())
|
|
flash('Delivery email sent with MP3 attachments.', 'success')
|
|
except Exception as e:
|
|
flash(f'Payment recorded, but delivery email failed: {e}', 'error')
|
|
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
return render_template(
|
|
'admin/request.html',
|
|
req=req,
|
|
style_parts=parse_style_genre(req.get('style_genre')),
|
|
decades=DECADES,
|
|
genres=MUSIC_GENRES,
|
|
statuses=STATUS_LABELS,
|
|
file_exists=file_exists,
|
|
basename=basename,
|
|
extra_files=extra_files,
|
|
callback_url=build_prompt_callback_url(rid),
|
|
revision_history=list_revision_history(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."""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
req = get_request_by_id(rid)
|
|
if not req:
|
|
abort(404)
|
|
|
|
# Delete uploaded files if they exist.
|
|
for field in ('song_a_path', 'song_b_path'):
|
|
path = req.get(field)
|
|
if path and Path(path).exists():
|
|
try:
|
|
Path(path).unlink()
|
|
except OSError:
|
|
pass
|
|
# Remove empty upload directory.
|
|
upload_dir = Path(current_app.config['UPLOAD_FOLDER']) / str(rid)
|
|
if upload_dir.exists():
|
|
try:
|
|
upload_dir.rmdir()
|
|
except OSError:
|
|
pass
|
|
|
|
delete_request(rid)
|
|
flash(f'Request #{rid} deleted.', 'success')
|
|
return redirect(url_for('admin_dashboard'))
|
|
|
|
|
|
@app.route('/admin/settings', methods=['GET', 'POST'])
|
|
def admin_settings():
|
|
"""
|
|
Settings / maintenance page for operators.
|
|
GET -> show database health, statistics, disk usage, runtime settings forms,
|
|
booth open/closed switch, SMTP/email config, MP3 metadata defaults, backup/restore, and reset.
|
|
POST -> handle one of: fix_db, reset_system, save_max_revisions, save_metadata,
|
|
save_email_config, send_test_email, save_refresh, save_booth_open, download_db, restore_db.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
|
|
db_path = Path(current_app.config['DATABASE'])
|
|
upload_root = Path(current_app.config['UPLOAD_FOLDER'])
|
|
|
|
# Compute database stats.
|
|
db_size = db_path.stat().st_size if db_path.exists() else 0
|
|
all_requests = list_requests()
|
|
total_records = len(all_requests)
|
|
status_counts = {}
|
|
for req in all_requests:
|
|
status_counts[req['status']] = status_counts.get(req['status'], 0) + 1
|
|
|
|
# Load persistent runtime settings (max_revisions overrides env var if set).
|
|
runtime_settings = load_booth_settings()
|
|
current_max_revisions = get_max_revisions()
|
|
current_refresh_seconds = runtime_settings.get('refresh_seconds', 10)
|
|
booth_open = runtime_settings.get('booth_open', True)
|
|
|
|
# Effective email config to show in the form (non-sensitive only; password left blank).
|
|
email_form = {
|
|
'smtp_host': runtime_settings.get('smtp_host', current_app.config['SMTP_HOST']),
|
|
'smtp_port': runtime_settings.get('smtp_port', str(current_app.config['SMTP_PORT'])),
|
|
'smtp_user': runtime_settings.get('smtp_user', current_app.config['SMTP_USER']),
|
|
'smtp_from': runtime_settings.get('smtp_from', current_app.config['SMTP_FROM']),
|
|
'smtp_pass_set': bool(runtime_settings.get('smtp_pass', '')),
|
|
}
|
|
|
|
# Hermes API key state for the settings page.
|
|
hermes_key = get_hermes_api_key()
|
|
hermes_key_masked = mask_api_key(hermes_key)
|
|
hermes_key_set = bool(hermes_key)
|
|
hermes_key_just_generated = session.pop('hermes_key_just_generated', None)
|
|
|
|
# Compute upload folder stats.
|
|
total_upload_size = 0
|
|
upload_file_count = 0
|
|
request_dir_count = 0
|
|
if upload_root.exists():
|
|
for entry in upload_root.iterdir():
|
|
if entry.is_dir():
|
|
request_dir_count += 1
|
|
for f in entry.iterdir():
|
|
if f.is_file():
|
|
total_upload_size += f.stat().st_size
|
|
upload_file_count += 1
|
|
elif entry.is_file():
|
|
total_upload_size += entry.stat().st_size
|
|
upload_file_count += 1
|
|
|
|
def format_bytes(n):
|
|
for unit in ['B', 'KB', 'MB', 'GB']:
|
|
if n < 1024:
|
|
return f"{n:.2f} {unit}"
|
|
n /= 1024
|
|
return f"{n:.2f} TB"
|
|
|
|
# Health check: verify expected columns exist and expected tables exist.
|
|
expected_cols = {
|
|
'id', 'created_at', 'name', 'email', 'hobbies', 'notable_facts',
|
|
'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', 'stems_interest'
|
|
}
|
|
expected_tables = {'requests', 'revision_history'}
|
|
health = {'ok': True, 'missing_columns': [], 'missing_tables': [], 'message': 'Database schema looks good.'}
|
|
try:
|
|
db = get_db()
|
|
cur = db.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
existing_tables = {row['name'] for row in cur.fetchall()}
|
|
missing_tables = sorted(expected_tables - existing_tables)
|
|
|
|
cur = db.execute('PRAGMA table_info(requests)')
|
|
existing_cols = {row['name'] for row in cur.fetchall()}
|
|
missing_cols = sorted(expected_cols - existing_cols)
|
|
|
|
if missing_tables or missing_cols:
|
|
parts = []
|
|
if missing_tables:
|
|
parts.append(f"missing tables: {', '.join(missing_tables)}")
|
|
if missing_cols:
|
|
parts.append(f"missing columns: {', '.join(missing_cols)}")
|
|
health = {'ok': False, 'missing_columns': missing_cols, 'missing_tables': missing_tables, 'message': 'Database schema issues: ' + '; '.join(parts)}
|
|
except Exception as e:
|
|
health = {'ok': False, 'missing_columns': [], 'missing_tables': [], 'message': f'Could not inspect database: {e}'}
|
|
|
|
if request.method == 'POST':
|
|
action = request.form.get('action')
|
|
|
|
if action == 'fix_db':
|
|
# Attempt to create missing tables and add missing columns via ALTER TABLE.
|
|
try:
|
|
db = get_db()
|
|
db.executescript(SCHEMA)
|
|
if not health['ok'] and health['missing_columns']:
|
|
for col in health['missing_columns']:
|
|
# Default to TEXT columns; adequate for current schema.
|
|
db.execute(f'ALTER TABLE requests ADD COLUMN {col} TEXT')
|
|
flash(f'Created missing tables and added columns: {", ".join(health["missing_columns"])}. Please refresh the page.', 'success')
|
|
else:
|
|
flash('Database schema is up to date.', 'success')
|
|
db.commit()
|
|
except Exception as e:
|
|
flash(f'Failed to fix database: {e}', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'reset_system':
|
|
# Same nuclear reset logic as the old /admin/reset endpoint.
|
|
if upload_root.exists():
|
|
for entry in upload_root.iterdir():
|
|
try:
|
|
if entry.is_file():
|
|
entry.unlink()
|
|
elif entry.is_dir():
|
|
shutil.rmtree(entry)
|
|
except OSError:
|
|
pass
|
|
reset_all_requests()
|
|
flash('System reset complete. All orders and files have been cleared.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_max_revisions':
|
|
# Update the MAX_REVISIONS config from the settings form.
|
|
try:
|
|
val = int(request.form.get('max_revisions', '2').strip())
|
|
if val < 0:
|
|
raise ValueError
|
|
cfg = load_booth_settings()
|
|
cfg['max_revisions'] = val
|
|
save_booth_settings(cfg)
|
|
flash(f'Maximum revisions set to {val}.', 'success')
|
|
except ValueError:
|
|
flash('Invalid revision limit. Please enter a non-negative number.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_metadata':
|
|
# Update MP3 metadata defaults from the settings form.
|
|
# Store empty strings (not None) so fields repopulate correctly on reload.
|
|
cfg = load_booth_settings()
|
|
for key in ('artist', 'album', 'year', 'comment'):
|
|
cfg[key] = request.form.get(key, '').strip()
|
|
save_booth_settings(cfg)
|
|
flash('MP3 metadata defaults saved.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_email_config':
|
|
# Update SMTP settings from the settings form. Password is encrypted.
|
|
# Empty values are stored as empty strings so the form repopulates.
|
|
cfg = load_booth_settings()
|
|
cfg['smtp_host'] = request.form.get('smtp_host', '').strip()
|
|
cfg['smtp_port'] = request.form.get('smtp_port', '').strip()
|
|
cfg['smtp_user'] = request.form.get('smtp_user', '').strip()
|
|
cfg['smtp_from'] = request.form.get('smtp_from', '').strip()
|
|
new_pass = request.form.get('smtp_pass', '').strip()
|
|
# Only overwrite the stored password if a new value was provided.
|
|
if new_pass:
|
|
cfg['smtp_pass'] = encrypt_value(new_pass)
|
|
save_booth_settings(cfg)
|
|
flash('Email (SMTP) settings saved. Password stored encrypted.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'send_test_email':
|
|
# Send a test email to the address provided in the form.
|
|
test_to = request.form.get('test_email_address', '').strip()
|
|
if not test_to:
|
|
flash('Enter a test email address first.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
try:
|
|
body = "Hi,\n\nThis is a test email from the Trollgorithm Theme Song Booth. If you're seeing this, SMTP is configured correctly."
|
|
send_email(test_to, 'SMTP Test from Theme Song Booth', body, inline_images=build_signature_images())
|
|
flash(f'Test email sent to {test_to}.', 'success')
|
|
except Exception as e:
|
|
flash(f'Failed to send test email: {e}', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_refresh':
|
|
# Update dashboard auto-refresh interval.
|
|
val = request.form.get('refresh_seconds', '10').strip()
|
|
if val not in ('0', '10', '20', '30'):
|
|
val = '10'
|
|
cfg = load_booth_settings()
|
|
cfg['refresh_seconds'] = int(val)
|
|
save_booth_settings(cfg)
|
|
flash(f'Dashboard auto-refresh set to {val} seconds.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_kiosk_cycle':
|
|
# Update kiosk slide cycle interval. Special values:
|
|
# -1 = show QR only, 0 = show pricing only, 5+ = cycle every N seconds.
|
|
raw = request.form.get('kiosk_cycle_seconds', '10').strip()
|
|
try:
|
|
val = int(raw)
|
|
except ValueError:
|
|
val = 10
|
|
if val not in (-1, 0) and val < 5:
|
|
val = 5
|
|
cfg = load_booth_settings()
|
|
cfg['kiosk_cycle_seconds'] = val
|
|
save_booth_settings(cfg)
|
|
if val == -1:
|
|
label = 'QR code only'
|
|
elif val == 0:
|
|
label = 'pricing only'
|
|
else:
|
|
label = f'cycle every {val} seconds'
|
|
flash(f'Kiosk mode set to {label}.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_booth_open':
|
|
# Toggle whether the public request form is accepting submissions.
|
|
cfg = load_booth_settings()
|
|
cfg['booth_open'] = request.form.get('booth_open', '1') == '1'
|
|
save_booth_settings(cfg)
|
|
state = 'open' if cfg['booth_open'] else 'closed'
|
|
flash(f'Booth is now {state}.', 'success')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'download_db':
|
|
# Send the SQLite database file as a download.
|
|
if db_path.exists():
|
|
return send_file(str(db_path), as_attachment=True, download_name='theme-song-booth.db')
|
|
flash('Database file not found.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'download_uploads_zip':
|
|
# Zip all files under UPLOAD_FOLDER and send as a download.
|
|
import zipfile
|
|
zip_path = db_path.with_suffix('.uploads-' + str(int(time.time())) + '.zip')
|
|
try:
|
|
with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zf:
|
|
if upload_root.exists():
|
|
for entry in upload_root.rglob('*'):
|
|
if entry.is_file():
|
|
zf.write(str(entry), str(entry.relative_to(upload_root)))
|
|
return send_file(str(zip_path), as_attachment=True, download_name='theme-song-booth-uploads.zip')
|
|
except Exception as e:
|
|
flash(f'Failed to create uploads ZIP: {e}', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
finally:
|
|
if zip_path.exists():
|
|
zip_path.unlink()
|
|
|
|
elif action == 'restore_db':
|
|
# Replace the current database file with an uploaded SQLite backup.
|
|
file_obj = request.files.get('db_backup')
|
|
if not file_obj or file_obj.filename == '':
|
|
flash('No database backup file selected.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
backup_path = db_path.with_suffix('.backup-restore')
|
|
try:
|
|
# Stream uploaded file directly to disk to avoid memory issues with large DBs.
|
|
file_obj.save(backup_path)
|
|
# Quick sanity check: try to open as SQLite and query sqlite_master.
|
|
import sqlite3
|
|
conn = sqlite3.connect(str(backup_path))
|
|
conn.execute("SELECT name FROM sqlite_master WHERE type='table'")
|
|
conn.close()
|
|
# Replace old database with backup.
|
|
old_backup = db_path.with_suffix('.backup-' + str(int(time.time())))
|
|
db_path.rename(old_backup)
|
|
backup_path.rename(db_path)
|
|
flash('Database restored successfully. Old database kept at ' + old_backup.name, 'success')
|
|
except Exception as e:
|
|
if backup_path.exists():
|
|
backup_path.unlink()
|
|
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')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
elif action == 'save_ntfy':
|
|
# Update ntfy push notification server/topic/access token from the settings form.
|
|
cfg = load_booth_settings()
|
|
cfg['ntfy_server'] = request.form.get('ntfy_server', '').strip().rstrip('/')
|
|
cfg['ntfy_topic'] = request.form.get('ntfy_topic', '').strip()
|
|
new_token = request.form.get('ntfy_token', '').strip()
|
|
if new_token:
|
|
cfg['ntfy_token'] = encrypt_value(new_token)
|
|
save_booth_settings(cfg)
|
|
flash('ntfy notification settings saved.', 'success')
|
|
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()
|
|
if not ntfy['server'] or not ntfy['topic']:
|
|
flash('Configure ntfy server and topic first.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
ok = send_ntfy('This is a test push notification from the Trollgorithm Theme Song Booth.', title='Test Notification', priority='high', tags='test_tube')
|
|
if ok:
|
|
flash('Test ntfy notification sent.', 'success')
|
|
else:
|
|
flash('Failed to send ntfy test notification. Check server, topic, and access token.', 'error')
|
|
return redirect(url_for('admin_settings'))
|
|
|
|
return render_template(
|
|
'admin/settings.html',
|
|
health=health,
|
|
db_size=format_bytes(db_size),
|
|
total_records=total_records,
|
|
status_counts=status_counts,
|
|
statuses=STATUS_LABELS,
|
|
upload_file_count=upload_file_count,
|
|
upload_dir_count=request_dir_count,
|
|
upload_size=format_bytes(total_upload_size),
|
|
db_path=str(db_path),
|
|
upload_path=str(upload_root),
|
|
current_max_revisions=current_max_revisions,
|
|
current_refresh_seconds=current_refresh_seconds,
|
|
current_kiosk_cycle_seconds=get_kiosk_cycle_seconds(),
|
|
booth_open=booth_open,
|
|
email_form=email_form,
|
|
metadata={
|
|
'artist': runtime_settings.get('artist', ''),
|
|
'album': runtime_settings.get('album', ''),
|
|
'year': runtime_settings.get('year', ''),
|
|
'comment': runtime_settings.get('comment', ''),
|
|
},
|
|
hermes_key_masked=hermes_key_masked,
|
|
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,
|
|
)
|
|
|
|
|
|
@app.route('/admin/reset', methods=['POST'])
|
|
def admin_reset_system():
|
|
"""
|
|
Nuclear reset for the start of an event.
|
|
Deletes all database rows and all files/directories under UPLOAD_FOLDER.
|
|
Requires clicking through a browser confirm dialog.
|
|
"""
|
|
redir = require_admin()
|
|
if redir:
|
|
return redir
|
|
|
|
upload_root = Path(current_app.config['UPLOAD_FOLDER'])
|
|
if upload_root.exists():
|
|
for entry in upload_root.iterdir():
|
|
try:
|
|
if entry.is_file():
|
|
entry.unlink()
|
|
elif entry.is_dir():
|
|
shutil.rmtree(entry)
|
|
except OSError:
|
|
pass
|
|
|
|
reset_all_requests()
|
|
flash('System reset complete. All orders and files have been cleared.', 'success')
|
|
return redirect(url_for('admin_dashboard'))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI and entry point
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@app.cli.command('init-db')
|
|
def init_db_command():
|
|
"""Flask CLI command: flask --app app init-db"""
|
|
init_db()
|
|
print('Database initialized.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
# Development-only entry point. Production uses gunicorn (see Dockerfile).
|
|
app.run(debug=True, host='0.0.0.0')
|
|
|
|
|
|
# Ensure the database file and expected tables exist when the app is imported by
|
|
# gunicorn in production. init_db() uses CREATE TABLE IF NOT EXISTS, so this is
|
|
# safe to run on every startup without wiping data.
|
|
with app.app_context():
|
|
try:
|
|
init_db()
|
|
except Exception:
|
|
# 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)
|