feat: add metadata options for file status checks and optimize permission filtering logic

domain/permission/service.py

@@ -1,3 +1,4 @@
+from dataclasses import dataclass
 from typing import List, Optional
 from fastapi import HTTPException
 
@@ -17,74 +18,169 @@ from .types import (
     PERMISSION_DEFINITIONS,
 )
 
+@dataclass(slots=True)
+class PermissionContext:
+    exists: bool
+    is_admin: bool
+    path_rules: List[PathRule]
+
 
 class PermissionService:
     """权限检查服务"""
 
     # 权限检查结果缓存（简单的内存缓存）
     _cache: dict[str, tuple[bool, float]] = {}
+    _context_cache: dict[int, tuple[PermissionContext, float]] = {}
     _cache_ttl = 300  # 5分钟缓存
 
+    @classmethod
+    def _now(cls) -> float:
+        import time
+
+        return time.time()
+
+    @classmethod
+    def _is_cache_valid(cls, timestamp: float) -> bool:
+        return cls._now() - timestamp < cls._cache_ttl
+
+    @classmethod
+    def _get_cached_result(cls, cache_key: str) -> Optional[bool]:
+        cached = cls._cache.get(cache_key)
+        if not cached:
+            return None
+        result, timestamp = cached
+        if cls._is_cache_valid(timestamp):
+            return result
+        cls._cache.pop(cache_key, None)
+        return None
+
+    @classmethod
+    def _sort_path_rules(cls, rules: List[PathRule]) -> List[PathRule]:
+        return sorted(
+            rules,
+            key=lambda r: (
+                r.priority,
+                PathMatcher.get_pattern_specificity(r.path_pattern, r.is_regex),
+            ),
+            reverse=True,
+        )
+
+    @classmethod
+    def _match_sorted_path_rules(
+        cls, path: str, action: str, sorted_rules: List[PathRule]
+    ) -> Optional[bool]:
+        for rule in sorted_rules:
+            if PathMatcher.match_pattern(path, rule.path_pattern, rule.is_regex):
+                if action == PathAction.READ:
+                    return rule.can_read
+                if action == PathAction.WRITE:
+                    return rule.can_write
+                if action == PathAction.DELETE:
+                    return rule.can_delete
+                if action == PathAction.SHARE:
+                    return rule.can_share
+                return False
+        return None
+
+    @classmethod
+    async def _get_permission_context(cls, user_id: int) -> PermissionContext:
+        cached = cls._context_cache.get(user_id)
+        if cached:
+            context, timestamp = cached
+            if cls._is_cache_valid(timestamp):
+                return context
+            cls._context_cache.pop(user_id, None)
+
+        user = await UserAccount.get_or_none(id=user_id)
+        if not user:
+            context = PermissionContext(exists=False, is_admin=False, path_rules=[])
+            cls._context_cache[user_id] = (context, cls._now())
+            return context
+
+        if user.is_admin:
+            context = PermissionContext(exists=True, is_admin=True, path_rules=[])
+            cls._context_cache[user_id] = (context, cls._now())
+            return context
+
+        user_roles = await UserRole.filter(user_id=user_id)
+        role_ids = [ur.role_id for ur in user_roles]
+        if not role_ids:
+            context = PermissionContext(exists=True, is_admin=False, path_rules=[])
+            cls._context_cache[user_id] = (context, cls._now())
+            return context
+
+        path_rules = await PathRule.filter(role_id__in=role_ids)
+        context = PermissionContext(
+            exists=True,
+            is_admin=False,
+            path_rules=cls._sort_path_rules(list(path_rules)),
+        )
+        cls._context_cache[user_id] = (context, cls._now())
+        return context
+
+    @classmethod
+    def _check_path_permission_with_context(
+        cls,
+        user_id: int,
+        normalized_path: str,
+        action: str,
+        context: PermissionContext,
+    ) -> bool:
+        if not context.exists:
+            return False
+        if context.is_admin:
+            return True
+
+        checked_cache_keys: List[str] = []
+        current_path = normalized_path
+
+        while True:
+            cache_key = f"{user_id}:{current_path}:{action}"
+            cached_result = cls._get_cached_result(cache_key)
+            if cached_result is not None:
+                result = cached_result
+                break
+
+            checked_cache_keys.append(cache_key)
+            result = cls._match_sorted_path_rules(current_path, action, context.path_rules)
+            if result is not None:
+                break
+
+            parent_path = PathMatcher.get_parent_path(current_path)
+            if not parent_path:
+                result = False
+                break
+            current_path = parent_path
+
+        timestamp = cls._now()
+        for cache_key in checked_cache_keys:
+            cls._cache[cache_key] = (result, timestamp)
+        return result
+
     @classmethod
     async def check_path_permission(
         cls, user_id: int, path: str, action: str
     ) -> bool:
         """
         检查用户对路径的操作权限
-        
+
         Args:
             user_id: 用户ID
             path: 要检查的路径
             action: 操作类型 (read/write/delete/share)
-            
+
         Returns:
             是否有权限
         """
