forked from erp-dev/erp
129 lines
4.4 KiB
Python
129 lines
4.4 KiB
Python
"""
|
||
七牛云 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()
|