影刀RPA AWS云服务自动化：EC2 S3管理实战

什么情况用什么 → 怎么做 → 有什么坑
作者：林焱 | 飞行社出品

---

什么情况用什么

企业用AWS（亚马逊云），每天要手动启停EC2实例、清理S3存储桶、监控账单，重复性工作多到爆。

这套方案适合：

• 企业运维团队自动化AWS资源管理
• 定时启停开发/测试环境（省钱！）
• S3存储桶自动清理（避免产生不必要的费用）

核心工具： 影刀RPA + Boto3（AWS SDK for Python） + CloudWatch监控

---

怎么做

第一步：配置AWS凭证

import boto3
import os
from botocore.exceptions import NoCredentialsError, ClientError

def init_aws_session(access_key, secret_key, region="us-east-1"):
    """
    初始化AWS会话
    access_key: AWS Access Key ID
    secret_key: AWS Secret Access Key
    region: AWS区域（如 "cn-north-1" 代表北京）
    """
    
    # 方法1：直接在代码里配置（不推荐，密钥会泄露）
    # session = boto3.Session(
    #     aws_access_key_id=access_key,
    #     aws_secret_access_key=secret_key,
    #     region_name=region
    # )
    
    # 方法2：从环境变量读取（推荐）
    # 先设置环境变量：
    # export AWS_ACCESS_KEY_ID="your_access_key"
    # export AWS_SECRET_ACCESS_KEY="your_secret_key"
    
    session = boto3.Session(
        aws_access_key_id=os.getenv("AWS_ACCESS_KEY_ID"),
        aws_secret_access_key=os.getenv("AWS_SECRET_ACCESS_KEY"),
        region_name=region
    )
    
    # 验证凭证是否有效
    try:
        sts = session.client('sts')
        identity = sts.get_caller_identity()
        print(f"AWS凭证有效，账号ID: {identity['Account']}")
        return session
    except NoCredentialsError:
        print("⚠️ AWS凭证未找到，请配置环境变量")
        return None
    except ClientError as e:
        print(f"⚠️ AWS凭证无效: {e}")
        return None

# 使用示例
session = init_aws_session(
    access_key="YOUR_ACCESS_KEY",
    secret_key="YOUR_SECRET_KEY",
    region="cn-north-1"  # 北京区域
)

第二步：自动化EC2实例管理

def list_ec2_instances(session):
    """列出所有EC2实例"""
    
    ec2 = session.client('ec2')
    
    response = ec2.describe_instances()
    
    instances = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            # 提取关键信息
            instance_info = {
                "实例ID": instance['InstanceId'],
                "实例类型": instance['InstanceType'],
                "状态": instance['State']['Name'],
                "公网IP": instance.get('PublicIpAddress', 'N/A'),
                "私网IP": instance.get('PrivateIpAddress', 'N/A'),
                "启动时间": instance.get('LaunchTime', 'N/A'),
                "名称": next((tag['Value'] for tag in instance.get('Tags', []) if tag['Key'] == 'Name'), '未命名')
            }
            
            instances.append(instance_info)
    
    return instances

def start_ec2_instances(session, instance_ids):
    """批量启动EC2实例"""
    
    ec2 = session.client('ec2')
    
    try:
        response = ec2.start_instances(InstanceIds=instance_ids)
        
        for instance in response['StartingInstances']:
            print(f"正在启动实例: {instance['InstanceId']}")
            print(f"  当前状态: {instance['CurrentState']['Name']}")
        
        return True
    
    except ClientError as e:
        print(f"启动实例失败: {e}")
        return False

def stop_ec2_instances(session, instance_ids, force=False):
    """批量停止EC2实例"""
    
    ec2 = session.client('ec2')
    
    try:
        response = ec2.stop_instances(
            InstanceIds=instance_ids,
            Force=force  # 是否强制关机
        )
        
        for instance in response['StoppingInstances']:
            print(f"正在停止实例: {instance['InstanceId']}")
            print(f"  当前状态: {instance['CurrentState']['Name']}")
        
        return True
    
    except ClientError as e:
        print(f"停止实例失败: {e}")
        return False

