SOONet部署安全加固：Gradio认证登录、API密钥控制、视频上传大小限制

1. 项目简介与安全加固背景

SOONet是一个基于自然语言输入的长视频时序片段定位系统。简单来说，你给它一段文字描述，它就能在长视频里快速找到匹配这个描述的时间片段。这个功能在视频内容分析、智能检索、媒体资产管理等场景下特别有用。

不过，当我们把这样一个强大的AI系统部署到实际环境中时，安全就成了必须考虑的问题。想象一下，如果你的SOONet服务直接暴露在公网上，任何人都能随意访问、上传视频、调用API，那会带来什么风险？

你可能遇到的安全问题：

• 未经授权的用户访问你的服务，消耗你的计算资源
• 恶意上传超大视频文件，导致服务器存储空间爆满
• API被滥用，影响正常服务的稳定性
• 敏感视频数据被随意访问

今天我就来分享一套完整的SOONet部署安全加固方案。我会手把手教你如何给SOONet的Gradio界面加上登录认证，如何控制API访问权限，以及如何限制视频上传大小。这些措施实施后，你的SOONet服务就能安全地对外提供服务了。

2. 安全加固方案概览

在开始具体操作之前，我们先看看整个安全加固方案包含哪些内容。这样你就能对整个改造有个清晰的了解。

2.1 三个核心安全功能

我们的安全加固主要围绕三个核心功能展开：

1. Gradio认证登录 - 给Web界面加把锁

   • 用户访问SOONet的Web界面时需要输入用户名和密码
   • 支持多用户管理，不同用户可以有不同的权限
   • 会话超时自动退出，防止长时间未操作的安全风险

2. API密钥控制 - 管理程序化访问

   • 为API调用生成唯一的访问密钥
   • 可以设置API密钥的有效期和使用次数限制
   • 记录API调用日志，方便监控和审计

3. 视频上传大小限制 - 防止资源滥用

   • 限制单个视频文件的最大大小
   • 限制总上传视频的数量和大小
   • 支持白名单机制，特定用户可以上传更大文件

2.2 改造前后的对比

为了让你更清楚地理解安全加固的价值，我做了个简单的对比：

功能	原始SOONet	安全加固后
Web访问	完全开放，无需认证	需要用户名密码登录
API调用	直接调用，无限制	需要有效的API密钥
文件上传	无大小限制	可配置大小限制
用户管理	无用户概念	支持多用户、多角色
访问日志	无记录	完整记录所有访问

2.3 技术实现思路

整个安全加固的实现思路其实很清晰：

1. 在Gradio应用启动前，先加载用户配置和API密钥配置
2. 修改app.py，在创建Gradio界面时加入认证参数
3. 添加中间件，在视频上传时检查文件大小
4. 封装API接口，在调用SOONet核心功能前验证API密钥
5. 添加配置文件，把用户信息、API密钥、限制参数都放在配置文件里

这样的设计有几个好处：配置和代码分离，修改安全设置不需要改代码；功能模块化，每个安全功能相对独立；扩展性强，后续要加新的安全功能也很方便。

3. Gradio认证登录实现

现在我们来具体实现第一个安全功能：给SOONet的Web界面加上登录认证。

3.1 创建用户配置文件

首先，我们需要创建一个用户配置文件。这个文件里存放所有可以访问SOONet的用户信息。

创建一个名为 users_config.yaml 的文件：

# SOONet用户配置
users:
  # 管理员用户 - 拥有所有权限
  admin:
    password: "admin_secure_password_123"  # 实际使用时请修改
    role: "admin"
    permissions:
      - "full_access"
      - "upload_large_files"
      - "manage_users"
      - "view_logs"
  
  # 普通用户 - 基础使用权限
  user1:
    password: "user1_password_456"  # 实际使用时请修改
    role: "user"
    permissions:
      - "basic_access"
      - "upload_standard_files"
  
  # 只读用户 - 只能查看，不能上传
  viewer:
    password: "viewer_password_789"  # 实际使用时请修改
    role: "viewer"
    permissions:
      - "view_only"

# 会话配置
session:
  timeout_minutes: 30  # 30分钟后自动退出
  max_sessions_per_user: 3  # 每个用户最多3个并发会话

# 登录尝试限制
login:
  max_attempts: 5  # 最大尝试次数
  lockout_minutes: 15  # 锁定时间

配置文件说明：

• 密码建议使用强密码，包含大小写字母、数字和特殊字符
• 角色分为admin、user、viewer三种，权限逐级递减
• 会话超时机制可以防止用户离开后会话被他人利用
• 登录尝试限制可以防止暴力破解

3.2 修改Gradio应用代码

接下来，我们需要修改SOONet的 app.py 文件，加入认证功能。下面是修改后的关键部分：

import gradio as gr
import yaml
import hashlib
import time
from functools import wraps
from typing import Dict, Optional

# 加载用户配置
def load_users_config(config_path: str = "users_config.yaml") -> Dict:
    """加载用户配置文件"""
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            config = yaml.safe_load(f)
        return config
    except FileNotFoundError:
        print(f"警告：配置文件 {config_path} 不存在，使用默认配置")
        return {
            "users": {
                "admin": {
                    "password": "admin",
                    "role": "admin",
                    "permissions": ["full_access"]
                }
            }
        }

