如何使用Python Boto3仅从S3下载与本地存在差异且更新的文件？

嘿，这个需求太常见了，我来帮你捋清楚怎么实现！要判断是否需要下载，其实结合最后修改时间和ETag来做是最靠谱的，既高效又能保证准确性，下面给你详细拆解：

先说说几个判断依据的优劣，方便你理解为什么这么选：

• 最后修改时间（LastModified）：最快速的判断方式，适合绝大多数场景。S3返回的LastModified是UTC时间，只要本地文件的修改时间早于这个时间，就说明S3文件更新。但要注意时区转换的坑，别因为本地时间和UTC不一致导致判断出错。
• ETag：S3对象的ETag默认是文件内容的MD5哈希值（单部分上传的文件），如果是分段上传的文件，ETag会变成类似abc123-4这样的格式（后面的数字是分段数）。所以如果你的文件都是单部分上传的，ETag可以用来精确验证内容是否一致；如果有分段上传的情况，就得换别的方式了。
• MD5哈希：最严格的验证方式，但计算大文件的MD5会比较耗时，而且S3不是所有对象都会返回ContentMD5值，所以一般只作为时间判断后的补充验证。

接下来是修改后的完整代码，我整合了时间判断+ETag验证的逻辑，兼顾效率和准确性：

import boto3
import os
import hashlib
from datetime import datetime, timezone

BUCKET_NAME = 'testing'
KEY = 'your-prefix-here'  # 记得替换成你的实际前缀
LOCAL_DIR = 'test_dir'

# 确保本地目录存在，避免报错
os.makedirs(LOCAL_DIR, exist_ok=True)

s3_client = boto3.client('s3')
response = s3_client.list_objects_v2(Bucket=BUCKET_NAME, Prefix=KEY)

if 'Contents' in response:
    for obj in response['Contents']:
        file_key = obj['Key']
        # 跳过S3上的目录（如果前缀是目录的话）
        if file_key.endswith('/'):
            continue
            
        file_name = os.path.basename(file_key)
        local_file_path = os.path.join(LOCAL_DIR, file_name)
        s3_last_modified = obj['LastModified']
        s3_etag = obj['ETag'].strip('"')  # 去掉ETag自带的双引号
        
        need_download = False
        
        if not os.path.exists(local_file_path):
            # 本地文件不存在，直接下载
            need_download = True
        else:
            # 把本地文件的修改时间转成UTC时间对象，和S3的时间对齐
            local_mtime = os.path.getmtime(local_file_path)
            local_last_modified = datetime.fromtimestamp(local_mtime, timezone.utc)
            
            # 先判断S3文件是否更新
            if s3_last_modified > local_last_modified:
                need_download = True
            else:
                # 如果时间一致，再验证内容是否相同（针对单部分上传的文件）
                def calculate_local_md5(file_path):
                    hash_md5 = hashlib.md5()
                    # 分块读取大文件，避免内存占用过高
                    with open(file_path, "rb") as f:
                        for chunk in iter(lambda: f.read(4096), b""):
                            hash_md5.update(chunk)
                    return hash_md5.hexdigest()
                
                local_md5 = calculate_local_md5(local_file_path)
                # 只有ETag不带分段后缀时，才和本地MD5对比
                if '-' not in s3_etag and local_md5 != s3_etag:
                    need_download = True
        
        if need_download:
            s3_client.download_file(BUCKET_NAME, file_key, local_file_path)
            print(f"已下载更新的文件: {file_name}")
        else:
            print(f"文件{file_name}已是最新，无需下载")

补充几个细节说明：

1. 代码里加了跳过S3目录的逻辑，避免把目录当成文件处理；
2. 处理了ETag自带的双引号，Boto3返回的ETag默认是带引号的字符串，需要去掉才能和MD5对比；
3. 计算本地MD5时用了分块读取，避免大文件占用过多内存；
4. 如果你的场景里有大量分段上传的文件，那ETag的验证逻辑需要调整——因为分段上传的ETag不是纯MD5，这时候可以考虑直接用时间判断，或者调用head_object接口获取ContentMD5（如果上传时提供了的话）。

备注：内容来源于stack exchange，提问作者user1769197