def auto_stop_dev_instances(session, tag_name="Environment", tag_value="Dev"):
    """
    自动停止开发环境实例（定时任务调用）
    例如：每天晚上8点自动停止所有Environment=Dev的实例
    """
    
    ec2 = session.client('ec2')
    
    # 查询符合条件的实例
    response = ec2.describe_instances(
        Filters=[
            {'Name': f'tag:{tag_name}', 'Values': [tag_value]},
            {'Name': 'instance-state-name', 'Values': ['running']}
        ]
    )
    
    instance_ids = []
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            instance_ids.append(instance['InstanceId'])
    
    if not instance_ids:
        print("没有需要停止的实例")
        return
    
    print(f"准备停止 {len(instance_ids)} 个开发环境实例...")
    stop_ec2_instances(session, instance_ids)

第三步：S3存储桶自动清理

def list_s3_buckets(session):
    """列出所有S3存储桶"""
    
    s3 = session.client('s3')
    
    response = s3.list_buckets()
    
    buckets = []
    for bucket in response['Buckets']:
        buckets.append({
            "存储桶名称": bucket['Name'],
            "创建时间": bucket['CreationDate']
        })
    
    return buckets

def clean_s3_bucket(session, bucket_name, prefix="", days=30):
    """
    清理S3存储桶中的旧文件
    prefix: 前缀（如 "logs/" 只清理logs目录）
    days: 删除多少天前的文件
    """
    
    import datetime
    
    s3 = session.client('s3')
    
    # 计算截止时间
    cutoff_date = datetime.datetime.now() - datetime.timedelta(days=days)
    
    # 列出所有对象
    paginator = s3.get_paginator('list_objects_v2')
    
    delete_list = []
    
    for page in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
        if 'Contents' in page:
            for obj in page['Contents']:
                # 检查文件修改时间
                if obj['LastModified'] < cutoff_date:
                    delete_list.append({'Key': obj['Key']})
                    
                    # AWS一次最多删除1000个对象
                    if len(delete_list) >= 1000:
                        delete_objects(session, bucket_name, delete_list)
                        delete_list = []
    
    # 删除剩余的对象
    if delete_list:
        delete_objects(session, bucket_name, delete_list)
    
    print(f"清理完成: {bucket_name}/{prefix}")

def delete_objects(session, bucket_name, delete_list):
    """批量删除S3对象"""
    
    s3 = session.client('s3')
    
    response = s3.delete_objects(
        Bucket=bucket_name,
        Delete={'Objects': delete_list}
    )
    
    deleted = len(response.get('Deleted', []))
    errors = len(response.get('Errors', []))
    
    print(f"  已删除: {deleted} 个文件")
    if errors > 0:
        print(f"  ⚠️ 删除失败: {errors} 个文件")

def get_s3_bucket_size(session, bucket_name):
    """获取S3存储桶大小（需要CloudWatch权限）"""
    
    cloudwatch = session.client('cloudwatch')
    
    response = cloudwatch.get_metric_statistics(
        Namespace='AWS/S3',
        MetricName='BucketSizeBytes',
        Dimensions=[{'Name': 'BucketName', 'Value': bucket_name}],
        StartTime=datetime.datetime.now() - datetime.timedelta(days=1),
        EndTime=datetime.datetime.now(),
        Period=86400,
        Statistics=['Average']
    )
    
    if response['Datapoints']:
        size_bytes = response['Datapoints'][0]['Average']
        size_gb = size_bytes / (1024 ** 3)
        return round(size_gb, 2)
    
    return 0

第四步：监控AWS账单（防止超支）

