第01章 云计算基础¶

import boto3
from datetime import datetime, timedelta

def create_ec2_instance():
    """
    创建AWS EC2实例
    """
    # 创建EC2客户端
    ec2 = boto3.client('ec2', region_name='us-east-1')

    # 实例配置
    instance_params = {
        'ImageId': 'ami-0c55b159cbfafe1f0',  # Amazon Linux 2 AMI
        'InstanceType': 't2.micro',
        'MinCount': 1,
        'MaxCount': 1,
        'KeyName': 'my-key-pair',
        'SecurityGroupIds': ['sg-12345678'],
        'SubnetId': 'subnet-12345678',
        'UserData': '''#!/bin/bash
            # 安装Docker
            yum install -y docker
            service docker start
            usermod -aG docker ec2-user
        ''',
        'TagSpecifications': [
            {
                'ResourceType': 'instance',
                'Tags': [
                    {'Key': 'Name', 'Value': 'my-web-server'},
                    {'Key': 'Environment', 'Value': 'production'}
                ]
            }
        ]
    }

    # 创建实例
    response = ec2.run_instances(**instance_params)

    instance_id = response['Instances'][0]['InstanceId']
    print(f"EC2实例创建成功: {instance_id}")

    # 等待实例运行
    waiter = ec2.get_waiter('instance_running')
    waiter.wait(InstanceIds=[instance_id])

    # 获取实例信息
    instance = ec2.describe_instances(InstanceIds=[instance_id])
    public_ip = instance['Reservations'][0]['Instances'][0]['PublicIpAddress']

    print(f"实例公网IP: {public_ip}")
    return instance_id

if __name__ == '__main__':
    create_ec2_instance()

import boto3

def deploy_application():
    """
    使用Elastic Beanstalk部署应用
    """
    eb = boto3.client('elasticbeanstalk', region_name='us-east-1')

    # 应用配置
    app_name = 'my-web-app'
    env_name = 'my-web-app-prod'

    # 创建应用
    try:
        response = eb.create_application(
            ApplicationName=app_name,
            Description='My Web Application'
        )
        print(f"应用创建成功: {app_name}")
    except eb.exceptions.ApplicationAlreadyExistsException:
        print(f"应用已存在: {app_name}")

    # 创建环境
    response = eb.create_environment(
        ApplicationName=app_name,
        EnvironmentName=env_name,
        SolutionStackName='64bit Amazon Linux 2 v5.8.0 running Python 3.11',
        OptionSettings=[
            {
                'Namespace': 'aws:elasticbeanstalk:container:python',
                'OptionName': 'WSGIPath',
                'Value': 'application.py'
            },
            {
                'Namespace': 'aws:autoscaling:asg',
                'OptionName': 'MinSize',
                'Value': '2'
            },
            {
                'Namespace': 'aws:autoscaling:asg',
                'OptionName': 'MaxSize',
                'Value': '6'
            }
        ]
    )

    print(f"环境创建成功: {env_name}")
    return response

if __name__ == '__main__':
    deploy_application()

import boto3

def setup_hybrid_cloud():
    """
    搭建混合云网络连接
    """
    # 创建AWS Direct Connect连接
    directconnect = boto3.client('directconnect', region_name='us-east-1')

    # 创建连接
    connection = directconnect.create_connection(
        locationName='EqDC2',
        bandwidth='1Gbps',
        connectionName='hybrid-cloud-connection'
    )

    connection_id = connection['connectionId']
    print(f"Direct Connect连接创建成功: {connection_id}")

    # 创建虚拟接口
    vif = directconnect.create_private_virtual_interface(
        connectionId=connection_id,
        newPrivateVirtualInterface={
            'virtualInterfaceName': 'hybrid-cloud-vif',
            'vlan': 100,
            'asn': 65000,
            'authKey': 'my-auth-key',
            'amazonAddress': '169.254.1.1/30',
            'customerAddress': '169.254.1.2/30',
            'virtualGatewayId': 'vgw-12345678'
        }
    )

    print(f"虚拟接口创建成功: {vif['virtualInterfaceId']}")

    # 配置路由
    ec2 = boto3.client('ec2', region_name='us-east-1')
    ec2.create_route(
        RouteTableId='rtb-12345678',
        DestinationCidrBlock='192.168.0.0/16',
        GatewayId='vgw-12345678'
    )

    print("混合云网络配置完成")
    return connection_id

