File size: 4,811 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """Module for folder binding base."""
from __future__ import annotations
import logging
from typing import ClassVar
from apps.core.utils.path import Path
from .browse_policy import FolderBrowsePolicy
from .filesystem_service import FolderFilesystemService
from .path_validator import FolderPathValidator
logger = logging.getLogger("apps.core.filesystem")
class BaseFolderBindingService:
DEFAULT_SUBDIRS: ClassVar[dict[str, str]] = {}
def __init__(
self,
filesystem_service: FolderFilesystemService | None = None,
path_validator: FolderPathValidator | None = None,
browse_policy: FolderBrowsePolicy | None = None,
roots_setting_name: str = "FOLDER_BROWSE_ROOTS",
fallback_roots_setting_name: str = "CONTRACT_FOLDER_BROWSE_ROOTS",
) -> None:
self._filesystem_service = filesystem_service
self._path_validator = path_validator
self._browse_policy = browse_policy
self._roots_setting_name = roots_setting_name
self._fallback_roots_setting_name = fallback_roots_setting_name
@property
def path_validator(self) -> FolderPathValidator:
if self._path_validator is None:
self._path_validator = FolderPathValidator()
return self._path_validator
@property
def filesystem_service(self) -> FolderFilesystemService:
if self._filesystem_service is None:
self._filesystem_service = FolderFilesystemService(validator=self.path_validator)
return self._filesystem_service
@property
def browse_policy(self) -> FolderBrowsePolicy:
if self._browse_policy is None:
self._browse_policy = FolderBrowsePolicy(
validator=self.path_validator,
roots_setting_name=self._roots_setting_name,
fallback_roots_setting_name=self._fallback_roots_setting_name,
)
return self._browse_policy
def validate_folder_path(self, path: str) -> tuple[bool, str | None]:
return self.path_validator.validate_folder_path(path)
def _is_network_path(self, path: str) -> bool:
return self.path_validator.is_network_path(path)
def get_browse_roots(self) -> list[Path]:
return self.browse_policy.get_browse_roots()
def get_default_browse_path(self) -> Path | None:
"""获取默认浏览路径（用户下载目录）"""
return self.browse_policy._get_user_downloads_path()
def resolve_under_allowed_roots(self, path: str) -> Path:
return self.browse_policy.resolve_under_allowed_roots(path)
def list_subdirs(self, path: str, include_hidden: bool = False) -> list[dict[str, str]]:
return self.browse_policy.list_subdirs(path, include_hidden=include_hidden)
def check_folder_accessible(self, path: str) -> bool:
try:
folder = Path(path)
if hasattr(folder, "exists") and folder.exists():
if hasattr(folder, "is_dir"):
return bool(folder.is_dir())
if hasattr(folder, "isdir"):
return folder.isdir()
return False
except (OSError, PermissionError):
return False
def ensure_subdirectories(self, base_path: str) -> bool:
ok = self.filesystem_service.ensure_subdirectories(base_path, self.DEFAULT_SUBDIRS.values())
if not ok:
logger.error("创建子目录失败", extra={"base_path": base_path})
return ok
def format_path_for_display(self, path: str, max_length: int = 50) -> str:
if not path:
return ""
if len(path) <= max_length:
return path
start_len = max_length // 3
end_len = max_length - start_len - 3
return f"{path[:start_len]}...{path[-end_len:]}"
def compute_parent_path(self, resolved: Path) -> str | None:
"""计算已解析路径的父路径(如果父路径在允许的根目录范围内).
Args:
resolved: 已解析的路径
Returns:
父路径字符串,如果父路径不在允许范围内则返回 None
"""
parent: Path = resolved.parent
for root in self.get_browse_roots():
try:
parent.relative_to(root)
return str(parent)
except ValueError:
continue
return None
def is_browsable_path(self, path: str) -> tuple[bool, str | None]:
"""检查路径是否可浏览.
Args:
path: 要检查的路径
Returns:
(is_browsable, message) 元组.如果不可浏览,message 包含原因.
"""
if self._is_network_path(path):
return False, "网络路径不支持浏览,但可以直接绑定"
return True, None
|