def get_billing_info(session):
    """获取AWS账单信息（需要开通Cost Explorer）"""
    
    ce = session.client('ce', region_name='us-east-1')  # Cost Explorer只在us-east-1
    
    # 查询本月支出
    from datetime import datetime, timedelta
    
    start_date = datetime.now().replace(day=1).strftime("%Y-%m-%d")
    end_date = datetime.now().strftime("%Y-%m-%d")
    
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='MONTHLY',
        Metrics=['BlendedCost'],
        GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
    )
    
    billing_info = []
    total_cost = 0
    
    for result in response['ResultsByTime'][0]['Groups']:
        service_name = result['Keys'][0]
        cost = float(result['Metrics']['BlendedCost']['Amount'])
        
        billing_info.append({
            "服务名称": service_name,
            "本月支出(USD)": round(cost, 2)
        })
        
        total_cost += cost
    
    print(f"本月总支出: ${round(total_cost, 2)}")
    
    return billing_info, round(total_cost, 2)

def check_billing_alert(session, threshold=1000):
    """
    检查账单是否超预算，超过阈值发送告警
    threshold: 预算阈值（美元）
    """
    
    _, total_cost = get_billing_info(session)
    
    if total_cost > threshold:
        message = f"⚠️ AWS账单告警！本月支出 ${total_cost}，已超过预算 ${threshold}！"
        
        # 发送到企微/钉钉
        send_alert_to_wecom(message)
        
        return True
    
    return False

def send_alert_to_wecom(message):
    """发送告警到企微"""
    
    webhook_url = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"
    
    payload = {
        "msgtype": "text",
        "text": {"content": message}
    }
    
    import requests
    response = requests.post(webhook_url, json=payload)
    
    if response.json().get("errcode") == 0:
        print("告警已发送到企微")
    else:
        print(f"发送告警失败: {response.text}")

第五步：影刀RPA完整流程编排

【定时触发】每天早上9点
    ↓
【Python节点】init_aws_session() → 初始化AWS会话
    ↓
【Python节点】list_ec2_instances() → 检查EC2实例状态
    ↓
【条件判断】是否有未使用的实例在运行？
    ├─ 是 → 【Python节点】stop_ec2_instances() → 停止实例（省钱！）
    └─ 否 → 继续
    ↓
【Python节点】list_s3_buckets() → 检查S3存储桶
    ↓
【Python节点】clean_s3_bucket() → 清理30天前的旧文件
    ↓
【Python节点】get_billing_info() → 获取本月账单
    ↓
【条件判断】账单是否超预算？
    ├─ 是 → 【企微通知】发送账单告警
    └─ 否 → 继续
    ↓
【生成报告】"AWS资源监控日报.xlsx"
    → 包含：EC2实例列表、S3存储桶大小、本月账单
    ↓
【发送邮件】将报告发送给运维团队

---

有什么坑

坑1：AWS凭证泄露风险

在代码里硬编码AWS Access Key和Secret Key，一旦代码泄露，别人可以用你的凭证启动大量EC2实例（挖矿！），产生巨额账单。

解决方案：

1. 使用IAM Role（推荐）：给EC2实例绑定IAM Role，无需在代码里配置凭证
2. 使用环境变量：export AWS_ACCESS_KEY_ID="xxx"
3. 使用AWS Secrets Manager：把凭证存在AWS Secrets Manager里，用时动态获取
4. 最小权限原则：IAM用户只给必要的权限（如只读权限、只能启停特定实例）

# 使用IAM Role（最安全）
# 在EC2实例上运行代码时，无需配置凭证，boto3会自动从实例元数据获取临时凭证
session = boto3.Session(region_name='cn-north-1')

坑2：批量操作失败，部分成功部分失败

批量启停EC2实例时，可能有的成功有的失败，需要记录失败的部分并重试。

解决方案： 实现重试机制：