if __name__ == '__main__':
    setup_hybrid_cloud()

import boto3
from botocore.exceptions import ClientError

def upload_file_to_s3(file_path, bucket_name, object_name=None):
    """
    上传文件到S3
    """
    # 创建S3客户端
    s3_client = boto3.client('s3', region_name='us-east-1')

    # 如果未指定对象名称，使用文件名
    if object_name is None:
        object_name = file_path.split('/')[-1]  # 负索引：从末尾倒数访问元素

    try:
        # 上传文件
        response = s3_client.upload_file(
            file_path,
            bucket_name,
            object_name,
            ExtraArgs={
                'ContentType': 'application/octet-stream',
                'Metadata': {
                    'uploaded-by': 'devops-team',
                    'environment': 'production'
                }
            }
        )

        # 设置生命周期策略
        s3_client.put_object_lifecycle_configuration(
            Bucket=bucket_name,
            LifecycleConfiguration={
                'Rules': [
                    {
                        'Id': 'DeleteOldFiles',
                        'Status': 'Enabled',
                        'Filter': {'Prefix': 'logs/'},
                        'Expiration': {'Days': 30}
                    }
                ]
            }
        )

        print(f"文件上传成功: {object_name}")
        return True

    except ClientError as e:
        print(f"上传失败: {e}")
        return False

def download_file_from_s3(bucket_name, object_name, file_path):
    """
    从S3下载文件
    """
    s3_client = boto3.client('s3', region_name='us-east-1')

    try:
        s3_client.download_file(bucket_name, object_name, file_path)
        print(f"文件下载成功: {file_path}")
        return True
    except ClientError as e:
        print(f"下载失败: {e}")
        return False

if __name__ == '__main__':
    # 上传文件
    upload_file_to_s3('data.csv', 'my-bucket', 'data/data.csv')

    # 下载文件
    download_file_from_s3('my-bucket', 'data/data.csv', 'downloaded_data.csv')

from azure.identity import DefaultAzureCredential
from azure.mgmt.compute import ComputeManagementClient
from azure.mgmt.resource import ResourceManagementClient

def create_azure_vm():
    """
    创建Azure虚拟机
    """
    # 认证
    credential = DefaultAzureCredential()

    # 创建客户端
    compute_client = ComputeManagementClient(
        credential,
        subscription_id='your-subscription-id'
    )
    resource_client = ResourceManagementClient(
        credential,
        subscription_id='your-subscription-id'
    )

    # 资源组配置
    RESOURCE_GROUP_NAME = 'my-resource-group'
    LOCATION = 'eastus'

    # 创建资源组
    resource_client.resource_groups.create_or_update(
        RESOURCE_GROUP_NAME,
        {'location': LOCATION}
    )

    # VM配置
    vm_parameters = {
        'location': LOCATION,
        'os_profile': {
            'computer_name': 'my-vm',
            'admin_username': 'azureuser',
            'admin_password': 'Password123!',
            'disable_password_authentication': False
        },
        'hardware_profile': {
            'vm_size': 'Standard_DS1_v2'
        },
        'storage_profile': {
            'image_reference': {
                'publisher': 'Canonical',
                'offer': '0001-com-ubuntu-server-jammy',
                'sku': '22_04-lts-gen2',
                'version': 'latest'
            },
            'os_disk': {
                'create_option': 'FromImage',
                'managed_disk': {
                    'storage_account_type': 'Standard_LRS'
                }
            }
        },
        'network_profile': {
            'network_interfaces': [{
                'id': '/subscriptions/{}/resourceGroups/{}/providers/Microsoft.Network/networkInterfaces/my-nic'.format(
                    'your-subscription-id',
                    RESOURCE_GROUP_NAME
                )
            }]
        }
    }

    # 创建VM
    poller = compute_client.virtual_machines.begin_create_or_update(
        RESOURCE_GROUP_NAME,
        'my-vm',
        vm_parameters
    )

    vm_result = poller.result()
    print(f"虚拟机创建成功: {vm_result.name}")

    return vm_result

if __name__ == '__main__':
    create_azure_vm()

from google.cloud import bigquery
from google.cloud.exceptions import NotFound

