如何从S3存储桶各子文件夹下载指定数量文件？

背景 #

假设我的S3存储桶结构如下：

Top Level Prefix
    Sub Prefix A
        Objects 1...100+
    Sub Prefix B
        Objects 1...100+

说明 #

我不会遇到子文件夹中对象与前缀共存的情况，例如以下结构不会出现：

Top Level Prefix
    Objects 1...100+
    Sub Prefix A
        Objects 1...100+
    etc.

目标 #

从每个子文件夹中下载指定数量N的文件。

尝试的思路 #

• 使用boto3的list_objects_v2方法，设置Delimiter为'/'，获取通用前缀列表。
• 再次调用list_objects_v2，循环传入上一步得到的前缀作为Prefix参数，同时将MaxKeys设为指定值N。
• 每次循环中将返回的键名添加到列表中（每个子文件夹对应N个文件）。
• 最后遍历该键名列表，使用boto3的.download_file()方法下载文件。

问题 #

现有代码只能获取根目录下的顶级前缀，无法识别嵌套的子文件夹，如何实现完整需求？

完整实现代码 #

1. 初始化依赖与核心函数 #

import boto3
import os

def get_final_prefixes(s3_client, bucket_name, parent_prefix=''):
    """递归获取所有最终前缀（即直接包含对象的子文件夹前缀）"""
    response = s3_client.list_objects_v2(
        Bucket=bucket_name,
        Prefix=parent_prefix,
        Delimiter='/'
    )
    common_prefixes = response.get('CommonPrefixes', [])
    if not common_prefixes:
        # 没有子前缀，当前前缀为最终前缀
        yield parent_prefix
    else:
        # 递归遍历每个子前缀
        for prefix_info in common_prefixes:
            yield from get_final_prefixes(s3_client, bucket_name, prefix_info['Prefix'])

def get_n_objects_per_prefix(s3_client, bucket_name, prefix, n):
    """获取指定前缀下的前N个对象键，处理分页情况"""
    object_keys = []
    continuation_token = None
    while len(object_keys) < n:
        # 每次请求最多取1000个（S3单次请求的最大限制），或剩余需要的数量
        batch_size = min(n - len(object_keys), 1000)
        kwargs = {
            'Bucket': bucket_name,
            'Prefix': prefix,
            'MaxKeys': batch_size
        }
        if continuation_token:
            kwargs['ContinuationToken'] = continuation_token
        
        response = s3_client.list_objects_v2(**kwargs)
        contents = response.get('Contents', [])
        object_keys.extend([obj['Key'] for obj in contents])
        
        continuation_token = response.get('NextContinuationToken')
        if not continuation_token:
            break  # 没有更多对象可获取
    
    # 返回最多N个对象键
    return object_keys[:n]

def download_objects(s3_client, bucket_name, object_keys, download_dir='./s3_downloads'):
    """批量下载S3对象到本地，保持原前缀目录结构"""
    os.makedirs(download_dir, exist_ok=True)
    for key in object_keys:
        local_file_path = os.path.join(download_dir, key)
        # 创建文件所在的父目录
        os.makedirs(os.path.dirname(local_file_path), exist_ok=True)
        s3_client.download_file(bucket_name, key, local_file_path)
        print(f"已完成下载: {key}")

2. 主执行逻辑 #

if __name__ == '__main__':
    # 配置参数
    BUCKET_NAME = 'your-bucket-name'  # 替换为你的存储桶名称
    BUCKET_REGION = 'us-east-1'       # 替换为你的存储桶区域
    DOWNLOAD_COUNT_PER_FOLDER = 5     # 每个子文件夹要下载的文件数量

    # 初始化S3客户端
    s3_client = boto3.client('s3', region_name=BUCKET_REGION)

    # 获取所有需要下载的子文件夹前缀
    final_prefixes = list(get_final_prefixes(s3_client, BUCKET_NAME))
    print(f"找到 {len(final_prefixes)} 个目标子文件夹")

    # 收集所有待下载的对象键
    all_download_keys = []
    for prefix in final_prefixes:
        folder_keys = get_n_objects_per_prefix(s3_client, BUCKET_NAME, prefix, DOWNLOAD_COUNT_PER_FOLDER)
        all_download_keys.extend(folder_keys)
        print(f"从文件夹 {prefix} 获取了 {len(folder_keys)} 个文件")

    # 执行下载
    if all_download_keys:
        download_objects(s3_client, BUCKET_NAME, all_download_keys)
        print(f"全部下载完成，共下载 {len(all_download_keys)} 个文件")
    else:
        print("没有找到可下载的文件")

关键说明 #

• get_final_prefixes：递归遍历存储桶的前缀结构，自动找到所有直接包含对象的子文件夹，适配任意层级的嵌套结构。
• get_n_objects_per_prefix：处理S3的分页限制（单次最多返回1000个对象），确保能获取到指定数量的文件，若子文件夹内对象不足N个，则返回所有对象。
• download_objects：自动创建本地目录结构，保持与S3前缀一致，避免文件重名或目录缺失问题。

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