253 lines
9.5 KiB
Python
253 lines
9.5 KiB
Python
|
"""
|
||
|
|
module_manager.py
|
||
|
|
|
||
|
|
This module provides the ModuleManager class which allows for the management
|
||
|
|
of dynamically loadable modules. The ModuleManager supports loading, unloading,
|
||
|
|
and execution of modules, as well as handling module-specific commands and hooks.
|
||
|
|
|
||
|
|
Classes:
|
||
|
|
ModuleManager: Manages loading, unloading, and executing modules and their commands.
|
||
|
|
|
||
|
|
Variables:
|
||
|
|
hook_manager (HookManager): An instance of HookManager for managing hooks.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
import importlib
|
||
|
|
import logging
|
||
|
|
import hashlib
|
||
|
|
from hook_manager import HookManager
|
||
|
|
|
||
|
|
hook_manager = HookManager()
|
||
|
|
|
||
|
|
class ModuleManager:
|
||
|
|
"""
|
||
|
|
A manager for handling dynamically loadable modules, including loading, unloading,
|
||
|
|
and executing module-specific commands and hooks.
|
||
|
|
|
||
|
|
Attributes:
|
||
|
|
module_dirs (list): A list of directories to search for modules.
|
||
|
|
loaded_modules (dict): A dictionary of loaded modules.
|
||
|
|
extra_commands (dict): A dictionary of additional commands provided by modules.
|
||
|
|
module_hashes (dict): A dictionary storing file paths and their hashes.
|
||
|
|
hook_manager (HookManager): An instance of HookManager for managing hooks.
|
||
|
|
"""
|
||
|
|
|
||
|
|
def __init__(self, module_dirs):
|
||
|
|
"""
|
||
|
|
Initializes a new instance of the ModuleManager class.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
module_dirs (list): A list of directories to search for modules.
|
||
|
|
"""
|
||
|
|
self.module_dirs = module_dirs
|
||
|
|
self.loaded_modules = {}
|
||
|
|
self.extra_commands = {}
|
||
|
|
self.module_hashes = {} # store file paths and hashes
|
||
|
|
self._update_sys_path()
|
||
|
|
self.hook_manager = hook_manager # Use the global hook_manager!
|
||
|
|
|
||
|
|
def _update_sys_path(self):
|
||
|
|
"""
|
||
|
|
Updates the system path to include the directories in module_dirs.
|
||
|
|
"""
|
||
|
|
for dir in self.module_dirs:
|
||
|
|
full_path = os.path.abspath(dir)
|
||
|
|
if full_path not in sys.path:
|
||
|
|
sys.path.append(full_path)
|
||
|
|
logging.info(f"Added {full_path} to sys.path")
|
||
|
|
|
||
|
|
def add_module_dir(self, new_dir):
|
||
|
|
"""
|
||
|
|
Adds a new directory to the list of module directories and updates the system path.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
new_dir (str): The new directory to add.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
tuple: A tuple containing a boolean indicating success and a message.
|
||
|
|
"""
|
||
|
|
try:
|
||
|
|
full_path = os.path.abspath(new_dir)
|
||
|
|
if full_path not in self.module_dirs:
|
||
|
|
self.module_dirs.append(full_path)
|
||
|
|
self._update_sys_path()
|
||
|
|
return True, f"Added module directory: {full_path}"
|
||
|
|
return False, f"Module directory already exists: {full_path}"
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f"Error adding module directory: {str(e)}")
|
||
|
|
return False, f"Error adding module directory: {str(e)}"
|
||
|
|
|
||
|
|
def calculate_file_hash(self, file_path):
|
||
|
|
"""
|
||
|
|
Calculates the SHA-256 hash of a file.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
file_path (str): The path to the file.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
str: The SHA-256 hash of the file.
|
||
|
|
"""
|
||
|
|
sha256_hash = hashlib.sha256()
|
||
|
|
try:
|
||
|
|
with open(file_path, "rb") as f:
|
||
|
|
for byte_block in iter(lambda: f.read(4096), b""):
|
||
|
|
sha256_hash.update(byte_block)
|
||
|
|
return sha256_hash.hexdigest()
|
||
|
|
except FileNotFoundError:
|
||
|
|
logging.error(f"File not found: {file_path}")
|
||
|
|
return None
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f"Error calculating hash for file '{file_path}': {str(e)}")
|
||
|
|
return None
|
||
|
|
|
||
|
|
def load_module(self, module_name):
|
||
|
|
"""
|
||
|
|
Loads a module by name, initializing it and registering its commands and hooks.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
module_name (str): The name of the module to load.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
tuple: A tuple containing a boolean indicating success and a message.
|
||
|
|
"""
|
||
|
|
self.hook_manager.execute_hook('pre_module_load', module_name)
|
||
|
|
for dir in self.module_dirs:
|
||
|
|
try:
|
||
|
|
module_path = os.path.join(dir, f"{module_name}.py")
|
||
|
|
if not os.path.exists(module_path):
|
||
|
|
continue
|
||
|
|
|
||
|
|
module = importlib.import_module(f'{os.path.basename(dir)}.{module_name}')
|
||
|
|
self.loaded_modules[module_name] = module
|
||
|
|
|
||
|
|
# Calculate and store the hash of the module file
|
||
|
|
module_hash = self.calculate_file_hash(module_path)
|
||
|
|
if module_hash:
|
||
|
|
self.module_hashes[module_path] = module_hash
|
||
|
|
|
||
|
|
self._initialize_module(module)
|
||
|
|
self.hook_manager.execute_hook('post_module_load', module_name)
|
||
|
|
return True, f"Module '{module_name}' loaded and initialized successfully from {dir}."
|
||
|
|
except ImportError as e:
|
||
|
|
logging.error(f"ImportError for module '{module_name}' from {dir}: {str(e)}")
|
||
|
|
continue
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f"Error loading module '{module_name}' from {dir}: {str(e)}")
|
||
|
|
return False, f"Error loading module '{module_name}' from {dir}: {str(e)}"
|
||
|
|
return False, f"Error: Unable to load module '{module_name}' from any of the module directories."
|
||
|
|
|
||
|
|
def _initialize_module(self, module):
|
||
|
|
"""
|
||
|
|
Initializes a loaded module by calling its initialize method, registering commands, and registering hooks.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
module: The loaded module.
|
||
|
|
"""
|
||
|
|
if hasattr(module, 'initialize'):
|
||
|
|
module.initialize()
|
||
|
|
if hasattr(module, 'get_commands'):
|
||
|
|
new_commands = module.get_commands()
|
||
|
|
self.extra_commands.update(new_commands)
|
||
|
|
if hasattr(module, 'register_hooks'):
|
||
|
|
module.register_hooks(self.hook_manager)
|
||
|
|
|
||
|
|
def unload_module(self, module_name):
|
||
|
|
"""
|
||
|
|
Unloads a module by name, shutting it down and removing its commands and hooks.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
module_name (str): The name of the module to unload.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
tuple: A tuple containing a boolean indicating success and a message.
|
||
|
|
"""
|
||
|
|
self.hook_manager.execute_hook('pre_module_unload', module_name)
|
||
|
|
if module_name in self.loaded_modules:
|
||
|
|
try:
|
||
|
|
module = self.loaded_modules[module_name]
|
||
|
|
self._shutdown_module(module)
|
||
|
|
self.hook_manager.execute_hook('post_module_unload', module_name)
|
||
|
|
# Remove the module's hash from our tracking
|
||
|
|
module_path = module.__file__
|
||
|
|
self.module_hashes.pop(module_path, None)
|
||
|
|
|
||
|
|
del self.loaded_modules[module_name]
|
||
|
|
logging.info(f"Module '{module_name}' unloaded successfully.")
|
||
|
|
return True, f"Module '{module_name}' unloaded and shut down."
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f"Error unloading module '{module_name}': {str(e)}")
|
||
|
|
return False, f"Error unloading module '{module_name}': {str(e)}"
|
||
|
|
return False, f"Module '{module_name}' is not loaded."
|
||
|
|
|
||
|
|
def _shutdown_module(self, module):
|
||
|
|
"""
|
||
|
|
Shuts down a loaded module by calling its shutdown method and removing its commands and hooks.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
module: The loaded module.
|
||
|
|
"""
|
||
|
|
if hasattr(module, 'shutdown'):
|
||
|
|
module.shutdown()
|
||
|
|
if hasattr(module, 'get_commands'):
|
||
|
|
commands_to_remove = module.get_commands().keys()
|
||
|
|
for cmd in commands_to_remove:
|
||
|
|
self.extra_commands.pop(cmd, None)
|
||
|
|
if hasattr(module, 'unregister_hooks'):
|
||
|
|
module.unregister_hooks(self.hook_manager)
|
||
|
|
|
||
|
|
def list_modules(self):
|
||
|
|
"""
|
||
|
|
Lists all currently loaded modules.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
list: A list of names of loaded modules.
|
||
|
|
"""
|
||
|
|
return list(self.loaded_modules.keys())
|
||
|
|
|
||
|
|
def list_commands(self):
|
||
|
|
"""
|
||
|
|
Lists all commands provided by the loaded modules.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
list: A list of command names.
|
||
|
|
"""
|
||
|
|
return list(self.extra_commands.keys())
|
||
|
|
|
||
|
|
def execute_command(self, command, args):
|
||
|
|
"""
|
||
|
|
Executes a command provided by a loaded module.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
command (str): The name of the command to execute.
|
||
|
|
args (list): The arguments to pass to the command.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
tuple: A tuple containing a boolean indicating success and the command result or an error message.
|
||
|
|
"""
|
||
|
|
if command in self.extra_commands:
|
||
|
|
try:
|
||
|
|
result = self.extra_commands[command](args)
|
||
|
|
return True, result
|
||
|
|
except Exception as e:
|
||
|
|
logging.error(f"Error executing command '{command}': {str(e)}")
|
||
|
|
return False, f"Error executing command '{command}': {str(e)}"
|
||
|
|
return False, f"Command '{command}' not found"
|
||
|
|
|
||
|
|
|
||
|
|
def check_modules_modified(self):
|
||
|
|
"""
|
||
|
|
Checks if any loaded modules have been modified on disk by comparing file hashes.
|
||
|
|
|
||
|
|
Returns:
|
||
|
|
list: A list of names of modified modules.
|
||
|
|
"""
|
||
|
|
modified_modules = []
|
||
|
|
for module_path, stored_hash in self.module_hashes.items():
|
||
|
|
current_hash = self.calculate_file_hash(module_path)
|
||
|
|
if current_hash and current_hash != stored_hash:
|
||
|
|
module_name = os.path.splitext(os.path.basename(module_path))[0]
|
||
|
|
modified_modules.append(module_name)
|
||
|
|
return modified_modules
|