def create_bigquery_dataset():
    """
    创建BigQuery数据集
    """
    # 创建客户端
    client = bigquery.Client()

    # 数据集配置
    dataset_id = "my-project.my_dataset"
    dataset = bigquery.Dataset(dataset_id)

    # 设置数据集属性
    dataset.description = "My production dataset"
    dataset.location = "US"
    dataset.default_table_expiration_ms = 30 * 24 * 60 * 60 * 1000  # 30天

    # 创建数据集
    try:
        dataset = client.create_dataset(dataset)
        print(f"数据集创建成功: {dataset.dataset_id}")
    except Exception as e:
        print(f"数据集创建失败: {e}")
        return None

    # 创建表
    table_id = f"{dataset_id}.user_events"
    schema = [
        bigquery.SchemaField("event_id", "STRING"),
        bigquery.SchemaField("user_id", "STRING"),
        bigquery.SchemaField("event_type", "STRING"),
        bigquery.SchemaField("event_time", "TIMESTAMP"),
        bigquery.SchemaField("properties", "JSON"),
    ]

    table = bigquery.Table(table_id, schema=schema)
    table.description = "User events tracking"
    table = client.create_table(table)

    print(f"表创建成功: {table.table_id}")

    return dataset

def query_bigquery():
    """
    查询BigQuery数据
    """
    client = bigquery.Client()

    query = """
        SELECT
            event_type,
            COUNT(*) as event_count,
            COUNT(DISTINCT user_id) as unique_users
        FROM
            `my-project.my_dataset.user_events`
        WHERE
            event_time >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 7 DAY)
        GROUP BY
            event_type
        ORDER BY
            event_count DESC
    """

    query_job = client.query(query)
    results = query_job.result()

    for row in results:
        print(f"{row.event_type}: {row.event_count} events, {row.unique_users} users")

if __name__ == '__main__':
    create_bigquery_dataset()
    query_bigquery()

from alibabacloud_ecs20140526.client import Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_ecs20140526 import models as ecs_models
import json

def create_aliyun_ecs():
    """
    创建阿里云ECS实例
    """
    # 配置客户端
    config = open_api_models.Config(
        access_key_id='your-access-key-id',
        access_key_secret='your-access-key-secret',
        region_id='cn-hangzhou'
    )
    config.endpoint = 'ecs.cn-hangzhou.aliyuncs.com'

    client = Client(config)

    # 实例配置
    create_instance_request = ecs_models.CreateInstanceRequest(
        region_id='cn-hangzhou',
        image_id='ubuntu_20_04_x64_20G_alibase_20230727.vhd',
        instance_type='ecs.t6-c1m1.large',
        security_group_id='sg-12345678',
        v_switch_id='vsw-12345678',
        instance_name='my-web-server',
        description='Web server for production',
        internet_max_bandwidth_out=5,
        system_disk_category='cloud_essd',
        system_disk_size=40,
        data_disks=[
            ecs_models.CreateInstanceRequestDataDisks(
                category='cloud_essd',
                size=100
            )
        ],
        tags=[
            ecs_models.CreateInstanceRequestTags(
                key='Environment',
                value='production'
            ),
            ecs_models.CreateInstanceRequestTags(
                key='Owner',
                value='devops-team'
            )
        ]
    )

    # 创建实例
    response = client.create_instance(create_instance_request)
    instance_id = response.body.instance_id

    print(f"ECS实例创建成功: {instance_id}")

    # 等待实例启动
    import time
    time.sleep(30)

    # 启动实例
    start_instance_request = ecs_models.StartInstanceRequest(
        instance_id=instance_id
    )
    client.start_instance(start_instance_request)

    print(f"ECS实例启动成功: {instance_id}")

    return instance_id

if __name__ == '__main__':
    create_aliyun_ecs()

import boto3
from datetime import datetime, timedelta

