"""DMS存储目录创建和安全路径解析。""" from __future__ import annotations from pathlib import Path, PurePosixPath STORAGE_SUBDIRECTORIES = ( "original", "preview", "extracted", "temporary", "quarantine", "recycle", ) class UnsafeStoragePathError(ValueError): """相对路径试图逃逸DMS存储根目录。""" def ensure_storage_directories(storage_root: str | Path) -> Path: root = Path(storage_root).expanduser().resolve() root.mkdir(parents=True, exist_ok=True) for directory in STORAGE_SUBDIRECTORIES: (root / directory).mkdir(parents=True, exist_ok=True) return root def resolve_storage_path( relative_path: str, storage_root: str | Path, ) -> Path: """安全解析数据库相对路径并保证结果仍位于存储根目录。""" if not relative_path or "\x00" in relative_path: raise UnsafeStoragePathError("存储相对路径不能为空或包含空字符") normalized = PurePosixPath(relative_path.replace("\\", "/")) has_drive_prefix = bool(normalized.parts and ":" in normalized.parts[0]) if normalized.is_absolute() or has_drive_prefix or ".." in normalized.parts: raise UnsafeStoragePathError("存储路径必须是根目录内的安全相对路径") root = Path(storage_root).expanduser().resolve() candidate = root.joinpath(*normalized.parts).resolve() if not candidate.is_relative_to(root): raise UnsafeStoragePathError("存储路径超出DMS存储根目录") return candidate