"""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 import uuid from datetime import datetime from flask import ( Blueprint, flash, redirect, render_template, request, send_file, url_for, ) from flask_babel import gettext as _ from flask_login import current_user, login_required from marshmallow import ValidationError from werkzeug.utils import secure_filename from app.extensions import db from app.forms import flash_validation_errors from app.models import ( DOCUMENT_KIND_ROLES, DOCUMENT_KINDS, Admin, AppSettings, AuditLog, BackupRecord, Document, DocumentAssignment, DocumentVersion, OrgTeam, TeamMatch, TeamMatchParticipant, TeamPlayer, Tryout, User, ) from app.responders import respond_after_change from app.routes.users._shared import ALLOWED_CONTRACT_EXTENSIONS, pdf_upload_error from app.services.documents import backfill_assignments_for_document from app.storage import GoogleDriveStorageError, open_document, store_master_document from app.supporting_scripts.backup import ( BackupError, backup_database, create_backup_dir, parse_database_url, ) from app.validators import DocumentCreateSchema 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 respond_after_change(url_for('admin.dashboard')) _log('backup_created', f'File: {record.filename} ({record.size_bytes} bytes)') flash(f'Backup created successfully: {record.filename}', 'success') return respond_after_change(url_for('admin.dashboard')) @admin_bp.route('/backup//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 respond_after_change(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//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 respond_after_change(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 respond_after_change(url_for('admin.dashboard')) @admin_bp.route('/backup//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 respond_after_change(url_for('admin.dashboard')) # Create a safety backup of the current state before restoring. try: _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 respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(url_for('admin.dashboard')) try: datetime.strptime(start_date, '%Y-%m-%d') except ValueError: flash('Invalid season start date format.', 'danger') return respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(url_for('admin.dashboard')) try: datetime.strptime(end_date, '%Y-%m-%d') except ValueError: flash('Invalid season end date format.', 'danger') return respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(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 respond_after_change(url_for('admin.dashboard')) # --------------------------------------------------------------------------- # Documents (canonical document store) # --------------------------------------------------------------------------- @admin_bp.route('/documents') @login_required def documents(): """List every document with its published flag and completion counts.""" if not require_admin(): return redirect(url_for('main.dashboard')) documents = Document.query.order_by(Document.kind, Document.title).all() return render_template( 'pages/admin_documents.html', documents=documents, kinds=DOCUMENT_KINDS, kind_roles=DOCUMENT_KIND_ROLES, ) @admin_bp.route('/documents/create', methods=['POST']) @login_required def create_document(): """Create a new canonical document (no file yet; publish one next).""" if not require_admin(): return redirect(url_for('main.dashboard')) try: data = DocumentCreateSchema().load(request.form) except ValidationError as err: flash_validation_errors(err) return respond_after_change(url_for('admin.documents')) document = Document( title=data['title'], kind=data['kind'], created_by=current_user.id, ) db.session.add(document) db.session.commit() _log('document_created', f'{data["kind"]}: {data["title"]}') flash(_('Document "%(title)s" created.', title=data['title']), 'success') return respond_after_change(url_for('admin.documents')) @admin_bp.route('/documents//publish-version', methods=['POST']) @login_required def publish_document_version(doc_id): """Store a new PDF once and record it as the current version.""" if not require_admin(): return redirect(url_for('main.dashboard')) document = db.get_or_404(Document, doc_id) file = request.files.get('version_file') error = pdf_upload_error(file, ALLOWED_CONTRACT_EXTENSIONS) if error: flash(error, 'danger') return respond_after_change(url_for('admin.document_detail', doc_id=document.id)) original_filename = secure_filename(file.filename) stored_filename = f'{uuid.uuid4()}.pdf' try: stored_path = store_master_document(file, stored_filename) except GoogleDriveStorageError: flash( _('Document storage is temporarily unavailable. Please try again later.'), 'danger', ) return respond_after_change(url_for('admin.document_detail', doc_id=document.id)) version = DocumentVersion( document_id=document.id, version=document.next_version_number(), original_filename=original_filename, stored_filename=stored_filename, file_path=stored_path, notes=request.form.get('notes', '').strip() or None, published_by=current_user.id, ) db.session.add(version) db.session.flush() document.current_version_id = version.id db.session.commit() backfill_assignments_for_document(document) _log('document_version_published', f'{document.title} v{version.version}') flash( _( 'Version %(number)s of "%(title)s" published.', number=version.version, title=document.title, ), 'success', ) return respond_after_change(url_for('admin.document_detail', doc_id=document.id)) @admin_bp.route('/documents//toggle-published', methods=['POST']) @login_required def toggle_document_published(doc_id): """Flip whether team members can see and download the document.""" if not require_admin(): return redirect(url_for('main.dashboard')) document = db.get_or_404(Document, doc_id) document.is_published = not document.is_published db.session.commit() if document.is_published: # Enabling a document must also backfill anyone who is nowadays # eligible but was not when versions were published earlier. created = backfill_assignments_for_document(document) _log('document_published', f'{document.title} (assignments: {created})') flash( _('"%(title)s" is now downloadable by team members.', title=document.title), 'success', ) else: _log('document_unpublished', document.title) flash( _('"%(title)s" is no longer downloadable.', title=document.title), 'success', ) return respond_after_change(url_for('admin.documents')) @admin_bp.route('/documents/version//download') @login_required def download_document_version(version_id): """Admin download of a stored document revision.""" if not require_admin(): return redirect(url_for('main.dashboard')) version = db.get_or_404(DocumentVersion, version_id) try: document = open_document(version.file_path) except GoogleDriveStorageError: flash( _('Document storage is temporarily unavailable. Please try again later.'), 'danger', ) return respond_after_change(url_for('admin.documents')) return send_file(document, as_attachment=True, download_name=version.original_filename) @admin_bp.route('/documents/') @login_required def document_detail(doc_id): """Revisions and per-person completion for one document.""" if not require_admin(): return redirect(url_for('main.dashboard')) document = db.get_or_404(Document, doc_id) versions = document.versions.order_by(DocumentVersion.version).all() assignments = ( db.session.query(DocumentAssignment, User) .join(User, User.id == DocumentAssignment.user_id) .filter(DocumentAssignment.document_id == document.id) .order_by(User.username) .all() ) return render_template( 'pages/admin_document_detail.html', document=document, versions=versions, assignments=assignments, kinds=DOCUMENT_KINDS, )