def analyze_cloud_costs():
    """
    分析云成本
    """
    # 创建Cost Explorer客户端
    ce = boto3.client('ce', region_name='us-east-1')

    # 查询最近30天的成本
    end_date = datetime.now().strftime('%Y-%m-%d')
    start_date = (datetime.now() - timedelta(days=30)).strftime('%Y-%m-%d')

    # 按服务查询成本
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='DAILY',
        Metrics=['BlendedCost', 'UnblendedCost'],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': 'SERVICE'
            }
        ]
    )

    # 分析结果
    print("成本分析报告")
    print("=" * 50)
    total_cost = 0

    for result in response['ResultsByTime']:
        date = result['TimePeriod']['Start']
        print(f"\n日期: {date}")

        for group in result['Groups']:
            service = group['Keys'][0]
            cost = float(group['Metrics']['BlendedCost']['Amount'])
            total_cost += cost
            print(f"  {service}: ${cost:.2f}")

    print(f"\n总成本: ${total_cost:.2f}")

    # 按标签查询成本
    tagged_response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='MONTHLY',
        Metrics=['BlendedCost'],
        GroupBy=[
            {
                'Type': 'TAG',
                'Key': 'Environment'
            }
        ]
    )

    print("\n按环境分摊成本:")
    print("-" * 50)
    for result in tagged_response['ResultsByTime']:
        for group in result['Groups']:
            if group['Keys'][0]:  # 忽略无标签的资源
                environment = group['Keys'][0]
                cost = float(group['Metrics']['BlendedCost']['Amount'])
                print(f"{environment}: ${cost:.2f}")

    return total_cost

def set_budget_alert():
    """
    设置成本告警
    """
    # 创建Budgets客户端
    budgets = boto3.client('budgets', region_name='us-east-1')

    # 创建月度预算
    budget = {
        'AccountId': '123456789012',
        'Budget': {
            'BudgetName': 'MonthlyCostBudget',
            'BudgetLimit': {
                'Amount': '1000.00',
                'Unit': 'USD'
            },
            'TimeUnit': 'MONTHLY',
            'BudgetType': 'COST',
            'CostFilters': {
                'Service': ['Amazon EC2', 'Amazon RDS', 'Amazon S3']
            }
        },
        'NotificationsWithSubscribers': [
            {
                'Notification': {
                    'NotificationType': 'ACTUAL',
                    'ComparisonOperator': 'GREATER_THAN',
                    'Threshold': 80.0,
                    'ThresholdType': 'PERCENTAGE'
                },
                'Subscribers': [
                    {
                        'SubscriptionType': 'EMAIL',
                        'Address': 'devops-team@example.com'
                    },
                    {
                        'SubscriptionType': 'SNS',
                        'Address': 'arn:aws:sns:us-east-1:123456789012:cost-alerts'
                    }
                ]
            }
        ]
    }

    try:
        budgets.create_budget(**budget)
        print("成本告警设置成功")
    except Exception as e:
        print(f"设置失败: {e}")

if __name__ == '__main__':
    analyze_cloud_costs()
    set_budget_alert()

import boto3
import json

def setup_iam_security():
    """
    设置IAM安全配置
    """
    iam = boto3.client('iam', region_name='us-east-1')

    # 1. 创建自定义策略（最小权限）
    policy_document = {
        "Version": "2012-10-17",
        "Statement": [
            {
                "Effect": "Allow",
                "Action": [
                    "s3:GetObject",
                    "s3:PutObject"
                ],
                "Resource": "arn:aws:s3:::my-bucket/*",
                "Condition": {
                    "IpAddress": {
                        "aws:SourceIp": ["192.0.2.0/24"]
                    }
                }
            }
        ]
    }

    try:
        policy = iam.create_policy(
            PolicyName='S3ReadOnlyForSpecificIP',
            PolicyDocument=json.dumps(policy_document),
            Description='Allow S3 access from specific IP range'
        )
        print(f"策略创建成功: {policy['Policy']['Arn']}")
    except Exception as e:
        print(f"策略创建失败: {e}")

    # 2. 创建用户并分配策略
    try:
        user = iam.create_user(
            UserName='devops-user',
            Path='/developers/'
        )
        print(f"用户创建成功: {user['User']['UserName']}")
    except Exception as e:
        print(f"用户创建失败: {e}")

    # 3. 分配策略
    iam.attach_user_policy(
        UserName='devops-user',
        PolicyArn='arn:aws:iam::123456789012:policy/S3ReadOnlyForSpecificIP'
    )

    # 4. 创建访问密钥（生产环境应使用临时凭证）
    try:
        keys = iam.create_access_key(UserName='devops-user')
        print(f"访问密钥创建成功:")
        print(f"  Access Key ID: {keys['AccessKey']['AccessKeyId']}")
        print(f"  Secret Access Key: {keys['AccessKey']['SecretAccessKey']}")
        print("请妥善保管密钥，不要泄露！")
    except Exception as e:
        print(f"访问密钥创建失败: {e}")

    # 5. 启用MFA（需要用户在控制台操作）
    print("\n安全提醒:")
    print("1. 请为用户devops-user启用MFA")
    print("2. 定期轮换访问密钥")
    print("3. 使用临时凭证而非长期密钥")
    print("4. 定期审计IAM权限")

