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

Add comprehensive comments to all source files and expand documentation

Browse source
This commit is contained in:
Troll (Hermes Agent) 2026-07-31 22:23:40 +00:00
parent 9e532a1920
commit b4f0222bd5
16 changed files with 1230 additions and 552 deletions
Download patch file Download diff file Expand all files Collapse all files

151
app.py
View file

@ -1,3 +1,30 @@
"""
app.py
======
Main Flask application for the Theme Song Booth.
This module defines all HTTP routes, helper functions, and the email layer.
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
- /admin/request/<id> -> detail/edit page for a single request
- /admin/request/<id>/delete -> deletes one request and its files
- /admin/reset -> deletes ALL requests and ALL files
"""
# Standard library imports
import os
import shutil
import smtplib
@ -5,16 +32,26 @@ import ssl
from email.message import EmailMessage
from pathlib import Path
# Flask and related imports
from flask import Flask, request, render_template, redirect, url_for, flash, session, send_from_directory, abort, current_app
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
# ---------------------------------------------------------------------------
# App setup
# ---------------------------------------------------------------------------
# Create the Flask app and load configuration from Config class.
app = Flask(__name__)
app.config.from_object(Config)
# 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',
@ -24,27 +61,46 @@ STATUS_LABELS = {
'delivered': 'Delivered',
}
# ---------------- helpers ----------------
# ---------------------------------------------------------------------------
# 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):
"""
Save an uploaded MP3 file for a request.
:param request_id: database ID of the request
:param file_obj: Flask FileStorage from request.files
:param version: 'a' or 'b'
:return: full filesystem path saved, or None on missing/invalid file
"""
if not file_obj or file_obj.filename == '':
return None
if not allowed_file(file_obj.filename):
@ -55,7 +111,15 @@ def save_upload(request_id, file_obj, version):
file_obj.save(p / filename)
return str(p / filename)
def send_email(to, subject, body, attachments=None):
"""
Send an email via SMTP_SSL.
:param to: recipient address
:param subject: email subject
:param body: plain-text body
:param attachments: optional list of (filepath, attachment_name) tuples
"""
cfg = current_app.config
if not cfg['SMTP_PASS']:
raise RuntimeError('SMTP_PASS is not configured')
@ -66,6 +130,7 @@ def send_email(to, subject, body, attachments=None):
msg['Subject'] = subject
msg.set_content(body)
# Attach any MP3 files as audio/mpeg attachments.
if attachments:
for path, name in attachments:
with open(path, 'rb') as f:
@ -76,14 +141,24 @@ def send_email(to, subject, body, attachments=None):
server.login(cfg['SMTP_USER'], cfg['SMTP_PASS'])
server.send_message(msg)
# ---------------- public ----------------
# ---------------------------------------------------------------------------
# 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'])
def request_form():
"""
Public request form.
GET -> shows the form with the banner image.
POST -> creates a database record and redirects to the thanks page.
"""
if request.method == 'POST':
rid = create_request(
name=request.form.get('name', '').strip(),
@ -97,22 +172,34 @@ def request_form():
return redirect(url_for('thanks', rid=rid))
return render_template('request.html')
@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('/play/<token>')
def play(token):
"""
Private player page for a customer.
The token is a cryptographically random URL-safe string generated at request time.
"""
req = get_request_by_token(token)
if not req:
abort(404)
return render_template('player.html', req=req)
@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.
"""
req = get_request_by_token(token)
if not req:
abort(404)
@ -123,28 +210,37 @@ def approve(token):
update_request(req['id'], customer_approved=choice, status='awaiting_payment', approval_notified_at=now_utc())
# alert operator (disabled — admin dashboard is the queue)
# NOTE: Operator email alerts are intentionally disabled. The admin dashboard is the single queue.
# alert_to = current_app.config['ADMIN_ALERT_EMAIL'] or current_app.config['SMTP_FROM']
# if alert_to: ...
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. Store the note and reset status to 'songs_uploaded'
so the operator sees it in the dashboard queue.
"""
req = get_request_by_token(token)
if not req:
abort(404)
note = request.form.get('revision_note', '').strip()
update_request(req['id'], revision_note=note, status='songs_uploaded')
# Revision feedback is stored in the DB and surfaced on the admin dashboard.
# No operator email is sent — the dashboard is the single queue.
# 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('/audio/<token>/<version>.mp3')
def audio(token, version):
"""
Serve an uploaded MP3 file for a specific request token and version ('a' or 'b').
This keeps the files off the public static path and ties them to the private token.
"""
req = get_request_by_token(token)
if not req:
abort(404)
@ -156,10 +252,14 @@ def audio(token, version):
abort(404)
return send_from_directory(Path(path).parent, Path(path).name)
# ---------------- admin ----------------
# ---------------------------------------------------------------------------
# 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':
@ -169,13 +269,20 @@ def admin_login():
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.
"""
redir = require_admin()
if redir:
return redir
@ -183,8 +290,15 @@ def admin_dashboard():
requests = list_requests(status_filter)
return render_template('admin/dashboard.html', requests=requests, statuses=STATUS_LABELS, current_status=status_filter)
@app.route('/admin/request/<int:rid>', methods=['GET', 'POST'])
def admin_request(rid):
"""
Detail/edit page for a single request.
GET -> render the request details and editing forms.
POST -> handle one of four actions:
save_prompt, upload_songs, notify_customer, mark_paid_deliver
"""
redir = require_admin()
if redir:
return redir
@ -192,6 +306,7 @@ def admin_request(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())
@ -202,6 +317,7 @@ def admin_request(rid):
action = request.form.get('action')
if 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(),
@ -211,6 +327,7 @@ def admin_request(rid):
flash('Prompt saved.', 'success')
elif action == 'upload_songs':
# Save uploaded MP3 files for Version A and/or Version B.
a_path = save_upload(rid, request.files.get('song_a'), 'a')
b_path = save_upload(rid, request.files.get('song_b'), 'b')
fields = {}
@ -224,6 +341,7 @@ def admin_request(rid):
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:
@ -237,6 +355,7 @@ def admin_request(rid):
flash(f'Failed to send preview email: {e}', 'error')
elif action == 'mark_paid_deliver':
# Finalize: record Square payment ref, attach approved MP3s, email customer.
if req['customer_approved'] == 'none':
flash('Customer has not approved a version yet.', 'error')
else:
@ -264,8 +383,10 @@ def admin_request(rid):
return render_template('admin/request.html', req=req, statuses=STATUS_LABELS, file_exists=file_exists, basename=basename)
@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
@ -273,7 +394,7 @@ def admin_delete_request(rid):
if not req:
abort(404)
# Delete uploaded files if they exist
# 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():
@ -281,7 +402,7 @@ def admin_delete_request(rid):
Path(path).unlink()
except OSError:
pass
# Remove empty upload directory
# Remove empty upload directory.
upload_dir = Path(current_app.config['UPLOAD_FOLDER']) / str(rid)
if upload_dir.exists():
try:
@ -293,8 +414,14 @@ def admin_delete_request(rid):
flash(f'Request #{rid} deleted.', 'success')
return redirect(url_for('admin_dashboard'))
@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
@ -314,12 +441,18 @@ def admin_reset_system():
flash('System reset complete. All orders and files have been cleared.', 'success')
return redirect(url_for('admin_dashboard'))
# ---------------- init ----------------
# ---------------------------------------------------------------------------
# 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')