Tianyang
542361c63a
Also add a reindex endpoint and button to rebuild the Meilisearch index from files already on disk, without needing to re-upload them.
87 lines
3.0 KiB
Python
87 lines
3.0 KiB
Python
import os
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
|
|
|
|
class LocalFileClient:
|
|
"""本地文件系统存储客户端"""
|
|
|
|
def __init__(self, base_dir: str):
|
|
self._base_dir = Path(base_dir).resolve()
|
|
self._base_dir.mkdir(parents=True, exist_ok=True)
|
|
|
|
def _resolve_path(self, key: str) -> Path:
|
|
"""将 key 转换为文件系统路径，确保不会逃出 base_dir"""
|
|
path = (self._base_dir / key).resolve()
|
|
if not path.is_relative_to(self._base_dir):
|
|
raise ValueError(f"Path traversal attempt: {key}")
|
|
return path
|
|
|
|
def list_objects(self, prefix: str) -> list[dict]:
|
|
"""列出指定前缀下的所有对象（文件和目录）"""
|
|
prefix_path = self._base_dir / prefix if prefix else self._base_dir
|
|
|
|
if not prefix_path.exists():
|
|
return []
|
|
|
|
result = []
|
|
try:
|
|
for entry in prefix_path.iterdir():
|
|
rel_path = entry.relative_to(self._base_dir)
|
|
key = str(rel_path).replace(os.sep, "/")
|
|
|
|
if entry.is_dir():
|
|
result.append({
|
|
"key": key + "/",
|
|
"size": None,
|
|
"is_dir": True,
|
|
"modifiedAt": None,
|
|
})
|
|
else:
|
|
stat = entry.stat()
|
|
result.append({
|
|
"key": key,
|
|
"size": stat.st_size,
|
|
"is_dir": False,
|
|
"modifiedAt": datetime.fromtimestamp(stat.st_mtime, tz=timezone.utc).isoformat(),
|
|
})
|
|
except (PermissionError, OSError):
|
|
pass
|
|
|
|
return result
|
|
|
|
def get_object(self, key: str) -> bytes:
|
|
path = self._resolve_path(key)
|
|
if not path.exists() or not path.is_file():
|
|
raise FileNotFoundError(key)
|
|
return path.read_bytes()
|
|
|
|
def put_object(self, key: str, data: bytes, content_type: str) -> None:
|
|
path = self._resolve_path(key)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
|
|
def delete_object(self, key: str) -> None:
|
|
path = self._resolve_path(key)
|
|
if not path.exists():
|
|
raise FileNotFoundError(key)
|
|
path.unlink()
|
|
# 清理空目录（逐层向上）
|
|
parent = path.parent
|
|
while parent != self._base_dir:
|
|
if not any(parent.iterdir()):
|
|
parent.rmdir()
|
|
parent = parent.parent
|
|
else:
|
|
break
|
|
|
|
def walk_files(self) -> list[str]:
|
|
"""递归遍历所有文件，返回相对于 base_dir 的 key 列表（不含目录）"""
|
|
result = []
|
|
for root, _dirs, filenames in os.walk(self._base_dir):
|
|
for filename in filenames:
|
|
full_path = Path(root) / filename
|
|
rel_path = full_path.relative_to(self._base_dir)
|
|
result.append(str(rel_path).replace(os.sep, "/"))
|
|
return result
|