""" 系统配置管理API """ from flask import Blueprint, jsonify, request, g from app.models.database.models import db, SystemConfig, CategoryConfig from app.services.auth.decorators import admin_required from app.services.config import get_config_loader from app.services.ai.tools import SandboxConfig system_config_bp = Blueprint('system_config', __name__, url_prefix='/api/admin/system-config') # 默认 AI 沙箱策略 DEFAULT_SANDBOX_POLICY = { 'allowed_commands': sorted(list(SandboxConfig.DEFAULT_ALLOWED_COMMANDS)), 'blocked_commands': [], 'allowed_read_dirs': [], 'allowed_write_dirs': [], 'knowledge_executable_dirs': [], # 知识库可执行目录范围 'max_file_size': 10 * 1024 * 1024, 'max_output_size': 8000, 'command_timeout': 300, 'docker_timeout': 600, } @system_config_bp.route('/available-options', methods=['GET']) @admin_required def get_available_options(): """从各方向的表单配置中动态获取所有可用选项""" try: # 获取所有启用的方向 categories = CategoryConfig.query.filter_by(enabled=True).order_by(CategoryConfig.sort_order).all() # 按方向组织的选项 category_options = {} for category in categories: form_fields = category.get_form_fields() category_data = { 'id': category.id, 'name': category.name, 'icon': category.icon, 'fields': {} } for field in form_fields: field_id = field.get('id') field_type = field.get('type') field_label = field.get('label') options = field.get('options', []) # 提取选项值 extracted_options = [] if field_type in ['select', 'multi_select']: for opt in options: if isinstance(opt, dict): extracted_options.append({ 'id': opt.get('value'), 'name': opt.get('label'), 'icon': opt.get('icon') }) else: extracted_options.append({'id': opt, 'name': opt}) elif field_type == 'multi_select_categorized': # 分类选项（如漏洞类型） for cat_group in options: cat_name = cat_group.get('category', '') for item in cat_group.get('items', []): extracted_options.append({ 'id': item.get('value'), 'name': item.get('label'), 'category': cat_name }) if extracted_options: category_data['fields'][field_id] = { 'label': field_label, 'type': field_type, 'options': extracted_options } category_options[category.id] = category_data return jsonify({ 'success': True, 'categories': category_options }) except Exception as e: import traceback traceback.print_exc() return jsonify({ 'success': False, 'message': str(e) }), 500 # ============================================================ # AI 沙箱策略配置 # ============================================================ def _normalize_list(value): """将输入转换为去重后的列表""" if value is None: return [] if isinstance(value, list): items = value elif isinstance(value, str): # 支持换行或逗号分隔 items = [v.strip() for v in value.replace(',', '\n').split('\n')] else: items = [] return [item for item in items if item] def _normalize_policy(data: dict) -> dict: """合并并规范化策略数据""" policy = DEFAULT_SANDBOX_POLICY.copy() policy.update({ 'allowed_commands': _normalize_list(data.get('allowed_commands')), 'blocked_commands': _normalize_list(data.get('blocked_commands')), 'allowed_read_dirs': _normalize_list(data.get('allowed_read_dirs')), 'allowed_write_dirs': _normalize_list(data.get('allowed_write_dirs')), 'knowledge_executable_dirs': _normalize_list(data.get('knowledge_executable_dirs')), }) for key in ['max_file_size', 'max_output_size', 'command_timeout', 'docker_timeout']: try: val = data.get(key, policy[key]) if val is not None: policy[key] = int(val) except Exception: # 保持默认值 policy[key] = DEFAULT_SANDBOX_POLICY[key] return policy @system_config_bp.route('/ai-sandbox', methods=['GET']) @admin_required def get_ai_sandbox_policy(): """获取 AI 沙箱策略""" policy = SystemConfig.get_config('ai_sandbox_policy', DEFAULT_SANDBOX_POLICY) or {} normalized = _normalize_policy(policy) return jsonify({'success': True, 'policy': normalized}) @system_config_bp.route('/ai-sandbox', methods=['POST']) @admin_required def update_ai_sandbox_policy(): """更新 AI 沙箱策略""" try: data = request.get_json() or {} policy = _normalize_policy(data) SystemConfig.set_config( 'ai_sandbox_policy', policy, 'AI 工具沙箱策略（命令白名单/黑名单、目录限制、阈值）', 'ai' ) return jsonify({'success': True, 'policy': policy, 'message': '沙箱策略已更新'}) except Exception as e: db.session.rollback() return jsonify({ 'success': False, 'message': f'更新失败: {str(e)}' }), 500 @system_config_bp.route('/generation-limits', methods=['GET']) @admin_required def get_generation_limits(): """获取题目生成限制配置（按方向）""" try: # 获取按方向的限制配置 category_limits = SystemConfig.get_config('category_generation_limits', {}) # 获取全局配置 global_config = { 'allow_registration': SystemConfig.is_registration_enabled() } return jsonify({ 'success': True, 'category_limits': category_limits, 'global_config': global_config }) except Exception as e: return jsonify({ 'success': False, 'message': str(e) }), 500 @system_config_bp.route('/generation-limits', methods=['POST']) @admin_required def update_generation_limits(): """更新题目生成限制配置（按方向）""" try: data = request.get_json() # 更新按方向的限制配置 if 'category_limits' in data: SystemConfig.set_config( 'category_generation_limits', data['category_limits'], '按方向的题目生成限制配置', 'generation' ) # 更新全局配置 if 'allow_registration' in data: SystemConfig.set_config( 'allow_registration', data['allow_registration'], '是否允许用户注册（关闭后只能由管理员创建用户）', 'security' ) return jsonify({ 'success': True, 'message': '配置更新成功' }) except Exception as e: db.session.rollback() return jsonify({ 'success': False, 'message': f'更新失败: {str(e)}' }), 500 # ============================================================ # 全局Token管理API # ============================================================ @system_config_bp.route('/upload-limit', methods=['GET']) @admin_required def get_upload_limit(): """获取文件上传大小限制""" try: from flask import current_app max_size = current_app.config.get('MAX_CONTENT_LENGTH') return jsonify({ 'success': True, 'max_size': max_size }) except Exception as e: return jsonify({ 'success': False, 'message': str(e) }), 500 @system_config_bp.route('/upload-limit', methods=['POST']) @admin_required def set_upload_limit(): """设置文件上传大小限制（字节）""" try: from flask import current_app data = request.get_json() max_size = data.get('max_size') if max_size is None: current_app.config['MAX_CONTENT_LENGTH'] = None else: current_app.config['MAX_CONTENT_LENGTH'] = int(max_size) # 同时保存到系统配置 SystemConfig.set_config( 'max_upload_size', max_size, '文件上传大小限制（字节，None表示无限制）', 'system' ) return jsonify({ 'success': True, 'message': '上传大小限制已更新' }) except Exception as e: return jsonify({ 'success': False, 'message': f'更新失败: {str(e)}' }), 500 @system_config_bp.route('/ui-config', methods=['GET']) @admin_required def get_ui_config(): """获取UI定制配置（统一管理）""" try: ui_config = SystemConfig.get_config('ui_customization', {}) return jsonify({ 'success': True, 'config': ui_config }) except Exception as e: return jsonify({ 'success': False, 'message': str(e) }), 500 @system_config_bp.route('/ui-config', methods=['POST']) @admin_required def update_ui_config(): """更新UI定制配置（统一管理）""" try: data = request.get_json() ui_config = data.get('config', {}) SystemConfig.set_config( 'ui_customization', ui_config, 'UI定制配置（页面标题、主色调、描述等）', 'ui' ) return jsonify({ 'success': True, 'message': 'UI配置更新成功' }) except Exception as e: db.session.rollback() return jsonify({ 'success': False, 'message': f'更新失败: {str(e)}' }), 500