-        import time
-
-        # 检查缓存
-        cache_key = f"{user_id}:{path}:{action}"
-        if cache_key in cls._cache:
-            result, timestamp = cls._cache[cache_key]
-            if time.time() - timestamp < cls._cache_ttl:
-                return result
-
-        # 获取用户
-        user = await UserAccount.get_or_none(id=user_id)
-        if not user:
-            return False
-
-        # 超级管理员直接放行
-        if user.is_admin:
-            cls._cache[cache_key] = (True, time.time())
-            return True
-
-        # 获取用户所有角色
-        user_roles = await UserRole.filter(user_id=user_id).prefetch_related("role")
-        role_ids = [ur.role_id for ur in user_roles]
-
-        if not role_ids:
-            cls._cache[cache_key] = (False, time.time())
-            return False
-
-        # 获取所有角色的路径规则
-        path_rules = await PathRule.filter(role_id__in=role_ids).order_by("-priority")
-
-        # 规范化路径
         normalized_path = PathMatcher.normalize_path(path)
+        cache_key = f"{user_id}:{normalized_path}:{action}"
+        cached_result = cls._get_cached_result(cache_key)
+        if cached_result is not None:
+            return cached_result
 
-        # 按优先级和具体程度匹配
-        result = cls._match_path_rules(normalized_path, action, list(path_rules))
-
-        # 如果没有匹配到规则，检查父目录（继承）
-        if result is None:
-            parent_path = PathMatcher.get_parent_path(normalized_path)
-            if parent_path:
-                result = await cls.check_path_permission(user_id, parent_path, action)
-            else:
-                result = False  # 默认拒绝
-
-        cls._cache[cache_key] = (result, time.time())
+        context = await cls._get_permission_context(user_id)
+        result = cls._check_path_permission_with_context(user_id, normalized_path, action, context)
+        cls._cache[cache_key] = (result, cls._now())
         return result
 
     @classmethod
@@ -97,31 +193,7 @@ class PermissionService:
         Returns:
             True/False 表示明确的权限结果，None 表示没有匹配到规则
         """
-        # 按优先级和具体程度排序
-        sorted_rules = sorted(
-            rules,
-            key=lambda r: (
-                r.priority,
-                PathMatcher.get_pattern_specificity(r.path_pattern, r.is_regex),
-            ),
-            reverse=True,
-        )
-
-        for rule in sorted_rules:
-            if PathMatcher.match_pattern(path, rule.path_pattern, rule.is_regex):
-                # 匹配到规则，检查具体操作权限
-                if action == PathAction.READ:
-                    return rule.can_read
-                elif action == PathAction.WRITE:
-                    return rule.can_write
-                elif action == PathAction.DELETE:
-                    return rule.can_delete
-                elif action == PathAction.SHARE:
-                    return rule.can_share
-                else:
-                    return False
-
-        return None
+        return cls._match_sorted_path_rules(path, action, cls._sort_path_rules(rules))
 
     @classmethod
     async def check_system_permission(cls, user_id: int, permission_code: str) -> bool:
@@ -251,35 +323,20 @@ class PermissionService:
         cls, user_id: int, path: str, action: str
     ) -> PathPermissionResult:
         """检查路径权限并返回详细结果"""
-        user = await UserAccount.get_or_none(id=user_id)
-        if not user:
+        context = await cls._get_permission_context(user_id)
+        if not context.exists:
             return PathPermissionResult(path=path, action=action, allowed=False)
 
-        # 超级管理员
-        if user.is_admin:
+        if context.is_admin:
             return PathPermissionResult(path=path, action=action, allowed=True)
 
-        # 获取用户角色
-        user_roles = await UserRole.filter(user_id=user_id)
-        role_ids = [ur.role_id for ur in user_roles]
-
-        if not role_ids:
+        if not context.path_rules:
             return PathPermissionResult(path=path, action=action, allowed=False)
 
-        # 获取路径规则
-        path_rules = await PathRule.filter(role_id__in=role_ids).order_by("-priority")
         normalized_path = PathMatcher.normalize_path(path)
 
-        # 查找匹配的规则
         matched_rule = None
-        for rule in sorted(
-            path_rules,
-            key=lambda r: (
-                r.priority,
-                PathMatcher.get_pattern_specificity(r.path_pattern, r.is_regex),
-            ),
-            reverse=True,
-        ):
+        for rule in context.path_rules:
             if PathMatcher.match_pattern(
                 normalized_path, rule.path_pattern, rule.is_regex
             ):
@@ -322,19 +379,30 @@ class PermissionService:
         """清除权限缓存"""
         if user_id is None:
             cls._cache.clear()
+            cls._context_cache.clear()
         else:
-            # 清除特定用户的缓存
             keys_to_delete = [k for k in cls._cache if k.startswith(f"{user_id}:")]
             for k in keys_to_delete:
                 del cls._cache[k]
+            cls._context_cache.pop(user_id, None)
 
     @classmethod
     async def filter_paths_by_permission(
         cls, user_id: int, paths: List[str], action: str
     ) -> List[str]:
         """过滤出用户有权限的路径列表"""
+        if not paths:
+            return []
+
+        context = await cls._get_permission_context(user_id)
+        if not context.exists:
+            return []
+        if context.is_admin:
+            return list(paths)
+
         result = []
         for path in paths:
-            if await cls.check_path_permission(user_id, path, action):
+            normalized_path = PathMatcher.normalize_path(path)
+            if cls._check_path_permission_with_context(user_id, normalized_path, action, context):
                 result.append(path)
         return result