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

refactor(semantics): migrate legacy @TIER to @COMPLEXITY annotations

Browse Source 
- Replaced @TIER: TRIVIAL with @COMPLEXITY: 1
- Replaced @TIER: STANDARD with @COMPLEXITY: 3
- Replaced @TIER: CRITICAL with @COMPLEXITY: 5
- Manually elevated specific critical/complex components to levels 2 and 4
- Ignored legacy, specs, and node_modules directories
- Updated generated semantic map
This commit is contained in:
busya
2026-03-16 10:06:44 +03:00
parent 7d8866f6ef
commit 67867f8220
321 changed files with 30101 additions and 58483 deletions
Download Patch File Download Diff File Expand all files Collapse all files

378
backend/src/core/plugin_loader.py
View File

@@ -1,192 +1,192 @@
import importlib.util
import os
import sys # Added this line
from typing import Dict, List, Optional
from .plugin_base import PluginBase, PluginConfig
from .logger import belief_scope
# [DEF:PluginLoader:Class]
# @TIER: STANDARD
# @SEMANTICS: plugin, loader, dynamic, import
# @PURPOSE: Scans a specified directory for Python modules, dynamically loads them, and registers any classes that are valid implementations of the PluginBase interface.
# @LAYER: Core
# @RELATION: Depends on PluginBase. It is used by the main application to discover and manage available plugins.
class PluginLoader:
"""
Scans a directory for Python modules, loads them, and identifies classes
that inherit from PluginBase.
"""
# [DEF:__init__:Function]
# @PURPOSE: Initializes the PluginLoader with a directory to scan.
# @PRE: plugin_dir is a valid directory path.
# @POST: Plugins are loaded and registered.
# @PARAM: plugin_dir (str) - The directory containing plugin modules.
def __init__(self, plugin_dir: str):
with belief_scope("__init__"):
self.plugin_dir = plugin_dir
self._plugins: Dict[str, PluginBase] = {}
self._plugin_configs: Dict[str, PluginConfig] = {}
self._load_plugins()
# [/DEF:__init__:Function]
# [DEF:_load_plugins:Function]
# @PURPOSE: Scans the plugin directory and loads all valid plugins.
# @PRE: plugin_dir exists or can be created.
# @POST: _load_module is called for each .py file.
def _load_plugins(self):
with belief_scope("_load_plugins"):
"""
Scans the plugin directory, imports modules, and registers valid plugins.
"""
if not os.path.exists(self.plugin_dir):
os.makedirs(self.plugin_dir)
# Add the plugin directory's parent to sys.path to enable relative imports within plugins
# This assumes plugin_dir is something like 'backend/src/plugins'
# and we want 'backend/src' to be on the path for 'from ..core...' imports
plugin_parent_dir = os.path.abspath(os.path.join(self.plugin_dir, os.pardir))
if plugin_parent_dir not in sys.path:
sys.path.insert(0, plugin_parent_dir)
for filename in os.listdir(self.plugin_dir):
file_path = os.path.join(self.plugin_dir, filename)
# Handle directory-based plugins (packages)
if os.path.isdir(file_path):
init_file = os.path.join(file_path, "__init__.py")
if os.path.exists(init_file):
self._load_module(filename, init_file)
continue
# Handle single-file plugins
if filename.endswith(".py") and filename != "__init__.py":
module_name = filename[:-3]
self._load_module(module_name, file_path)
# [/DEF:_load_plugins:Function]
# [DEF:_load_module:Function]
# @PURPOSE: Loads a single Python module and discovers PluginBase implementations.
# @PRE: module_name and file_path are valid.
# @POST: Plugin classes are instantiated and registered.
# @PARAM: module_name (str) - The name of the module.
# @PARAM: file_path (str) - The path to the module file.
def _load_module(self, module_name: str, file_path: str):
with belief_scope("_load_module"):
"""
Loads a single Python module and extracts PluginBase subclasses.
"""
import importlib.util
import os
import sys # Added this line
from typing import Dict, List, Optional
from .plugin_base import PluginBase, PluginConfig
from .logger import belief_scope
# [DEF:PluginLoader:Class]
# @COMPLEXITY: 3
# @SEMANTICS: plugin, loader, dynamic, import
# @PURPOSE: Scans a specified directory for Python modules, dynamically loads them, and registers any classes that are valid implementations of the PluginBase interface.
# @LAYER: Core
# @RELATION: Depends on PluginBase. It is used by the main application to discover and manage available plugins.
class PluginLoader:
"""
Scans a directory for Python modules, loads them, and identifies classes
that inherit from PluginBase.
"""
# [DEF:__init__:Function]
# @PURPOSE: Initializes the PluginLoader with a directory to scan.
# @PRE: plugin_dir is a valid directory path.
# @POST: Plugins are loaded and registered.
# @PARAM: plugin_dir (str) - The directory containing plugin modules.
def __init__(self, plugin_dir: str):
with belief_scope("__init__"):
self.plugin_dir = plugin_dir
self._plugins: Dict[str, PluginBase] = {}
self._plugin_configs: Dict[str, PluginConfig] = {}
self._load_plugins()
# [/DEF:__init__:Function]
# [DEF:_load_plugins:Function]
# @PURPOSE: Scans the plugin directory and loads all valid plugins.
# @PRE: plugin_dir exists or can be created.
# @POST: _load_module is called for each .py file.
def _load_plugins(self):
with belief_scope("_load_plugins"):
"""
Scans the plugin directory, imports modules, and registers valid plugins.
"""
if not os.path.exists(self.plugin_dir):
os.makedirs(self.plugin_dir)
# Add the plugin directory's parent to sys.path to enable relative imports within plugins
# This assumes plugin_dir is something like 'backend/src/plugins'
# and we want 'backend/src' to be on the path for 'from ..core...' imports
plugin_parent_dir = os.path.abspath(os.path.join(self.plugin_dir, os.pardir))
if plugin_parent_dir not in sys.path:
sys.path.insert(0, plugin_parent_dir)
for filename in os.listdir(self.plugin_dir):
file_path = os.path.join(self.plugin_dir, filename)
# Handle directory-based plugins (packages)
if os.path.isdir(file_path):
init_file = os.path.join(file_path, "__init__.py")
if os.path.exists(init_file):
self._load_module(filename, init_file)
continue
# Handle single-file plugins
if filename.endswith(".py") and filename != "__init__.py":
module_name = filename[:-3]
self._load_module(module_name, file_path)
# [/DEF:_load_plugins:Function]
# [DEF:_load_module:Function]
# @PURPOSE: Loads a single Python module and discovers PluginBase implementations.
# @PRE: module_name and file_path are valid.
# @POST: Plugin classes are instantiated and registered.
# @PARAM: module_name (str) - The name of the module.
# @PARAM: file_path (str) - The path to the module file.
def _load_module(self, module_name: str, file_path: str):
with belief_scope("_load_module"):
"""
Loads a single Python module and extracts PluginBase subclasses.
"""
# All runtime code is imported through the canonical `src` package root.
package_name = f"src.plugins.{module_name}"
# print(f"DEBUG: Loading plugin {module_name} as {package_name}")
spec = importlib.util.spec_from_file_location(package_name, file_path)
if spec is None or spec.loader is None:
print(f"Could not load module spec for {package_name}") # Replace with proper logging
return
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except Exception as e:
print(f"Error loading plugin module {module_name}: {e}") # Replace with proper logging
return
for attribute_name in dir(module):
attribute = getattr(module, attribute_name)
if (
isinstance(attribute, type)
and issubclass(attribute, PluginBase)
and attribute is not PluginBase
):
try:
plugin_instance = attribute()
self._register_plugin(plugin_instance)
except Exception as e:
print(f"Error instantiating plugin {attribute_name} in {module_name}: {e}") # Replace with proper logging
# [/DEF:_load_module:Function]
# [DEF:_register_plugin:Function]
# @PURPOSE: Registers a PluginBase instance and its configuration.
# @PRE: plugin_instance is a valid implementation of PluginBase.
# @POST: Plugin is added to _plugins and _plugin_configs.
# @PARAM: plugin_instance (PluginBase) - The plugin instance to register.
def _register_plugin(self, plugin_instance: PluginBase):
with belief_scope("_register_plugin"):
"""
Registers a valid plugin instance.
"""
plugin_id = plugin_instance.id
if plugin_id in self._plugins:
print(f"Warning: Duplicate plugin ID '{plugin_id}' found. Skipping.") # Replace with proper logging
return
try:
schema = plugin_instance.get_schema()
# Basic validation to ensure it's a dictionary
if not isinstance(schema, dict):
raise TypeError("get_schema() must return a dictionary.")
plugin_config = PluginConfig(
id=plugin_instance.id,
name=plugin_instance.name,
description=plugin_instance.description,
version=plugin_instance.version,
ui_route=plugin_instance.ui_route,
schema=schema,
)
# The following line is commented out because it requires a schema to be passed to validate against.
# The schema provided by the plugin is the one being validated, not the data.
# validate(instance={}, schema=schema)
self._plugins[plugin_id] = plugin_instance
self._plugin_configs[plugin_id] = plugin_config
from ..core.logger import logger
logger.info(f"Plugin '{plugin_instance.name}' (ID: {plugin_id}) loaded successfully.")
except Exception as e:
from ..core.logger import logger
logger.error(f"Error validating plugin '{plugin_instance.name}' (ID: {plugin_id}): {e}")
# [/DEF:_register_plugin:Function]
# [DEF:get_plugin:Function]
# @PURPOSE: Retrieves a loaded plugin instance by its ID.
# @PRE: plugin_id is a string.
# @POST: Returns plugin instance or None.
# @PARAM: plugin_id (str) - The unique identifier of the plugin.
# @RETURN: Optional[PluginBase] - The plugin instance if found, otherwise None.
def get_plugin(self, plugin_id: str) -> Optional[PluginBase]:
with belief_scope("get_plugin"):
"""
Returns a loaded plugin instance by its ID.
"""
return self._plugins.get(plugin_id)
# [/DEF:get_plugin:Function]
# [DEF:get_all_plugin_configs:Function]
# @PURPOSE: Returns a list of all registered plugin configurations.
# @PRE: None.
# @POST: Returns list of all PluginConfig objects.
# @RETURN: List[PluginConfig] - A list of plugin configurations.
def get_all_plugin_configs(self) -> List[PluginConfig]:
with belief_scope("get_all_plugin_configs"):
"""
Returns a list of all loaded plugin configurations.
"""
return list(self._plugin_configs.values())
# [/DEF:get_all_plugin_configs:Function]
# [DEF:has_plugin:Function]
# @PURPOSE: Checks if a plugin with the given ID is registered.
# @PRE: plugin_id is a string.
# @POST: Returns True if plugin exists.
# @PARAM: plugin_id (str) - The unique identifier of the plugin.
# @RETURN: bool - True if the plugin is registered, False otherwise.
def has_plugin(self, plugin_id: str) -> bool:
with belief_scope("has_plugin"):
"""
Checks if a plugin with the given ID is loaded.
"""
return plugin_id in self._plugins
# [/DEF:has_plugin:Function]
# print(f"DEBUG: Loading plugin {module_name} as {package_name}")
spec = importlib.util.spec_from_file_location(package_name, file_path)
if spec is None or spec.loader is None:
print(f"Could not load module spec for {package_name}") # Replace with proper logging
return
module = importlib.util.module_from_spec(spec)
try:
spec.loader.exec_module(module)
except Exception as e:
print(f"Error loading plugin module {module_name}: {e}") # Replace with proper logging
return
for attribute_name in dir(module):
attribute = getattr(module, attribute_name)
if (
isinstance(attribute, type)
and issubclass(attribute, PluginBase)
and attribute is not PluginBase
):
try:
plugin_instance = attribute()
self._register_plugin(plugin_instance)
except Exception as e:
print(f"Error instantiating plugin {attribute_name} in {module_name}: {e}") # Replace with proper logging
# [/DEF:_load_module:Function]
# [DEF:_register_plugin:Function]
# @PURPOSE: Registers a PluginBase instance and its configuration.
# @PRE: plugin_instance is a valid implementation of PluginBase.
# @POST: Plugin is added to _plugins and _plugin_configs.
# @PARAM: plugin_instance (PluginBase) - The plugin instance to register.
def _register_plugin(self, plugin_instance: PluginBase):
with belief_scope("_register_plugin"):
"""
Registers a valid plugin instance.
"""
plugin_id = plugin_instance.id
if plugin_id in self._plugins:
print(f"Warning: Duplicate plugin ID '{plugin_id}' found. Skipping.") # Replace with proper logging
return
try:
schema = plugin_instance.get_schema()
# Basic validation to ensure it's a dictionary
if not isinstance(schema, dict):
raise TypeError("get_schema() must return a dictionary.")
plugin_config = PluginConfig(
id=plugin_instance.id,
name=plugin_instance.name,
description=plugin_instance.description,
version=plugin_instance.version,
ui_route=plugin_instance.ui_route,
schema=schema,
)
# The following line is commented out because it requires a schema to be passed to validate against.
# The schema provided by the plugin is the one being validated, not the data.
# validate(instance={}, schema=schema)
self._plugins[plugin_id] = plugin_instance
self._plugin_configs[plugin_id] = plugin_config
from ..core.logger import logger
logger.info(f"Plugin '{plugin_instance.name}' (ID: {plugin_id}) loaded successfully.")
except Exception as e:
from ..core.logger import logger
logger.error(f"Error validating plugin '{plugin_instance.name}' (ID: {plugin_id}): {e}")
# [/DEF:_register_plugin:Function]
# [DEF:get_plugin:Function]
# @PURPOSE: Retrieves a loaded plugin instance by its ID.
# @PRE: plugin_id is a string.
# @POST: Returns plugin instance or None.
# @PARAM: plugin_id (str) - The unique identifier of the plugin.
# @RETURN: Optional[PluginBase] - The plugin instance if found, otherwise None.
def get_plugin(self, plugin_id: str) -> Optional[PluginBase]:
with belief_scope("get_plugin"):
"""
Returns a loaded plugin instance by its ID.
"""
return self._plugins.get(plugin_id)
# [/DEF:get_plugin:Function]
# [DEF:get_all_plugin_configs:Function]
# @PURPOSE: Returns a list of all registered plugin configurations.
# @PRE: None.
# @POST: Returns list of all PluginConfig objects.
# @RETURN: List[PluginConfig] - A list of plugin configurations.
def get_all_plugin_configs(self) -> List[PluginConfig]:
with belief_scope("get_all_plugin_configs"):
"""
Returns a list of all loaded plugin configurations.
"""
return list(self._plugin_configs.values())
# [/DEF:get_all_plugin_configs:Function]
# [DEF:has_plugin:Function]
# @PURPOSE: Checks if a plugin with the given ID is registered.
# @PRE: plugin_id is a string.
# @POST: Returns True if plugin exists.
# @PARAM: plugin_id (str) - The unique identifier of the plugin.
# @RETURN: bool - True if the plugin is registered, False otherwise.
def has_plugin(self, plugin_id: str) -> bool:
with belief_scope("has_plugin"):
"""
Checks if a plugin with the given ID is loaded.
"""
return plugin_id in self._plugins
# [/DEF:has_plugin:Function]
# [/DEF:PluginLoader:Class]