# #region StoragePlugin [TYPE Module] [SEMANTICS fastapi, storage, filesystem, backup, archive] # @BRIEF Provides core filesystem operations for managing backups and repositories. # @LAYER App # @RELATION DEPENDS_ON -> [TaskContext] # @RELATION DEPENDS_ON -> [TaskContext] # @RELATION DEPENDS_ON -> [TaskContext] # @INVARIANT All file operations must be restricted to the configured storage root. # @RATIONALE Replaced Path(__file__).parents[3] with BASE_DIR import from database.py for path resolution consistency. from datetime import datetime import os from pathlib import Path import shutil from typing import Any from fastapi import UploadFile from ...core.logger import belief_scope, logger from ...core.plugin_base import PluginBase from ...core.task_manager.context import TaskContext from ...dependencies import get_config_manager from ...models.storage import FileCategory, StoredFile # #region StoragePlugin [TYPE Class] # @BRIEF Implementation of the storage management plugin. class StoragePlugin(PluginBase): """ Plugin for managing local file storage for backups and repositories. """ # region __init__ [TYPE Function] # @PURPOSE Initializes the StoragePlugin and ensures required directories exist. # @PRE Configuration manager must be accessible. # @POST Storage root and category directories are created on disk. def __init__(self): with belief_scope("StoragePlugin:init"): self.ensure_directories() # endregion __init__ @property # region id [TYPE Function] # @PURPOSE Returns the unique identifier for the storage plugin. # @PRE None. # @POST Returns the plugin ID string. # @RETURN str - "storage-manager" def id(self) -> str: with belief_scope("StoragePlugin:id"): return "storage-manager" # endregion id @property # region name [TYPE Function] # @PURPOSE Returns the human-readable name of the storage plugin. # @PRE None. # @POST Returns the plugin name string. # @RETURN str - "Storage Manager" def name(self) -> str: with belief_scope("StoragePlugin:name"): return "Storage Manager" # endregion name @property # region description [TYPE Function] # @PURPOSE Returns a description of the storage plugin. # @PRE None. # @POST Returns the plugin description string. # @RETURN str - Plugin description. def description(self) -> str: with belief_scope("StoragePlugin:description"): return "Manages local file storage for backups and repositories." # endregion description @property # region version [TYPE Function] # @PURPOSE Returns the version of the storage plugin. # @PRE None. # @POST Returns the version string. # @RETURN str - "1.0.0" def version(self) -> str: with belief_scope("StoragePlugin:version"): return "1.0.0" # endregion version @property # region ui_route [TYPE Function] # @PURPOSE Returns the frontend route for the storage plugin. # @RETURN str - "/tools/storage" def ui_route(self) -> str: with belief_scope("StoragePlugin:ui_route"): return "/tools/storage" # endregion ui_route # region get_schema [TYPE Function] # @PURPOSE Returns the JSON schema for storage plugin parameters. # @PRE None. # @POST Returns a dictionary representing the JSON schema. # @RETURN Dict[str, Any] - JSON schema. def get_schema(self) -> dict[str, Any]: with belief_scope("StoragePlugin:get_schema"): return { "type": "object", "properties": { "category": { "type": "string", "enum": [c.value for c in FileCategory], "title": "Category" } }, "required": ["category"] } # endregion get_schema # region execute [TYPE Function] # @PURPOSE Executes storage-related tasks with TaskContext support. # @PARAM params (Dict[str, Any]) - Storage parameters. # @PARAM context (Optional[TaskContext]) - Task context for logging with source attribution. # @PRE params must match the plugin schema. # @POST Task is executed and logged. async def execute(self, params: dict[str, Any], context: TaskContext | None = None): with belief_scope("StoragePlugin:execute"): # Use TaskContext logger if available, otherwise fall back to app logger log = context.logger if context else logger # Create sub-loggers for different components storage_log = log.with_source("storage") if context else log log.with_source("filesystem") if context else log storage_log.info(f"Executing with params: {params}") # endregion execute # region get_storage_root [TYPE Function] # @PURPOSE Resolves the absolute path to the storage root. # @PRE Settings must define a storage root path. # @POST Returns a Path object representing the storage root. def get_storage_root(self) -> Path: with belief_scope("StoragePlugin:get_storage_root"): config_manager = get_config_manager() global_settings = config_manager.get_config().settings # Use storage.root_path as the source of truth for storage UI root = Path(global_settings.storage.root_path) if not root.is_absolute(): # Resolve relative to the backend directory # Path(__file__) is backend/src/plugins/storage/plugin.py # parents[3] is the project root (ss-tools) # We need to ensure it's relative to where backend/ is project_root = Path(__file__).parents[3] root = (project_root / root).resolve() return root # endregion get_storage_root # region resolve_path [TYPE Function] # @PURPOSE Resolves a dynamic path pattern using provided variables. # @PARAM pattern (str) - The path pattern to resolve. # @PARAM variables (Dict[str, str]) - Variables to substitute in the pattern. # @PRE pattern must be a valid format string. # @POST Returns the resolved path string. # @RETURN str - The resolved path. def resolve_path(self, pattern: str, variables: dict[str, str]) -> str: with belief_scope("StoragePlugin:resolve_path"): # Add common variables vars_with_defaults = { "timestamp": datetime.now().strftime("%Y%m%dT%H%M%S"), **variables } try: resolved = pattern.format(**vars_with_defaults) # Clean up any double slashes or leading/trailing slashes for relative path return os.path.normpath(resolved).strip("/") except KeyError as e: logger.warning(f"[StoragePlugin][Coherence:Failed] Missing variable for path resolution: {e}") # Fallback to literal pattern if formatting fails partially (or handle as needed) return pattern.replace("{", "").replace("}", "") # endregion resolve_path # region ensure_directories [TYPE Function] # @PURPOSE Creates the storage root and category subdirectories if they don't exist. # @PRE Storage root must be resolvable. # @POST Directories are created on the filesystem. # @SIDE_EFFECT Creates directories on the filesystem. def ensure_directories(self): with belief_scope("StoragePlugin:ensure_directories"): root = self.get_storage_root() for category in FileCategory: # Use singular name for consistency with BackupPlugin and GitService path = root / category.value path.mkdir(parents=True, exist_ok=True) logger.reason(f"Ensured directory: {path}", extra={"src": "StoragePlugin.ensure_directories"}) # endregion ensure_directories # region validate_path [TYPE Function] # @PURPOSE Prevents path traversal attacks by ensuring the path is within the storage root. # @PRE path must be a Path object. # @POST Returns the resolved absolute path if valid, otherwise raises ValueError. def validate_path(self, path: Path) -> Path: with belief_scope("StoragePlugin:validate_path"): root = self.get_storage_root().resolve() resolved = path.resolve() try: resolved.relative_to(root) except ValueError: logger.error(f"[StoragePlugin][Coherence:Failed] Path traversal detected: {resolved} is not under {root}") raise ValueError("Access denied: Path is outside of storage root.") return resolved # endregion validate_path # region list_files [TYPE Function] # @PURPOSE Lists all files and directories in a specific category and subpath. # @PARAM category (Optional[FileCategory]) - The category to list. # @PARAM subpath (Optional[str]) - Nested path within the category. # @PARAM recursive (bool) - Whether to scan nested subdirectories recursively. # @PRE Storage root must exist. # @POST Returns a list of StoredFile objects. # @RETURN List[StoredFile] - List of file and directory metadata objects. def list_files( self, category: FileCategory | None = None, subpath: str | None = None, recursive: bool = False, ) -> list[StoredFile]: with belief_scope("StoragePlugin:list_files"): root = self.get_storage_root() logger.reason( f"Listing files in root: {root}", extra={"src": "StoragePlugin.list_files", "payload": {"root": str(root), "category": category.value if category else None, "subpath": subpath, "recursive": recursive}}, ) files = [] # Root view contract: show category directories only. if category is None and not subpath: for cat in FileCategory: base_dir = root / cat.value if not base_dir.exists(): continue stat = base_dir.stat() files.append( StoredFile( name=cat.value, path=cat.value, size=0, created_at=datetime.fromtimestamp(stat.st_ctime), category=cat, mime_type="directory", ) ) return sorted(files, key=lambda x: x.name) categories = [category] if category else list(FileCategory) for cat in categories: # Scan the category subfolder + optional subpath base_dir = root / cat.value if subpath: target_dir = self.validate_path(base_dir / subpath) else: target_dir = base_dir if not target_dir.exists(): continue logger.reason(f"Scanning directory: {target_dir}", extra={"src": "StoragePlugin.list_files"}) if recursive: for current_root, dirs, filenames in os.walk(target_dir): dirs[:] = [d for d in dirs if "Logs" not in d] for filename in filenames: file_path = Path(current_root) / filename if "Logs" in str(file_path): continue stat = file_path.stat() files.append( StoredFile( name=filename, path=str(file_path.relative_to(root)), size=stat.st_size, created_at=datetime.fromtimestamp(stat.st_ctime), category=cat, mime_type=None, ) ) continue # Use os.scandir for better performance and to distinguish files vs dirs with os.scandir(target_dir) as it: for entry in it: # Skip logs if "Logs" in entry.path: continue stat = entry.stat() is_dir = entry.is_dir() files.append(StoredFile( name=entry.name, path=str(Path(entry.path).relative_to(root)), size=stat.st_size if not is_dir else 0, created_at=datetime.fromtimestamp(stat.st_ctime), category=cat, mime_type="directory" if is_dir else None )) # Sort: directories first, then by name return sorted(files, key=lambda x: (x.mime_type != "directory", x.name)) # endregion list_files # region save_file [TYPE Function] # @PURPOSE Saves an uploaded file to the specified category and optional subpath. # @PARAM file (UploadFile) - The uploaded file. # @PARAM category (FileCategory) - The target category. # @PARAM subpath (Optional[str]) - The target subpath. # @PRE file must be a valid UploadFile; category must be valid. # @POST File is written to disk and metadata is returned. # @RETURN StoredFile - Metadata of the saved file. # @SIDE_EFFECT Writes file to disk. async def save_file(self, file: UploadFile, category: FileCategory, subpath: str | None = None) -> StoredFile: with belief_scope("StoragePlugin:save_file"): root = self.get_storage_root() dest_dir = root / category.value if subpath: dest_dir = dest_dir / subpath dest_dir.mkdir(parents=True, exist_ok=True) dest_path = self.validate_path(dest_dir / file.filename) with dest_path.open("wb") as buffer: shutil.copyfileobj(file.file, buffer) stat = dest_path.stat() return StoredFile( name=dest_path.name, path=str(dest_path.relative_to(root)), size=stat.st_size, created_at=datetime.fromtimestamp(stat.st_ctime), category=category, mime_type=file.content_type ) # endregion save_file # region delete_file [TYPE Function] # @PURPOSE Deletes a file or directory from the specified category and path. # @PARAM category (FileCategory) - The category. # @PARAM path (str) - The relative path of the file or directory. # @PRE path must belong to the specified category and exist on disk. # @POST The file or directory is removed from disk. # @SIDE_EFFECT Removes item from disk. def delete_file(self, category: FileCategory, path: str): with belief_scope("StoragePlugin:delete_file"): root = self.get_storage_root() # path is relative to root, but we ensure it starts with category full_path = self.validate_path(root / path) if not str(Path(path)).startswith(category.value): raise ValueError(f"Path {path} does not belong to category {category}") if full_path.exists(): if full_path.is_dir(): shutil.rmtree(full_path) else: full_path.unlink() logger.reason(f"Deleted: {full_path}", extra={"src": "StoragePlugin.delete_file"}) else: raise FileNotFoundError(f"Item {path} not found") # endregion delete_file # region get_file_path [TYPE Function] # @PURPOSE Returns the absolute path of a file for download. # @PARAM category (FileCategory) - The category. # @PARAM path (str) - The relative path of the file. # @PRE path must belong to the specified category and be a file. # @POST Returns the absolute Path to the file. # @RETURN Path - Absolute path to the file. def get_file_path(self, category: FileCategory, path: str) -> Path: with belief_scope("StoragePlugin:get_file_path"): root = self.get_storage_root() file_path = self.validate_path(root / path) if not str(Path(path)).startswith(category.value): raise ValueError(f"Path {path} does not belong to category {category}") if not file_path.exists() or file_path.is_dir(): raise FileNotFoundError(f"File {path} not found") return file_path # endregion get_file_path # #endregion StoragePlugin # #endregion StoragePlugin