You need to enable JavaScript to run this app.
最新活动
大模型
产品
解决方案
定价
生态与合作
支持与服务
开发者
了解我们

如何使用boto从S3批量下载最后更新/上传的文件

解决S3批量下载最后更新/上传文件的问题

嘿,我看你已经有了获取S3文件列表的基础代码,现在要实现批量下载最新更新的文件对吧?咱们一步步来搞定它:

首先先把你现有代码里的重复导入清理掉,然后添加筛选最新文件和下载的逻辑。核心思路是先找到所有文件里最新的LastModified时间(S3里这个字段同时代表最后上传/修改时间),再把所有匹配这个时间的文件筛选出来,最后批量下载到本地。

完整实现代码

import os
import boto3
from datetime import datetime, timezone

# 初始化S3客户端
aws_access_key_id = '***'
aws_secret_access_key = '***'
client = boto3.client(
    's3',
    aws_access_key_id=aws_access_key_id,
    aws_secret_access_key=aws_secret_access_key
)

def download_latest_s3_files(bucket_name, local_save_dir):
    # 创建本地保存目录(不存在则自动创建)
    os.makedirs(local_save_dir, exist_ok=True)
    
    try:
        # 获取Bucket中的所有文件对象
        response = client.list_objects_v2(Bucket=bucket_name)
        if 'Contents' not in response:
            print("Bucket中没有任何文件")
            return
        
        files = response['Contents']
        
        # 找到所有文件里最新的修改时间
        latest_modified = max(file['LastModified'] for file in files)
        
        # 筛选出所有最后更新时间等于最新时间的文件
        latest_files = [file for file in files if file['LastModified'] == latest_modified]
        
        if not latest_files:
            print("没有找到符合条件的最新文件")
            return
        
        print(f"找到{len(latest_files)}个最新更新的文件,开始下载...")
        
        # 遍历下载每个文件
        for file in latest_files:
            file_key = file['Key']
            # 处理本地文件路径,保留文件名
            local_file_path = os.path.join(local_save_dir, os.path.basename(file_key))
            
            try:
                # 调用S3客户端的下载方法
                client.download_file(bucket_name, file_key, local_file_path)
                print(f"成功下载: {file_key} -> {local_file_path}")
            except Exception as e:
                print(f"下载{file_key}失败: {str(e)}")
                
    except Exception as e:
        print(f"整体操作出错: {str(e)}")

# 调用函数,替换成你的Bucket名和本地保存路径
if __name__ == "__main__":
    download_latest_s3_files(bucket_name='mybuycket', local_save_dir='./s3_latest_files')

关键逻辑说明

  • 目录预处理:用os.makedirs确保本地保存目录存在,避免下载时因路径不存在报错
  • 筛选最新文件:通过max()快速定位最晚的更新时间,再筛选出所有同时间更新的文件(支持批量同批次上传的文件)
  • 文件下载:使用client.download_file()方法,直接完成S3文件到本地的写入,无需手动处理流操作
  • 异常捕获:多层异常处理,方便排查单个文件下载失败或整体操作出错的问题

额外提示

  1. 生产环境别把AWS密钥硬编码!建议用环境变量、IAM角色或者AWS凭证管理器来管理密钥
  2. 如果你的Bucket里文件超过1000个,list_objects_v2会分页返回结果,这时候需要循环处理NextContinuationToken来获取全部文件,要是需要我可以补充分页逻辑~
  3. 确保你的AWS账号拥有对应Bucket的s3:ListBucket(查看文件列表)和s3:GetObject(下载文件)权限

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

火山引擎 最新活动