python oss上传（纯代码无贴图）

阿里云OSS存储工具类用于文件上传、下载和生成签名URL等操作import oss2import base64from urllib.parse import unquote_plus, urlparseclass AliyunOSSBucket(object):阿里云OSS存储操作类功能1. 创建OSS连接2. 上传文件到OSS3. 生成文件下载链接支持代理和签名返回格式所有方法统一返回字典格式{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.comself.aliyun_oss_bucket_name xxxx # 存储空间名称self.aliyun_oss_access_key_id xxxx # 访问密钥IDself.aliyun_oss_access_key_secret xxxx # 访问密钥Secret# 下载代理配置可选# 如果设置了代理地址下载链接会通过代理转发self.aliyun_oss_download xxx # 代理服务器地址如https://download.example.comdef _response(self, code, message, dataNone):构建统一响应格式Args:code (int): 响应码0表示成功非0表示失败message (str): 响应消息data (any, optional): 返回的数据默认为NoneReturns:dict: 统一格式的响应字典response {code: code,message: message}if data is not None:response[data] datareturn responsedef get_aliyun_oss_bucket(self):创建并返回阿里云OSS Bucket对象Returns:dict: 统一格式的响应- 成功时: {code: 0, message: success, data: bucket对象}- 失败时: {code: 非0, message: 错误信息}try:# 获取配置参数endpoint self.aliyun_oss_endpointbucket_name self.aliyun_oss_bucket_nameaccess_key_id self.aliyun_oss_access_key_idaccess_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 fhttps://{endpoint}# 创建认证对象auth oss2.Auth(access_key_id, access_key_secret)# 创建Bucket对象bucket oss2.Bucket(auth, endpoint, bucket_name)# 验证连接是否正常try:bucket.list_objects(max_keys1)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):上传文件到阿里云OSSArgs:unique_file_name (str): 文件在OSS中存储的唯一名称支持路径如images/photo.jpgfile_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,fBase64解码失败: {str(e)})# 上传文件到OSSbucket.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_time60 * 60 * 24 * 365, use_agentTrue):获取文件的下载链接支持签名和代理转发Args:unique_file_name (str): OSS中的文件名OSS keyexpire_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_time3600)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]# 生成签名URLurl bucket.sign_url(GET, unique_file_name, expire_time)# 获取代理配置download_url self.aliyun_oss_download# 如果启用代理且配置了代理地址则替换为代理链接if download_url and use_agent:# 解析原始OSS URLparsed_url urlparse(url)# 提取路径和查询参数path_and_query parsed_url.pathif parsed_url.query:path_and_query ? parsed_url.query# 拼接代理URLurl 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_time3600)# 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]})