cedrick2711
bfb25f4d16
CI - Security, Lint & Tests / validate (push) Failing after 1m3s
ajout de compteur d'heure disponible ajout du calendrier de dispo dans les profils publics
396 lines
14 KiB
Python
396 lines
14 KiB
Python
"""Admin panel routes for system-level management.
|
|
|
|
Provides backup/restore, tryout open/close toggling, season lifecycle
|
|
management, team wiping, and audit log viewing. All routes are restricted
|
|
to administrators.
|
|
"""
|
|
|
|
import os
|
|
from datetime import datetime
|
|
|
|
from flask import (
|
|
Blueprint, render_template, redirect, url_for, flash, request,
|
|
send_file,
|
|
)
|
|
from flask_login import login_required, current_user
|
|
|
|
from app.extensions import db
|
|
from app.models import (
|
|
Admin, User, Tryout, OrgTeam, TeamPlayer, TeamMatch,
|
|
TeamMatchParticipant, AppSettings, BackupRecord, AuditLog,
|
|
)
|
|
from app.supporting_scripts.backup import (
|
|
BACKUP_DIR, BackupError, backup_database, backup_documents,
|
|
create_backup_dir, parse_database_url, verify_backup,
|
|
)
|
|
|
|
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
|
|
|
|
|
|
def require_admin():
|
|
"""Return True if current user is an Admin, else flash and redirect."""
|
|
if isinstance(current_user, Admin):
|
|
return True
|
|
flash('Only the president can access the admin panel.', 'danger')
|
|
return False
|
|
|
|
|
|
def _client_ip():
|
|
"""Best-effort client IP for audit logging."""
|
|
if request.headers.get('X-Forwarded-For'):
|
|
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
|
|
return request.remote_addr
|
|
|
|
|
|
def _log(action, details=None):
|
|
"""Record an audit log entry for the current user."""
|
|
AuditLog.record(
|
|
user_id=current_user.id,
|
|
action=action,
|
|
details=details,
|
|
ip_address=_client_ip(),
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Dashboard
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@admin_bp.route('')
|
|
@login_required
|
|
def dashboard():
|
|
"""Render the admin panel dashboard."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
stats = {
|
|
'total_users': User.query.count(),
|
|
'total_players': User.query.filter_by(role='player').count(),
|
|
'total_org_teams': OrgTeam.query.count(),
|
|
'active_tryouts': Tryout.query.filter_by(status='in_progress').count(),
|
|
'upcoming_tryouts': Tryout.query.filter_by(status='upcoming').count(),
|
|
}
|
|
|
|
settings = {
|
|
'tryouts_open': AppSettings.get_bool('tryouts_open', default=True),
|
|
'document_uploads_enabled': AppSettings.get_bool('document_uploads_enabled', default=True),
|
|
'season_active': AppSettings.get_bool('season_active', default=False),
|
|
'season_name': AppSettings.get('season_name') or 'Not set',
|
|
'season_start': AppSettings.get('season_start') or 'Not set',
|
|
'season_end': AppSettings.get('season_end') or 'Not set',
|
|
}
|
|
|
|
backups = BackupRecord.query.order_by(BackupRecord.created_at.desc()).limit(20).all()
|
|
audit_logs = AuditLog.query.order_by(AuditLog.created_at.desc()).limit(20).all()
|
|
|
|
return render_template(
|
|
'pages/admin.html',
|
|
stats=stats,
|
|
settings=settings,
|
|
backups=backups,
|
|
audit_logs=audit_logs,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Backups
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _create_backup_record(backup_type='manual', notes=None):
|
|
"""Run a database backup and return a BackupRecord, or raise BackupError."""
|
|
create_backup_dir()
|
|
conn = parse_database_url(os.getenv('DATABASE_URL'))
|
|
file_path = backup_database(conn)
|
|
size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
|
|
filename = os.path.basename(file_path)
|
|
|
|
record = BackupRecord(
|
|
filename=filename,
|
|
file_path=file_path,
|
|
size_bytes=size,
|
|
backup_type=backup_type,
|
|
notes=notes,
|
|
created_by_id=current_user.id,
|
|
)
|
|
db.session.add(record)
|
|
db.session.commit()
|
|
return record
|
|
|
|
|
|
@admin_bp.route('/backup/create', methods=['POST'])
|
|
@login_required
|
|
def create_backup():
|
|
"""Create a manual database backup."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
notes = request.form.get('notes', '').strip() or None
|
|
try:
|
|
record = _create_backup_record(backup_type='manual', notes=notes)
|
|
except BackupError as e:
|
|
flash(f'Backup failed: {e}', 'danger')
|
|
_log('backup_failed', f'Error: {e}')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
_log('backup_created', f'File: {record.filename} ({record.size_bytes} bytes)')
|
|
flash(f'Backup created successfully: {record.filename}', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
@admin_bp.route('/backup/<int:backup_id>/download')
|
|
@login_required
|
|
def download_backup(backup_id):
|
|
"""Download a backup file."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
record = BackupRecord.query.get_or_404(backup_id)
|
|
if not os.path.exists(record.file_path):
|
|
flash('Backup file is missing from disk.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
_log('backup_downloaded', f'File: {record.filename}')
|
|
return send_file(record.file_path, as_attachment=True, download_name=record.filename)
|
|
|
|
|
|
@admin_bp.route('/backup/<int:backup_id>/delete', methods=['POST'])
|
|
@login_required
|
|
def delete_backup(backup_id):
|
|
"""Delete a backup file and its record."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
record = BackupRecord.query.get_or_404(backup_id)
|
|
if os.path.exists(record.file_path):
|
|
try:
|
|
os.remove(record.file_path)
|
|
except OSError as e:
|
|
flash(f'Could not remove backup file: {e}', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
_log('backup_deleted', f'File: {record.filename}')
|
|
db.session.delete(record)
|
|
db.session.commit()
|
|
flash(f'Backup {record.filename} deleted.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
@admin_bp.route('/backup/<int:backup_id>/restore', methods=['POST'])
|
|
@login_required
|
|
def restore_backup(backup_id):
|
|
"""Restore a selected backup.
|
|
|
|
A safety backup of the current state is created first, then the
|
|
selected dump is restored via pg_restore.
|
|
"""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
record = BackupRecord.query.get_or_404(backup_id)
|
|
if not os.path.exists(record.file_path):
|
|
flash('Backup file is missing from disk.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
# Create a safety backup of the current state before restoring.
|
|
try:
|
|
safety = _create_backup_record(
|
|
backup_type='pre_restore',
|
|
notes='Automatic safety backup before restoring ' + record.filename,
|
|
)
|
|
except BackupError as e:
|
|
flash(f'Could not create safety backup, restore aborted: {e}', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
# Restore using pg_restore
|
|
import subprocess
|
|
from app.supporting_scripts.backup import (
|
|
PG_RESTORE, dump_environment, parse_database_url,
|
|
)
|
|
|
|
conn = parse_database_url(os.getenv('DATABASE_URL'))
|
|
cmd = [
|
|
PG_RESTORE,
|
|
'--host', conn['host'],
|
|
'--port', conn['port'],
|
|
'--username', conn['user'],
|
|
'--dbname', conn['dbname'],
|
|
'--clean', '--if-exists', '--no-owner',
|
|
record.file_path,
|
|
]
|
|
try:
|
|
result = subprocess.run(
|
|
cmd,
|
|
env=dump_environment(conn),
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=900,
|
|
)
|
|
if result.returncode != 0:
|
|
raise RuntimeError(result.stderr.strip() or 'pg_restore failed')
|
|
except (subprocess.TimeoutExpired, FileNotFoundError, RuntimeError) as e:
|
|
flash(f'Restore failed: {e}', 'danger')
|
|
_log('backup_restore_failed', f'File: {record.filename}, Error: {e}')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
_log('backup_restored', f'File: {record.filename}')
|
|
flash(
|
|
f'Backup {record.filename} restored successfully. The database has been rolled back.',
|
|
'success',
|
|
)
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tryouts open/close toggle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@admin_bp.route('/toggle-tryouts', methods=['POST'])
|
|
@login_required
|
|
def toggle_tryouts():
|
|
"""Toggle the global tryout open/close state."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
current = AppSettings.get_bool('tryouts_open', default=True)
|
|
new_value = not current
|
|
AppSettings.set_bool('tryouts_open', new_value)
|
|
|
|
state = 'opened' if new_value else 'closed'
|
|
_log('tryouts_toggled', f'Tryouts {state}')
|
|
flash(f'Tryouts are now {state}.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Season lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@admin_bp.route('/season/start', methods=['POST'])
|
|
@login_required
|
|
def start_season():
|
|
"""Begin a new regular season."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
if AppSettings.get_bool('season_active', default=False):
|
|
flash('A season is already active. End it before starting a new one.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
name = request.form.get('season_name', '').strip()
|
|
start_date = request.form.get('season_start', '').strip()
|
|
if not start_date:
|
|
flash('A season start date is required.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
try:
|
|
datetime.strptime(start_date, '%Y-%m-%d')
|
|
except ValueError:
|
|
flash('Invalid season start date format.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
AppSettings.set('season_name', name or 'Untitled Season')
|
|
AppSettings.set('season_start', start_date)
|
|
AppSettings.set('season_end', None)
|
|
AppSettings.set_bool('season_active', True)
|
|
|
|
_log('season_started', f'Season: {name or "Untitled Season"}, Start: {start_date}')
|
|
flash(f'Season "{name or "Untitled Season"}" has begun.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
@admin_bp.route('/season/end', methods=['POST'])
|
|
@login_required
|
|
def end_season():
|
|
"""End the current regular season."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
if not AppSettings.get_bool('season_active', default=False):
|
|
flash('No season is currently active.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
end_date = request.form.get('season_end', '').strip()
|
|
if not end_date:
|
|
flash('A season end date is required.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
try:
|
|
datetime.strptime(end_date, '%Y-%m-%d')
|
|
except ValueError:
|
|
flash('Invalid season end date format.', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
name = AppSettings.get('season_name')
|
|
AppSettings.set('season_end', end_date)
|
|
AppSettings.set_bool('season_active', False)
|
|
|
|
_log('season_ended', f'Season: {name}, End: {end_date}')
|
|
flash(f'Season "{name}" has ended.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
@admin_bp.route('/toggle-document-uploads', methods=['POST'])
|
|
@login_required
|
|
def toggle_document_uploads():
|
|
"""Enable or disable document uploads for non-admin users."""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
current = AppSettings.get_bool('document_uploads_enabled', default=True)
|
|
new_value = not current
|
|
AppSettings.set_bool('document_uploads_enabled', new_value)
|
|
|
|
state = 'enabled' if new_value else 'disabled'
|
|
_log('document_uploads_toggled', f'Document uploads {state}')
|
|
flash(f'Document uploads are now {state}.', 'success')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
|
|
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Wipe teams for new season
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@admin_bp.route('/teams/wipe', methods=['POST'])
|
|
@login_required
|
|
def wipe_teams():
|
|
"""Wipe team rosters and season matches for a new season.
|
|
|
|
Players are removed from org teams (TeamPlayer records), regular-season
|
|
matches and their participants are deleted. OrgTeam structures, coaches,
|
|
and managers are preserved.
|
|
"""
|
|
if not require_admin():
|
|
return redirect(url_for('main.dashboard'))
|
|
|
|
confirm = request.form.get('confirm', '').strip()
|
|
if confirm != 'WIPE':
|
|
flash("Type 'WIPE' in the confirmation box to proceed.", 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
# Safety backup before destructive operation
|
|
try:
|
|
_create_backup_record(
|
|
backup_type='pre_wipe',
|
|
notes='Automatic safety backup before wiping teams',
|
|
)
|
|
except BackupError as e:
|
|
flash(f'Wipe aborted — could not create safety backup: {e}', 'danger')
|
|
return redirect(url_for('admin.dashboard'))
|
|
|
|
roster_count = TeamPlayer.query.count()
|
|
match_count = TeamMatch.query.count()
|
|
|
|
# Delete team match participants first (FK order)
|
|
TeamMatchParticipant.query.delete()
|
|
TeamMatch.query.delete()
|
|
TeamPlayer.query.delete()
|
|
db.session.commit()
|
|
|
|
_log('teams_wiped', f'Removed {roster_count} roster entries and {match_count} season matches')
|
|
flash(
|
|
f'Teams wiped for the new season. Removed {roster_count} roster entries '
|
|
f'and {match_count} regular-season matches.',
|
|
'success',
|
|
)
|
|
return redirect(url_for('admin.dashboard')) |