def audit_iam_permissions():
    """
    审计IAM权限
    """
    iam = boto3.client('iam', region_name='us-east-1')

    # 获取所有用户
    users = iam.list_users()

    print("IAM权限审计报告")
    print("=" * 50)

    for user in users['Users']:
        username = user['UserName']
        print(f"\n用户: {username}")

        # 检查访问密钥
        access_keys = iam.list_access_keys(UserName=username)
        for key in access_keys['AccessKeyMetadata']:
            key_id = key['AccessKeyId']
            status = key['Status']
            create_date = key['CreateDate']

            # 检查密钥年龄
            key_age = (datetime.now() - create_date.replace(tzinfo=None)).days

            print(f"  密钥: {key_id}")
            print(f"    状态: {status}")
            print(f"    创建时间: {create_date}")
            print(f"    密钥年龄: {key_age}天")

            if key_age > 90:
                print("    ⚠️ 警告: 密钥超过90天，建议轮换")

        # 检查MFA
        try:
            mfa_devices = iam.list_mfa_devices(UserName=username)
            if not mfa_devices['MFADevices']:
                print("  ⚠️ 警告: 未启用MFA")
            else:
                print("  ✓ MFA已启用")
        except Exception as e:
            print(f"  MFA检查失败: {e}")

        # 检查附加的策略
        policies = iam.list_attached_user_policies(UserName=username)
        print(f"  附加策略数量: {len(policies['AttachedPolicies'])}")

if __name__ == '__main__':
    setup_iam_security()
    print("\n")
    audit_iam_permissions()

import boto3
import time
from datetime import datetime

