Add admin Settings page with DB health check, stats, and system reset

This commit is contained in:
Troll (Hermes Agent) 2026-08-01 20:55:20 +00:00
parent 2b960bd3d5
commit 51d1b4a4d9
3 changed files with 333 additions and 5 deletions

117
app.py
View file

@ -19,6 +19,7 @@ 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
@ -38,7 +39,8 @@ 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
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
# ---------------------------------------------------------------------------
# App setup
@ -425,6 +427,119 @@ def admin_delete_request(rid):
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
# 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', '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'
}
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'))
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)
)
@app.route('/admin/reset', methods=['POST'])
def admin_reset_system():
"""