# 用户认证装饰器
def require_auth(func):
    """认证装饰器，保护需要登录的功能"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 这里简化处理，实际应该检查会话
        # 完整的实现需要会话管理
        return func(*args, **kwargs)
    return wrapper

# 修改Gradio界面创建部分
def create_auth_interface():
    """创建带认证的Gradio界面"""
    
    # 加载用户配置
    users_config = load_users_config()
    
    # 创建登录状态管理
    login_status = {"username": None, "logged_in": False}
    
    def check_login(username: str, password: str) -> tuple:
        """检查用户名和密码"""
        if username not in users_config["users"]:
            return False, "用户不存在"
        
        user_info = users_config["users"][username]
        # 实际应该使用密码哈希，这里简化处理
        if password == user_info["password"]:
            login_status["username"] = username
            login_status["logged_in"] = True
            return True, f"登录成功！欢迎 {username}"
        else:
            return False, "密码错误"
    
    def logout():
        """退出登录"""
        login_status["username"] = None
        login_status["logged_in"] = False
        return "已退出登录"
    
    # 创建登录界面
    with gr.Blocks(title="SOONet - 视频时序定位系统") as demo:
        # 登录界面
        with gr.Column(visible=True) as login_section:
            gr.Markdown("## 🔐 SOONet 登录")
            username_input = gr.Textbox(label="用户名", placeholder="请输入用户名")
            password_input = gr.Textbox(
                label="密码", 
                placeholder="请输入密码",
                type="password"
            )
            login_button = gr.Button("登录", variant="primary")
            login_status_output = gr.Textbox(label="登录状态", interactive=False)
        
        # 主功能界面（初始隐藏）
        with gr.Column(visible=False) as main_section:
            # 原有的SOONet功能界面
            gr.Markdown("## 🎯 SOONet 视频时序定位")
            
            # 查询文本输入
            query_text = gr.Textbox(
                label="查询文本",
                placeholder="输入英文描述，如：a man takes food out of the refrigerator",
                lines=2
            )
            
            # 视频上传
            video_input = gr.Video(label="上传视频")
            
            # 定位按钮
            locate_button = gr.Button("🔍 开始定位", variant="primary")
            
            # 结果显示
            result_output = gr.Textbox(label="定位结果", interactive=False)
            
            # 退出按钮
            logout_button = gr.Button("退出登录", variant="secondary")
        
        # 登录按钮事件
        def on_login(username, password):
            success, message = check_login(username, password)
            if success:
                return [
                    gr.update(visible=False),  # 隐藏登录界面
                    gr.update(visible=True),   # 显示主界面
                    message
                ]
            else:
                return [
                    gr.update(visible=True),   # 保持登录界面
                    gr.update(visible=False),  # 隐藏主界面
                    message
                ]
        
        # 退出按钮事件
        def on_logout():
            logout()
            return [
                gr.update(visible=True),   # 显示登录界面
                gr.update(visible=False),  # 隐藏主界面
                "已退出登录"
            ]
        
        # 绑定事件
        login_button.click(
            fn=on_login,
            inputs=[username_input, password_input],
            outputs=[login_section, main_section, login_status_output]
        )
        
        logout_button.click(
            fn=on_logout,
            inputs=[],
            outputs=[login_section, main_section, login_status_output]
        )
        
        # 原有的定位功能（需要认证）
        @require_auth
        def locate_video(query, video):
            """需要认证的视频定位功能"""
            # 这里调用原有的SOONet定位逻辑
            # 为了简化示例，这里返回模拟结果
            if not login_status["logged_in"]:
                return "请先登录"
            
            # 模拟定位结果
            result = f"查询: {query}\n"
            result += f"视频: {video}\n"
            result += "定位结果: 00:01:15 - 00:01:45 (置信度: 0.87)"
            return result
        
        # 绑定定位功能
        locate_button.click(
            fn=locate_video,
            inputs=[query_text, video_input],
            outputs=result_output
        )
    
    return demo

# 启动应用
if __name__ == "__main__":
    demo = create_auth_interface()
    demo.launch(
        server_name="0.0.0.0",
        server_port=7860,
        share=False  # 关闭Gradio的分享功能，增强安全性
    )

3.3 密码安全增强

上面的示例中，密码是明文存储的，这在实际生产环境中是不安全的。我们需要对密码进行哈希处理：

import bcrypt
import base64

def hash_password(password: str) -> str:
    """使用bcrypt哈希密码"""
    salt = bcrypt.gensalt()
    hashed = bcrypt.hashpw(password.encode('utf-8'), salt)
    return hashed.decode('utf-8')

def verify_password(password: str, hashed_password: str) -> bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode('utf-8'), hashed_password.encode('utf-8'))

# 更新用户配置，存储哈希后的密码
hashed_config = {
    "users": {
        "admin": {
            "password_hash": hash_password("your_secure_password"),
            "role": "admin",
            # ... 其他配置
        }
    }
}

3.4 会话管理

为了完善认证系统，我们还需要实现会话管理：

import uuid
from datetime import datetime, timedelta

class SessionManager:
    """会话管理器"""
    
    def __init__(self):
        self.sessions = {}  # session_id -> session_data
        self.user_sessions = {}  # username -> [session_ids]
    
    def create_session(self, username: str) -> str:
        """创建新会话"""
        session_id = str(uuid.uuid4())
        now = datetime.now()
        
        self.sessions[session_id] = {
            "username": username,
            "created_at": now,
            "last_activity": now,
            "expires_at": now + timedelta(minutes=30)
        }
        
        # 更新用户会话列表
        if username not in self.user_sessions:
            self.user_sessions[username] = []
        self.user_sessions[username].append(session_id)
        
        return session_id
    
    def validate_session(self, session_id: str) -> bool:
        """验证会话是否有效"""
        if session_id not in self.sessions:
            return False
        
        session = self.sessions[session_id]
        if datetime.now() > session["expires_at"]:
            # 会话过期，清理
            self.remove_session(session_id)
            return False
        
        # 更新最后活动时间
        session["last_activity"] = datetime.now()
        return True
    
    def remove_session(self, session_id: str):
        """移除会话"""
        if session_id in self.sessions:
            username = self.sessions[session_id]["username"]
            if username in self.user_sessions:
                self.user_sessions[username] = [
                    sid for sid in self.user_sessions[username] 
                    if sid != session_id
                ]
            del self.sessions[session_id]
    
    def cleanup_expired_sessions(self):
        """清理过期会话"""
        now = datetime.now()
        expired_sessions = [
            session_id for session_id, session in self.sessions.items()
            if now > session["expires_at"]
        ]
        
        for session_id in expired_sessions:
            self.remove_session(session_id)

4. API密钥控制实现

对于需要通过API调用SOONet服务的场景，我们需要实现API密钥控制。这样只有持有有效API密钥的程序才能调用我们的服务。

4.1 API密钥管理配置

创建 api_keys_config.yaml 配置文件：

# API密钥配置
api_keys:
  # 开发环境密钥
  dev_key_001:
    key: "sk_dev_abc123xyz789"
    name: "开发测试密钥"
    owner: "dev_team"
    created_at: "2024-01-15"
    expires_at: "2024-12-31"  # 过期时间，空表示永不过期
    rate_limit: 100  # 每分钟最大请求数
    total_calls: 0  # 总调用次数
    last_used: null
    permissions:
      - "video_upload"
      - "query_execution"
      - "result_retrieval"
    enabled: true
  
  # 生产环境密钥
  prod_key_001:
    key: "sk_prod_def456uvw012"
    name: "生产环境密钥"
    owner: "production"
    created_at: "2024-01-10"
    expires_at: null  # 永不过期
    rate_limit: 50  # 更严格的限流
    total_calls: 1250
    last_used: "2024-01-20 14:30:00"
    permissions:
      - "query_execution"
      - "result_retrieval"
    enabled: true
  
  # 只读密钥
  readonly_key_001:
    key: "sk_read_ghi789rst345"
    name: "只读访问密钥"
    owner: "analytics"
    created_at: "2024-01-05"
    expires_at: "2024-06-30"
    rate_limit: 30
    total_calls: 320
    last_used: "2024-01-19 09:15:00"
    permissions:
      - "result_retrieval"  # 只能获取结果，不能执行查询
    enabled: true

# API全局配置
api_config:
  default_rate_limit: 60  # 默认每分钟60次
  require_https: true  # 是否要求HTTPS
  enable_logging: true  # 是否启用日志
  log_file: "api_access.log"
  blocked_ips: []  # 被阻止的IP列表

4.2 API密钥验证中间件

创建一个API密钥验证的中间件，在请求到达SOONet处理逻辑之前进行验证：

import time
import json
from datetime import datetime
from functools import wraps
from typing import Dict, Optional, Tuple
import yaml
import redis  # 用于分布式环境下的限流

class APIKeyManager:
    """API密钥管理器"""
    
    def __init__(self, config_path: str = "api_keys_config.yaml"):
        self.config_path = config_path
        self.api_keys = self.load_api_keys()
        
        # 初始化Redis连接（用于分布式限流）
        try:
            self.redis_client = redis.Redis(
                host='localhost', 
                port=6379, 
                db=0,
                decode_responses=True
            )
            self.redis_available = True
        except:
            self.redis_available = False
            print("警告：Redis不可用，使用内存限流")
            self.local_rate_limits = {}
    
    def load_api_keys(self) -> Dict:
        """加载API密钥配置"""
        try:
            with open(self.config_path, 'r', encoding='utf-8') as f:
                config = yaml.safe_load(f)
            return config.get('api_keys', {})
        except FileNotFoundError:
            print(f"警告：API密钥配置文件 {self.config_path} 不存在")
            return {}
    
    def validate_api_key(self, api_key: str) -> Tuple[bool, Optional[Dict], str]:
        """验证API密钥有效性"""
        # 查找对应的密钥配置
        key_config = None
        key_id = None
        
        for key_id, config in self.api_keys.items():
            if config['key'] == api_key:
                key_config = config
                break
        
        if not key_config:
            return False, None, "无效的API密钥"
        
        # 检查是否启用
        if not key_config.get('enabled', True):
            return False, key_config, "API密钥已被禁用"
        
        # 检查过期时间
        expires_at = key_config.get('expires_at')
        if expires_at:
            try:
                expire_date = datetime.strptime(expires_at, "%Y-%m-%d")
                if datetime.now() > expire_date:
                    return False, key_config, "API密钥已过期"
            except ValueError:
                pass  # 日期格式错误，跳过检查
        
        return True, key_config, "验证成功"
    
    def check_rate_limit(self, api_key: str, key_config: Dict) -> Tuple[bool, int]:
        """检查速率限制"""
        rate_limit = key_config.get('rate_limit', 60)
        
        if self.redis_available:
            # 使用Redis进行分布式限流
            key = f"rate_limit:{api_key}:{int(time.time() // 60)}"
            current = self.redis_client.incr(key)
            if current == 1:
                self.redis_client.expire(key, 60)  # 60秒过期
            return current <= rate_limit, rate_limit - current
        else:
            # 使用内存限流（单机版）
            current_minute = int(time.time() // 60)
            key = f"{api_key}:{current_minute}"
            
            if key not in self.local_rate_limits:
                self.local_rate_limits[key] = {
                    'count': 1,
                    'minute': current_minute
                }
            else:
                self.local_rate_limits[key]['count'] += 1
            
            # 清理过期的记录
            for k in list(self.local_rate_limits.keys()):
                if self.local_rate_limits[k]['minute'] < current_minute - 1:
                    del self.local_rate_limits[k]
            
            current_count = self.local_rate_limits[key]['count']
            return current_count <= rate_limit, rate_limit - current_count
    
    def update_usage(self, key_id: str, key_config: Dict):
        """更新使用统计"""
        # 更新总调用次数
        key_config['total_calls'] = key_config.get('total_calls', 0) + 1
        key_config['last_used'] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        # 保存回配置文件
        self.save_api_keys()
    
    def save_api_keys(self):
        """保存API密钥配置"""
        config = {
            'api_keys': self.api_keys,
            'api_config': {
                'default_rate_limit': 60,
                'require_https': True,
                'enable_logging': True,
                'log_file': 'api_access.log'
            }
        }
        
        with open(self.config_path, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False, allow_unicode=True)
    
    def log_api_access(self, api_key: str, endpoint: str, status: str, ip: str):
        """记录API访问日志"""
        if not self.api_keys.get('api_config', {}).get('enable_logging', True):
            return
        
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'api_key': api_key[:8] + '...',  # 只记录部分密钥
            'endpoint': endpoint,
            'status': status,
            'ip': ip
        }
        
        log_file = self.api_keys.get('api_config', {}).get('log_file', 'api_access.log')
        with open(log_file, 'a', encoding='utf-8') as f:
            f.write(json.dumps(log_entry) + '\n')

# API密钥验证装饰器
def require_api_key(func):
    """API密钥验证装饰器"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        # 从请求中获取API密钥
        # 这里假设API密钥通过HTTP头传递
        # 实际实现取决于你的Web框架
        
        # 示例：Flask框架的实现
        from flask import request, jsonify
        
        api_key = request.headers.get('X-API-Key')
        if not api_key:
            return jsonify({
                'error': '缺少API密钥',
                'message': '请在请求头中包含X-API-Key'
            }), 401
        
        # 初始化密钥管理器
        key_manager = APIKeyManager()
        
        # 验证密钥
        is_valid, key_config, message = key_manager.validate_api_key(api_key)
        if not is_valid:
            key_manager.log_api_access(
                api_key, 
                request.path, 
                'REJECTED', 
                request.remote_addr
            )
            return jsonify({
                'error': 'API密钥验证失败',
                'message': message
            }), 403
        
        # 检查速率限制
        rate_ok, remaining = key_manager.check_rate_limit(api_key, key_config)
        if not rate_ok:
            key_manager.log_api_access(
                api_key, 
                request.path, 
                'RATE_LIMITED', 
                request.remote_addr
            )
            return jsonify({
                'error': '请求频率超限',
                'message': f'每分钟最多{key_config.get("rate_limit", 60)}次请求',
                'retry_after': 60
            }), 429
        
        # 更新使用统计
        key_id = None
        for k, v in key_manager.api_keys.items():
            if v['key'] == api_key:
                key_id = k
                break
        
        if key_id:
            key_manager.update_usage(key_id, key_config)
        
        # 记录成功的访问
        key_manager.log_api_access(
            api_key, 
            request.path, 
            'SUCCESS', 
            request.remote_addr
        )
        
        # 将密钥信息添加到请求上下文
        request.api_key_info = {
            'key_id': key_id,
            'config': key_config,
            'remaining_quota': remaining
        }
        
        # 添加剩余配额到响应头
        response = func(*args, **kwargs)
        if hasattr(response, 'headers'):
            response.headers['X-RateLimit-Remaining'] = str(remaining)
            response.headers['X-RateLimit-Limit'] = str(key_config.get('rate_limit', 60))
        
        return response
    
    return wrapper

