""" 七牛云 SDK 兼容性补丁 修复 qiniu SDK 中 FormUploader 对空数据的判断问题 """ def patch_qiniu_form_uploader(): """ 修复 qiniu.services.storage.uploaders.form_uploader.FormUploader.upload 方法 问题:原方法使用 `if not data` 来判断是否提供了 data 参数, 但空字节串 b'' 在布尔上下文中为 False,导致误判。 解决:Monkey patch 修改方法内部的检查逻辑。 """ try: import types from qiniu.services.storage.uploaders import form_uploader from qiniu.services.storage.uploaders.form_uploader import FormUploader # 获取原始方法的代码 original_upload = FormUploader.upload original_code = original_upload.__code__ # 创建一个包装器来修复参数检查 def upload_wrapper( self, key, file_path=None, data=None, data_size=None, modify_time=None, part_size=None, mime_type=None, metadata=None, file_name=None, custom_vars=None, **kwargs ): """包装原始 upload 方法,修复空数据检查问题""" from io import BytesIO from os import path from time import time from qiniu.compat import is_seekable from qiniu.utils import b, io_crc32 from qiniu.auth import Auth # 复制原始方法的逻辑,但修复检查 bucket_name = kwargs.get('bucket_name', self.bucket_name) up_token = kwargs.get('up_token', None) if not up_token: up_token = self.get_up_token(**kwargs) access_key = self.auth.get_access_key() else: access_key, _, _ = Auth.up_token_decode(up_token) crc32_int = kwargs.get('crc32_int', None) if not file_name and file_path: file_name = path.basename(file_path) # *** 修复:使用 is None 进行检查 *** if file_path is None and data is None: raise TypeError('Must provide one of file_path or data.') if file_path is not None and data is not None: raise TypeError('Must provide only one of file_path or data.') if not modify_time: if file_path: modify_time = int(path.getmtime(file_path)) else: modify_time = int(time()) # 继续执行上传逻辑 try: if file_path: data_size = path.getsize(file_path) data = open(file_path, 'rb') elif isinstance(data, bytes): data_size = len(data) data = BytesIO(data) elif isinstance(data, str): data_size = len(data) data = BytesIO(b(data)) if not crc32_int: crc32_int = self._FormUploader__get_crc32_int(data) fields = self._FormUploader__get_form_fields( up_token=up_token, key=key, crc32_int=crc32_int, custom_vars=custom_vars, metadata=metadata ) ret, resp = self._FormUploader__upload_data_with_retrier( access_key=access_key, bucket_name=bucket_name, fields=fields, file_name=file_name, data=data, data_size=data_size, mime_type=mime_type ) finally: if file_path: data.close() return ret, resp # 替换方法 FormUploader.upload = upload_wrapper print("✓ 七牛云 FormUploader 补丁已应用") return True except Exception as e: print(f"⚠ 应用七牛云补丁时出错: {e}") import traceback traceback.print_exc() return False def apply_all_patches(): """应用所有第三方库补丁""" patch_qiniu_form_uploader()