deploy_all.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
驾驶员手部检测系统 - 一键部署脚本

此脚本将完成以下操作：
1. 部署CDK堆栈（包括所有AWS资源）
2. 创建统一的Parameter Store参数
3. 构建并部署Fargate服务（可选）
4. 验证部署状态
"""

import os
import sys
import subprocess
import boto3
import time
import json
from botocore.exceptions import ClientError
from typing import Dict, Any, List, Optional

# 添加src目录到路径
sys.path.append(os.path.join(os.path.dirname(__file__), 'src'))

# 导入配置管理器
try:
    from utils.parameter_manager import ConfigManager
except ImportError as e:
    print(f"警告: 无法导入模块: {e}")
    ConfigManager = None

class OneClickDeployer:
    def __init__(self, region='us-east-1'):
        self.region = region
        self.ssm = boto3.client('ssm', region_name=region)
        self.base_path = '/driver-hand-detection'
        self.config_manager = None
        
        # 尝试初始化配置管理器
        if ConfigManager:
            try:
                self.config_manager = ConfigManager(environment='config', region_name=region)
                self.log("✅ 配置管理器初始化成功")
            except Exception as e:
                self.log(f"⚠️ 配置管理器初始化失败: {e}", "WARNING")
        
    def log(self, message, level="INFO"):
        """输出日志"""
        timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{timestamp}] {level}: {message}")
        
    def run_command(self, command, description):
        """执行命令"""
        self.log(f"执行: {description}")
        self.log(f"命令: {command}")
        
        result = subprocess.run(command, shell=True, capture_output=True, text=True)
        
        if result.returncode == 0:
            self.log(f"✅ {description} - 成功")
            if result.stdout.strip():
                print(result.stdout)
            return True
        else:
            self.log(f"❌ {description} - 失败", "ERROR")
            if result.stderr.strip():
                print(result.stderr)
            return False
    
    def create_parameter(self, name, value, description="", param_type="String"):
        """创建Parameter Store参数"""
        try:
            full_name = f"{self.base_path}/{name}"
            self.ssm.put_parameter(
                Name=full_name,
                Value=str(value),
                Type=param_type,
                Description=description,
                Overwrite=True
            )
            # 对于SecureString类型，不显示实际值
            display_value = "[SecureString]" if param_type == "SecureString" else value
            self.log(f"✅ 参数已设置: {name} = {display_value}")
            return True
        except ClientError as e:
            self.log(f"❌ 设置参数失败 {name}: {e}", "ERROR")
            return False
    
    def create_unified_parameters(self, env='config'):
        """创建统一的参数配置"""
        self.log(f"🔧 创建统一的Parameter Store参数 (环境: {env})...")
        
        # 检查是否为中国区域
        is_china_region = self.region.startswith('cn-')
        if is_china_region:
            self.log("🇨🇳 检测到中国区域，将设置中国区域特定参数")
        
        if self.config_manager:
            # 使用配置管理器初始化参数
            self.log("使用配置管理器初始化参数...")
            self.config_manager.init_default_config()
            
            # 如果是中国区域，设置中国区域特定参数
            if self.region.startswith('cn-'):
                self.log("设置中国区域特定参数...")
                china_params = {
                    'ai/model_id': 'Pro/deepseek-ai/DeepSeek-R1',
                    'api-endpoint': 'https://api.siliconflow.cn/v1',
                    'monitoring/log_level': 'INFO',
                    'error/max_retry_attempts': '5',
                }
                
                for key, value in china_params.items():
                    self.config_manager.update_config(key, value)
                    self.log(f"✅ {key}: {value}")
            
            # 验证参数
            try:
                validation = self.config_manager.validate_config()
                if validation['valid']:
                    self.log("✅ 参数验证通过")
                else:
                    self.log("⚠️ 参数验证存在问题:", "WARNING")
                    for error in validation.get('errors', []):
                        self.log(f"  - {error}", "ERROR")
                    for warning in validation.get('warnings', []):
                        self.log(f"  - {warning}", "WARNING")
            except Exception as e:
                self.log(f"⚠️ 参数验证失败: {e}", "WARNING")
            
            return True
        else:
            # 使用内置方法创建参数
            return self.create_all_parameters(self.region.startswith('cn-'))
    
    def create_all_parameters(self, is_china_region=False):
        """创建所有必需的参数（内置方法）"""
        self.log("🔧 创建Parameter Store参数（内置方法）...")
        
        # 基本参数列表
        parameters = [
            # 视频处理配置
            ("config/video/frames_per_segment", "30", "每个Lambda处理的帧数"),
            ("config/video/frame_interval", "2", "帧提取间隔（秒）"),
            ("config/video/supported_formats", "mp4,avi,mov,mkv", "支持的视频格式"),
            ("config/video/max_video_size_gb", "2", "最大视频文件大小(GB)"),
            ("config/video/frame_quality", "90", "JPEG质量(0-100)"),
            
            # Lambda配置
            ("config/lambda/concurrency_limit", "20", "Lambda并发处理数量限制"),
            ("config/lambda/timeout", "900", "Lambda函数超时时间（秒）"),
            ("config/lambda/video_processor_memory", "2048", "视频处理器内存"),
            ("config/lambda/frame_extractor_memory", "1024", "帧提取器内存"),
            ("config/lambda/ai_analyzer_memory", "1024", "AI分析器内存"),
            ("config/lambda/result_processor_memory", "512", "结果处理器内存"),
            
            # S3配置
            ("config/s3/input_prefix", "input/videos/", "输入视频前缀"),
            ("config/s3/processing_prefix", "processing/jobs/", "处理中文件前缀"),
            ("config/s3/output_prefix", "output/jobs/", "输出结果前缀"),
            ("config/s3/cleanup_temp_files", "true", "是否清理临时文件"),
            
            # Step Functions配置
            ("config/stepfunctions/max_retries", "3", "最大重试次数"),
            ("config/stepfunctions/retry_interval", "2", "重试间隔"),
            ("config/stepfunctions/execution_timeout", "3600", "执行超时时间"),
            
            # 通知配置
            ("config/notification/enabled", "true", "是否启用通知"),
            ("config/notification/email_subject", "驾驶员手部检测分析完成", "邮件主题"),
            
            # 监控配置
            ("config/monitoring/log_level", "INFO", "日志级别"),
            ("config/monitoring/metrics_enabled", "true", "是否启用指标"),
            
            # 性能优化配置
            ("config/performance/enable_parallel_processing", "true", "是否启用并行处理"),
            ("config/performance/batch_size", "5", "批处理大小"),
            
            # 错误处理配置
            ("config/error/max_retry_attempts", "3", "最大重试次数"),
            ("config/error/exponential_backoff", "true", "是否使用指数退避"),
            
            # 向后兼容的旧参数路径
            ("config/frame-interval", "2", "帧提取间隔（秒）- 旧路径"),
            ("config/frames-per-segment", "30", "每个Lambda处理的帧数 - 旧路径"),
            ("config/frame-quality", "90", "JPEG质量 - 旧路径"),
        ]
        
        # 根据区域设置AI分析配置
        if is_china_region:
            self.log("🇨🇳 设置中国区域特定参数...")
            ai_parameters = [
                # 中国区域AI分析配置
                ("config/ai/model_id", "Pro/deepseek-ai/DeepSeek-R1", "DeepSeek R1模型ID"),
                ("config/ai/max_tokens", "4096", "最大输出令牌数"),
                ("config/ai/temperature", "0.1", "温度参数"),
                ("config/ai/confidence_threshold", "0.7", "AI分析置信度阈值"),
                ("config/ai/max_retries", "5", "最大重试次数"),
                ("config/ai/retry_delay", "2", "重试延迟(秒)"),
                ("config/api-endpoint", "https://api.siliconflow.cn/v1", "硅基流动API端点"),
                ("config/api-key", "0", "API密钥（请在AWS控制台Parameter Store中设置实际值）", "SecureString"),
            ]
        else:
            ai_parameters = [
                # 全球区域AI分析配置
                ("config/ai/model_id", "us.anthropic.claude-3-7-sonnet-20250219-v1:0", "Claude AI模型ID"),
                ("config/ai/max_tokens", "4096", "最大输出令牌数"),
                ("config/ai/temperature", "0.0", "温度参数"),
                ("config/ai/top_p", "0.9", "Top-p参数"),
                ("config/ai/confidence_threshold", "0.7", "AI分析置信度阈值"),
                ("config/ai/max_retries", "3", "最大重试次数"),
                ("config/ai/retry_delay", "2", "重试延迟(秒)"),
                ("config/api-endpoint", "bedrock", "API端点"),
                ("config/api-key", "aws-role", "API密钥（全球区域使用AWS IAM角色）", "String"),
            ]
        
        # 合并参数列表
        parameters.extend(ai_parameters)
        
        success_count = 0
        for param in parameters:
            if len(param) == 4:
                name, value, desc, param_type = param
            else:
                name, value, desc = param
                param_type = "String"
            
            if self.create_parameter(name, value, desc, param_type):
                success_count += 1
        
        self.log(f"📊 参数创建完成: {success_count}/{len(parameters)} 成功")
        return success_count == len(parameters)
    
    def deploy_cdk_stack(self):
        """部署CDK堆栈"""
        self.log("🏗️ 开始部署CDK堆栈...")
        
        # 检查CDK是否已安装
        if not self.run_command("cdk --version", "检查CDK版本"):
            self.log("❌ CDK未安装，请先安装CDK", "ERROR")
            return False
        
        # 安装依赖
        if not self.run_command("pip install -r requirements.txt", "安装Python依赖"):
            self.log("⚠️ 依赖安装失败，继续尝试部署", "WARNING")
        
        # 检查是否为中国区域
        is_china_region = self.region.startswith('cn-')
        
        # 选择合适的上下文文件
        context_file = "cdk.context.cn.json" if is_china_region else "cdk.context.json"
        context_arg = ""
        
        if os.path.exists(context_file):
            self.log(f"使用区域特定上下文文件: {context_file}")
            context_arg = f"--context config-file={context_file}"
        
        # 部署堆栈
        deploy_cmd = f"cdk deploy --require-approval never --region {self.region} {context_arg}"
        return self.run_command(deploy_cmd, "部署CDK堆栈")
    
    def build_and_deploy_fargate(self):
        """构建并部署Fargate服务"""
        self.log("🐳 开始构建并部署Fargate服务...")
        
        # 检查Docker是否已安装
        if not self.run_command("docker --version", "检查Docker版本"):
            self.log("❌ Docker未安装，无法构建Fargate镜像", "ERROR")
            return False
        
        # 构建并推送Docker镜像
        if not self.run_command("python build_and_push_fargate.py", "构建并推送Fargate镜像"):
            self.log("❌ Fargate镜像构建失败", "ERROR")
            return False
        
        # 部署Fargate服务
        return self.run_command("python deploy_fargate.py", "部署Fargate服务")
    
    def verify_deployment(self):
        """验证部署状态"""
        self.log("🔍 验证部署状态...")
        
        try:
            # 检查参数
            response = self.ssm.get_parameters_by_path(
                Path=self.base_path,
                Recursive=True
            )
            param_count = len(response['Parameters'])
            self.log(f"✅ Parameter Store: {param_count} 个参数")
            
            # 检查Lambda函数
            lambda_client = boto3.client('lambda', region_name=self.region)
            functions = lambda_client.list_functions()
            
            driver_functions = [f for f in functions['Functions'] 
                             if 'DriverHandDetection' in f['FunctionName']]
            
            self.log(f"✅ Lambda函数: {len(driver_functions)} 个")
            for func in driver_functions[:5]:  # 只显示前5个
                self.log(f"   - {func['FunctionName']}")
            
            if len(driver_functions) > 5:
                self.log(f"   ... 以及 {len(driver_functions) - 5} 个其他函数")
            
            # 检查S3桶
            s3_client = boto3.client('s3', region_name=self.region)
            buckets = s3_client.list_buckets()
            
            driver_buckets = [b for b in buckets['Buckets'] 
                            if 'driverhanddetection' in b['Name'].lower()]
            
            self.log(f"✅ S3桶: {len(driver_buckets)} 个")
            for bucket in driver_buckets:
                self.log(f"   - {bucket['Name']}")
            
            # 检查ECS服务
            try:
                ecs_client = boto3.client('ecs', region_name=self.region)
                clusters = ecs_client.list_clusters()
                
                for cluster_arn in clusters['clusterArns']:
                    if 'driverhanddetection' in cluster_arn.lower():
                        cluster_name = cluster_arn.split('/')[-1]
                        self.log(f"✅ ECS集群: {cluster_name}")
                        
                        services = ecs_client.list_services(cluster=cluster_name)
                        for service_arn in services['serviceArns']:
                            service_name = service_arn.split('/')[-1]
                            self.log(f"   - 服务: {service_name}")
            except Exception as e:
                self.log(f"⚠️ ECS服务检查失败: {e}", "WARNING")
            
            # 检查Step Functions
            try:
                sfn_client = boto3.client('stepfunctions', region_name=self.region)
                state_machines = sfn_client.list_state_machines()
                
                driver_state_machines = [sm for sm in state_machines['stateMachines'] 
                                      if 'DriverHandDetection' in sm['name']]
                
                self.log(f"✅ Step Functions: {len(driver_state_machines)} 个")
                for sm in driver_state_machines:
                    self.log(f"   - {sm['name']}")
            except Exception as e:
                self.log(f"⚠️ Step Functions检查失败: {e}", "WARNING")
            
            return True
            
        except Exception as e:
            self.log(f"❌ 验证失败: {e}", "ERROR")
            return False
    
    def deploy_all(self, deploy_fargate=False, env='config'):
        """执行完整部署"""
        self.log("🚀 开始一键部署驾驶员手部检测系统...")
        self.log(f"📍 部署区域: {self.region}")
        
        # 检查是否为中国区域
        is_china_region = self.region.startswith('cn-')
        if is_china_region:
            self.log("🇨🇳 检测到中国区域部署模式")
        
        # 步骤1: 创建统一参数
        if not self.create_unified_parameters(env=env):
            self.log("❌ 参数创建失败，但将继续部署", "WARNING")
        
        # 步骤2: 部署CDK堆栈
        if not self.deploy_cdk_stack():
            self.log("❌ CDK堆栈部署失败", "ERROR")
            return False
        
        # 步骤3: 构建并部署Fargate服务（可选）
        if deploy_fargate:
            if not self.build_and_deploy_fargate():
                self.log("❌ Fargate服务部署失败", "ERROR")
                return False
        
        # 步骤4: 验证部署
        if not self.verify_deployment():
            self.log("⚠️ 部署验证失败，但堆栈可能已成功部署", "WARNING")
        
        self.log("🎉 一键部署完成！")
        self.log("\n⚠️  重要：部署后配置")
        self.log("=" * 50)
        
        if is_china_region:
            self.log("🇨🇳 中国区域 - 必需配置API密钥:")
            self.log("   方法1（推荐）：AWS控制台")
            self.log("   1. 进入 Systems Manager > Parameter Store")
            self.log("   2. 找到 /driver-hand-detection/config/api-key")
            self.log("   3. 将值从 '0' 改为你的硅基流动API密钥")
            self.log("")
            self.log("   方法2：命令行")
            self.log(f"   aws ssm put-parameter --name '/driver-hand-detection/config/api-key' \\")
            self.log(f"     --value 'sk-your-api-key' --type SecureString --region {self.region} --overwrite")
            self.log("")
            self.log("   获取API密钥：https://siliconflow.cn/")
        else:
            self.log("🌍 全球区域 - API密钥已设置为使用AWS IAM角色")
            self.log("   如需使用OpenAI等第三方服务，请在Parameter Store中修改:")
            self.log("   - /driver-hand-detection/config/api-key")
            self.log("   - /driver-hand-detection/config/api-endpoint")
            self.log("   - /driver-hand-detection/config/ai/model_id")
        
        self.log("\n📋 系统使用步骤:")
        self.log("   1. 设置API密钥（见上方说明）")
        self.log("   2. 上传视频到S3桶的 input/videos/ 目录")
        self.log("   3. 系统自动处理，结果保存在 output/reports/ 目录")
        
        if not deploy_fargate:
            self.log(f"\n🐳 Fargate部署:")
            self.log(f"   python build_and_push_fargate.py --region {self.region}")
            self.log(f"   python deploy_fargate.py --region {self.region}")
        
        return True

def main():
    """主函数"""
    import argparse
    
    parser = argparse.ArgumentParser(description='驾驶员手部检测系统一键部署')
    parser.add_argument('--region', default='us-east-1', help='AWS区域')
    parser.add_argument('--params-only', action='store_true', help='仅创建参数，不部署CDK')
    parser.add_argument('--cdk-only', action='store_true', help='仅部署CDK，不创建参数')
    parser.add_argument('--with-fargate', action='store_true', help='同时部署Fargate服务')
    parser.add_argument('--verify', action='store_true', help='仅验证部署状态')
    parser.add_argument('--reference-region', help='参考区域，从该区域复制参数')
    parser.add_argument('--china', action='store_true', help='中国区域部署模式')
    parser.add_argument('--env', default='config', help='环境名称')
    
    args = parser.parse_args()
    
    # 检查是否为中国区域
    is_china_region = args.china or args.region.startswith('cn-')
    
    # 如果指定了--china但没有指定区域，默认使用cn-north-1
    if args.china and not args.region.startswith('cn-'):
        args.region = 'cn-north-1'
        print(f"⚠️ 检测到中国区域部署模式，但未指定区域，默认使用: {args.region}")
    
    deployer = OneClickDeployer(region=args.region)
    
    if is_china_region:
        deployer.log(f"🇨🇳 中国区域部署模式: {args.region}")
    
    if args.reference_region:
        deployer.log(f"🔄 从参考区域 {args.reference_region} 复制参数...")
        # 实现从参考区域复制参数的逻辑
        # TODO: 实现参数复制功能
        deployer.log("⚠️ 参数复制功能尚未实现", "WARNING")
    
    if args.verify:
        deployer.log("🔍 仅验证部署状态")
        success = deployer.verify_deployment()
    elif args.params_only:
        deployer.log("🔧 仅创建参数模式")
        # 传递环境参数
        success = deployer.create_unified_parameters(env=args.env)
    elif args.cdk_only:
        deployer.log("🏗️ 仅部署CDK模式")
        success = deployer.deploy_cdk_stack()
    else:
        success = deployer.deploy_all(deploy_fargate=args.with_fargate, env=args.env)
    
    sys.exit(0 if success else 1)

if __name__ == "__main__":
    main()