使用Python Lambda函数复制S3桶中文件名时间晚于指定时间的多份文件

没问题，我来帮你搞定这个需求！下面是符合要求的Python Lambda函数基础框架，专门处理你提到的文件名带时间戳的场景：

基础代码框架 #

import boto3
from datetime import datetime

def lambda_handler(event, context):
    # 配置参数，根据你的实际情况修改
    SOURCE_BUCKET = '你的源S3桶名称'
    DESTINATION_BUCKET = '你的目标S3桶名称'
    # 设定截止时间，注意格式要和文件名的时间对应
    CUTOFF_TIME = datetime.strptime('2021-09-28 04:32:21', '%Y-%m-%d %H:%M:%S')
    
    # 初始化S3客户端
    s3 = boto3.client('s3')
    
    try:
        # 列出源桶中的所有文件
        response = s3.list_objects_v2(Bucket=SOURCE_BUCKET)
        if 'Contents' not in response:
            print("源桶中没有文件")
            return
        
        for obj in response['Contents']:
            file_name = obj['Key']
            # 跳过文件夹（如果有的话）
            if file_name.endswith('/'):
                continue
            
            # 解析文件名中的时间戳
            # 文件名格式：abc_ddmmyyyy_hhmmss.csv
            try:
                # 按下划线分割文件名
                parts = file_name.split('_')
                if len(parts) < 3:
                    print(f"文件名格式不符合要求：{file_name}")
                    continue
                # 拼接时间部分：ddmmyyyy_hhmmss
                timestamp_str = f"{parts[1]}_{parts[2].split('.')[0]}"
                # 转成datetime对象
                file_time = datetime.strptime(timestamp_str, '%d%m%Y_%H%M%S')
                
                # 对比时间，晚于截止时间则复制
                if file_time > CUTOFF_TIME:
                    print(f"正在复制文件：{file_name}")
                    # 执行复制操作
                    s3.copy_object(
                        Bucket=DESTINATION_BUCKET,
                        Key=file_name,
                        CopySource={'Bucket': SOURCE_BUCKET, 'Key': file_name}
                    )
                    print(f"文件{file_name}复制成功")
            except Exception as e:
                print(f"解析文件名{file_name}时出错：{str(e)}")
                continue
        
        return {
            'statusCode': 200,
            'body': '文件复制任务完成'
        }
    except Exception as e:
        print(f"执行过程中出错：{str(e)}")
        return {
            'statusCode': 500,
            'body': f"任务失败：{str(e)}"
        }

关键部分说明 #

• 时间戳解析：因为你的文件名是abc_ddmmyyyy_hhmmss.csv格式，所以用%d%m%Y_%H%M%S作为解析格式，如果你后续文件名格式有变化，记得同步修改这个格式字符串。
• Lambda权限配置：要确保你的Lambda角色拥有以下权限：
  • 源S3桶的s3:ListBucket权限（用来列出文件）
  • 源S3桶的s3:GetObject权限（复制时需要读取源文件）
  • 目标S3桶的s3:PutObject权限（用来写入复制后的文件）
• 分页处理：如果你的源桶里文件很多（超过1000个），上面的list_objects_v2只会返回第一页结果，你需要添加分页逻辑（检查NextContinuationToken）来遍历所有文件，这个可以根据你的实际文件数量来补充。
• 触发方式：你可以把这个Lambda设置为定时触发（比如用CloudWatch Events），或者当源桶有新文件上传时触发，根据你的需求调整即可。

内容的提问来源于stack exchange，提问作者Sri Vidhya Pavani