使用Boto3和Python高效遍历S3存储桶对象：深入解析s3list生成器

import boto3
from typing import Iterator, Dict, Any

def s3list(bucket_name: str, path: str, start: str = None, end: str = None, list_dirs: bool = False) -> Iterator[Dict[str, Any]]:
    """
    一个生成器函数，用于列出S3存储桶中指定路径（前缀）下的对象。
    支持按键后缀进行起始和结束过滤，并可选择是否包含目录。

    Args:
        bucket_name (str): S3存储桶的名称。
        path (str): 要列出对象的S3前缀（目录路径），例如 'folder/subfolder/'.
        start (str, optional): 一个键后缀，用于指定从哪个键开始列出（包含）。
                               例如，如果path是'logs/'，start='2023/05/01'，则从'logs/2023/05/01'开始。
        end (str, optional): 一个键后缀，用于指定到哪个键结束列出（不包含）。
                             例如，如果path是'logs/'，end='2023/06'，则列出到'logs/2023/06'之前。
        list_dirs (bool): 如果为True，将包含以'/'结尾的目录键。如果为False，则跳过。默认为False。

    Yields:
        dict: S3对象的字典表示（来自list_objects_v2的'Contents'字段）。
              包含 'Key', 'LastModified', 'ETag', 'Size' 等信息。
    """
    s3_client = boto3.client('s3')
    paginator = s3_client.get_paginator('list_objects_v2')

    list_kwargs = {'Bucket': bucket_name, 'Prefix': path}
    if start:
        # StartAfter参数是独占的，所以如果我们要包含start，我们需要调整
        # 然而，原始答案的s3list在内部处理了start，我们这里模拟其行为
        # 对于list_objects_v2，StartAfter是字典序的，所以直接拼接即可
        list_kwargs['StartAfter'] = path + start

    for page in paginator.paginate(**list_kwargs):
        for obj_summary in page.get('Contents', []):
            key = obj_summary['Key']

            # 应用结束键过滤 (end是独占的)
            if end and key >= path + end:
                return  # 达到结束条件，停止生成

            # 过滤目录键
            if not list_dirs and key.endswith('/'):
                continue

            yield obj_summary

# 替换为您的S3存储桶名称
my_bucket_name = 'your-s3-bucket-name'
target_path = 'splunk-kinesis-firehose/splunk-failed/' # 注意路径末尾的'/'，确保是前缀

print(f"Listing all objects under: {target_path}")
for s3obj in s3list(my_bucket_name, target_path, list_dirs=False):
    key = s3obj['Key']
    print(f"Found object: {key}")
    # 在这里可以对key进行进一步处理，例如下载、分析等

# 替换为您的S3存储桶名称
my_bucket_name = 'your-s3-bucket-name'
target_path = 'splunk-kinesis-firehose/splunk-failed/'

# 获取2023年5月1日到2023年5月31日的文件
# start='2023/05/01' 表示从'splunk-kinesis-firehose/splunk-failed/2023/05/01'开始
# end='2023/06' 表示到'splunk-kinesis-firehose/splunk-failed/2023/06'之前（不包含6月份）
print(f"\nListing objects for May 2023 under: {target_path}")
for s3obj in s3list(my_bucket_name, target_path, start='2023/05/01', end='2023/06', list_dirs=False):
    key = s3obj['Key']
    print(f"Found object (May 2023): {key}")

import urllib.parse

# 假设从S3事件中获取到一个URL编码的键
encoded_key = "Folder%2FFolder%2FYear%2FMonth%2FDay%2FHH%2Ffailedlogs.gz"
decoded_key = urllib.parse.unquote_plus(encoded_key, encoding='utf-8')
print(f"\nDecoded key: {decoded_key}")