如何使用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文件到本地的写入,无需手动处理流操作 - 异常捕获:多层异常处理,方便排查单个文件下载失败或整体操作出错的问题
额外提示
- 生产环境别把AWS密钥硬编码!建议用环境变量、IAM角色或者AWS凭证管理器来管理密钥
- 如果你的Bucket里文件超过1000个,
list_objects_v2会分页返回结果,这时候需要循环处理NextContinuationToken来获取全部文件,要是需要我可以补充分页逻辑~ - 确保你的AWS账号拥有对应Bucket的
s3:ListBucket(查看文件列表)和s3:GetObject(下载文件)权限
内容的提问来源于stack exchange,提问作者user11669928




