67 lines
1.9 KiB
Python
67 lines
1.9 KiB
Python
"""本地文件系统存储实现（MVP）。"""
|
|
|
|
from pathlib import Path
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.logging import get_logger
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class LocalStorageService:
|
|
"""文件存储到服务器本地 data/ 目录。
|
|
|
|
key 格式示例：
|
|
users/{uid}/kbs/{kb_id}/original/{rand}_{safe_name}
|
|
users/{uid}/kbs/{kb_id}/markdown/{rand}.md
|
|
|
|
实际路径 = storage_root / key
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
settings = get_settings()
|
|
self._root = settings.storage_root_path
|
|
self._root.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _resolve(self, key: str) -> Path:
|
|
"""解析 key 为绝对路径，同时防路径穿越。"""
|
|
path = (self._root / key).resolve()
|
|
# 安全：确保解析后路径仍在 root 内
|
|
if not str(path).startswith(str(self._root)):
|
|
raise ValueError(f"路径穿越检测：key={key}")
|
|
return path
|
|
|
|
def save(self, key: str, data: bytes) -> str:
|
|
path = self._resolve(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
logger.debug("Saved file: %s (%d bytes)", key, len(data))
|
|
return key
|
|
|
|
def read(self, key: str) -> bytes:
|
|
path = self._resolve(key)
|
|
return path.read_bytes()
|
|
|
|
def delete(self, key: str) -> None:
|
|
path = self._resolve(key)
|
|
if path.exists():
|
|
path.unlink()
|
|
logger.debug("Deleted file: %s", key)
|
|
|
|
def exists(self, key: str) -> bool:
|
|
return self._resolve(key).exists()
|
|
|
|
def get_size(self, key: str) -> int:
|
|
path = self._resolve(key)
|
|
return path.stat().st_size
|
|
|
|
|
|
# 单例
|
|
_instance: LocalStorageService | None = None
|
|
|
|
|
|
def get_storage() -> LocalStorageService:
|
|
global _instance
|
|
if _instance is None:
|
|
_instance = LocalStorageService()
|
|
return _instance |