影刀RPA AWS云服务自动化：EC2 S3管理实战-拓冰建站

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

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

什么情况用什么

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

这套方案适合：

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

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

怎么做

第一步：配置AWS凭证

拼多多店群自动化报活动上架！

importboto3importosfrombotocore.exceptionsimportNoCredentialsError,ClientErrordefinit_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']}")returnsessionexceptNoCredentialsError:print("⚠️ AWS凭证未找到，请配置环境变量")returnNoneexceptClientErrorase:print(f"⚠️ AWS凭证无效:{e}")returnNone# 使用示例session=init_aws_session(access_key="YOUR_ACCESS_KEY",secret_key="YOUR_SECRET_KEY",region="cn-north-1"# 北京区域)

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

deflist_ec2_instances(session):"""列出所有EC2实例"""ec2=session.client('ec2')response=ec2.describe_instances()instances=[]forreservationinresponse['Reservations']:forinstanceinreservation['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']fortagininstance.get('Tags',[])iftag['Key']=='Name'),'未命名')}instances.append(instance_info)returninstancesdefstart_ec2_instances(session,instance_ids):"""批量启动EC2实例"""ec2=session.client('ec2')try:response=ec2.start_instances(InstanceIds=instance_ids)forinstanceinresponse['StartingInstances']:print(f"正在启动实例:{instance['InstanceId']}")print(f" 当前状态:{instance['CurrentState']['Name']}")returnTrueexceptClientErrorase:print(f"启动实例失败:{e}")returnFalsedefstop_ec2_instances(session,instance_ids,force=False):"""批量停止EC2实例"""ec2=session.client('ec2')try:response=ec2.stop_instances(InstanceIds=instance_ids,Force=force# 是否强制关机)forinstanceinresponse['StoppingInstances']:print(f"正在停止实例:{instance['InstanceId']}")print(f" 当前状态:{instance['CurrentState']['Name']}")returnTrueexceptClientErrorase:print(f"停止实例失败:{e}")returnFalsedefauto_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=[]forreservationinresponse['Reservations']:forinstanceinreservation['Instances']:instance_ids.append(instance['InstanceId'])ifnotinstance_ids:print("没有需要停止的实例")returnprint(f"准备停止{len(instance_ids)}个开发环境实例...")stop_ec2_instances(session,instance_ids)

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

deflist_s3_buckets(session):"""列出所有S3存储桶"""s3=session.client('s3')response=s3.list_buckets()buckets=[]forbucketinresponse['Buckets']:buckets.append({"存储桶名称":bucket['Name'],"创建时间":bucket['CreationDate']})returnbucketsdefclean_s3_bucket(session,bucket_name,prefix="",days=30):""" 清理S3存储桶中的旧文件 prefix: 前缀（如 "logs/" 只清理logs目录） days: 删除多少天前的文件 """importdatetime s3=session.client('s3')# 计算截止时间cutoff_date=datetime.datetime.now()-datetime.timedelta(days=days)# 列出所有对象paginator=s3.get_paginator('list_objects_v2')delete_list=[]forpageinpaginator.paginate(Bucket=bucket_name,Prefix=prefix):if'Contents'inpage:forobjinpage['Contents']:# 检查文件修改时间ifobj['LastModified']<cutoff_date:delete_list.append({'Key':obj['Key']})# AWS一次最多删除1000个对象iflen(delete_list)>=1000:delete_objects(session,bucket_name,delete_list)delete_list=[]# 删除剩余的对象ifdelete_list:delete_objects(session,bucket_name,delete_list)print(f"清理完成:{bucket_name}/{prefix}")defdelete_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}个文件")iferrors>0:print(f" ⚠️ 删除失败:{errors}个文件")defget_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'])ifresponse['Datapoints']:size_bytes=response['Datapoints'][0]['Average']size_gb=size_bytes/(1024**3)returnround(size_gb,2)return0

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

defget_billing_info(session):"""获取AWS账单信息（需要开通Cost Explorer）"""ce=session.client('ce',region_name='us-east-1')# Cost Explorer只在us-east-1# 查询本月支出fromdatetimeimportdatetime,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=0forresultinresponse['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+=costprint(f"本月总支出: ${round(total_cost,2)}")returnbilling_info,round(total_cost,2)defcheck_billing_alert(session,threshold=1000):""" 检查账单是否超预算，超过阈值发送告警 threshold: 预算阈值（美元） """_,total_cost=get_billing_info(session)iftotal_cost>threshold:message=f"⚠️ AWS账单告警！本月支出 ${total_cost}，已超过预算 ${threshold}！"# 发送到企微/钉钉send_alert_to_wecom(message)returnTruereturnFalsedefsend_alert_to_wecom(message):"""发送告警到企微"""webhook_url="https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=YOUR_KEY"payload={"msgtype":"text","text":{"content":message}}importrequests response=requests.post(webhook_url,json=payload)ifresponse.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实例（挖矿！），产生巨额账单。

解决方案：

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

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

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

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

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

defstart_ec2_instances_with_retry(session,instance_ids,max_retries=3):"""带重试的批量启动实例"""failed_ids=[]forattemptinrange(max_retries):ifnotinstance_ids:breakprint(f"第{attempt+1}次尝试启动{len(instance_ids)}个实例...")success_ids=[]forinstance_idininstance_ids:try:ec2=session.client('ec2')ec2.start_instances(InstanceIds=[instance_id])success_ids.append(instance_id)exceptClientErrorase:print(f" 启动{instance_id}失败:{e}")failed_ids.append(instance_id)# 更新待重试列表instance_ids=failed_ids.copy()failed_ids=[]ifinstance_ids:importtime time.sleep(5)# 等待5秒后重试ifinstance_ids:print(f"⚠️ 以下实例启动失败（已达最大重试次数）:{instance_ids}")returnlen(instance_ids)==0# 是否全部成功

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

TEMU店群矩阵自动化运营核价报活动

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

解决方案：

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

# 开启S3版本控制（防止误删）defenable_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("（开启后，删除文件不会真正删除，可以恢复）")# 删除前先备份defsafe_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错误。

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

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