busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

semantic

Browse Source
This commit is contained in:
busya
2026-05-20 23:54:53 +03:00
parent a43886106e
commit 68740cd9ed
167 changed files with 10620 additions and 29041 deletions
Download Patch File Download Diff File Expand all files Collapse all files

161
backend/src/plugins/storage/plugin.py
View File

@@ -1,17 +1,16 @@
# #region StoragePlugin [TYPE Module] [SEMANTICS fastapi, storage, filesystem, backup, archive]
#
# @BRIEF Provides core filesystem operations for managing backups and repositories.
# @LAYER: App
# @RELATION IMPLEMENTS -> PluginBase
# @RELATION DEPENDS_ON -> backend.src.models.storage
# @LAYER App
# @RELATION USES -> TaskContext
#
# @INVARIANT: All file operations must be restricted to the configured storage root.
# @RELATION USES -> TaskContext
# @RELATION USES -> 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.
import os
import shutil
from datetime import datetime
import os
from pathlib import Path
import shutil
from typing import Any
from fastapi import UploadFile
@@ -31,9 +30,9 @@ class StoragePlugin(PluginBase):
"""
# 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.
# @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()
@@ -41,10 +40,10 @@ class StoragePlugin(PluginBase):
@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"
# @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"
@@ -52,10 +51,10 @@ class StoragePlugin(PluginBase):
@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"
# @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"
@@ -63,10 +62,10 @@ class StoragePlugin(PluginBase):
@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.
# @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."
@@ -74,10 +73,10 @@ class StoragePlugin(PluginBase):
@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"
# @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"
@@ -85,18 +84,18 @@ class StoragePlugin(PluginBase):
@property
# region ui_route [TYPE Function]
# @PURPOSE: Returns the frontend route for the storage plugin.
# @RETURN: str - "/tools/storage"
# @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.
# @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 {
@@ -113,11 +112,11 @@ class StoragePlugin(PluginBase):
# 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.
# @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
@@ -131,9 +130,9 @@ class StoragePlugin(PluginBase):
# 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.
# @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()
@@ -153,12 +152,12 @@ class StoragePlugin(PluginBase):
# 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.
# @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
@@ -177,10 +176,10 @@ class StoragePlugin(PluginBase):
# 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.
# @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()
@@ -192,9 +191,9 @@ class StoragePlugin(PluginBase):
# 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.
# @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()
@@ -208,13 +207,13 @@ class StoragePlugin(PluginBase):
# 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.
# @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,
@@ -307,14 +306,14 @@ class StoragePlugin(PluginBase):
# 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.
# @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()
@@ -341,12 +340,12 @@ class StoragePlugin(PluginBase):
# 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.
# @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()
@@ -367,12 +366,12 @@ class StoragePlugin(PluginBase):
# 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.
# @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()