| """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 |
|
|
