如何在Lambda成功复制文件到目标S3桶后删除源S3桶文件？

以下是满足你需求的修改后代码，保留原循环结构，仅删除成功复制的源文件，同时输出每个文件的处理状态：

import os
import logging
import boto3

LOGGER = logging.getLogger()
LOGGER.setLevel(logging.INFO)

DST_BUCKET = os.environ.get('DST_BUCKET')
REGION = os.environ.get('REGION')

s3 = boto3.resource('s3', region_name=REGION)

def handler(event, context):
    LOGGER.info('Event structure: %s', event)
    LOGGER.info('DST_BUCKET: %s', DST_BUCKET)
    
    # 记录成功复制的文件（桶名+Key），用于后续删除
    successful_files = []
    # 记录每个文件的处理结果
    processing_results = []

    for record in event['Records']:
        src_bucket = record['s3']['bucket']['name']
        src_key = record['s3']['object']['key']
        
        # 跳过S3虚拟文件夹（以/结尾的Key，并非实际文件）
        if src_key.endswith('/'):
            LOGGER.info('Skipping virtual folder: %s', src_key)
            processing_results.append({
                'filename': src_key,
                'success': False,
                'reason': 'Virtual folder, not a file'
            })
            continue

        copy_source = {
            'Bucket': src_bucket,
            'Key': src_key
        }
        LOGGER.info('Processing file: %s', copy_source)

        try:
            # 同步复制文件，复制完成后才会执行后续代码
            bucket = s3.Bucket(DST_BUCKET)
            bucket.copy(copy_source, src_key)
            
            LOGGER.info('Successfully copied file: %s', src_key)
            successful_files.append((src_bucket, src_key))
            processing_results.append({
                'filename': src_key,
                'success': True
            })
        except Exception as e:
            error_msg = f'Failed to copy file {src_key}: {str(e)}'
            LOGGER.error(error_msg)
            processing_results.append({
                'filename': src_key,
                'success': False,
                'reason': error_msg
            })

    # 批量删除所有成功复制的源文件
    if successful_files:
        # 按源桶分组删除，避免跨桶操作错误
        bucket_files_map = {}
        for bucket_name, key in successful_files:
            if bucket_name not in bucket_files_map:
                bucket_files_map[bucket_name] = []
            bucket_files_map[bucket_name].append({'Key': key})
        
        for bucket_name, delete_objs in bucket_files_map.items():
            s3_bucket = s3.Bucket(bucket_name)
            s3_bucket.delete_objects(Delete={'Objects': delete_objs})
            LOGGER.info('Deleted %d successful files from source bucket %s', len(delete_objs), bucket_name)

    return {
        'status': 'completed',
        'processing_results': processing_results
    }

核心解决逻辑说明 #

1. 异常隔离处理：每个文件的复制操作单独用try-except包裹，单个文件复制失败不会中断整个循环，其他文件仍能正常处理。
2. 精准跟踪成功文件：用successful_files列表记录复制成功的文件（桶名+Key），确保只有确认复制完成的文件才会被删除，彻底避免“提前删除”的同步问题。
3. 过滤虚拟文件夹：通过判断src_key是否以/结尾，跳过S3的虚拟文件夹对象（这类对象不是实际文件，删除会导致文件夹“消失”，不符合你的需求）。
4. 批量删除优化：按源桶分组批量删除成功文件，减少API调用次数，提升执行效率。
5. 详细状态返回：返回结果包含每个文件的处理状态（成功/失败+原因），方便你排查问题，符合多文件上传场景的需求。

内容的提问来源于stack exchange，提问作者stonewalker747