python oss上传（纯代码无贴图）
"""
阿里云OSS存储工具类
用于文件上传、下载和生成签名URL等操作
"""
import oss2
import base64
from urllib.parse import unquote_plus, urlparse
class AliyunOSSBucket(object):
"""
阿里云OSS存储操作类
功能：
1. 创建OSS连接
2. 上传文件到OSS
3. 生成文件下载链接（支持代理和签名）
返回格式：
所有方法统一返回字典格式：
{
'code': 0, # 0-成功，1-失败，2-参数错误等
'message': '', # 提示信息
'data': {} # 返回的数据（可选）
}
使用示例：
oss = AliyunOSSBucket()
# 上传文件
result = oss.put_file_to_oss("test.jpg", base64_string)
if result['code'] == 0:
print("上传成功")
else:
print(f"上传失败: {result['message']}")
# 获取下载链接
result = oss.get_download_url("test.jpg")
if result['code'] == 0:
url = result['data']['url']
print(f"下载链接: {url}")
"""
# 定义响应码常量
CODE_SUCCESS = 0 # 成功
CODE_ERROR = 1 # 一般错误
CODE_PARAM_ERROR = 2 # 参数错误
CODE_CONFIG_ERROR = 3 # 配置错误
CODE_CONNECTION_ERROR = 4 # 连接错误
def __init__(self):
"""
初始化OSS配置参数
注意：实际使用时应从配置文件或环境变量中读取，避免硬编码
"""
# OSS服务配置
self.aliyun_oss_endpoint = "xxxx" # OSS访问域名，如：oss-cn-xxx.aliyuncs.com
self.aliyun_oss_bucket_name = "xxxx" # 存储空间名称
self.aliyun_oss_access_key_id = "xxxx" # 访问密钥ID
self.aliyun_oss_access_key_secret = "xxxx" # 访问密钥Secret
# 下载代理配置（可选）
# 如果设置了代理地址，下载链接会通过代理转发
self.aliyun_oss_download = "xxx" # 代理服务器地址，如：https://download.example.com
def _response(self, code, message, data=None):
"""
构建统一响应格式
Args:
code (int): 响应码，0表示成功，非0表示失败
message (str): 响应消息
data (any, optional): 返回的数据，默认为None
Returns:
dict: 统一格式的响应字典
"""
response = {
'code': code,
'message': message
}
if data is not None:
response['data'] = data
return response
def get_aliyun_oss_bucket(self):
"""
创建并返回阿里云OSS Bucket对象
Returns:
dict: 统一格式的响应
- 成功时: {'code': 0, 'message': 'success', 'data': bucket对象}
- 失败时: {'code': 非0, 'message': 错误信息}
"""
try:
# 获取配置参数
endpoint = self.aliyun_oss_endpoint
bucket_name = self.aliyun_oss_bucket_name
access_key_id = self.aliyun_oss_access_key_id
access_key_secret = self.aliyun_oss_access_key_secret
# 验证参数完整性
if not all([endpoint, bucket_name, access_key_id, access_key_secret]):
return self._response(
self.CODE_CONFIG_ERROR,
"OSS配置参数不完整，请检查endpoint、bucket_name、access_key_id、access_key_secret"
)
# 确保Endpoint包含协议头
if not endpoint.startswith(('http://', 'https://')):
endpoint = f'https://{endpoint}'
# 创建认证对象
auth = oss2.Auth(access_key_id, access_key_secret)
# 创建Bucket对象
bucket = oss2.Bucket(auth, endpoint, bucket_name)
# 验证连接是否正常
try:
bucket.list_objects(max_keys=1)
except Exception as e:
return self._response(
self.CODE_CONNECTION_ERROR,
f"阿里云OSS连接验证失败: {str(e)}"
)
return self._response(
self.CODE_SUCCESS,
"OSS连接成功",
bucket
)
except Exception as e:
return self._response(
self.CODE_ERROR,
f"创建OSS Bucket对象失败: {str(e)}"
)
def put_file_to_oss(self, unique_file_name, file_content_base64):
"""
上传文件到阿里云OSS
Args:
unique_file_name (str): 文件在OSS中存储的唯一名称（支持路径，如：images/photo.jpg）
file_content_base64 (str): Base64编码的文件内容
Returns:
dict: 统一格式的响应
- 成功时: {'code': 0, 'message': '文件上传成功', 'data': {'file_name': 'xxx'}}
- 失败时: {'code': 非0, 'message': 错误信息}
"""
# 参数验证
if not unique_file_name:
return self._response(
self.CODE_PARAM_ERROR,
"文件名不能为空"
)
if not file_content_base64:
return self._response(
self.CODE_PARAM_ERROR,
"文件内容不能为空"
)
try:
# 获取Bucket对象
bucket_result = self.get_aliyun_oss_bucket()
# 检查是否获取成功
if bucket_result['code'] != self.CODE_SUCCESS:
return self._response(
bucket_result['code'],
f"获取OSS连接失败: {bucket_result['message']}"
)
bucket = bucket_result['data']
# 解码Base64内容
try:
file_content = base64.b64decode(file_content_base64)
except base64.binascii.Error as e:
return self._response(
self.CODE_PARAM_ERROR,
f"Base64解码失败: {str(e)}"
)
# 上传文件到OSS
bucket.put_object(unique_file_name, file_content)
return self._response(
self.CODE_SUCCESS,
"文件上传成功",
{'file_name': unique_file_name}
)
except Exception as e:
return self._response(
self.CODE_ERROR,
f"文件上传失败: {str(e)}"
)
def get_download_url(self, unique_file_name, expire_time=60 * 60 * 24 * 365, use_agent=True):
"""
获取文件的下载链接（支持签名和代理转发）
Args:
unique_file_name (str): OSS中的文件名（OSS key）
expire_time (int): 链接过期时间（秒），默认365天（31536000秒）
use_agent (bool): 是否使用下载代理，默认True
- True: 如果配置了代理地址，使用代理链接；否则使用OSS原链接
- False: 直接使用OSS原链接
Returns:
dict: 统一格式的响应
- 成功时: {'code': 0, 'message': 'success', 'data': {'url': 'xxx', 'expire_time': 3600}}
- 失败时: {'code': 非0, 'message': 错误信息}
示例：
# 获取1小时有效的链接，使用代理
result = oss.get_download_url("test.jpg", expire_time=3600)
if result['code'] == 0:
url = result['data']['url']
print(f"下载链接: {url}")
"""
# 参数验证
if not unique_file_name:
return self._response(
self.CODE_PARAM_ERROR,
"文件名不能为空"
)
if expire_time <= 0:
return self._response(
self.CODE_PARAM_ERROR,
"过期时间必须大于0"
)
try:
# 获取Bucket对象
bucket_result = self.get_aliyun_oss_bucket()
# 检查是否获取成功
if bucket_result['code'] != self.CODE_SUCCESS:
return self._response(
bucket_result['code'],
f"获取OSS连接失败: {bucket_result['message']}"
)
bucket = bucket_result['data']
# 生成签名URL
url = bucket.sign_url('GET', unique_file_name, expire_time)
# 获取代理配置
download_url = self.aliyun_oss_download
# 如果启用代理且配置了代理地址，则替换为代理链接
if download_url and use_agent:
# 解析原始OSS URL
parsed_url = urlparse(url)
# 提取路径和查询参数
path_and_query = parsed_url.path
if parsed_url.query:
path_and_query += '?' + parsed_url.query
# 拼接代理URL
url = download_url.rstrip('/') + path_and_query
# URL编码处理,防止通过接口拿到的下载链接出现乱码不能下载的情况
url = unquote_plus(url)
url = url.replace('+', '%2B')
return self._response(
self.CODE_SUCCESS,
"生成下载链接成功",
{
'url': url,
'expire_time': expire_time,
'use_agent': use_agent
}
)
except Exception as e:
return self._response(
self.CODE_ERROR,
f"生成下载链接失败: {str(e)}"
)
def delete_file(self, unique_file_name):
"""
删除OSS中的文件
Args:
unique_file_name (str): OSS中的文件名
Returns:
dict: 统一格式的响应
"""
if not unique_file_name:
return self._response(
self.CODE_PARAM_ERROR,
"文件名不能为空"
)
try:
# 获取Bucket对象
bucket_result = self.get_aliyun_oss_bucket()
if bucket_result['code'] != self.CODE_SUCCESS:
return self._response(
bucket_result['code'],
f"获取OSS连接失败: {bucket_result['message']}"
)
bucket = bucket_result['data']
# 删除文件
bucket.delete_object(unique_file_name)
return self._response(
self.CODE_SUCCESS,
"文件删除成功",
{'file_name': unique_file_name}
)
except Exception as e:
return self._response(
self.CODE_ERROR,
f"文件删除失败: {str(e)}"
)
# 使用示例
if __name__ == '__main__':
# 初始化OSS工具类
oss_tool = AliyunOSSBucket()
# 示例1：上传文件
# with open('test.jpg', 'rb') as f:
# base64_content = base64.b64encode(f.read()).decode('utf-8')
# result = oss_tool.put_file_to_oss('test.jpg', base64_content)
# if result['code'] == 0:
# print(f"上传成功: {result['data']}")
# else:
# print(f"上传失败: {result['message']}")
# 示例2：获取下载链接
# result = oss_tool.get_download_url('test.jpg', expire_time=3600)
# if result['code'] == 0:
# print(f"下载链接: {result['data']['url']}")
# print(f"过期时间: {result['data']['expire_time']}秒")
# else:
# print(f"获取链接失败: {result['message']}")
# 示例3：删除文件
# result = oss_tool.delete_file('test.jpg')
# if result['code'] == 0:
# print("删除成功")
# else:
# print(f"删除失败: {result['message']}")