如何遍历带日期层级的S3桶并按文件夹为文件标记日期

我来帮你搞定这个需求！这种按年/月/日层级存文件但文件名不带日期的场景太常见了，用boto3完全可以轻松实现给文件标记对应日期的目标。下面我给你两种实用的解决方案，还有完整的代码示例：

我们要做的就是：遍历S3桶里的所有文件 → 从文件的路径（也就是S3对象的Key）里提取年、月、日 → 把日期标记到文件上（要么加元数据，要么改文件名）。

第一步：准备工作 #

首先确保你已经配置好了AWS凭证（可以通过~/.aws/credentials文件、环境变量，或者IAM角色如果是在EC2/EKS上运行的话），然后安装boto3：

pip install boto3

第二步：遍历S3对象并提取日期 #

S3的文件夹其实是前缀（Prefix），不是真实的目录，所以我们需要遍历所有对象，然后从它们的Key中解析日期。比如2018/3/24/sales_data.csv，分割后前三个部分就是2018、3、24。

方案一：给文件添加自定义元数据（推荐） #

这种方式不会修改文件名，而是给文件附加一个日期元数据，后续加载数据库时直接读取这个元数据即可，非常灵活。

完整代码：

import boto3
from botocore.exceptions import ClientError

def add_date_metadata_to_s3_objects(bucket_name):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(Bucket=bucket_name)

    for page in page_iterator:
        if 'Contents' not in page:
            continue  # 桶里没有对象，跳过

        for obj in page['Contents']:
            obj_key = obj['Key']
            # 跳过文件夹前缀（S3里的文件夹是末尾带/的空对象）
            if obj_key.endswith('/'):
                continue

            # 分割路径提取年、月、日
            path_parts = obj_key.split('/')
            # 确保路径是年/月/日/文件名的结构
            if len(path_parts) < 4:
                print(f"跳过不符合层级的文件: {obj_key}")
                continue

            year, month, day = path_parts[0], path_parts[1], path_parts[2]
            # 格式化日期为YYYY-MM-DD，补零处理（比如3→03）
            formatted_date = f"{year}-{month.zfill(2)}-{day.zfill(2)}"

            try:
                # 获取文件当前的元数据
                response = s3.head_object(Bucket=bucket_name, Key=obj_key)
                current_metadata = response.get('Metadata', {})

                # 如果已经有日期元数据，跳过避免重复操作
                if 'date' in current_metadata and current_metadata['date'] == formatted_date:
                    print(f"文件 {obj_key} 已存在日期元数据，跳过")
                    continue

                # 复制文件并添加新的元数据（S3修改元数据需要复制对象）
                s3.copy_object(
                    Bucket=bucket_name,
                    Key=obj_key,
                    CopySource={'Bucket': bucket_name, 'Key': obj_key},
                    Metadata={**current_metadata, 'date': formatted_date},
                    MetadataDirective='REPLACE'
                )
                print(f"成功给文件 {obj_key} 添加日期元数据: {formatted_date}")
            except ClientError as e:
                print(f"处理文件 {obj_key} 出错: {e}")

# 替换成你的桶名
add_date_metadata_to_s3_objects('your-bucket-name')

方案二：重命名文件（修改文件名） #

如果你希望直接把日期加到文件名里，比如把sales_data.csv改成2018-03-24-sales_data.csv，可以用下面的代码：

import boto3
from botocore.exceptions import ClientError
import os

def rename_s3_files_with_date(bucket_name):
    s3 = boto3.client('s3')
    paginator = s3.get_paginator('list_objects_v2')
    page_iterator = paginator.paginate(Bucket=bucket_name)

    for page in page_iterator:
        if 'Contents' not in page:
            continue

        for obj in page['Contents']:
            obj_key = obj['Key']
            if obj_key.endswith('/'):
                continue

            path_parts = obj_key.split('/')
            if len(path_parts) < 4:
                print(f"跳过不符合层级的文件: {obj_key}")
                continue

            year, month, day = path_parts[0], path_parts[1], path_parts[2]
            formatted_date = f"{year}-{month.zfill(2)}-{day.zfill(2)}"
            # 分离文件名和扩展名
            filename = os.path.basename(obj_key)
            name, ext = os.path.splitext(filename)
            # 新的文件名：日期+原文件名
            new_filename = f"{formatted_date}-{name}{ext}"
            # 新的Key：原路径的前三个层级+新文件名
            new_obj_key = '/'.join(path_parts[:3] + [new_filename])

            # 避免重名（如果新Key已存在则跳过）
            try:
                s3.head_object(Bucket=bucket_name, Key=new_obj_key)
                print(f"新文件名 {new_obj_key} 已存在，跳过原文件 {obj_key}")
                continue
            except ClientError as e:
                if e.response['Error']['Code'] == '404':
                    # 新Key不存在，可以重命名
                    try:
                        s3.copy_object(
                            Bucket=bucket_name,
                            Key=new_obj_key,
                            CopySource={'Bucket': bucket_name, 'Key': obj_key}
                        )
                        # 复制完成后删除原文件（可选，根据你的需求）
                        s3.delete_object(Bucket=bucket_name, Key=obj_key)
                        print(f"成功重命名文件: {obj_key} → {new_obj_key}")
                    except ClientError as e:
                        print(f"重命名文件 {obj_key} 出错: {e}")
                else:
                    print(f"检查新文件是否存在时出错: {e}")

# 替换成你的桶名
rename_s3_files_with_date('your-bucket-name')

关键细节说明 #

• 分页处理：用paginator处理超过1000个对象的情况，避免一次性请求返回数据过多。
• 补零处理：用zfill(2)把单数字的月份/日期改成两位（比如3→03），保证日期格式统一。
• 跳过文件夹：S3里的文件夹是末尾带/的空对象，所以要跳过这些，只处理实际的文件。
• 错误处理：添加了ClientError捕获，处理权限不足、文件不存在等异常情况。

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