735 lines
28 KiB
Python
735 lines
28 KiB
Python
"""
|
|
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/settings -> health check, DB stats, disk usage, system reset
|
|
- /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
|
|
import ssl
|
|
from email.message import EmailMessage
|
|
from pathlib import Path
|
|
|
|
import json
|
|
|
|
# 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, get_db
|
|
from models import SCHEMA
|
|
|
|
# Audio metadata import
|
|
from mutagen.mp3 import MP3
|
|
from mutagen.easyid3 import EasyID3
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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',
|
|
'songs_uploaded': 'Songs Uploaded — Awaiting Approval',
|
|
'revisions_requested': 'Revisions Requested',
|
|
'awaiting_payment': 'Awaiting Payment',
|
|
'paid': 'Paid',
|
|
'delivered': 'Delivered',
|
|
}
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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, genre, comment).
|
|
:param request_id: database ID of the request
|
|
:param file_obj: Flask FileStorage from request.files
|
|
:param version: 'a' or 'b'
|
|
:param song_title: title to write into the MP3 title tag
|
|
: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):
|
|
flash('Only MP3 files are allowed.', 'error')
|
|
return None
|
|
|
|
# Use Werkzeug's secure_filename to strip unsafe characters.
|
|
original = secure_filename(file_obj.filename)
|
|
# Prefix with A or B so the operator knows which version it is.
|
|
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 or overwrite common ID3 tags on an MP3 file using values from
|
|
runtime booth settings. The saved Suno title is written to the Title tag.
|
|
"""
|
|
cfg = load_booth_settings()
|
|
try:
|
|
audio = MP3(path)
|
|
# Ensure EasyID3 wrapper is available for the file.
|
|
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'])
|
|
if cfg.get('genre'):
|
|
tags['genre'] = cfg['genre']
|
|
|
|
audio.save()
|
|
|
|
# Comment requires a real ID3 frame, not EasyID3.
|
|
if cfg.get('comment'):
|
|
from mutagen.id3 import ID3, COMM
|
|
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.save()
|
|
except Exception as e:
|
|
# Don't fail the upload just because tagging failed; log/flash only if inside request context.
|
|
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 load_booth_settings():
|
|
"""Load persistent runtime settings from JSON file inside the upload parent."""
|
|
cfg_path = Path(current_app.config['UPLOAD_FOLDER']).parent / 'booth_settings.json'
|
|
if cfg_path.exists():
|
|
try:
|
|
return json.loads(cfg_path.read_text())
|
|
except (json.JSONDecodeError, OSError):
|
|
pass
|
|
return {}
|
|
|
|
|
|
def save_booth_settings(settings):
|
|
"""Persist runtime settings to JSON file."""
|
|
cfg_path = Path(current_app.config['UPLOAD_FOLDER']).parent / 'booth_settings.json'
|
|
try:
|
|
cfg_path.write_text(json.dumps(settings, indent=2))
|
|
except OSError as e:
|
|
flash(f'Warning: could not save settings: {e}', 'error')
|
|
|
|
|
|
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')
|
|
|
|
msg = EmailMessage()
|
|
msg['From'] = cfg['SMTP_FROM']
|
|
msg['To'] = to
|
|
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:
|
|
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)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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(),
|
|
email=request.form.get('email', '').strip(),
|
|
hobbies=request.form.get('hobbies', '').strip(),
|
|
notable_facts=request.form.get('notable_facts', '').strip(),
|
|
style_genre=request.form.get('style_genre', '').strip(),
|
|
extra_requests=request.form.get('extra_requests', '').strip(),
|
|
vocal_gender=request.form.get('vocal_gender', '').strip(),
|
|
)
|
|
flash('Your request has been submitted! Check your email soon.', 'success')
|
|
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)
|
|
|
|
# Load runtime max revisions setting.
|
|
runtime_settings = load_booth_settings()
|
|
max_revisions = runtime_settings.get('max_revisions', current_app.config['MAX_REVISIONS'])
|
|
revisions_left = max(0, max_revisions - (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.
|
|
"""
|
|
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())
|
|
|
|
# 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.
|
|
"""
|
|
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.
|
|
runtime_settings = load_booth_settings()
|
|
max_revisions = runtime_settings.get('max_revisions', current_app.config['MAX_REVISIONS'])
|
|
current_count = 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'])
|
|
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)
|
|
req[field] = str(archived)
|
|
except OSError:
|
|
pass
|
|
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=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('/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)
|
|
if version not in ('a', 'b'):
|
|
abort(404)
|
|
field = f'song_{version}_path'
|
|
path = req.get(field)
|
|
if not path or not Path(path).exists():
|
|
abort(404)
|
|
return send_from_directory(Path(path).parent, Path(path).name)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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.
|
|
"""
|
|
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)
|
|
|
|
|
|
@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
|
|
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 == '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 == '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)
|
|
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 == 'mark_paid_deliver':
|
|
# Finalize: record Square payment ref, attach approved MP3s, email customer.
|
|
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))
|
|
|
|
attachments = []
|
|
for path in selected:
|
|
p = Path(path)
|
|
if p.exists():
|
|
attachments.append((str(p), p.name))
|
|
|
|
player_link = f"{current_app.config['PUBLIC_BASE_URL']}/play/{req['player_token']}"
|
|
body = f"Hi {req['name']},\n\nThanks for your payment! Your selected song(s) are attached to this email.\n\nYou can also keep streaming them here: {player_link}\n\nEnjoy!\n\n— {current_app.config['BOOTH_NAME']}"
|
|
try:
|
|
send_email(req['email'], 'Your theme song files are here!', body, attachments=attachments)
|
|
update_request(rid, square_payment_ref=payment_ref, delivery_sent_at=now_utc(), status='delivered')
|
|
flash('Delivery email sent with MP3 attachments.', 'success')
|
|
except Exception as e:
|
|
flash(f'Failed to send delivery email: {e}', 'error')
|
|
|
|
return redirect(url_for('admin_request', rid=rid))
|
|
|
|
return render_template('admin/request.html', req=req, statuses=STATUS_LABELS, file_exists=file_exists, basename=basename, extra_files=extra_files)
|
|
|
|
|
|
@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, and reset button.
|
|
POST -> either run a health check/fix or reset the system.
|
|
"""
|
|
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).
|
|
cfg_path = Path(current_app.config['UPLOAD_FOLDER']).parent / 'booth_settings.json'
|
|
runtime_settings = load_booth_settings()
|
|
current_max_revisions = runtime_settings.get('max_revisions', current_app.config['MAX_REVISIONS'])
|
|
|
|
# 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.
|
|
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'
|
|
}
|
|
health = {'ok': True, 'missing_columns': [], 'message': 'Database schema looks good.'}
|
|
try:
|
|
db = get_db()
|
|
cur = db.execute('PRAGMA table_info(requests)')
|
|
existing_cols = {row['name'] for row in cur.fetchall()}
|
|
missing = sorted(expected_cols - existing_cols)
|
|
if missing:
|
|
health = {'ok': False, 'missing_columns': missing, 'message': f'Missing columns: {", ".join(missing)}'}
|
|
except Exception as e:
|
|
health = {'ok': False, 'missing_columns': [], 'message': f'Could not inspect table: {e}'}
|
|
|
|
if request.method == 'POST':
|
|
action = request.form.get('action')
|
|
|
|
if action == 'fix_db':
|
|
# Attempt to add missing columns via ALTER TABLE.
|
|
if not health['ok'] and health['missing_columns']:
|
|
try:
|
|
db = get_db()
|
|
for col in health['missing_columns']:
|
|
# Default to TEXT columns; adequate for current schema.
|
|
db.execute(f'ALTER TABLE requests ADD COLUMN {col} TEXT')
|
|
db.commit()
|
|
flash(f'Added missing columns: {", ".join(health["missing_columns"])}. Please refresh the page.', 'success')
|
|
except Exception as e:
|
|
flash(f'Failed to fix database: {e}', 'error')
|
|
else:
|
|
flash('No columns need fixing.', 'success')
|
|
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.
|
|
cfg = load_booth_settings()
|
|
cfg['artist'] = request.form.get('artist', '').strip() or None
|
|
cfg['album'] = request.form.get('album', '').strip() or None
|
|
cfg['year'] = request.form.get('year', '').strip() or None
|
|
cfg['genre'] = request.form.get('genre', '').strip() or None
|
|
cfg['comment'] = request.form.get('comment', '').strip() or None
|
|
save_booth_settings(cfg)
|
|
flash('MP3 metadata defaults saved.', 'success')
|
|
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,
|
|
metadata=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')
|