如何用Python获取S3存储桶中最新2个最深层级子文件夹的所有文件

核心思路 #

无需硬编码路径，通过S3的前缀机制识别最深层级的"文件夹"，再筛选出最新的2个，最后读取其中所有文件：

1. 拉取桶内所有对象的键，提取所有可能的前缀（模拟文件夹路径）
2. 筛选出没有子前缀的最深层级前缀
3. 按前缀下文件的最新修改时间排序，取前2个目标前缀
4. 遍历目标前缀下的所有文件，读取内容

代码实现 #

import boto3

def get_deepest_latest_prefixes(bucket_name, count=2):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    
    # 1. 获取桶内所有对象的键
    all_keys = []
    for response in paginator.paginate(Bucket=bucket_name):
        if 'Contents' in response:
            all_keys.extend([obj['Key'] for obj in response['Contents']])
    
    # 2. 提取所有前缀（模拟文件夹的路径）
    prefixes = set()
    for key in all_keys:
        if '/' not in key:
            continue  # 跳过根目录下的单个文件，不视为文件夹
        parts = key.split('/')
        for i in range(1, len(parts)):
            prefix = '/'.join(parts[:i]) + '/'
            prefixes.add(prefix)
    
    # 3. 筛选最深层级的前缀：没有子前缀的前缀
    deepest_prefixes = []
    for prefix in prefixes:
        has_child = any(p.startswith(prefix) and p != prefix for p in prefixes)
        if not has_child:
            deepest_prefixes.append(prefix)
    
    # 4. 获取每个最深前缀的最新修改时间
    prefix_time_map = {}
    for prefix in deepest_prefixes:
        latest_time = None
        for response in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
            if 'Contents' in response:
                for obj in response['Contents']:
                    if not latest_time or obj['LastModified'] > latest_time:
                        latest_time = obj['LastModified']
        if latest_time:
            prefix_time_map[prefix] = latest_time
    
    # 5. 按时间降序排序，取前count个前缀
    sorted_prefixes = sorted(prefix_time_map.items(), key=lambda x: x[1], reverse=True)[:count]
    return [prefix for prefix, _ in sorted_prefixes]

def read_files_in_prefixes(bucket_name, target_prefixes):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    
    for prefix in target_prefixes:
        print(f"=== 处理前缀：{prefix} ===")
        for response in paginator.paginate(Bucket=bucket_name, Prefix=prefix):
            if 'Contents' in response:
                for obj in response['Contents']:
                    if obj['Key'] == prefix:
                        continue  # 跳过模拟文件夹的占位对象（如果存在）
                    # 读取文件内容，二进制文件可去掉decode
                    file_response = s3.get_object(Bucket=bucket_name, Key=obj['Key'])
                    content = file_response['Body'].read().decode('utf-8')
                    print(f"文件 {obj['Key']} 内容：")
                    print(content)
                    print("-"*50)

# 示例调用
if __name__ == "__main__":
    BUCKET_NAME = "your-bucket-name"
    # 获取最新2个最深层级前缀
    target_prefixes = get_deepest_latest_prefixes(BUCKET_NAME, count=2)
    # 读取这些前缀下的所有文件
    read_files_in_prefixes(BUCKET_NAME, target_prefixes)

关键说明 #

• 前缀处理：S3没有真实文件夹，所有"文件夹"都是以/结尾的前缀，代码通过拆分键来提取所有可能的前缀
• 最深层级判断：检查每个前缀是否存在更长的子前缀，没有则视为最深层级
• 时间排序：取每个前缀下所有文件的最新修改时间作为该"文件夹"的更新时间，确保排序准确
• 分页处理：使用boto3的分页器处理大量对象，避免单次API调用的数量限制
• 编码适配：如果读取二进制文件（如图片、压缩包），去掉decode('utf-8')直接读取字节流即可

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