feat: stock module

flower/qiniu_patch.py

@@ -0,0 +1,128 @@
+"""
+七牛云 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()