4.3 安全的API端点实现

现在我们可以创建安全的API端点，这些端点要求有效的API密钥才能访问：

from flask import Flask, request, jsonify
import os

app = Flask(__name__)

# 初始化SOONet pipeline
soonet_pipeline = None  # 这里应该初始化你的SOONet模型

@app.route('/api/v1/locate', methods=['POST'])
@require_api_key
def locate_video():
    """视频时序定位API端点"""
    try:
        # 检查权限
        if 'query_execution' not in request.api_key_info['config']['permissions']:
            return jsonify({
                'error': '权限不足',
                'message': '该API密钥没有执行查询的权限'
            }), 403
        
        # 获取请求数据
        data = request.json
        if not data:
            return jsonify({'error': '请求体必须为JSON格式'}), 400
        
        query_text = data.get('query_text')
        video_path = data.get('video_path')
        
        if not query_text or not video_path:
            return jsonify({
                'error': '参数缺失',
                'message': 'query_text和video_path为必填参数'
            }), 400
        
        # 检查视频文件是否存在
        if not os.path.exists(video_path):
            return jsonify({
                'error': '文件不存在',
                'message': f'视频文件不存在: {video_path}'
            }), 404
        
        # 执行定位（这里调用SOONet的核心功能）
        # result = soonet_pipeline((query_text, video_path))
        
        # 模拟结果
        result = {
            'query': query_text,
            'video': video_path,
            'timestamps': [
                {'start': 75, 'end': 105, 'score': 0.87},
                {'start': 210, 'end': 240, 'score': 0.65}
            ],
            'status': 'success'
        }
        
        return jsonify(result), 200
        
    except Exception as e:
        return jsonify({
            'error': '处理失败',
            'message': str(e)
        }), 500

