from __future__ import annotations import hashlib import shutil from pathlib import Path from typing import Any from app.core.config import get_settings from app.core.errors import AppError class StorageService: @staticmethod def _base_dir() -> Path: return Path(get_settings().storage_root).resolve() @staticmethod def assert_within_storage_root( path: str | Path, *, label: str = "artifact", settings: Any = None, ) -> Path: """Resolve a path and refuse anything outside the configured storage root. ``tile_manifest_path`` arrives in the analysis request and a manifest entry may name an absolute tile path, so without this an API field is an unbounded reference to the host filesystem. It is also the persistence rule the product already states: only a governed, runtime-produced artifact may be consumed, and a file outside the root is not one. Resolution happens before the comparison, so ``..`` cannot climb out and a sibling that merely shares a name prefix does not pass. """ from app.core.config import get_settings settings = settings or get_settings() raw = str(path or "").strip() if not raw: raise AppError( code="STORAGE_PATH_OUTSIDE_ROOT", message=f"A {label} path is required.", status_code=400, ) resolved = Path(raw).expanduser().resolve() if getattr(settings, "allow_external_artifact_paths", False): return resolved root = Path(settings.storage_root).resolve() if resolved != root and root not in resolved.parents: raise AppError( code="STORAGE_PATH_OUTSIDE_ROOT", message=( f"The {label} path lies outside the configured storage root and is therefore not a " "governed artifact." ), details={"storage_root": str(root)}, status_code=400, ) return resolved @staticmethod def normalize_dataset_type(dataset_type: str) -> str: normalized = dataset_type.strip().lower() if normalized == "geojson": return "vector" return normalized @staticmethod def _safe_filename(value: str) -> str: value = value.strip().replace("\\", "/").split("/")[-1] fallback = "upload" if not value: return fallback allowed = [] for char in value: if char.isalnum() or char in "-_ .": allowed.append(char) else: allowed.append("_") cleaned = "".join(allowed) cleaned = cleaned.strip(" .") return cleaned or fallback @staticmethod def dataset_root(project_id: str, dataset_id: str, dataset_type: str) -> Path: return StorageService._base_dir() / "uploads" / project_id / dataset_type / dataset_id @staticmethod def derived_raster_root(project_id: str, dataset_id: str) -> Path: return StorageService._base_dir() / "rasters" / "derived" / project_id / dataset_id @staticmethod def preview_root(project_id: str, dataset_id: str) -> Path: return StorageService._base_dir() / "previews" / project_id / dataset_id @staticmethod def raster_tiles_root(project_id: str, source_dataset_id: str, tile_set_id: str) -> Path: return StorageService._base_dir() / "tiles" / project_id / source_dataset_id / tile_set_id @staticmethod def dataset_file_path( project_id: str, dataset_id: str, dataset_type: str, original_filename: str, ) -> str: safe_original = StorageService._safe_filename(original_filename) stored_filename = f"{dataset_id}_{safe_original}" return str(StorageService.dataset_root(project_id, dataset_id, dataset_type) / stored_filename) @staticmethod def calculate_checksum_sha256(content: bytes) -> str: digest = hashlib.sha256() digest.update(content) return digest.hexdigest() @staticmethod def persist_dataset_file( project_id: str, dataset_id: str, dataset_type: str, original_filename: str, content: bytes, content_type: str | None, ) -> dict[str, Any]: normalized_type = StorageService.normalize_dataset_type(dataset_type) file_path = Path(StorageService.dataset_file_path(project_id, dataset_id, normalized_type, original_filename)) file_path.parent.mkdir(parents=True, exist_ok=True) with file_path.open("wb") as stream: stream.write(content) metadata: dict[str, Any] = { "original_filename": StorageService._safe_filename(original_filename), "stored_filename": file_path.name, "content_type": content_type or "application/octet-stream", "size_bytes": len(content), "checksum_sha256": StorageService.calculate_checksum_sha256(content), "storage_path": str(file_path), } return metadata @staticmethod def persist_dataset_file_from_path( project_id: str, dataset_id: str, dataset_type: str, original_filename: str, source_path: str | Path, content_type: str | None, ) -> dict[str, Any]: source = Path(source_path).resolve() if not source.is_file(): raise FileNotFoundError(f"Dataset source artifact does not exist: {source}") normalized_type = StorageService.normalize_dataset_type(dataset_type) file_path = Path(StorageService.dataset_file_path(project_id, dataset_id, normalized_type, original_filename)) file_path.parent.mkdir(parents=True, exist_ok=True) digest = hashlib.sha256() size_bytes = 0 with source.open("rb") as input_stream, file_path.open("wb") as output_stream: for chunk in iter(lambda: input_stream.read(8 * 1024 * 1024), b""): output_stream.write(chunk) digest.update(chunk) size_bytes += len(chunk) return { "original_filename": StorageService._safe_filename(original_filename), "stored_filename": file_path.name, "content_type": content_type or "application/octet-stream", "size_bytes": size_bytes, "checksum_sha256": digest.hexdigest(), "storage_path": str(file_path), } @staticmethod def persist_file( storage_path: str, content: bytes, original_filename: str, content_type: str | None, ) -> dict[str, Any]: target = Path(storage_path) target.parent.mkdir(parents=True, exist_ok=True) with target.open("wb") as stream: stream.write(content) metadata: dict[str, Any] = { "original_filename": StorageService._safe_filename(original_filename), "stored_filename": target.name, "content_type": content_type or "application/octet-stream", "size_bytes": len(content), "checksum_sha256": StorageService.calculate_checksum_sha256(content), "storage_path": str(target), } return metadata @staticmethod def remove_dataset_file(path: str) -> None: target = Path(path) if target.exists(): target.unlink(missing_ok=True) dataset_parent = target.parent if dataset_parent.exists() and dataset_parent.is_dir(): has_files = any(dataset_parent.iterdir()) if not has_files: shutil.rmtree(dataset_parent, ignore_errors=True) @staticmethod def dataset_export_path(project_id: str, dataset_id: str, filename: str) -> str: output_dir = StorageService._base_dir() / "exports" / project_id / "datasets" output_dir.mkdir(parents=True, exist_ok=True) return str(output_dir / f"{dataset_id}_{StorageService._safe_filename(filename)}")