from flask import render_template, redirect, url_for, flash, request, jsonify, g from functools import wraps from . import admin_bp from app.services.auth.decorators import admin_required, login_required, superadmin_required from app.services.auth import AuthService from app.models.database.models import db, User, ChallengeRecord, DeploymentRecord, CategoryConfig, CategoryAdmin, Role from werkzeug.security import generate_password_hash def to_bool(value, default=False): """ 将值转换为布尔值 处理以下情况： - True/False -> 直接返回 - 'true'/'True'/'TRUE' -> True - 'false'/'False'/'FALSE' -> False - 1/0 -> True/False - None -> 返回 default """ if value is None: return default if isinstance(value, bool): return value if isinstance(value, (int, float)): return bool(value) if isinstance(value, str): value_lower = value.lower().strip() if value_lower in ('true', '1', 'yes', 'on'): return True elif value_lower in ('false', '0', 'no', 'off', ''): return False return bool(value) def category_access_required(min_role='viewer'): """ 方向访问权限装饰器 - 用于视图层 权限级别: - viewer: 只读权限 - editor: 编辑权限 - owner: 管理权限 """ def decorator(f): @wraps(f) def decorated_function(category_id, *args, **kwargs): user = g.user if not user: flash('请先登录', 'warning') return redirect(url_for('auth.login')) # 系统管理员拥有所有权限 if user.role == Role.ADMIN: return f(category_id, *args, **kwargs) # 检查用户在该方向的权限 admin_record = CategoryAdmin.query.filter_by( category_id=category_id, user_id=user.id ).first() if not admin_record: flash('您没有该方向的访问权限', 'danger') return redirect(url_for('admin.categories')) # 权限级别检查 role_levels = {'viewer': 1, 'editor': 2, 'owner': 3} user_level = role_levels.get(admin_record.role, 0) required_level = role_levels.get(min_role, 1) if user_level < required_level: role_names = {'viewer': '查看', 'editor': '编辑', 'owner': '管理'} flash(f'需要{role_names.get(min_role, min_role)}权限', 'danger') return redirect(url_for('admin.categories')) return f(category_id, *args, **kwargs) return decorated_function return decorator @admin_bp.route('/') @admin_required def index(): """管理员首页""" is_superadmin = g.user.role == Role.ADMIN # 获取方向概览 categories = [] if is_superadmin: categories = CategoryConfig.query.order_by(CategoryConfig.name).all() else: # 普通管理员只能看到自己管理的方向 categories = CategoryAdmin.get_user_categories(g.user.id) return render_template('pages/admin/index.html', categories=categories, is_superadmin=is_superadmin) @admin_bp.route('/settings') @admin_required def settings(): """系统设置页面（合并了AI配置和系统工具）""" import sys return render_template('pages/admin/settings.html', python_version=sys.version.split()[0]) @admin_bp.route('/data') @admin_required def data(): """数据管理页面 - 合并用户、题目、部署管理""" active_tab = request.args.get('tab', 'users') users_page = request.args.get('users_page', 1, type=int) challenges_page = request.args.get('challenges_page', 1, type=int) challenges_category = request.args.get('challenges_category', 'all') # 题目方向筛选 deployments_page = request.args.get('deployments_page', 1, type=int) per_page = 15 # 获取用户数据 users_pagination = User.query.order_by(User.created_at.desc()).paginate( page=users_page, per_page=per_page, error_out=False ) # 获取题目数据（支持按方向筛选） challenges_query = ChallengeRecord.query if challenges_category and challenges_category != 'all': challenges_query = challenges_query.filter_by(challenge_type=challenges_category) challenges_pagination = challenges_query.order_by(ChallengeRecord.created_at.desc()).paginate( page=challenges_page, per_page=per_page, error_out=False ) # 获取所有方向配置 try: from app.models.database.models import CategoryConfig all_categories = CategoryConfig.get_all_categories() categories_dict = {cat.id: {'name': cat.name, 'icon': cat.icon} for cat in all_categories} except: categories_dict = {} # 统计各方向题目数量 challenge_counts = {} challenge_counts['all'] = ChallengeRecord.query.count() for cat_id in categories_dict.keys(): challenge_counts[cat_id] = ChallengeRecord.query.filter_by(challenge_type=cat_id).count() # 获取部署数据 deployments_pagination = DeploymentRecord.query.order_by(DeploymentRecord.created_at.desc()).paginate( page=deployments_page, per_page=per_page, error_out=False ) # 获取所有方向配置（用于权限管理） categories_list = [] try: from app.models.database.models import CategoryConfig all_categories = CategoryConfig.query.order_by(CategoryConfig.name).all() categories_list = [{'id': c.id, 'name': c.name} for c in all_categories] except: categories_list = [] return render_template('pages/admin/data.html', active_tab=active_tab, users=users_pagination.items, users_pagination=users_pagination, challenges=challenges_pagination.items, challenges_pagination=challenges_pagination, challenges_category=challenges_category, categories_dict=categories_dict, challenge_counts=challenge_counts, deployments=deployments_pagination.items, deployments_pagination=deployments_pagination, categories_list=categories_list, is_superadmin=g.user.role == Role.ADMIN) @admin_bp.route('/users') @admin_required def users(): """用户管理页面 - 重定向到数据管理""" return redirect(url_for('admin.data', tab='users')) @admin_bp.route('/challenges') @admin_required def challenges(): """题目管理页面 - 重定向到数据管理""" return redirect(url_for('admin.data', tab='challenges')) @admin_bp.route('/deployments') @admin_required def deployments(): """部署管理页面 - 重定向到数据管理""" return redirect(url_for('admin.data', tab='deployments')) # ==================== 用户管理 API ==================== @admin_bp.route('/api/user/', methods=['GET']) @admin_required def api_get_user(user_id): """获取用户详情""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'message': '用户不存在'}), 404 # 统计用户的题目和部署数量 challenge_count = ChallengeRecord.query.filter_by(user_id=user_id).count() deployment_count = DeploymentRecord.query.filter_by(user_id=user_id).count() user_data = { 'id': user.id, 'username': user.username, 'email': user.email, 'role': user.role, 'is_active': user.is_active, 'created_at': user.created_at.strftime('%Y-%m-%d %H:%M') if user.created_at else None, 'last_login': user.last_login.strftime('%Y-%m-%d %H:%M') if user.last_login else None, 'challenge_count': challenge_count, 'deployment_count': deployment_count } return jsonify({'success': True, 'data': user_data}) except Exception as e: return jsonify({'success': False, 'message': f'错误: {str(e)}'}), 500 @admin_bp.route('/api/user/', methods=['POST']) @admin_required def api_update_user(user_id): """更新用户信息""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'message': '用户不存在'}), 404 data = request.get_json() # 更新邮箱 if 'email' in data and data['email']: # 检查邮箱是否已被其他用户使用 existing_user = User.query.filter_by(email=data['email']).first() if existing_user and existing_user.id != user_id: return jsonify({'success': False, 'message': '该邮箱已被使用'}), 400 user.email = data['email'] # 更新角色 if 'role' in data and data['role'] in ['user', 'moderator', 'admin']: user.role = data['role'] # 更新状态 if 'is_active' in data: user.is_active = to_bool(data['is_active'], default=True) # 更新密码（如果提供） if 'password' in data and data['password']: user.password_hash = generate_password_hash(data['password'], method='pbkdf2:sha256') db.session.commit() return jsonify({'success': True, 'message': '用户信息更新成功'}) except Exception as e: db.session.rollback() return jsonify({'success': False, 'message': f'更新失败: {str(e)}'}), 500 @admin_bp.route('/api/user/create', methods=['POST']) @admin_required def api_create_user(): """创建新用户""" try: data = request.get_json() # 验证必填字段 if not data.get('username'): return jsonify({'success': False, 'message': '用户名不能为空'}), 400 if not data.get('email'): return jsonify({'success': False, 'message': '邮箱不能为空'}), 400 if not data.get('password'): return jsonify({'success': False, 'message': '密码不能为空'}), 400 # 检查用户名是否已存在 if User.query.filter_by(username=data['username']).first(): return jsonify({'success': False, 'message': '用户名已存在'}), 400 # 检查邮箱是否已存在 if User.query.filter_by(email=data['email']).first(): return jsonify({'success': False, 'message': '邮箱已被使用'}), 400 # 创建新用户 # 确保 is_active 是布尔值 is_active = to_bool(data.get('is_active'), default=True) new_user = User( username=data['username'], email=data['email'], password_hash=generate_password_hash(data['password'], method='pbkdf2:sha256'), role=data.get('role', 'user'), is_active=is_active ) db.session.add(new_user) db.session.commit() return jsonify({'success': True, 'message': '用户创建成功', 'user_id': new_user.id}) except Exception as e: db.session.rollback() return jsonify({'success': False, 'message': f'创建失败: {str(e)}'}), 500 @admin_bp.route('/api/user/', methods=['DELETE']) @superadmin_required def api_delete_user(user_id): """删除用户 - 仅超级管理员可操作""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'message': '用户不存在'}), 404 # 不能删除超级管理员 if user.role == Role.ADMIN: return jsonify({'success': False, 'message': '不能删除超级管理员'}), 403 # 删除用户的方向权限 CategoryAdmin.query.filter_by(user_id=user_id).delete() # 删除用户 db.session.delete(user) db.session.commit() return jsonify({'success': True, 'message': f'用户 {user.username} 已删除'}) except Exception as e: db.session.rollback() return jsonify({'success': False, 'message': f'删除失败: {str(e)}'}), 500 # ==================== 方向配置管理 ==================== @admin_bp.route('/categories') @login_required def categories(): """方向管理页面""" # 非管理员用户也可以访问，但只能看到自己有权限的方向 is_admin = g.user and g.user.role == Role.ADMIN return render_template('pages/admin/categories.html', is_admin=is_admin) @admin_bp.route('/categories/') @login_required @category_access_required('viewer') def category_detail(category_id): """方向详情配置页面""" category = CategoryConfig.query.get(category_id) if not category: flash('方向不存在', 'danger') return redirect(url_for('admin.categories')) # 获取当前用户在该方向的权限级别 user_role = 'owner' # 默认管理员权限 if g.user.role != Role.ADMIN: admin_record = CategoryAdmin.query.filter_by( category_id=category_id, user_id=g.user.id ).first() user_role = admin_record.role if admin_record else 'viewer' # 获取 category 数据并清理控制字符 category_data = category.to_dict(include_config=True) # 清理数据中的控制字符，确保 JSON 序列化安全 import json as json_lib import re def clean_for_json(obj, path=''): """清理对象中的控制字符 Args: obj: 要清理的对象 path: 当前字段路径（用于识别 prompt 相关字段） """ if isinstance(obj, dict): return {k: clean_for_json(v, f"{path}.{k}" if path else k) for k, v in obj.items()} elif isinstance(obj, list): return [clean_for_json(item, f"{path}[{i}]" if path else f"[{i}]") for i, item in enumerate(obj)] elif isinstance(obj, str): # 对于 prompt 相关字段，保留合法的空白字符（换行、回车、制表符） # 只移除真正有问题的控制字符 is_prompt_field = any(keyword in path.lower() for keyword in [ 'prompt', 'content', 'description', 'instruction', 'script' ]) if is_prompt_field: # Prompt 字段：只移除有问题的控制字符，保留 \n, \r, \t # 移除：NULL(\x00), SOH-STX(\x01-\x02), EOT-ENQ(\x04-\x05), # ACK(\x06), BEL(\x07), BS(\x08), VT(\x0B), FF(\x0C), # SO-SI(\x0E-\x0F), DLE-DC4(\x10-\x14), NAK-SYN(\x15-\x16), # ETB-CAN(\x17-\x18), EM(\x19), SUB-ESC(\x1A-\x1B), # FS-GS(\x1C-\x1D), RS-US(\x1E-\x1F), DEL(\x7F) # 保留：TAB(\x09=\t), LF(\x0A=\n), CR(\x0D=\r) cleaned = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', obj) # 不合并空格，不 strip，保留原始格式 return cleaned else: # 非 prompt 字段：移除所有控制字符，替换为空格 cleaned = re.sub(r'[\x00-\x1F\x7F]', ' ', obj) cleaned = re.sub(r' +', ' ', cleaned) return cleaned.strip() else: return obj category_data_cleaned = clean_for_json(category_data) # 预先序列化为 JSON 字符串 category_json_str = json_lib.dumps(category_data_cleaned, ensure_ascii=False, separators=(',', ':')) category_json_str = category_json_str.replace('', '<\\/script>') return render_template('pages/admin/category_detail.html', category=category_data_cleaned, category_json=category_json_str, user_permission=user_role) @admin_bp.route('/categories//form-designer') @login_required @category_access_required('editor') def form_designer(category_id): """表单设计器页面""" category = CategoryConfig.query.get(category_id) if not category: flash('方向不存在', 'danger') return redirect(url_for('admin.categories')) category_dict = category.to_dict(include_config=True) return render_template('pages/admin/form_designer.html', category=category_dict, form_fields=category_dict.get('form_fields', [])) # ==================== 用户权限管理 ==================== @admin_bp.route('/user-permissions') @superadmin_required def user_permissions(): """用户权限管理页面 - 重定向到数据管理页面的用户标签""" return redirect(url_for('admin.data', tab='users')) @admin_bp.route('/api/user//permissions', methods=['GET']) @superadmin_required def api_get_user_permissions(user_id): """获取用户的方向权限""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'error': '用户不存在'}), 404 # 获取用户的所有方向权限 permissions = CategoryAdmin.query.filter_by(user_id=user_id).all() return jsonify({ 'success': True, 'user': { 'id': user.id, 'username': user.username, 'email': user.email, 'role': user.role }, 'permissions': [p.to_dict() for p in permissions] }) except Exception as e: return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/user//permissions', methods=['POST']) @superadmin_required def api_set_user_permission(user_id): """设置用户的方向权限""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'error': '用户不存在'}), 404 data = request.get_json() category_id = data.get('category_id') role = data.get('role', 'editor') # viewer, editor, owner if not category_id: return jsonify({'success': False, 'error': '方向ID不能为空'}), 400 # 检查方向是否存在 category = CategoryConfig.query.get(category_id) if not category: return jsonify({'success': False, 'error': '方向不存在'}), 404 # 检查是否已有权限记录 existing = CategoryAdmin.query.filter_by( user_id=user_id, category_id=category_id ).first() if existing: # 更新现有权限 existing.role = role else: # 创建新权限 new_permission = CategoryAdmin( user_id=user_id, category_id=category_id, role=role ) db.session.add(new_permission) db.session.commit() return jsonify({ 'success': True, 'message': f'已授予 {user.username} 对 {category.name} 的{role}权限' }) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/user//permissions/', methods=['DELETE']) @superadmin_required def api_delete_user_permission(user_id, category_id): """删除用户的方向权限""" try: permission = CategoryAdmin.query.filter_by( user_id=user_id, category_id=category_id ).first() if not permission: return jsonify({'success': False, 'error': '权限记录不存在'}), 404 db.session.delete(permission) db.session.commit() return jsonify({'success': True, 'message': '权限已删除'}) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/user//permissions/batch', methods=['POST']) @superadmin_required def api_batch_set_permissions(user_id): """批量设置用户权限""" try: user = User.query.get(user_id) if not user: return jsonify({'success': False, 'error': '用户不存在'}), 404 data = request.get_json() permissions = data.get('permissions', []) # [{category_id, role}, ...] # 先删除该用户的所有权限 CategoryAdmin.query.filter_by(user_id=user_id).delete() # 添加新权限 for perm in permissions: category_id = perm.get('category_id') role = perm.get('role', 'editor') if category_id: new_permission = CategoryAdmin( user_id=user_id, category_id=category_id, role=role ) db.session.add(new_permission) db.session.commit() return jsonify({ 'success': True, 'message': f'已更新 {user.username} 的权限配置' }) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 # ============================================================ # 系统工具 API # ============================================================ @admin_bp.route('/api/cleanup-deployments', methods=['POST']) @superadmin_required def api_cleanup_deployments(): """清理过期部署""" try: from datetime import datetime, timedelta from app.services.deployment.core.execution_service import DockerExecutionService now = datetime.utcnow() cutoff = now - timedelta(hours=24) # 查找需要清理的部署： # 1. 已经是 expired 状态的部署（直接删除） # 2. 根据 expires_at 字段判断已过期的部署 # 3. 创建时间超过24小时且状态为 running 或 stopped 的部署 expired_deployments = DeploymentRecord.query.filter( db.or_( # 已经是 expired 状态的部署 DeploymentRecord.status == 'expired', # 根据 expires_at 判断已过期 db.and_( DeploymentRecord.expires_at.isnot(None), DeploymentRecord.expires_at < now ), # 创建时间超过24小时且状态为 running 或 stopped db.and_( DeploymentRecord.created_at < cutoff, DeploymentRecord.status.in_(['running', 'stopped']) ) ) ).all() docker_service = DockerExecutionService() deleted_count = 0 marked_count = 0 for deployment in expired_deployments: try: # 如果已经是 expired 状态，直接删除记录 if deployment.status == 'expired': # 尝试清理容器和文件 if deployment.working_directory: try: docker_service.delete_deployment( deployment.deployment_uuid, deployment.working_directory, deployment.external_port ) except Exception as e: print(f"清理部署 {deployment.deployment_uuid} 失败: {str(e)}") # 删除数据库记录 db.session.delete(deployment) deleted_count += 1 else: # 对于其他状态的过期部署，先清理容器，然后标记为 expired if deployment.working_directory: try: docker_service.delete_deployment( deployment.deployment_uuid, deployment.working_directory, deployment.external_port ) except Exception as e: print(f"清理部署 {deployment.deployment_uuid} 失败: {str(e)}") deployment.status = 'expired' marked_count += 1 except Exception as e: print(f"处理部署 {deployment.deployment_uuid} 时出错: {str(e)}") continue db.session.commit() total_count = deleted_count + marked_count message = f'已清理 {total_count} 个过期部署' if deleted_count > 0: message += f'（删除 {deleted_count} 个，标记 {marked_count} 个）' return jsonify({ 'success': True, 'count': total_count, 'deleted': deleted_count, 'marked': marked_count, 'message': message }) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/cleanup-invalid-challenges', methods=['POST']) @superadmin_required def api_cleanup_invalid_challenges(): """清理无效题目记录""" try: import os # 查找没有文件的题目记录 challenges = ChallengeRecord.query.all() count = 0 for challenge in challenges: # 检查输出目录是否存在（使用 output_dir 字段） if challenge.output_dir and not os.path.exists(challenge.output_dir): db.session.delete(challenge) count += 1 db.session.commit() return jsonify({ 'success': True, 'count': count, 'message': f'已清理 {count} 个无效题目记录' }) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/backup-database', methods=['POST']) @superadmin_required def api_backup_database(): """备份数据库""" try: import os import shutil import gzip from datetime import datetime from pathlib import Path from config import Config # 获取项目根目录 project_root = Path(Config.BASE_DIR) backup_dir = project_root / 'backups' / 'database' backup_dir.mkdir(parents=True, exist_ok=True) # 生成备份文件名 timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') if Config.DATABASE_TYPE == 'sqlite': # SQLite 备份 db_path = Path(Config.DB_PATH) if not db_path.exists(): return jsonify({ 'success': False, 'error': f'数据库文件不存在: {db_path}' }), 400 # 获取数据库大小 db_size = db_path.stat().st_size # 创建备份文件路径 backup_file = backup_dir / f'ctf_{timestamp}.db' # 使用 SQLite 的备份命令 import sqlite3 source_conn = sqlite3.connect(str(db_path)) backup_conn = sqlite3.connect(str(backup_file)) source_conn.backup(backup_conn) backup_conn.close() source_conn.close() # 压缩备份 compressed_file = backup_dir / f'ctf_{timestamp}.db.gz' with open(backup_file, 'rb') as f_in: with gzip.open(compressed_file, 'wb') as f_out: shutil.copyfileobj(f_in, f_out) # 删除未压缩的文件 backup_file.unlink() # 获取压缩后的大小 compressed_size = compressed_file.stat().st_size # 清理旧备份（保留7天） keep_days = 7 cutoff_time = datetime.now().timestamp() - (keep_days * 24 * 60 * 60) deleted_count = 0 for old_backup in backup_dir.glob('*.gz'): if old_backup.stat().st_mtime < cutoff_time: old_backup.unlink() deleted_count += 1 return jsonify({ 'success': True, 'message': '数据库备份成功', 'backup_file': str(compressed_file.relative_to(project_root)), 'backup_size': compressed_size, 'db_size': db_size, 'deleted_old': deleted_count }) elif Config.DATABASE_TYPE == 'postgresql': # PostgreSQL 备份 - 使用 Python 直接连接，避免版本不匹配问题 try: import psycopg2 from psycopg2.extensions import ISOLATION_LEVEL_AUTOCOMMIT except ImportError: return jsonify({ 'success': False, 'error': '未安装 psycopg2，请运行: pip install psycopg2-binary' }), 500 # 创建备份文件路径 backup_file = backup_dir / f'ctf_{timestamp}.sql' try: # 使用 psycopg2 连接数据库 conn = psycopg2.connect( host=Config.POSTGRES_HOST, port=Config.POSTGRES_PORT, database=Config.POSTGRES_DB, user=Config.POSTGRES_USER, password=Config.POSTGRES_PASSWORD ) # 尝试使用 pg_dump（如果可用且版本匹配） import subprocess env = os.environ.copy() env['PGPASSWORD'] = Config.POSTGRES_PASSWORD cmd = [ 'pg_dump', '-h', Config.POSTGRES_HOST, '-p', str(Config.POSTGRES_PORT), '-U', Config.POSTGRES_USER, '-d', Config.POSTGRES_DB, '-F', 'p', # 纯文本格式 '-f', str(backup_file) ] try: result = subprocess.run( cmd, env=env, capture_output=True, text=True, check=True, timeout=300 # 5分钟超时 ) except subprocess.TimeoutExpired: conn.close() return jsonify({ 'success': False, 'error': '备份超时（超过5分钟）' }), 500 except (subprocess.CalledProcessError, FileNotFoundError) as e: # pg_dump 不可用或版本不匹配，使用 Python 方式备份 conn.close() # 使用 SQLAlchemy 连接来备份 from sqlalchemy import create_engine, text engine = create_engine(Config.SQLALCHEMY_DATABASE_URI) with engine.connect() as connection: # 获取所有表名 from sqlalchemy import inspect inspector = inspect(engine) tables = inspector.get_table_names() with open(backup_file, 'w', encoding='utf-8') as f: f.write("-- PostgreSQL 数据库备份\n") f.write(f"-- 备份时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"-- 数据库: {Config.POSTGRES_DB}\n\n") # 备份每个表的数据 for table in tables: f.write(f"\n-- 表: {table}\n") f.write(f"TRUNCATE TABLE \"{table}\" CASCADE;\n\n") # 获取表数据 result = connection.execute(text(f'SELECT * FROM "{table}"')) rows = result.fetchall() if rows: # 获取列名 columns = result.keys() col_names = ', '.join([f'"{col}"' for col in columns]) for row in rows: values = [] for val in row: if val is None: values.append('NULL') elif isinstance(val, str): # 转义单引号 val_escaped = val.replace("'", "''") values.append(f"'{val_escaped}'") elif isinstance(val, (int, float)): values.append(str(val)) elif isinstance(val, bool): values.append('TRUE' if val else 'FALSE') elif isinstance(val, datetime): values.append(f"'{val.isoformat()}'") else: val_str = str(val).replace("'", "''") values.append(f"'{val_str}'") values_str = ', '.join(values) f.write(f'INSERT INTO "{table}" ({col_names}) VALUES ({values_str});\n') engine.dispose() conn.close() except Exception as e: if 'conn' in locals(): conn.close() return jsonify({ 'success': False, 'error': f'PostgreSQL 备份失败: {str(e)}' }), 500 # 压缩备份文件 compressed_file = backup_dir / f'ctf_{timestamp}.sql.gz' with open(backup_file, 'rb') as f_in: with gzip.open(compressed_file, 'wb') as f_out: shutil.copyfileobj(f_in, f_out) # 删除未压缩的文件 backup_file.unlink() # 获取压缩后的大小 compressed_size = compressed_file.stat().st_size # 清理旧备份（保留7天） keep_days = 7 cutoff_time = datetime.now().timestamp() - (keep_days * 24 * 60 * 60) deleted_count = 0 for old_backup in backup_dir.glob('*.sql.gz'): if old_backup.stat().st_mtime < cutoff_time: old_backup.unlink() deleted_count += 1 return jsonify({ 'success': True, 'message': '数据库备份成功', 'backup_file': str(compressed_file.relative_to(project_root)), 'backup_size': compressed_size, 'deleted_old': deleted_count }) else: return jsonify({ 'success': False, 'error': f'不支持的数据库类型: {Config.DATABASE_TYPE}' }), 400 except Exception as e: import traceback traceback.print_exc() return jsonify({ 'success': False, 'error': f'备份失败: {str(e)}' }), 500 @admin_bp.route('/api/reset-database', methods=['POST']) @superadmin_required def api_reset_database(): """重置数据库（危险操作）""" try: # 删除所有题目记录 ChallengeRecord.query.delete() # 删除所有部署记录 DeploymentRecord.query.delete() # 删除所有非管理员用户 User.query.filter(User.role != Role.ADMIN).delete() # 删除所有权限分配 CategoryAdmin.query.delete() db.session.commit() return jsonify({ 'success': True, 'message': '数据库已重置' }) except Exception as e: db.session.rollback() return jsonify({'success': False, 'error': str(e)}), 500 @admin_bp.route('/api/challenge/', methods=['DELETE']) @admin_required def api_delete_challenge(challenge_id): """删除题目 - 管理员可操作""" try: from app.models.database.operations import delete_challenge success = delete_challenge(challenge_id) if success: return jsonify({'success': True, 'message': '题目删除成功'}) else: return jsonify({'success': False, 'message': '题目不存在或删除失败'}), 404 except Exception as e: return jsonify({'success': False, 'message': f'删除失败: {str(e)}'}), 500