如何用Python Boto3校验本地文件与S3文件是否一致？

下面是一套实用的实现方案，核心通过校验和对比判断文件是否变更，只上传修改过的文件：

1. 初始化Boto3客户端 #

先配置好AWS凭证（可通过环境变量、~/.aws/credentials文件或IAM角色），再初始化S3客户端：

import boto3
import hashlib
import os

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

2. 计算本地文件的MD5校验和 #

为避免大文件占用过多内存，采用分块读取的方式计算MD5：

def calculate_local_md5(file_path, chunk_size=8192):
    md5 = hashlib.md5()
    with open(file_path, 'rb') as f:
        while chunk := f.read(chunk_size):
            md5.update(chunk)
    return md5.hexdigest()

3. 获取S3文件的ETag #

S3对象的ETag通常对应文件的MD5（单块上传时），但分块上传时格式会变成MD5值-分块数，我们先获取这个标识：

def get_s3_etag(bucket_name, s3_key):
    try:
        response = s3.head_object(Bucket=bucket_name, Key=s3_key)
        # 去掉ETag前后的引号
        return response['ETag'].strip('"')
    except s3.exceptions.ClientError as e:
        # 文件不存在返回None
        if e.response['Error']['Code'] == '404':
            return None
        else:
            raise

4. 对比校验和并同步文件 #

遍历本地文件夹的所有文件，对比本地MD5和S3的ETag，仅当文件不存在或已修改时上传：

def sync_local_to_s3(local_dir, bucket_name, s3_prefix=''):
    for root, dirs, files in os.walk(local_dir):
        for file in files:
            local_path = os.path.join(root, file)
            # 构造S3存储键：转换为S3兼容的路径格式（用/分隔）
            relative_path = os.path.relpath(local_path, local_dir)
            s3_key = os.path.join(s3_prefix, relative_path).replace('\\', '/')

            local_md5 = calculate_local_md5(local_path)
            s3_etag = get_s3_etag(bucket_name, s3_key)

            if not s3_etag:
                print(f"上传新文件: {local_path}")
                s3.upload_file(local_path, bucket_name, s3_key)
            else:
                # 单块上传的ETag可直接对比MD5
                if '-' not in s3_etag:
                    if local_md5 == s3_etag:
                        print(f"文件未变更: {local_path}")
                    else:
                        print(f"上传修改后的文件: {local_path}")
                        s3.upload_file(local_path, bucket_name, s3_key)
                else:
                    # 分块上传的ETag无法直接对比MD5，改用修改时间判断
                    local_mtime = os.path.getmtime(local_path)
                    s3_last_modified = s3.head_object(Bucket=bucket_name, Key=s3_key)['LastModified'].timestamp()
                    if local_mtime > s3_last_modified:
                        print(f"文件修改时间更新，上传: {local_path}")
                        s3.upload_file(local_path, bucket_name, s3_key)
                    else:
                        print(f"文件未变更: {local_path}")

# 调用示例
if __name__ == '__main__':
    # 替换成你的本地目录、S3桶名和可选前缀
    LOCAL_FOLDER = '/your/local/folder/path'
    S3_BUCKET = 'your-bucket-name'
    S3_PATH_PREFIX = 'docs/backup'
    
    sync_local_to_s3(LOCAL_FOLDER, S3_BUCKET, S3_PATH_PREFIX)

补充说明 #

• 分块ETag处理：如果文件较大（超过8MB，boto3默认分块阈值），S3会生成带分块数的ETag，此时用修改时间作为辅助判断是简单有效的方案。若需要更精确校验，可实现分块MD5计算（按相同分块大小分割文件，计算每个块的MD5后拼接再做整体MD5，最后加上分块数，和S3的ETag对比）。
• 性能优化：同步大量文件时，可使用concurrent.futures.ThreadPoolExecutor实现并行处理，提升同步速度。
• 权限要求：确保运行代码的身份拥有S3的head_object（读取元数据）和put_object（上传文件）权限。

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