def start_ec2_instances_with_retry(session, instance_ids, max_retries=3):
    """带重试的批量启动实例"""
    
    failed_ids = []
    
    for attempt in range(max_retries):
        if not instance_ids:
            break
        
        print(f"第 {attempt + 1} 次尝试启动 {len(instance_ids)} 个实例...")
        
        success_ids = []
        for instance_id in instance_ids:
            try:
                ec2 = session.client('ec2')
                ec2.start_instances(InstanceIds=[instance_id])
                success_ids.append(instance_id)
            except ClientError as e:
                print(f"  启动 {instance_id} 失败: {e}")
                failed_ids.append(instance_id)
        
        # 更新待重试列表
        instance_ids = failed_ids.copy()
        failed_ids = []
        
        if instance_ids:
            import time
            time.sleep(5)  # 等待5秒后重试
    
    if instance_ids:
        print(f"⚠️ 以下实例启动失败（已达最大重试次数）: {instance_ids}")
    
    return len(instance_ids) == 0  # 是否全部成功

坑3：S3删除操作不可逆，误删数据无法恢复

S3删除文件后，默认无法恢复（除非开启了版本控制）。

解决方案：

1. 开启S3版本控制（推荐）
2. 删除前先移动到"回收站"：先复制到另一个存储桶，再删除原文件
3. 使用生命周期策略：配置S3自动清理规则，而不是手动删除

# 开启S3版本控制（防止误删）
def enable_s3_versioning(session, bucket_name):
    """开启S3版本控制"""
    
    s3 = session.client('s3')
    
    s3.put_bucket_versioning(
        Bucket=bucket_name,
        VersioningConfiguration={'Status': 'Enabled'}
    )
    
    print(f"S3版本控制已开启: {bucket_name}")
    print("（开启后，删除文件不会真正删除，可以恢复）")

# 删除前先备份
def safe_delete_s3_object(session, bucket_name, object_key, backup_bucket):
    """安全删除S3对象（先备份）"""
    
    s3 = session.client('s3')
    
    # 1. 先复制到备份存储桶
    copy_source = {'Bucket': bucket_name, 'Key': object_key}
    s3.copy_object(
        CopySource=copy_source,
        Bucket=backup_bucket,
        Key=f"backup/{object_key}"
    )
    
    # 2. 再删除原文件
    s3.delete_object(Bucket=bucket_name, Key=object_key)
    
    print(f"安全删除: {bucket_name}/{object_key}（已备份到 {backup_bucket}/backup/）")

坑4：AWS API调用频率限制（Throttling）

AWS有API调用频率限制，批量操作可能触发Throttling错误。

解决方案： 加入随机延迟，避免触发频率限制：

import time
import random

def batch_operation_with_throttling(session, instance_ids):
    """批量操作（避免触发频率限制）"""
    
    for instance_id in instance_ids:
        try:
            # 执行操作
            operate_on_instance(session, instance_id)
            
            # 随机延迟0.5-2秒
            time.sleep(random.uniform(0.5, 2.0))
        
        except ClientError as e:
            if 'Throttling' in str(e):
                print(f"触发频率限制，等待5秒后重试...")
                time.sleep(5)
                operate_on_instance(session, instance_id)
            else:
                raise

---

总结

功能	节省成本	附加价值
定时启停EC2	最高省70%算力费用	避免忘记关机产生的浪费
S3自动清理	省存储费用	避免旧文件堆积

| 账单监控 | — | 及时发现异常支出 |
| 资源监控 | — | 优化资源使用 |

实际落地建议：

1. 先开启CloudTrail：记录所有API调用，便于审计和问题排查
2. 使用AWS Organizations：统一管理多个AWS账号的账单
3. 设置Billing Alarm：在AWS控制台设置账单告警（双重保障）
4. 定期审查IAM权限：移除不必要的权限，降低安全风险

AWS自动化能为运维团队节省50%以上的日常操作时间，同时避免因人为疏忽导致的资源浪费和安全风险。