@app.route('/api/v1/status', methods=['GET'])
@require_api_key
def get_status():
    """获取服务状态"""
    return jsonify({
        'status': 'running',
        'version': '1.0.0',
        'timestamp': datetime.now().isoformat()
    }), 200

@app.route('/api/v1/usage', methods=['GET'])
@require_api_key
def get_usage():
    """获取API使用情况（仅管理员）"""
    # 检查管理员权限
    if 'view_logs' not in request.api_key_info['config']['permissions']:
        return jsonify({
            'error': '权限不足',
            'message': '需要管理员权限'
        }), 403
    
    key_manager = APIKeyManager()
    key_id = request.api_key_info['key_id']
    
    if key_id in key_manager.api_keys:
        usage_info = {
            'key_id': key_id,
            'total_calls': key_manager.api_keys[key_id].get('total_calls', 0),
            'last_used': key_manager.api_keys[key_id].get('last_used'),
            'rate_limit': key_manager.api_keys[key_id].get('rate_limit', 60)
        }
        return jsonify(usage_info), 200
    else:
        return jsonify({'error': '密钥不存在'}), 404

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000, ssl_context='adhoc')  # 启用HTTPS

5. 视频上传大小限制实现

视频文件上传是SOONet的重要功能，但如果不加限制，恶意用户可能会上传超大文件耗尽服务器资源。下面我们来实现视频上传的大小限制。

5.1 上传限制配置

创建 upload_config.yaml 配置文件：

# 视频上传配置
upload:
  # 全局限制
  global:
    max_file_size_mb: 500  # 单个文件最大500MB
    max_total_size_mb: 5000  # 总上传大小限制5GB
    allowed_extensions:  # 允许的文件扩展名
      - ".mp4"
      - ".avi"
      - ".mov"
      - ".mkv"
      - ".webm"
    max_files_per_hour: 10  # 每小时最多上传文件数
  
  # 用户组限制
  user_groups:
    admin:
      max_file_size_mb: 2000  # 管理员可上传2GB文件
      max_total_size_mb: 20000  # 总限制20GB
      max_files_per_hour: 50
    
    premium:
      max_file_size_mb: 1000  # 高级用户可上传1GB文件
      max_total_size_mb: 10000  # 总限制10GB
      max_files_per_hour: 30
    
    standard:
      max_file_size_mb: 500  # 标准用户500MB限制
      max_total_size_mb: 5000  # 总限制5GB
      max_files_per_hour: 10
    
    trial:
      max_file_size_mb: 100  # 试用用户100MB限制
      max_total_size_mb: 1000  # 总限制1GB
      max_files_per_hour: 5
  
  # 存储配置
  storage:
    upload_dir: "/var/soonet/uploads"  # 上传目录
    temp_dir: "/tmp/soonet_uploads"  # 临时目录
    cleanup_days: 7  # 7天后清理旧文件
    enable_compression: true  # 是否启用压缩
    compression_quality: 85  # 压缩质量（0-100）
  
  # 安全检查
  security:
    scan_for_malware: true  # 扫描恶意软件
    check_file_type: true  # 检查文件类型
    block_executables: true  # 阻止可执行文件
    max_filename_length: 255  # 最大文件名长度
  
  # 监控配置
  monitoring:
    enable_logging: true
    log_file: "/var/log/soonet_uploads.log"
    alert_on_limit: true  # 达到限制时告警
    alert_email: "admin@example.com"

5.2 文件上传处理器

创建一个专门处理文件上传的类，包含所有安全检查：

import os
import shutil
import magic  # 用于文件类型检测
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, Tuple, Optional
import yaml
import hashlib