class CloudWebApp:
    """云端Web应用部署类"""

    def __init__(self, region='us-east-1'):
        self.region = region
        self.ec2 = boto3.client('ec2', region_name=region)
        self.s3 = boto3.client('s3', region_name=region)
        self.rds = boto3.client('rds', region_name=region)
        self.elbv2 = boto3.client('elbv2', region_name=region)
        self.autoscaling = boto3.client('autoscaling', region_name=region)
        self.cloudfront = boto3.client('cloudfront')

        self.vpc_id = None
        self.subnet_ids = []
        self.security_group_id = None
        self.instance_ids = []
        self.load_balancer_arn = None
        self.target_group_arn = None

    def create_vpc(self):
        """创建VPC网络"""
        print("创建VPC...")

        # 创建VPC
        vpc_response = self.ec2.create_vpc(
            CidrBlock='10.0.0.0/16',
            AmazonProvidedIpv6CidrBlock=False,
            InstanceTenancy='default'
        )
        self.vpc_id = vpc_response['Vpc']['VpcId']

        # 启用DNS支持
        self.ec2.modify_vpc_attribute(
            VpcId=self.vpc_id,
            EnableDnsSupport={'Value': True}
        )
        self.ec2.modify_vpc_attribute(
            VpcId=self.vpc_id,
            EnableDnsHostnames={'Value': True}
        )

        # 创建子网
        for i in range(3):
            subnet_response = self.ec2.create_subnet(
                VpcId=self.vpc_id,
                CidrBlock=f'10.0.{i}.0/24',
                AvailabilityZone=f'{self.region}{chr(97 + i)}'
            )
            self.subnet_ids.append(subnet_response['Subnet']['SubnetId'])

        # 创建Internet网关
        igw_response = self.ec2.create_internet_gateway()
        igw_id = igw_response['InternetGateway']['InternetGatewayId']

        # 关联Internet网关
        self.ec2.attach_internet_gateway(
            InternetGatewayId=igw_id,
            VpcId=self.vpc_id
        )

        # 创建路由表
        route_table_response = self.ec2.create_route_table(VpcId=self.vpc_id)
        route_table_id = route_table_response['RouteTable']['RouteTableId']

        # 添加路由
        self.ec2.create_route(
            RouteTableId=route_table_id,
            DestinationCidrBlock='0.0.0.0/0',
            GatewayId=igw_id
        )

        # 关联子网
        for subnet_id in self.subnet_ids:
            self.ec2.associate_route_table(
                RouteTableId=route_table_id,
                SubnetId=subnet_id
            )

        print(f"VPC创建完成: {self.vpc_id}")

    def create_security_group(self):
        """创建安全组"""
        print("创建安全组...")

        sg_response = self.ec2.create_security_group(
            GroupName='web-app-sg',
            Description='Security group for web application',
            VpcId=self.vpc_id
        )
        self.security_group_id = sg_response['GroupId']

        # 添加入站规则
        self.ec2.authorize_security_group_ingress(
            GroupId=self.security_group_id,
            IpPermissions=[
                {
                    'IpProtocol': 'tcp',
                    'FromPort': 80,
                    'ToPort': 80,
                    'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
                },
                {
                    'IpProtocol': 'tcp',
                    'FromPort': 443,
                    'ToPort': 443,
                    'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
                },
                {
                    'IpProtocol': 'tcp',
                    'FromPort': 22,
                    'ToPort': 22,
                    'IpRanges': [{'CidrIp': '0.0.0.0/0'}]
                }
            ]
        )

        print(f"安全组创建完成: {self.security_group_id}")

    def create_launch_template(self):
        """创建启动模板"""
        print("创建启动模板...")

        # 用户数据脚本
        user_data = '''#!/bin/bash
            # 更新系统
            yum update -y

            # 安装Docker
            yum install -y docker
            service docker start
            usermod -aG docker ec2-user

            # 拉取并运行Web应用
            docker run -d -p 80:80 nginx:latest

            # 安装CloudWatch Agent
            yum install -y amazon-cloudwatch-agent
            /opt/aws/amazon-cloudwatch-agent/bin/amazon-cloudwatch-agent-ctl \
                -a fetch-config -m ec2 -s -c file:/opt/aws/amazon-cloudwatch-agent/etc/config.json
        '''

        # 创建启动模板
        ec2 = boto3.client('ec2', region_name=self.region)
        template_response = ec2.create_launch_template(
            LaunchTemplateName='web-app-template',
            LaunchTemplateData={
                'ImageId': 'ami-0c55b159cbfafe1f0',  # Amazon Linux 2
                'InstanceType': 't3.micro',
                'KeyName': 'my-key-pair',
                'SecurityGroupIds': [self.security_group_id],
                'UserData': user_data,
                'TagSpecifications': [
                    {
                        'ResourceType': 'instance',
                        'Tags': [
                            {'Key': 'Name', 'Value': 'web-app-instance'},
                            {'Key': 'Environment', 'Value': 'production'}
                        ]
                    }
                ]
            }
        )

        print("启动模板创建完成")
        return template_response['LaunchTemplate']['LaunchTemplateId']

    def create_load_balancer(self):
        """创建负载均衡器"""
        print("创建负载均衡器...")

        # 创建目标组
        target_group_response = self.elbv2.create_target_group(
            Name='web-app-targets',
            Protocol='HTTP',
            Port=80,
            VpcId=self.vpc_id,
            HealthCheckPath='/',
            HealthCheckIntervalSeconds=30,
            HealthCheckTimeoutSeconds=5,
            HealthyThresholdCount=3,
            UnhealthyThresholdCount=2
        )
        self.target_group_arn = target_group_response['TargetGroups'][0]['TargetGroupArn']

        # 创建负载均衡器
        lb_response = self.elbv2.create_load_balancer(
            Name='web-app-lb',
            Subnets=self.subnet_ids,
            SecurityGroups=[self.security_group_id],
            Scheme='internet-facing',
            Type='application',
            IpAddressType='ipv4'
        )
        self.load_balancer_arn = lb_response['LoadBalancers'][0]['LoadBalancerArn']

        # 等待负载均衡器可用
        waiter = self.elbv2.get_waiter('load_balancer_available')
        waiter.wait(LoadBalancerArns=[self.load_balancer_arn])

        # 创建监听器
        self.elbv2.create_listener(
            LoadBalancerArn=self.load_balancer_arn,
            Protocol='HTTP',
            Port=80,
            DefaultActions=[
                {
                    'Type': 'forward',
                    'TargetGroupArn': self.target_group_arn
                }
            ]
        )

        print(f"负载均衡器创建完成: {self.load_balancer_arn}")

    def create_auto_scaling_group(self, launch_template_id):
        """创建自动扩缩容组"""
        print("创建自动扩缩容组...")

        self.autoscaling.create_auto_scaling_group(
            AutoScalingGroupName='web-app-asg',
            LaunchTemplate={
                'LaunchTemplateId': launch_template_id,
                'Version': '$Latest'
            },
            MinSize=2,
            MaxSize=6,
            DesiredCapacity=2,
            VPCZoneIdentifier=','.join(self.subnet_ids),
            TargetGroupARNs=[self.target_group_arn],
            HealthCheckType='ELB',
            HealthCheckGracePeriod=300,
            DefaultInstanceWarmup=300,
            Tags=[
                {
                    'Key': 'Name',
                    'Value': 'web-app-instance',
                    'PropagateAtLaunch': True
                },
                {
                    'Key': 'Environment',
                    'Value': 'production',
                    'PropagateAtLaunch': True
                }
            ]
        )

        # 创建扩缩容策略
        self.autoscaling.put_scaling_policy(
            AutoScalingGroupName='web-app-asg',
            PolicyName='scale-out-policy',
            PolicyType='SimpleScaling',
            AdjustmentType='ChangeInCapacity',
            ScalingAdjustment=1,
            Cooldown=300
        )

        self.autoscaling.put_scaling_policy(
            AutoScalingGroupName='web-app-asg',
            PolicyName='scale-in-policy',
            PolicyType='SimpleScaling',
            AdjustmentType='ChangeInCapacity',
            ScalingAdjustment=-1,
            Cooldown=300
        )

        print("自动扩缩容组创建完成")

    def create_s3_bucket(self):
        """创建S3存储桶"""
        print("创建S3存储桶...")

        bucket_name = f'web-app-static-{int(time.time())}'

        self.s3.create_bucket(
            Bucket=bucket_name,
            CreateBucketConfiguration={
                'LocationConstraint': self.region
            }
        )

        # 设置存储桶策略
        bucket_policy = {
            "Version": "2012-10-17",
            "Statement": [
                {
                    "Sid": "PublicReadGetObject",
                    "Effect": "Allow",
                    "Principal": "*",
                    "Action": "s3:GetObject",
                    "Resource": f"arn:aws:s3:::{bucket_name}/*"
                }
            ]
        }

        self.s3.put_bucket_policy(
            Bucket=bucket_name,
            Policy=json.dumps(bucket_policy)  # json.dumps将Python对象转为JSON字符串
        )

        # 启用版本控制
        self.s3.put_bucket_versioning(
            Bucket=bucket_name,
            VersioningConfiguration={
                'Status': 'Enabled'
            }
        )

        # 配置生命周期规则
        lifecycle_config = {
            'Rules': [
                {
                    'Id': 'DeleteOldVersions',
                    'Status': 'Enabled',
                    'NoncurrentVersionExpiration': {'NoncurrentDays': 30}
                }
            ]
        }

        self.s3.put_bucket_lifecycle_configuration(
            Bucket=bucket_name,
            LifecycleConfiguration=lifecycle_config
        )

        print(f"S3存储桶创建完成: {bucket_name}")
        return bucket_name

    def create_rds_instance(self):
        """创建RDS数据库实例"""
        print("创建RDS数据库实例...")

        # 创建子网组
        subnet_group_name = 'web-app-db-subnet-group'
        self.rds.create_db_subnet_group(
            DBSubnetGroupName=subnet_group_name,
            DBSubnetGroupDescription='Subnet group for web app database',
            SubnetIds=self.subnet_ids[:2]  # 切片操作：[start:end:step]提取子序列
        )

        # 创建数据库实例
        # ⚠️ 安全警告：以下代码示例包含硬编码密码，仅用于演示目的。
        # 在生产环境中，请务必使用环境变量或密钥管理服务（如 AWS Secrets Manager、Azure Key Vault 等）来存储敏感信息。
        db_response = self.rds.create_db_instance(
            DBName='webapp',
            DBInstanceIdentifier='web-app-db',
            DBInstanceClass='db.t3.micro',
            Engine='mysql',
            MasterUsername='admin',
            MasterUserPassword='SecurePassword123!',
            AllocatedStorage=20,
            StorageType='gp2',
            VpcSecurityGroupIds=[self.security_group_id],
            DBSubnetGroupName=subnet_group_name,
            BackupRetentionPeriod=7,
            MultiAZ=False,
            PubliclyAccessible=False,
            Tags=[
                {
                    'Key': 'Name',
                    'Value': 'web-app-database'
                },
                {
                    'Key': 'Environment',
                    'Value': 'production'
                }
            ]
        )

        print(f"RDS实例创建中: {db_response['DBInstance']['DBInstanceIdentifier']}")
        return db_response['DBInstance']['DBInstanceIdentifier']

    def create_cloudfront_distribution(self, s3_bucket_name):
        """创建CloudFront分发"""
        print("创建CloudFront分发...")

        # 获取S3存储桶的Origin ID
        origin_id = f'S3-{s3_bucket_name}'

        distribution_config = {
            'CallerReference': str(int(time.time())),
            'Comment': 'CloudFront distribution for web app',
            'DefaultCacheBehavior': {
                'TargetOriginId': origin_id,
                'ViewerProtocolPolicy': 'allow-all',
                'AllowedMethods': {
                    'Quantity': 2,
                    'Items': ['GET', 'HEAD'],
                    'CachedMethods': {
                        'Quantity': 2,
                        'Items': ['GET', 'HEAD']
                    }
                },
                'ForwardedValues': {
                    'QueryString': False,
                    'Cookies': {'Forward': 'none'}
                },
                'MinTTL': 0,
                'DefaultTTL': 86400,
                'MaxTTL': 31536000
            },
            'Origins': {
                'Quantity': 1,
                'Items': [
                    {
                        'Id': origin_id,
                        'DomainName': f'{s3_bucket_name}.s3.amazonaws.com',
                        'S3OriginConfig': {
                            'OriginAccessIdentity': ''
                        }
                    }
                ]
            },
            'Enabled': True,
            'DefaultRootObject': 'index.html',
            'PriceClass': 'PriceClass_All'
        }

        distribution_response = self.cloudfront.create_distribution(
            DistributionConfig=distribution_config
        )

        distribution_id = distribution_response['Distribution']['Id']
        domain_name = distribution_response['Distribution']['DomainName']

        print(f"CloudFront分发创建完成: {distribution_id}")
        print(f"域名: {domain_name}")

        return distribution_id, domain_name

    def deploy(self):
        """部署完整应用"""
        print("开始部署云端Web应用...")
        print("=" * 50)

        try:  # try/except捕获异常
            # 1. 创建VPC
            self.create_vpc()
            time.sleep(2)

            # 2. 创建安全组
            self.create_security_group()
            time.sleep(2)

            # 3. 创建启动模板
            launch_template_id = self.create_launch_template()
            time.sleep(2)

            # 4. 创建负载均衡器
            self.create_load_balancer()
            time.sleep(5)

            # 5. 创建自动扩缩容组
            self.create_auto_scaling_group(launch_template_id)
            time.sleep(10)

            # 6. 创建S3存储桶
            s3_bucket_name = self.create_s3_bucket()
            time.sleep(2)

            # 7. 创建RDS实例
            db_instance_id = self.create_rds_instance()
            time.sleep(2)

            # 8. 创建CloudFront分发
            distribution_id, domain_name = self.create_cloudfront_distribution(s3_bucket_name)

            print("\n" + "=" * 50)
            print("部署完成！")
            print("=" * 50)
            print(f"VPC ID: {self.vpc_id}")
            print(f"安全组 ID: {self.security_group_id}")
            print(f"负载均衡器 ARN: {self.load_balancer_arn}")
            print(f"自动扩缩容组: web-app-asg")
            print(f"S3存储桶: {s3_bucket_name}")
            print(f"RDS实例: {db_instance_id}")
            print(f"CloudFront域名: {domain_name}")
            print("\n请等待几分钟让所有资源启动完成...")

            return {
                'vpc_id': self.vpc_id,
                'security_group_id': self.security_group_id,
                'load_balancer_arn': self.load_balancer_arn,
                's3_bucket_name': s3_bucket_name,
                'db_instance_id': db_instance_id,
                'cloudfront_domain': domain_name
            }

        except Exception as e:
            print(f"部署失败: {e}")
            raise

if __name__ == '__main__':
    # 部署应用
    app = CloudWebApp(region='us-east-1')
    result = app.deploy()

    print("\n部署信息已保存，请妥善保管！")