如何使用同一套代码实现带MD5校验的大文件分片上传至AWS S3与阿里云OSS？

嘿，刚好搞定了类似的需求！我整理了一套能同时给AWS S3和阿里云OSS传大文件的分片上传代码，还自带MD5校验（其实用的是两家都兼容的ETAG，和MD5校验效果一致），直接看代码和解析吧！

#!/usr/bin/env python3.10

"""
这个小工具用于实现带校验的大文件分片上传，同时支持AWS S3和阿里云OSS
是基于相关技术讨论改造的版本
"""

import os
import sys
import hashlib
import oss2

import boto3
from botocore.exceptions import ClientError
from botocore.client import Config
from boto3.s3.transfer import TransferConfig

# 分片大小设置为8MB，S3和OSS都支持这个范围
chunk_size = 2**23  # 8 * 1024 * 1024


def calculate_etag(file_path, chunk_size=chunk_size):
    """计算文件的ETAG，兼容S3和OSS的分片ETAG规则"""
    md5s = []
    with open(file_path, 'rb') as f:
        while True:
            data = f.read(chunk_size)
            if not data:
                break
            md5s.append(hashlib.md5(data))
    
    if len(md5s) == 1:
        # 单文件上传，ETAG就是文件的MD5值
        return f'"{md5s[0].hexdigest()}"'
    else:
        # 分片上传，ETAG是所有分片MD5拼接后的MD5，加上分片数
        digests = b''.join(m.digest() for m in md5s)
        digests_md5 = hashlib.md5(digests)
        return f'"{digests_md5.hexdigest()}-{len(md5s)}"'


def upload_to_s3(file_path, bucket_name, object_key, aws_access_key, aws_secret_key, region='us-east-1'):
    """上传文件到AWS S3，并校验ETAG"""
    # 初始化S3客户端
    s3_client = boto3.client(
        's3',
        aws_access_key_id=aws_access_key,
        aws_secret_access_key=aws_secret_key,
        region_name=region,
        config=Config(signature_version='s3v4')
    )
    
    # 配置分片上传参数
    transfer_config = TransferConfig(
        multipart_threshold=chunk_size,
        multipart_chunksize=chunk_size
    )
    
    try:
        # 执行上传
        s3_client.upload_file(
            file_path,
            bucket_name,
            object_key,
            Config=transfer_config
        )
        
        # 获取云端的ETAG
        response = s3_client.head_object(Bucket=bucket_name, Key=object_key)
        cloud_etag = response['ETag']
        
        # 计算本地文件的ETAG并对比
        local_etag = calculate_etag(file_path)
        if cloud_etag == local_etag:
            print(f"S3上传成功，ETAG校验通过：{local_etag}")
            return True
        else:
            print(f"S3上传失败，ETAG不匹配！云端：{cloud_etag}，本地：{local_etag}")
            # 可以选择删除上传失败的文件
            s3_client.delete_object(Bucket=bucket_name, Key=object_key)
            return False
    except ClientError as e:
        print(f"S3上传出错：{e}")
        return False


def upload_to_oss(file_path, bucket_name, object_key, oss_access_key, oss_secret_key, endpoint='oss-cn-hangzhou.aliyuncs.com'):
    """上传文件到阿里云OSS，并校验ETAG"""
    # 初始化OSS客户端
    auth = oss2.Auth(oss_access_key, oss_secret_key)
    bucket = oss2.Bucket(auth, endpoint, bucket_name)
    
    try:
        # 执行分片上传
        result = bucket.put_object_from_file(object_key, file_path, multipart_threshold=chunk_size, part_size=chunk_size)
        
        # 获取云端的ETAG
        cloud_etag = result.etag
        
        # 计算本地文件的ETAG并对比
        local_etag = calculate_etag(file_path)
        # OSS返回的ETAG不带引号，需要处理一下
        if f'"{cloud_etag}"' == local_etag:
            print(f"OSS上传成功，ETAG校验通过：{local_etag}")
            return True
        else:
            print(f"OSS上传失败，ETAG不匹配！云端：{cloud_etag}，本地：{local_etag}")
            # 可以选择删除上传失败的文件
            bucket.delete_object(object_key)
            return False
    except oss2.exceptions.OssError as e:
        print(f"OSS上传出错：{e}")
        return False


if __name__ == "__main__":
    # 示例用法，实际使用时替换为你的配置
    FILE_PATH = "large_file.bin"
    S3_BUCKET = "your-s3-bucket"
    OSS_BUCKET = "your-oss-bucket"
    OBJECT_KEY = "large_file.bin"
    
    # AWS配置
    AWS_ACCESS_KEY = "your-aws-access-key"
    AWS_SECRET_KEY = "your-aws-secret-key"
    
    # 阿里云OSS配置
    OSS_ACCESS_KEY = "your-oss-access-key"
    OSS_SECRET_KEY = "your-oss-secret-key"
    OSS_ENDPOINT = "oss-cn-hangzhou.aliyuncs.com"
    
    # 上传到S3
    upload_to_s3(FILE_PATH, S3_BUCKET, OBJECT_KEY, AWS_ACCESS_KEY, AWS_SECRET_KEY)
    
    # 上传到OSS
    upload_to_oss(FILE_PATH, OSS_BUCKET, OBJECT_KEY, OSS_ACCESS_KEY, OSS_SECRET_KEY)

关键要点解析 #

• 统一ETAG校验逻辑：我把原来的S3专属ETAG计算函数改成了通用版，因为S3和OSS的分片ETAG生成规则完全一致——单文件上传时ETAG就是文件MD5；分片上传时，是所有分片MD5值拼接后的MD5，再加上分片数。这样一个函数就能同时适配两家服务商。
• 分片大小选择：设置为8MB（2**23），这个大小在S3和OSS的分片大小允许范围内（S3要求分片最小5MB，OSS最小100KB），平衡了上传效率和请求次数。
• 客户端分片配置：
  • S3用boto3的TransferConfig指定分片阈值和大小，自动触发分片上传；
  • OSS用oss2的put_object_from_file，通过multipart_threshold和part_size参数开启分片上传，用法和S3很接近。
• 校验与容错：上传完成后，对比本地计算的ETAG和云端返回的ETAG，不一致则删除云端文件，避免无效存储，同时输出明确的错误信息方便排查。

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