You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何用boto3获取S3桶指定目录中符合时间过滤的文件并解决分页限制?

解决方案

1. 解决分页问题(获取全量文件)

S3的list_objects/list_objects_v2接口单次默认最多返回1000条数据,要拿到所有文件,推荐用**分页器(Paginator)**自动处理分页逻辑,无需手动循环判断分页标记。

2. 按时间戳过滤文件

先计算目标时间范围(比如昨天下午的起止时间),遍历每个S3对象时,对比其LastModified属性,筛选符合条件的文件即可。

完整代码示例

from boto3 import client
from datetime import datetime, timedelta, timezone

# 初始化S3客户端
s3_client = client('s3')

# 定义桶名和目标前缀
bucket_name = 'foo-bar'
prefix = 'prod/'

# 计算时间范围:昨天下午12点到当天23:59(统一用UTC时间,匹配S3的时间格式)
yesterday_utc = datetime.now(timezone.utc) - timedelta(days=1)
start_time = yesterday_utc.replace(hour=12, minute=0, second=0, microsecond=0)
end_time = yesterday_utc.replace(hour=23, minute=59, second=59, microsecond=999999)

# 用分页器获取所有前缀匹配的对象
paginator = s3_client.get_paginator('list_objects_v2')
page_iterator = paginator.paginate(Bucket=bucket_name, Prefix=prefix)

# 筛选符合时间条件的文件
filtered_files = []
for page in page_iterator:
    if 'Contents' not in page:
        continue
    for obj in page['Contents']:
        if start_time <= obj['LastModified'] <= end_time:
            filtered_files.append({
                '文件路径': obj['Key'],
                '修改时间': obj['LastModified'],
                '文件大小': obj['Size']
            })

# 输出结果
print(f"找到符合条件的文件共{len(filtered_files)}个:")
for file in filtered_files:
    print(f"{file['文件路径']} | 修改时间:{file['修改时间']}")

关键细节说明

  • 优先用list_objects_v2:它是list_objects的升级版,功能更完善,分页逻辑更清晰。
  • 时区统一:S3对象的LastModified是UTC时间,计算时间范围时必须用UTC时间,避免时区偏差导致过滤错误。
  • 灵活调整时间规则:如果只需要"比某个特定时间戳新的文件",直接把判断条件改成obj['LastModified'] > 目标时间戳即可。

内容的提问来源于stack exchange,提问作者Unknowntiou

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.31 04:39:34