141 lines
5.1 KiB
Python
141 lines
5.1 KiB
Python
from pathlib import Path
|
|
import hashlib
|
|
|
|
from app.storage.base import StorageProvider
|
|
from app.storage.exceptions import StorageFileNotFoundError, StorageValidationError
|
|
from app.storage.schemas import FileMetadata, StoredFileMetadata
|
|
|
|
|
|
class LocalDiskStorageProvider(StorageProvider):
|
|
def __init__(
|
|
self,
|
|
*,
|
|
base_path: str,
|
|
legacy_base_paths: list[str] | None = None,
|
|
) -> None:
|
|
self.base_path = Path(base_path).expanduser().resolve()
|
|
self.legacy_base_paths = [
|
|
Path(path).expanduser().resolve()
|
|
for path in legacy_base_paths or []
|
|
if path
|
|
]
|
|
self._ensure_layout()
|
|
|
|
def save_file(
|
|
self,
|
|
*,
|
|
namespace: str,
|
|
content: bytes,
|
|
original_filename: str,
|
|
stored_filename: str,
|
|
mime_type: str,
|
|
) -> StoredFileMetadata:
|
|
if Path(stored_filename).name != stored_filename or ".." in Path(stored_filename).parts:
|
|
raise StorageValidationError("Ungültiger gespeicherter Dateiname")
|
|
namespace_path = self._safe_namespace_path(namespace)
|
|
namespace_path.mkdir(parents=True, exist_ok=True)
|
|
absolute_path = self._safe_child(namespace_path / stored_filename)
|
|
absolute_path.write_bytes(content)
|
|
relative_path = absolute_path.relative_to(self.base_path).as_posix()
|
|
checksum = self._checksum(absolute_path)
|
|
return StoredFileMetadata(
|
|
storage_key=relative_path,
|
|
original_filename=original_filename,
|
|
stored_filename=stored_filename,
|
|
relative_path=relative_path,
|
|
absolute_path=absolute_path,
|
|
mime_type=mime_type,
|
|
size=len(content),
|
|
checksum_sha256=checksum,
|
|
)
|
|
|
|
def open_file(self, storage_key: str) -> Path:
|
|
path = self._resolve_storage_key(storage_key)
|
|
if not path.exists() or not path.is_file():
|
|
raise StorageFileNotFoundError("Datei nicht gefunden")
|
|
return path
|
|
|
|
def delete_file(self, storage_key: str) -> None:
|
|
path = self._resolve_storage_key(storage_key)
|
|
if not path.exists():
|
|
return
|
|
if not path.is_file():
|
|
raise StorageValidationError("Storage-Key verweist nicht auf eine Datei")
|
|
path.unlink()
|
|
|
|
def file_exists(self, storage_key: str) -> bool:
|
|
try:
|
|
path = self._resolve_storage_key(storage_key)
|
|
except StorageValidationError:
|
|
return False
|
|
return path.exists() and path.is_file()
|
|
|
|
def get_file_metadata(self, storage_key: str) -> FileMetadata:
|
|
path = self.open_file(storage_key)
|
|
return FileMetadata(
|
|
storage_key=storage_key,
|
|
relative_path=self._relative_storage_key(path),
|
|
absolute_path=path,
|
|
size=path.stat().st_size,
|
|
checksum_sha256=self._checksum(path),
|
|
)
|
|
|
|
def _ensure_layout(self) -> None:
|
|
for relative_path in [
|
|
"knowledge/documents",
|
|
"knowledge/thumbnails",
|
|
"customers",
|
|
"projects",
|
|
"tickets",
|
|
"imports",
|
|
"temp",
|
|
]:
|
|
(self.base_path / relative_path).mkdir(parents=True, exist_ok=True)
|
|
|
|
def _safe_namespace_path(self, namespace: str) -> Path:
|
|
if namespace.startswith("/") or ".." in Path(namespace).parts:
|
|
raise StorageValidationError("Ungültiger Storage-Namespace")
|
|
return self._safe_child(self.base_path / namespace)
|
|
|
|
def _safe_child(self, path: Path) -> Path:
|
|
resolved = path.resolve()
|
|
if self.base_path != resolved and self.base_path not in resolved.parents:
|
|
raise StorageValidationError("Ungültiger Storage-Pfad")
|
|
return resolved
|
|
|
|
def _resolve_storage_key(self, storage_key: str) -> Path:
|
|
raw_path = Path(storage_key)
|
|
candidate_paths: list[Path] = []
|
|
|
|
if raw_path.is_absolute():
|
|
candidate_paths.append(raw_path)
|
|
else:
|
|
candidate_paths.append(self.base_path / raw_path)
|
|
candidate_paths.extend(legacy_base / raw_path for legacy_base in self.legacy_base_paths)
|
|
|
|
for candidate in candidate_paths:
|
|
resolved = candidate.expanduser().resolve()
|
|
if self._is_allowed_path(resolved):
|
|
return resolved
|
|
|
|
raise StorageValidationError("Ungültiger Storage-Key")
|
|
|
|
def _is_allowed_path(self, path: Path) -> bool:
|
|
roots = [self.base_path, *self.legacy_base_paths]
|
|
return any(root == path or root in path.parents for root in roots)
|
|
|
|
def _relative_storage_key(self, path: Path) -> str:
|
|
for root in [self.base_path, *self.legacy_base_paths]:
|
|
try:
|
|
return path.relative_to(root).as_posix()
|
|
except ValueError:
|
|
continue
|
|
return path.name
|
|
|
|
@staticmethod
|
|
def _checksum(path: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
with path.open("rb") as handle:
|
|
for chunk in iter(lambda: handle.read(1024 * 1024), b""):
|
|
digest.update(chunk)
|
|
return digest.hexdigest()
|