Update hf_bucket_provider.py

hf_bucket_provider.py

@@ -1,76 +1,68 @@
 import os
 import tempfile
 import time
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 from huggingface_hub import (
     batch_bucket_files,
     download_bucket_files,
     list_bucket_tree,
 )
-from wsgidav.dav_provider import DAVProvider, _DAVResource
+from wsgidav.dav_provider import DAVProvider, DAVNonCollection, DAVCollection
 from wsgidav.dav_error import DAVError, HTTP_FORBIDDEN, HTTP_NOT_FOUND
 
 
-class HfBucketResource(_DAVResource):
-    """表示 Hugging Face Bucket 中的文件或目录"""
+class HfBucketFile(DAVNonCollection):
+    """表示 Hugging Face Bucket 中的文件"""
 
-    def __init__(self, path, environ, provider):
+    def __init__(self, path, environ, provider, info):
         super().__init__(path, environ)
         self.provider = provider
-        self._info = None
-
-    def _get_info(self):
-        if self._info is None:
-            self._info = self.provider._get_file_info(self.path)
-        return self._info
-
-    @property
-    def exists(self):
-        info = self._get_info()
-        return info.get('exists', False)
+        self.info = info
 
     def get_content_length(self):
-        info = self._get_info()
-        if not info or info.get('is_dir', False):
-            return None
-        return info.get('size', 0)
+        return self.info.get('size', 0)
 
     def get_content_type(self):
-        return None
+        return None  # 由 wsgidav 根据扩展名自动判断
 
     def get_creation_date(self):
-        info = self._get_info()
-        return info.get('last_modified', 0) if info else 0
+        return self.info.get('last_modified', 0)
 
     def get_display_name(self):
         return self.name
 
     def get_etag(self):
-        info = self._get_info()
-        if info and info.get('last_modified'):
-            return f'"{info["last_modified"]}"'
-        return None
+        lm = self.info.get('last_modified', 0)
+        return f'"{lm}"' if lm else None
 
     def get_last_modified(self):
-        info = self._get_info()
-        return info.get('last_modified', 0) if info else 0
+        return self.info.get('last_modified', 0)
+
+    def get_content(self):
+        return self.provider._get_file_content(self.path)
 
-    def is_collection(self):
-        info = self._get_info()
-        return info.get('is_dir', False) if info else False
+    def begin_write(self, content_type=None):
+        return HfBucketWriteFile(self.provider, self.path)
 
-    def is_link(self):
-        return False
 
-    def support_etag(self):
-        return True
+class HfBucketFolder(DAVCollection):
+    """表示 Hugging Face Bucket 中的目录"""
+
+    def __init__(self, path, environ, provider, info):
+        super().__init__(path, environ)
+        self.provider = provider
+        self.info = info
+
+    def get_creation_date(self):
+        return self.info.get('last_modified', 0)
+
+    def get_display_name(self):
+        return self.name
 
-    def support_ranges(self):
-        return False
+    def get_last_modified(self):
+        return self.info.get('last_modified', 0)
 
     def get_member_names(self):
-        if not self.is_collection():
-            return []
         return self.provider._list_directory(self.path)
 
     def get_member(self, name):
@@ -84,54 +76,18 @@ class HfBucketResource(_DAVResource):
 
     def create_empty_resource(self, name):
         new_path = self.util.join_uri(self.path, name)
+        # 空文件将在写入时创建，这里只需返回资源对象
         return self.provider.get_resource_inst(new_path, self.environ)
 
     def delete(self):
-        if not self.exists:
-            raise DAVError(HTTP_NOT_FOUND)
-        if self.is_collection():
-            for member_name in self.get_member_names():
-                member = self.get_member(member_name)
-                member.delete()
-        else:
-            self.provider._delete_file(self.path)
-
-    def move_recursive(self, dest_path):
-        self.copy_recursive(dest_path)
-        self.delete()
-
-    def copy_recursive(self, dest_path):
-        if not self.exists:
-            raise DAVError(HTTP_NOT_FOUND)
-        if self.is_collection():
-            dest_resource = self.provider.get_resource_inst(dest_path, self.environ)
-            if not dest_resource.exists:
-                self.provider._create_collection(dest_path)
-                dest_resource = self.provider.get_resource_inst(dest_path, self.environ)
-            for member_name in self.get_member_names():
-                member = self.get_member(member_name)
-                dest_member_path = self.util.join_uri(dest_path, member_name)
-                member.copy_recursive(dest_member_path)
-        else:
-            self.provider._copy_file(self.path, dest_path)
-
-    def get_content(self):
-        if not self.exists:
-            raise DAVError(HTTP_NOT_FOUND)
-        if self.is_collection():
-            raise DAVError(HTTP_FORBIDDEN)
-        return self.provider._get_file_content(self.path)
-
-    def begin_write(self, content_type=None):
-        if not self.exists:
-            # 新文件允许写入
-            pass
-        if self.is_collection():
-            raise DAVError(HTTP_FORBIDDEN)
-        return HfBucketWriteFile(self.provider, self.path)
+        # 递归删除所有子项
+        for member_name in self.get_member_names():
+            member = self.get_member(member_name)
+            member.delete()
 
 
 class HfBucketWriteFile:
+    """辅助类：处理文件写入"""
     def __init__(self, provider, path):
         self.provider = provider
         self.path = path
@@ -147,6 +103,8 @@ class HfBucketWriteFile:
 
 
 class HfBucketProvider(DAVProvider):
+    """将 Hugging Face Bucket 适配为 DAVProvider"""
+
     def __init__(self, bucket_id: str, token: str):
         super().__init__()
         self.bucket_id = bucket_id
@@ -159,8 +117,8 @@ class HfBucketProvider(DAVProvider):
             path = path[:-1]
         return path
 
-    def _get_file_info(self, path: str) -> Dict[str, Any]:
-        """返回字典，包含 exists, is_dir, size, last_modified"""
+    def _get_file_info(self, path: str) -> Optional[Dict[str, Any]]:
+        """获取文件/目录信息，如果不存在则返回 None"""
         path = self._normalize_path(path)
         if path == '/':
             return {'exists': True, 'is_dir': True, 'size': 0, 'last_modified': time.time()}
@@ -190,9 +148,9 @@ class HfBucketProvider(DAVProvider):
                         }
             if items:
                 return {'exists': True, 'is_dir': True, 'size': 0, 'last_modified': time.time()}
-            return {'exists': False, 'is_dir': False, 'size': 0, 'last_modified': 0}
+            return None
         except Exception:
-            return {'exists': False, 'is_dir': False, 'size': 0, 'last_modified': 0}
+            return None
 
     def _list_directory(self, path: str) -> List[str]:
         path = self._normalize_path(path)
@@ -279,8 +237,16 @@ class HfBucketProvider(DAVProvider):
                     os.unlink(tmp.name)
 
     def get_resource_inst(self, path, environ):
-        # 始终返回资源对象（即使不存在）
-        return HfBucketResource(path, environ, self)
+        """返回资源实例（文件或目录）"""
+        info = self._get_file_info(path)
+        if info is None:
+            # 路径不存在，返回一个占位的文件资源（允许后续写入）
+            # 使用默认信息
+            info = {'exists': False, 'is_dir': False, 'size': 0, 'last_modified': 0}
+        if info['is_dir']:
+            return HfBucketFolder(path, environ, self, info)
+        else:
+            return HfBucketFile(path, environ, self, info)
 
     def exists(self, path, environ):
-        return self.get_resource_inst(path, environ).exists
+        return self._get_file_info(path) is not None