class VideoUploadHandler:
    """视频上传处理器"""
    
    def __init__(self, config_path: str = "upload_config.yaml"):
        self.config = self.load_config(config_path)
        self.ensure_directories()
        self.upload_stats = self.load_upload_stats()
    
    def load_config(self, config_path: str) -> Dict:
        """加载上传配置"""
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        except FileNotFoundError:
            print(f"警告：上传配置文件 {config_path} 不存在，使用默认配置")
            return {
                'upload': {
                    'global': {
                        'max_file_size_mb': 500,
                        'allowed_extensions': ['.mp4', '.avi', '.mov', '.mkv']
                    },
                    'storage': {
                        'upload_dir': './uploads',
                        'temp_dir': './temp'
                    }
                }
            }
    
    def ensure_directories(self):
        """确保必要的目录存在"""
        upload_dir = self.config['upload']['storage']['upload_dir']
        temp_dir = self.config['upload']['storage']['temp_dir']
        
        os.makedirs(upload_dir, exist_ok=True)
        os.makedirs(temp_dir, exist_ok=True)
    
    def load_upload_stats(self) -> Dict:
        """加载上传统计"""
        stats_file = os.path.join(
            self.config['upload']['storage']['upload_dir'],
            '.upload_stats.json'
        )
        
        if os.path.exists(stats_file):
            import json
            with open(stats_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        else:
            return {}
    
    def save_upload_stats(self):
        """保存上传统计"""
        stats_file = os.path.join(
            self.config['upload']['storage']['upload_dir'],
            '.upload_stats.json'
        )
        
        import json
        with open(stats_file, 'w', encoding='utf-8') as f:
            json.dump(self.upload_stats, f, indent=2)
    
    def get_user_limits(self, username: str) -> Dict:
        """获取用户的上传限制"""
        user_group = self.get_user_group(username)
        group_limits = self.config['upload']['user_groups'].get(
            user_group, 
            self.config['upload']['user_groups']['standard']
        )
        
        return {
            'max_file_size': group_limits['max_file_size_mb'] * 1024 * 1024,  # 转换为字节
            'max_total_size': group_limits['max_total_size_mb'] * 1024 * 1024,
            'max_files_per_hour': group_limits['max_files_per_hour'],
            'allowed_extensions': self.config['upload']['global']['allowed_extensions']
        }
    
    def get_user_group(self, username: str) -> str:
        """获取用户所属组（简化实现）"""
        # 这里应该根据用户信息确定用户组
        # 简化实现：根据用户名前缀判断
        if username.startswith('admin_'):
            return 'admin'
        elif username.startswith('premium_'):
            return 'premium'
        elif username.startswith('trial_'):
            return 'trial'
        else:
            return 'standard'
    
    def validate_file(self, file_path: str, username: str) -> Tuple[bool, str]:
        """验证上传文件"""
        try:
            # 检查文件是否存在
            if not os.path.exists(file_path):
                return False, "文件不存在"
            
            # 获取文件信息
            file_size = os.path.getsize(file_path)
            file_ext = os.path.splitext(file_path)[1].lower()
            
            # 获取用户限制
            limits = self.get_user_limits(username)
            
            # 检查文件大小
            if file_size > limits['max_file_size']:
                max_mb = limits['max_file_size'] / (1024 * 1024)
                return False, f"文件大小超过限制（最大{max_mb}MB）"
            
            # 检查文件扩展名
            if file_ext not in limits['allowed_extensions']:
                allowed = ', '.join(limits['allowed_extensions'])
                return False, f"不支持的文件类型，允许的类型：{allowed}"
            
            # 检查文件类型（防止文件扩展名欺骗）
            if self.config['upload']['security']['check_file_type']:
                file_type = self.get_file_type(file_path)
                if not self.is_video_file(file_type):
                    return False, f"检测到非视频文件：{file_type}"
            
            # 检查文件名长度
            filename = os.path.basename(file_path)
            if len(filename) > self.config['upload']['security']['max_filename_length']:
                return False, "文件名过长"
            
            # 检查恶意文件（简化实现）
            if self.config['upload']['security']['scan_for_malware']:
                if self.detect_malicious_patterns(file_path):
                    return False, "文件可能包含恶意内容"
            
            # 检查可执行文件
            if self.config['upload']['security']['block_executables']:
                if self.is_executable(file_path):
                    return False, "可执行文件被禁止上传"
            
            # 检查用户上传频率
            if not self.check_upload_rate(username):
                return False, "上传频率超限，请稍后再试"
            
            # 检查用户总上传量
            if not self.check_total_usage(username, file_size):
                user_total_mb = self.get_user_total_usage(username) / (1024 * 1024)
                max_total_mb = limits['max_total_size'] / (1024 * 1024)
                return False, f"总上传量超限（已用{user_total_mb:.1f}MB，限制{max_total_mb}MB）"
            
            return True, "验证通过"
            
        except Exception as e:
            return False, f"文件验证失败：{str(e)}"
    
    def get_file_type(self, file_path: str) -> str:
        """获取文件真实类型"""
        try:
            import magic
            mime = magic.Magic(mime=True)
            return mime.from_file(file_path)
        except:
            # 如果magic不可用，使用备用方法
            import mimetypes
            mime_type, _ = mimetypes.guess_type(file_path)
            return mime_type or "application/octet-stream"
    
    def is_video_file(self, mime_type: str) -> bool:
        """检查是否为视频文件"""
        video_mimes = [
            'video/mp4',
            'video/avi',
            'video/quicktime',
            'video/x-matroska',
            'video/webm',
            'video/x-msvideo'
        ]
        return mime_type in video_mimes
    
    def detect_malicious_patterns(self, file_path: str) -> bool:
        """检测恶意模式（简化实现）"""
        # 这里应该实现更复杂的恶意文件检测
        # 简化实现：检查文件头
        try:
            with open(file_path, 'rb') as f:
                header = f.read(100)  # 读取前100字节
            
            # 检查常见的恶意模式（简化）
            malicious_patterns = [
                b'MZ',  # Windows可执行文件
                b'#!/',  # 脚本文件
                b'%PDF',  # PDF可能包含恶意代码
            ]
            
            for pattern in malicious_patterns:
                if header.startswith(pattern):
                    return True
                    
            return False
        except:
            return False
    
    def is_executable(self, file_path: str) -> bool:
        """检查是否为可执行文件"""
        # 检查文件扩展名
        executable_extensions = ['.exe', '.bat', '.sh', '.bin', '.app']
        file_ext = os.path.splitext(file_path)[1].lower()
        
        if file_ext in executable_extensions:
            return True
        
        # 检查文件权限（Unix系统）
        if os.name == 'posix':
            import stat
            mode = os.stat(file_path).st_mode
            if mode & stat.S_IXUSR or mode & stat.S_IXGRP or mode & stat.S_IXOTH:
                return True
        
        return False
    
    def check_upload_rate(self, username: str) -> bool:
        """检查用户上传频率"""
        current_hour = datetime.now().strftime("%Y-%m-%d %H")
        hour_key = f"{username}:{current_hour}"
        
        if hour_key not in self.upload_stats:
            self.upload_stats[hour_key] = {
                'count': 0,
                'total_size': 0
            }
        
        user_limits = self.get_user_limits(username)
        current_count = self.upload_stats[hour_key]['count']
        
        if current_count >= user_limits['max_files_per_hour']:
            return False
        
        # 更新计数
        self.upload_stats[hour_key]['count'] += 1
        self.save_upload_stats()
        
        return True
    
    def check_total_usage(self, username: str, new_file_size: int) -> bool:
        """检查用户总上传量"""
        user_total = self.get_user_total_usage(username)
        user_limits = self.get_user_limits(username)
        
        return (user_total + new_file_size) <= user_limits['max_total_size']
    
    def get_user_total_usage(self, username: str) -> int:
        """获取用户总上传量"""
        total = 0
        for key, stats in self.upload_stats.items():
            if key.startswith(f"{username}:"):
                total += stats.get('total_size', 0)
        return total
    
    def generate_unique_filename(self, original_filename: str) -> str:
        """生成唯一文件名"""
        import uuid
        import time
        
        # 获取文件扩展名
        ext = os.path.splitext(original_filename)[1]
        
        # 生成唯一ID（时间戳+UUID）
        timestamp = int(time.time())
        unique_id = str(uuid.uuid4())[:8]
        
        # 添加原始文件名哈希（前8位）
        name_hash = hashlib.md5(original_filename.encode()).hexdigest()[:8]
        
        return f"{timestamp}_{unique_id}_{name_hash}{ext}"
    
    def save_uploaded_file(self, temp_path: str, username: str, original_filename: str) -> Tuple[bool, str, str]:
        """保存上传的文件"""
        try:
            # 验证文件
            is_valid, message = self.validate_file(temp_path, username)
            if not is_valid:
                return False, message, ""
            
            # 生成唯一文件名
            unique_filename = self.generate_unique_filename(original_filename)
            upload_dir = self.config['upload']['storage']['upload_dir']
            final_path = os.path.join(upload_dir, unique_filename)
            
            # 如果需要压缩，先压缩
            if self.config['upload']['storage']['enable_compression']:
                compressed_path = self.compress_video(temp_path)
                if compressed_path:
                    shutil.move(compressed_path, final_path)
                    os.remove(temp_path)  # 删除临时文件
                else:
                    shutil.move(temp_path, final_path)
            else:
                shutil.move(temp_path, final_path)
            
            # 更新上传统计
            file_size = os.path.getsize(final_path)
            current_hour = datetime.now().strftime("%Y-%m-%d %H")
            hour_key = f"{username}:{current_hour}"
            
            if hour_key not in self.upload_stats:
                self.upload_stats[hour_key] = {
                    'count': 0,
                    'total_size': 0
                }
            
            self.upload_stats[hour_key]['total_size'] += file_size
            self.save_upload_stats()
            
            # 记录上传日志
            self.log_upload(username, original_filename, unique_filename, file_size)
            
            return True, "上传成功", final_path
            
        except Exception as e:
            return False, f"保存文件失败：{str(e)}", ""
    
    def compress_video(self, input_path: str) -> Optional[str]:
        """压缩视频文件（简化实现）"""
        # 这里应该使用ffmpeg等工具进行实际压缩
        # 简化实现：直接返回原路径
        return input_path
    
    def log_upload(self, username: str, original_name: str, saved_name: str, file_size: int):
        """记录上传日志"""
        if not self.config['upload']['monitoring']['enable_logging']:
            return
        
        log_entry = {
            'timestamp': datetime.now().isoformat(),
            'username': username,
            'original_filename': original_name,
            'saved_filename': saved_name,
            'file_size': file_size,
            'file_size_mb': file_size / (1024 * 1024)
        }
        
        log_file = self.config['upload']['monitoring']['log_file']
        try:
            import json
            with open(log_file, 'a', encoding='utf-8') as f:
                f.write(json.dumps(log_entry) + '\n')
        except:
            pass  # 日志记录失败不影响主要功能
    
    def cleanup_old_files(self):
        """清理旧文件"""
        try:
            upload_dir = self.config['upload']['storage']['upload_dir']
            cleanup_days = self.config['upload']['storage']['cleanup_days']
            
            cutoff_time = datetime.now() - timedelta(days=cleanup_days)
            
            for file_path in Path(upload_dir).glob('*'):
                if file_path.is_file():
                    file_time = datetime.fromtimestamp(file_path.stat().st_mtime)
                    if file_time < cutoff_time:
                        os.remove(file_path)
                        print(f"清理旧文件：{file_path}")
        except Exception as e:
            print(f"清理文件时出错：{str(e)}")

5.3 集成到Gradio界面

现在我们将文件上传限制集成到Gradio界面中：

import gradio as gr
from upload_handler import VideoUploadHandler

class SecureSOONetApp:
    """安全的SOONet应用"""
    
    def __init__(self):
        self.upload_handler = VideoUploadHandler()
        self.current_user = None
    
    def create_interface(self):
        """创建带上传限制的界面"""
        
        with gr.Blocks(title="SOONet - 安全视频时序定位系统") as demo:
            # 登录界面（同上，略）
            
            # 主功能界面
            with gr.Column(visible=False) as main_section:
                gr.Markdown("## 🎯 SOONet 视频时序定位")
                
                # 显示当前用户和限制信息
                with gr.Row():
                    user_info = gr.Markdown("")
                    limit_info = gr.Markdown("")
                
                # 查询文本输入
                query_text = gr.Textbox(
                    label="查询文本",
                    placeholder="输入英文描述，如：a man takes food out of the refrigerator",
                    lines=2
                )
                
                # 安全的视频上传组件
                video_input = gr.File(
                    label="上传视频",
                    file_types=["video"],
                    file_count="single"
                )
                
                # 上传状态显示
                upload_status = gr.Textbox(
                    label="上传状态",
                    interactive=False,
                    visible=False
                )
                
                # 定位按钮
                locate_button = gr.Button("🔍 开始定位", variant="primary")
                
                # 结果显示
                result_output = gr.Textbox(label="定位结果", interactive=False)
                
                # 退出按钮
                logout_button = gr.Button("退出登录", variant="secondary")
            
            # 更新用户信息
            def update_user_info(username):
                """更新用户信息和限制显示"""
                if not username:
                    return "", ""
                
                limits = self.upload_handler.get_user_limits(username)
                max_mb = limits['max_file_size'] / (1024 * 1024)
                max_total_mb = limits['max_total_size'] / (1024 * 1024)
                
                user_text = f"**当前用户**：{username}"
                limit_text = f"**上传限制**：单文件≤{max_mb}MB，总量≤{max_total_mb}MB"
                
                return user_text, limit_text
            
            # 文件上传处理
            def handle_file_upload(file, username):
                """处理文件上传"""
                if not file:
                    return None, "请选择文件", gr.update(visible=True)
                
                if not username:
                    return None, "请先登录", gr.update(visible=True)
                
                # 保存上传的文件
                success, message, saved_path = self.upload_handler.save_uploaded_file(
                    file.name, username, os.path.basename(file.name)
                )
                
                if success:
                    return saved_path, f"✅ {message}", gr.update(visible=False)
                else:
                    return None, f"❌ {message}", gr.update(visible=True)
            
            # 绑定事件
            # ...（登录/退出事件绑定，同上）
            
            # 文件上传事件
            video_input.upload(
                fn=lambda file: handle_file_upload(file, self.current_user),
                inputs=[video_input],
                outputs=[video_input, upload_status, upload_status]
            )
            
            # 定位功能（需要检查文件大小）
            def secure_locate(query, video_path):
                """安全的定位功能"""
                if not self.current_user:
                    return "请先登录"
                
                if not video_path:
                    return "请先上传视频"
                
                # 检查文件大小（再次验证）
                if os.path.exists(video_path):
                    file_size = os.path.getsize(video_path)
                    limits = self.upload_handler.get_user_limits(self.current_user)
                    
                    if file_size > limits['max_file_size']:
                        max_mb = limits['max_file_size'] / (1024 * 1024)
                        return f"文件大小超过限制（最大{max_mb}MB）"
                
                # 调用SOONet定位功能
                # result = soonet_pipeline((query, video_path))
                
                # 模拟结果
                result = f"查询: {query}\n"
                result += f"视频: {os.path.basename(video_path)}\n"
                result += "定位结果: 00:01:15 - 00:01:45 (置信度: 0.87)"
                
                return result
            
            locate_button.click(
                fn=secure_locate,
                inputs=[query_text, video_input],
                outputs=result_output
            )
        
        return demo
    
    def run(self):
        """运行应用"""
        # 定期清理旧文件
        import threading
        import time
        
        def cleanup_task():
            while True:
                time.sleep(3600)  # 每小时清理一次
                self.upload_handler.cleanup_old_files()
        
        # 启动清理线程
        cleanup_thread = threading.Thread(target=cleanup_task, daemon=True)
        cleanup_thread.start()
        
        # 启动Gradio应用
        demo = self.create_interface()
        demo.launch(
            server_name="0.0.0.0",
            server_port=7860,
            share=False
        )

# 启动应用
if __name__ == "__main__":
    app = SecureSOONetApp()
    app.run()

6. 完整部署与配置指南

现在我们已经实现了所有安全功能，接下来看看如何完整部署这个加固版的SOONet系统。

6.1 项目结构

部署前，先确保项目结构正确：

/root/multi-modal_soonet_video-temporal-grounding/
├── app.py                          # 主应用文件（已修改）
├── requirements.txt                # Python依赖
├── users_config.yaml              # 用户认证配置
├── api_keys_config.yaml           # API密钥配置
├── upload_config.yaml             # 上传限制配置
├── auth/                          # 认证模块
│   ├── __init__.py
│   ├── session_manager.py         # 会话管理
│   └── password_utils.py          # 密码工具
├── api/                           # API模块
│   ├── __init__.py
│   ├── key_manager.py             # API密钥管理
│   └── rate_limiter.py           # 速率限制
├── upload/                        # 上传模块
│   ├── __init__.py
│   ├── upload_handler.py          # 上传处理器
│   └── file_validator.py          # 文件验证
├── logs/                          # 日志目录
│   ├── access.log                 # 访问日志
│   └── uploads.log               # 上传日志
└── uploads/                       # 上传文件存储
    └── .gitkeep

6.2 依赖安装

创建 requirements.txt 文件：

# 基础依赖
torch>=1.10.0
torchvision>=0.11.0
modelscope>=1.0.0
gradio==6.4.0
opencv-python>=4.5.0

# 安全加固新增依赖
Flask>=2.0.0
Flask-CORS>=3.0.0
PyYAML>=6.0
bcrypt>=4.0.0
redis>=4.0.0
python-magic>=0.4.0
python-magic-bin>=0.4.0  # Windows系统需要

# 文本处理
ftfy>=6.0.0
regex>=2021.0.0

# 注意：numpy需要<2.0版本
numpy<2.0

安装所有依赖：

pip install -r requirements.txt

6.3 配置文件设置

1. 用户认证配置 (users_config.yaml)：

# 生产环境用户配置
users:
  # 管理员
  admin:
    password_hash: "$2b$12$YourHashedPasswordHere"  # 使用bcrypt哈希
    role: "admin"
    permissions:
      - "full_access"
      - "upload_large_files"
      - "manage_users"
      - "view_logs"
    email: "admin@yourcompany.com"
    created_at: "2024-01-01"
  
  # 开发人员
  developer:
    password_hash: "$2b$12$AnotherHashedPassword"
    role: "premium"
    permissions:
      - "video_upload"
      - "query_execution"
      - "result_retrieval"
    email: "dev@yourcompany.com"
    created_at: "2024-01-01"
  
  # 普通用户
  user:
    password_hash: "$2b$12$UserHashedPassword"
    role: "standard"
    permissions:
      - "video_upload"
      - "query_execution"
    email: "user@yourcompany.com"
    created_at: "2024-01-01"

session:
  timeout_minutes: 30
  max_sessions_per_user: 3
  secure_cookies: true  # 仅HTTPS传输cookie

login:
  max_attempts: 5
  lockout_minutes: 15
  require_2fa: false  # 是否要求双因素认证

2. API密钥配置 (api_keys_config.yaml)：

api_keys:
  # 内部服务密钥
  internal_service_001:
    key: "sk_live_xxxxxxxxxxxxxxxx"
    name: "内部服务密钥"
    owner: "backend_service"
    created_at: "2024-01-01"
    expires_at: "2024-12-31"
    rate_limit: 1000
    total_calls: 0
    last_used: null
    permissions:
      - "video_upload"
      - "query_execution"
      - "result_retrieval"
      - "batch_processing"
    enabled: true
    ip_whitelist:  # IP白名单
      - "10.0.0.0/8"
      - "192.168.1.0/24"
  
  # 合作伙伴API密钥
  partner_001:
    key: "sk_partner_yyyyyyyyyyyy"
    name: "合作伙伴A"
    owner: "partner_a"
    created_at: "2024-01-01"
    expires_at: "2024-06-30"
    rate_limit: 100
    total_calls: 0
    last_used: null
    permissions:
      - "query_execution"
      - "result_retrieval"
    enabled: true
    monthly_quota: 10000  # 每月调用限额
  
  # 移动应用密钥
  mobile_app_001:
    key: "sk_mobile_zzzzzzzzzzzz"
    name: "移动应用"
    owner: "mobile_team"
    created_at: "2024-01-01"
    expires_at: null
    rate_limit: 60
    total_calls: 0
    last_used: null
    permissions:
      - "query_execution"
    enabled: true
    device_limit: 1000  # 最大设备数

api_config:
  default_rate_limit: 60
  require_https: true
  enable_logging: true
  log_file: "/var/log/soonet_api.log"
  log_retention_days: 30
  blocked_ips: []
  enable_metrics: true
  metrics_port: 9090

3. 上传限制配置 (upload_config.yaml)：

upload:
  global:
    max_file_size_mb: 500
    max_total_size_mb: 5000
    allowed_extensions:
      - ".mp4"
      - ".avi"
      - ".mov"
      - ".mkv"
      - ".webm"
      - ".flv"
      - ".wmv"
    max_files_per_hour: 10
    max_files_per_day: 50
  
  user_groups:
    admin:
      max_file_size_mb: 5000
      max_total_size_mb: 50000
      max_files_per_hour: 100
      max_files_per_day: 500
    
    premium:
      max_file_size_mb: 2000
      max_total_size_mb: 20000
      max_files_per_hour: 50
      max_files_per_day: 200
    
    standard:
      max_file_size_mb: 500
      max_total_size_mb: 5000
      max_files_per_hour: 10
      max_files_per_day: 50
    
    trial:
      max_file_size_mb: 100
      max_total_size_mb: 1000
      max_files_per_hour: 5
      max_files_per_day: 20
  
  storage:
    upload_dir: "/data/soonet/uploads"
    temp_dir: "/tmp/soonet_uploads"
    cleanup_days: 7
    enable_compression: true
    compression_quality: 85
    max_storage_gb: 100  # 最大存储空间
    enable_encryption: true  # 启用文件加密
    
    # 存储分区配置
    partitions:
      - path: "/data/soonet/uploads/partition1"
        max_size_gb: 50
      - path: "/data/soonet/uploads/partition2"
        max_size_gb: 50
  
  security:
    scan_for_malware: true
    check_file_type: true
    block_executables: true
    max_filename_length: 255
    block_suspicious_names: true  # 阻止可疑文件名
    virus_scan_command: "clamscan --no-summary"  # 病毒扫描命令
    
    # 文件类型深度检查
    deep_file_check: true
    allowed_mime_types:
      - "video/mp4"
      - "video/avi"
      - "video/quicktime"
      - "video/x-matroska"
      - "video/webm"
      - "video/x-msvideo"
  
  monitoring:
    enable_logging: true
    log_file: "/var/log/soonet_uploads.log"
    alert_on_limit: true
    alert_email: "alerts@yourcompany.com"
    alert_webhook: "https://hooks.slack.com/services/xxx"
    
    # 监控指标
    enable_metrics: true
    metrics_port: 9091
    
    # 自动清理
    auto_cleanup: true
    cleanup_schedule: "0 2 * * *"  # 每天凌晨2点

6.4 启动脚本

创建启动脚本 start_soonet.sh：

#!/bin/bash

# SOONet安全加固版启动脚本

set -e

echo "========================================"
echo "启动 SOONet 安全加固版"
echo "========================================"

# 检查Python版本
PYTHON_VERSION=$(python3 --version | cut -d' ' -f2)
echo "Python版本: $PYTHON_VERSION"

if [[ "$PYTHON_VERSION" < "3.7" ]]; then
    echo "错误: 需要Python 3.7或更高版本"
    exit 1
fi

# 检查依赖
echo "检查依赖..."
pip install -r requirements.txt

# 创建必要的目录
echo "创建目录..."
mkdir -p logs
mkdir -p uploads
mkdir -p /tmp/soonet_uploads

# 设置权限
echo "设置权限..."
chmod 755 uploads
chmod 755 logs
chmod 600 *.yaml  # 配置文件设置为只读

# 检查配置文件
echo "检查配置文件..."
if [ ! -f "users_config.yaml" ]; then
    echo "警告: users_config.yaml 不存在，创建示例配置"
    cp users_config.example.yaml users_config.yaml
fi

if [ ! -f "api_keys_config.yaml" ]; then
    echo "警告: api_keys_config.yaml 不存在，创建示例配置"
    cp api_keys_config.example.yaml api_keys_config.yaml
fi

if [ ! -f "upload_config.yaml" ]; then
    echo "警告: upload_config.yaml 不存在，创建示例配置"
    cp upload_config.example.yaml upload_config.yaml
fi

# 检查模型文件
echo "检查模型文件..."
MODEL_DIR="/root/ai-models/iic/multi-modal_soonet_video-temporal-grounding"
if [ ! -d "$MODEL_DIR" ]; then
    echo "错误: 模型目录不存在: $MODEL_DIR"
    exit 1
fi

# 设置环境变量
export SOONET_MODEL_PATH="$MODEL_DIR"
export PYTHONPATH="/root/multi-modal_soonet_video-temporal-grounding:$PYTHONPATH"

# 启动Redis（用于速率限制）
echo "启动Redis..."
if ! command -v redis-server &> /dev/null; then
    echo "警告: Redis未安装，速率限制将使用内存模式"
else
    # 检查Redis是否已在运行
    if ! pgrep -x "redis-server" > /dev/null; then
        redis-server --daemonize yes
        echo "Redis已启动"
    else
        echo "Redis已在运行"
    fi
fi

# 启动监控服务（可选）
echo "启动监控服务..."
python monitoring_service.py &

# 启动SOONet应用
echo "启动SOONet应用..."
cd /root/multi-modal_soonet_video-temporal-grounding

# 使用gunicorn生产环境部署（推荐）
if command -v gunicorn &> /dev/null; then
    echo "使用gunicorn启动..."
    gunicorn \
        --bind 0.0.0.0:7860 \
        --workers 4 \
        --threads 2 \
        --timeout 120 \
        --access-logfile logs/access.log \
        --error-logfile logs/error.log \
        app:app
else
    echo "使用Python直接启动..."
    python app.py
fi

echo "========================================"
echo "SOONet 已启动"
echo "访问地址: http://localhost:7860"
echo "API地址: http://localhost:5000/api/v1"
echo "========================================"

6.5 监控与维护脚本

创建监控脚本 monitor_soonet.py：

#!/usr/bin/env python3
"""
SOONet监控脚本
监控系统状态、资源使用、安全事件等
"""

import psutil
import requests
import json
import time
from datetime import datetime
import logging
from typing import Dict, Any

class SOONetMonitor:
    """SOONet监控器"""
    
    def __init__(self, config_path: str = "monitor_config.yaml"):
        self.config = self.load_config(config_path)
        self.setup_logging()
        
    def load_config(self, config_path: str) -> Dict[str, Any]:
        """加载监控配置"""
        import yaml
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                return yaml.safe_load(f)
        except FileNotFoundError:
            return {
                'monitoring': {
                    'check_interval': 60,  # 检查间隔（秒）
                    'alerts': {
                        'cpu_threshold': 80,  # CPU使用率阈值
                        'memory_threshold': 85,  # 内存使用率阈值
                        'disk_threshold': 90,  # 磁盘使用率阈值
                        'api_error_threshold': 10,  # API错误阈值
                        'upload_failure_threshold': 5  # 上传失败阈值
                    },
                    'notifications': {
                        'enable_email': False,
                        'enable_slack': False,
                        'enable_webhook': False
                    }
                }
            }
    
    def setup_logging(self):
        """设置日志"""
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler('logs/monitor.log'),
                logging.StreamHandler()
            ]
        )
        self.logger = logging.getLogger('SOONetMonitor')
    
    def check_system_resources(self) -> Dict[str, Any]:
        """检查系统资源"""
        resources = {
            'timestamp': datetime.now().isoformat(),
            'cpu_percent': psutil.cpu_percent(interval=1),
            'memory_percent': psutil.virtual_memory().percent,
            'disk_percent': psutil.disk_usage('/').percent,
            'network_io': psutil.net_io_counters()._asdict(),
            'process_count': len(psutil.pids())
        }
        
        # 检查阈值
        alerts = []
        if resources['cpu_percent'] > self.config['monitoring']['alerts']['cpu_threshold']:
            alerts.append(f"CPU使用率过高: {resources['cpu_percent']}%")
        
        if resources['memory_percent'] > self.config['monitoring']['alerts']['memory_threshold']:
            alerts.append(f"内存使用率过高: {resources['memory_percent']}%")
        
        if resources['disk_percent'] > self.config['monitoring']['alerts']['disk_threshold']:
            alerts.append(f"磁盘使用率过高: {resources['disk_percent']}%")
        
        resources['alerts'] = alerts
        return resources
    
    def check_service_health(self) -> Dict[str, Any]:
        """检查服务健康状态"""
        health = {
            'timestamp': datetime.now().isoformat(),
            'services': {}
        }
        
        # 检查Gradio服务
        try:
            response = requests.get('http://localhost:7860', timeout=5)
            health['services']['gradio'] = {
                'status': 'up' if response.status_code == 200 else 'down',
                'response_time': response.elapsed.total_seconds(),
                'status_code': response.status_code
            }
        except requests.RequestException as e:
            health['services']['gradio'] = {
                'status': 'down',
                'error': str(e)
            }
        
        # 检查API服务
        try:
            response = requests.get('http://localhost:5000/api/v1/status', timeout=5)
            health['services']['api'] = {
                'status': 'up' if response.status_code == 200 else 'down',
                'response_time': response.elapsed.total_seconds(),
                'status_code': response.status_code
            }
        except requests.RequestException as e:
            health['services']['api'] = {
                'status': 'down',
                'error': str(e)
            }
        
        return health
    
    def check_security_events(self) -> Dict[str, Any]:
        """检查安全事件"""
        import os
        import glob
        
        security = {
            'timestamp': datetime.now().isoformat(),
            'events': []
        }
        
        # 检查失败登录尝试
        auth_log = 'logs/auth.log'
        if os.path.exists(auth_log):
            with open(auth_log, 'r', encoding='utf-8') as f:
                lines = f.readlines()[-100:]  # 最近100行
                
            failed_attempts = 0
            for line in lines:
                if 'FAILED' in line or '失败' in line:
                    failed_attempts += 1
            
            if failed_attempts > 0:
                security['events'].append({
                    'type': 'failed_login',
                    'count': failed_attempts,
                    'message': f'最近有{failed_attempts}次失败登录尝试'
                })
        
        # 检查API滥用
        api_log = 'logs/api_access.log'
        if os.path.exists(api_log):
            with open(api_log, 'r', encoding='utf-8') as f:
                lines = f.readlines()[-100:]
            
            rate_limited = 0
            for line in lines:
                if 'RATE_LIMITED' in line:
                    rate_limited += 1
            
            if rate_limited > self.config['monitoring']['alerts']['api_error_threshold']:
                security['events'].append({
                    'type': 'api_abuse',
                    'count': rate_limited,
                    'message': f'API限流次数过多: {rate_limited}次'
                })
        
        return security
    
    def send_alert(self, alert_type: str, message: str, level: str = 'warning'):
        """发送