影刀RPA AWS云服务自动化EC2 S3管理实战什么情况用什么 → 怎么做 → 有什么坑作者林焱 | 飞行社出品什么情况用什么企业用AWS亚马逊云每天要手动启停EC2实例、清理S3存储桶、监控账单重复性工作多到爆。这套方案适合企业运维团队自动化AWS资源管理定时启停开发/测试环境省钱S3存储桶自动清理避免产生不必要的费用核心工具影刀RPA Boto3AWS SDK for Python CloudWatch监控怎么做第一步配置AWS凭证拼多多店群自动化报活动上架importboto3importosfrombotocore.exceptionsimportNoCredentialsError,ClientErrordefinit_aws_session(access_key,secret_key,regionus-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_idaccess_key,# aws_secret_access_keysecret_key,# region_nameregion# )# 方法2从环境变量读取推荐# 先设置环境变量# export AWS_ACCESS_KEY_IDyour_access_key# export AWS_SECRET_ACCESS_KEYyour_secret_keysessionboto3.Session(aws_access_key_idos.getenv(AWS_ACCESS_KEY_ID),aws_secret_access_keyos.getenv(AWS_SECRET_ACCESS_KEY),region_nameregion)# 验证凭证是否有效try:stssession.client(sts)identitysts.get_caller_identity()print(fAWS凭证有效账号ID:{identity[Account]})returnsessionexceptNoCredentialsError:print(⚠️ AWS凭证未找到请配置环境变量)returnNoneexceptClientErrorase:print(f⚠️ AWS凭证无效:{e})returnNone# 使用示例sessioninit_aws_session(access_keyYOUR_ACCESS_KEY,secret_keyYOUR_SECRET_KEY,regioncn-north-1# 北京区域)第二步自动化EC2实例管理deflist_ec2_instances(session):列出所有EC2实例ec2session.client(ec2)responseec2.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实例ec2session.client(ec2)try:responseec2.start_instances(InstanceIdsinstance_ids)forinstanceinresponse[StartingInstances]:print(f正在启动实例:{instance[InstanceId]})print(f 当前状态:{instance[CurrentState][Name]})returnTrueexceptClientErrorase:print(f启动实例失败:{e})returnFalsedefstop_ec2_instances(session,instance_ids,forceFalse):批量停止EC2实例ec2session.client(ec2)try:responseec2.stop_instances(InstanceIdsinstance_ids,Forceforce# 是否强制关机)forinstanceinresponse[StoppingInstances]:print(f正在停止实例:{instance[InstanceId]})print(f 当前状态:{instance[CurrentState][Name]})returnTrueexceptClientErrorase:print(f停止实例失败:{e})returnFalsedefauto_stop_dev_instances(session,tag_nameEnvironment,tag_valueDev): 自动停止开发环境实例定时任务调用 例如每天晚上8点自动停止所有EnvironmentDev的实例 ec2session.client(ec2)# 查询符合条件的实例responseec2.describe_instances(Filters[{Name:ftag:{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存储桶s3session.client(s3)responses3.list_buckets()buckets[]forbucketinresponse[Buckets]:buckets.append({存储桶名称:bucket[Name],创建时间:bucket[CreationDate]})returnbucketsdefclean_s3_bucket(session,bucket_name,prefix,days30): 清理S3存储桶中的旧文件 prefix: 前缀如 logs/ 只清理logs目录 days: 删除多少天前的文件 importdatetime s3session.client(s3)# 计算截止时间cutoff_datedatetime.datetime.now()-datetime.timedelta(daysdays)# 列出所有对象paginators3.get_paginator(list_objects_v2)delete_list[]forpageinpaginator.paginate(Bucketbucket_name,Prefixprefix):ifContentsinpage: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对象s3session.client(s3)responses3.delete_objects(Bucketbucket_name,Delete{Objects:delete_list})deletedlen(response.get(Deleted,[]))errorslen(response.get(Errors,[]))print(f 已删除:{deleted}个文件)iferrors0:print(f ⚠️ 删除失败:{errors}个文件)defget_s3_bucket_size(session,bucket_name):获取S3存储桶大小需要CloudWatch权限cloudwatchsession.client(cloudwatch)responsecloudwatch.get_metric_statistics(NamespaceAWS/S3,MetricNameBucketSizeBytes,Dimensions[{Name:BucketName,Value:bucket_name}],StartTimedatetime.datetime.now()-datetime.timedelta(days1),EndTimedatetime.datetime.now(),Period86400,Statistics[Average])ifresponse[Datapoints]:size_bytesresponse[Datapoints][0][Average]size_gbsize_bytes/(1024**3)returnround(size_gb,2)return0第四步监控AWS账单防止超支defget_billing_info(session):获取AWS账单信息需要开通Cost Explorercesession.client(ce,region_nameus-east-1)# Cost Explorer只在us-east-1# 查询本月支出fromdatetimeimportdatetime,timedelta start_datedatetime.now().replace(day1).strftime(%Y-%m-%d)end_datedatetime.now().strftime(%Y-%m-%d)responsece.get_cost_and_usage(TimePeriod{Start:start_date,End:end_date},GranularityMONTHLY,Metrics[BlendedCost],GroupBy[{Type:DIMENSION,Key:SERVICE}])billing_info[]total_cost0forresultinresponse[ResultsByTime][0][Groups]:service_nameresult[Keys][0]costfloat(result[Metrics][BlendedCost][Amount])billing_info.append({服务名称:service_name,本月支出(USD):round(cost,2)})total_costcostprint(f本月总支出: ${round(total_cost,2)})returnbilling_info,round(total_cost,2)defcheck_billing_alert(session,threshold1000): 检查账单是否超预算超过阈值发送告警 threshold: 预算阈值美元 _,total_costget_billing_info(session)iftotal_costthreshold:messagef⚠️ AWS账单告警本月支出 ${total_cost}已超过预算 ${threshold}# 发送到企微/钉钉send_alert_to_wecom(message)returnTruereturnFalsedefsend_alert_to_wecom(message):发送告警到企微webhook_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyYOUR_KEYpayload{msgtype:text,text:{content:message}}importrequests responserequests.post(webhook_url,jsonpayload)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存储桶大小、本月账单 ↓ 【发送邮件】将报告发送给运维团队有什么坑坑1AWS凭证泄露风险在代码里硬编码AWS Access Key和Secret Key一旦代码泄露别人可以用你的凭证启动大量EC2实例挖矿产生巨额账单。解决方案使用IAM Role推荐给EC2实例绑定IAM Role无需在代码里配置凭证使用环境变量export AWS_ACCESS_KEY_IDxxx使用AWS Secrets Manager把凭证存在AWS Secrets Manager里用时动态获取最小权限原则IAM用户只给必要的权限如只读权限、只能启停特定实例# 使用IAM Role最安全# 在EC2实例上运行代码时无需配置凭证boto3会自动从实例元数据获取临时凭证sessionboto3.Session(region_namecn-north-1)坑2批量操作失败部分成功部分失败批量启停EC2实例时可能有的成功有的失败需要记录失败的部分并重试。解决方案实现重试机制defstart_ec2_instances_with_retry(session,instance_ids,max_retries3):带重试的批量启动实例failed_ids[]forattemptinrange(max_retries):ifnotinstance_ids:breakprint(f第{attempt1}次尝试启动{len(instance_ids)}个实例...)success_ids[]forinstance_idininstance_ids:try:ec2session.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_idsfailed_ids.copy()failed_ids[]ifinstance_ids:importtime time.sleep(5)# 等待5秒后重试ifinstance_ids:print(f⚠️ 以下实例启动失败已达最大重试次数:{instance_ids})returnlen(instance_ids)0# 是否全部成功坑3S3删除操作不可逆误删数据无法恢复TEMU店群矩阵自动化运营核价报活动S3删除文件后默认无法恢复除非开启了版本控制。解决方案开启S3版本控制推荐删除前先移动到回收站先复制到另一个存储桶再删除原文件使用生命周期策略配置S3自动清理规则而不是手动删除# 开启S3版本控制防止误删defenable_s3_versioning(session,bucket_name):开启S3版本控制s3session.client(s3)s3.put_bucket_versioning(Bucketbucket_name,VersioningConfiguration{Status:Enabled})print(fS3版本控制已开启:{bucket_name})print(开启后删除文件不会真正删除可以恢复)# 删除前先备份defsafe_delete_s3_object(session,bucket_name,object_key,backup_bucket):安全删除S3对象先备份s3session.client(s3)# 1. 先复制到备份存储桶copy_source{Bucket:bucket_name,Key:object_key}s3.copy_object(CopySourcecopy_source,Bucketbackup_bucket,Keyfbackup/{object_key})# 2. 再删除原文件s3.delete_object(Bucketbucket_name,Keyobject_key)print(f安全删除:{bucket_name}/{object_key}已备份到{backup_bucket}/backup/)坑4AWS API调用频率限制ThrottlingAWS有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:ifThrottlinginstr(e):print(f触发频率限制等待5秒后重试...)time.sleep(5)operate_on_instance(session,instance_id)else:raise总结功能节省成本附加价值定时启停EC2最高省70%算力费用避免忘记关机产生的浪费S3自动清理省存储费用避免旧文件堆积| 账单监控 | — | 及时发现异常支出 || 资源监控 | — | 优化资源使用 |实际落地建议先开启CloudTrail记录所有API调用便于审计和问题排查使用AWS Organizations统一管理多个AWS账号的账单设置Billing Alarm在AWS控制台设置账单告警双重保障定期审查IAM权限移除不必要的权限降低安全风险AWS自动化能为运维团队节省50%以上的日常操作时间同时避免因人为疏忽导致的资源浪费和安全风险。