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

Передаем на тест

Browse Source
This commit is contained in:
busya
2026-01-25 18:33:00 +03:00
parent a863807cf2
commit a542e7d2df
17 changed files with 954 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
backend/src/plugins/storage/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .plugin import StoragePlugin
__all__ = ["StoragePlugin"]

230
backend/src/plugins/storage/plugin.py Normal file
View File

@@ -0,0 +1,230 @@
# [DEF:StoragePlugin:Module]
#
# @SEMANTICS: storage, files, filesystem, plugin
# @PURPOSE: Provides core filesystem operations for managing backups and repositories.
# @LAYER: App
# @RELATION: IMPLEMENTS -> PluginBase
# @RELATION: DEPENDS_ON -> backend.src.models.storage
#
# @INVARIANT: All file operations must be restricted to the configured storage root.
# [SECTION: IMPORTS]
import os
import shutil
from pathlib import Path
from datetime import datetime
from typing import Dict, Any, List, Optional
from fastapi import UploadFile
from ...core.plugin_base import PluginBase
from ...core.logger import belief_scope, logger
from ...models.storage import StoredFile, FileCategory, StorageConfig
from ...dependencies import get_config_manager
# [/SECTION]
# [DEF:StoragePlugin:Class]
# @PURPOSE: Implementation of the storage management plugin.
class StoragePlugin(PluginBase):
"""
Plugin for managing local file storage for backups and repositories.
"""
# [DEF:__init__:Function]
# @PURPOSE: Initializes the StoragePlugin and ensures required directories exist.
def __init__(self):
with belief_scope("StoragePlugin:init"):
self.ensure_directories()
# [/DEF:__init__:Function]
@property
# [DEF:id:Function]
# @PURPOSE: Returns the unique identifier for the storage plugin.
# @RETURN: str - "storage-manager"
def id(self) -> str:
return "storage-manager"
# [/DEF:id:Function]
@property
# [DEF:name:Function]
# @PURPOSE: Returns the human-readable name of the storage plugin.
# @RETURN: str - "Storage Manager"
def name(self) -> str:
return "Storage Manager"
# [/DEF:name:Function]
@property
# [DEF:description:Function]
# @PURPOSE: Returns a description of the storage plugin.
# @RETURN: str - Plugin description.
def description(self) -> str:
return "Manages local file storage for backups and repositories."
# [/DEF:description:Function]
@property
# [DEF:version:Function]
# @PURPOSE: Returns the version of the storage plugin.
# @RETURN: str - "1.0.0"
def version(self) -> str:
return "1.0.0"
# [/DEF:version:Function]
# [DEF:get_schema:Function]
# @PURPOSE: Returns the JSON schema for storage plugin parameters.
# @RETURN: Dict[str, Any] - JSON schema.
def get_schema(self) -> Dict[str, Any]:
return {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": [c.value for c in FileCategory],
"title": "Category"
}
},
"required": ["category"]
}
# [/DEF:get_schema:Function]
# [DEF:execute:Function]
# @PURPOSE: Executes storage-related tasks (placeholder for PluginBase compliance).
async def execute(self, params: Dict[str, Any]):
with belief_scope("StoragePlugin:execute"):
logger.info(f"[StoragePlugin][Action] Executing with params: {params}")
# [/DEF:execute:Function]
# [DEF:get_storage_root:Function]
# @PURPOSE: Resolves the absolute path to the storage root.
# @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()
storage_config = config_manager.get_config().settings.storage
root = Path(storage_config.root_path)
if not root.is_absolute():
# Resolve relative to the workspace root (ss-tools)
root = (Path(__file__).parents[4] / root).resolve()
return root
# [/DEF:get_storage_root:Function]
# [DEF:ensure_directories:Function]
# @PURPOSE: Creates the storage root and category subdirectories if they don't exist.
# @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:
path = root / f"{category.value}s"
path.mkdir(parents=True, exist_ok=True)
logger.debug(f"[StoragePlugin][Action] Ensured directory: {path}")
# [/DEF:ensure_directories:Function]
# [DEF:validate_path: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
# [/DEF:validate_path:Function]
# [DEF:list_files:Function]
# @PURPOSE: Lists all files in a specific category.
# @PARAM: category (Optional[FileCategory]) - The category to list.
# @RETURN: List[StoredFile] - List of file metadata objects.
def list_files(self, category: Optional[FileCategory] = None) -> List[StoredFile]:
with belief_scope("StoragePlugin:list_files"):
root = self.get_storage_root()
files = []
categories = [category] if category else list(FileCategory)
for cat in categories:
cat_dir = root / f"{cat.value}s"
if not cat_dir.exists():
continue
for item in cat_dir.iterdir():
if item.is_file():
stat = item.stat()
files.append(StoredFile(
name=item.name,
path=str(item.relative_to(root)),
size=stat.st_size,
created_at=datetime.fromtimestamp(stat.st_ctime),
category=cat,
mime_type=None # Could use python-magic here if needed
))
return sorted(files, key=lambda x: x.created_at, reverse=True)
# [/DEF:list_files:Function]
# [DEF:save_file:Function]
# @PURPOSE: Saves an uploaded file to the specified category.
# @PARAM: file (UploadFile) - The uploaded file.
# @PARAM: category (FileCategory) - The target category.
# @RETURN: StoredFile - Metadata of the saved file.
# @SIDE_EFFECT: Writes file to disk.
async def save_file(self, file: UploadFile, category: FileCategory) -> StoredFile:
with belief_scope("StoragePlugin:save_file"):
root = self.get_storage_root()
dest_dir = root / f"{category.value}s"
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
)
# [/DEF:save_file:Function]
# [DEF:delete_file:Function]
# @PURPOSE: Deletes a file from the specified category.
# @PARAM: category (FileCategory) - The category.
# @PARAM: filename (str) - The name of the file.
# @SIDE_EFFECT: Removes file from disk.
def delete_file(self, category: FileCategory, filename: str):
with belief_scope("StoragePlugin:delete_file"):
root = self.get_storage_root()
file_path = self.validate_path(root / f"{category.value}s" / filename)
if file_path.exists():
file_path.unlink()
logger.info(f"[StoragePlugin][Action] Deleted file: {file_path}")
else:
raise FileNotFoundError(f"File {filename} not found in {category.value}s")
# [/DEF:delete_file:Function]
# [DEF:get_file_path:Function]
# @PURPOSE: Returns the absolute path of a file for download.
# @PARAM: category (FileCategory) - The category.
# @PARAM: filename (str) - The name of the file.
# @RETURN: Path - Absolute path to the file.
def get_file_path(self, category: FileCategory, filename: str) -> Path:
with belief_scope("StoragePlugin:get_file_path"):
root = self.get_storage_root()
file_path = self.validate_path(root / f"{category.value}s" / filename)
if not file_path.exists():
raise FileNotFoundError(f"File {filename} not found in {category.value}s")
return file_path
# [/DEF:get_file_path:Function]
# [/DEF:StoragePlugin:Class]
# [/DEF:StoragePlugin:Module]