Add support for plugin overrides in configuration, allowing alternative plugin implementations. Enhanced weather display settings with configurable units for temperature, wind speed, and precipitation. Updated repeater_manager to improve contact removal logic with better error handling and logging.

This commit is contained in:
agessaman committed 2025-11-15 15:07:43 -08:00
1 parent a93ce069a7
commit ed7f5691c7
9 files changed
+1056 -39

No files matched your search

+154 -12
View File
@@ -22,10 +22,27 @@ class PluginLoader:
self.bot = bot
self.logger = bot.logger
self.commands_dir = commands_dir or os.path.join(os.path.dirname(__file__), 'commands')
self.alternatives_dir = os.path.join(self.commands_dir, 'alternatives')
self.loaded_plugins: Dict[str, BaseCommand] = {}
self.plugin_metadata: Dict[str, Dict[str, Any]] = {}
self.keyword_mappings: Dict[str, str] = {} # keyword -> plugin_name
self.plugin_overrides: Dict[str, str] = {} # plugin_name -> alternative_file_name
self._load_plugin_overrides()
def _load_plugin_overrides(self):
"""Load plugin override configuration from config file"""
self.plugin_overrides = {}
try:
if self.bot.config.has_section('Plugin_Overrides'):
for command_name, alternative_file in self.bot.config.items('Plugin_Overrides'):
# Remove .py extension if present
if alternative_file.endswith('.py'):
alternative_file = alternative_file[:-3]
self.plugin_overrides[command_name.strip()] = alternative_file.strip()
self.logger.info(f"Plugin override configured: {command_name} -> {alternative_file}")
except Exception as e:
self.logger.warning(f"Error loading plugin overrides: {e}")
def discover_plugins(self) -> List[str]:
"""Discover all Python files in the commands directory that could be plugins"""
plugin_files = []
@@ -43,11 +60,41 @@ class PluginLoader:
self.logger.info(f"Discovered {len(plugin_files)} potential plugin files: {plugin_files}")
return plugin_files
def load_plugin(self, plugin_name: str) -> Optional[BaseCommand]:
"""Load a single plugin by name"""
def discover_alternative_plugins(self) -> List[str]:
"""Discover all Python files in the alternatives directory that could be plugins
Note: Plugins in the 'inactive' subdirectory are ignored
"""
plugin_files = []
alternatives_path = Path(self.alternatives_dir)
if not alternatives_path.exists():
# Alternatives directory doesn't exist yet, that's okay
return plugin_files
# Scan for Python files (excluding __init__.py)
# Note: glob("*.py") only matches files in the current directory, not subdirectories
# So the 'inactive' subdirectory is automatically excluded
for file_path in alternatives_path.glob("*.py"):
if file_path.name not in ["__init__.py"]:
plugin_files.append(file_path.stem)
if plugin_files:
self.logger.info(f"Discovered {len(plugin_files)} alternative plugin files: {plugin_files}")
return plugin_files
def load_plugin(self, plugin_name: str, from_alternatives: bool = False) -> Optional[BaseCommand]:
"""Load a single plugin by name
Args:
plugin_name: Name of the plugin file (without .py extension)
from_alternatives: If True, load from alternatives directory; if False, load from commands directory
"""
try:
# Construct the full module path
module_path = f"modules.commands.{plugin_name}"
if from_alternatives:
module_path = f"modules.commands.alternatives.{plugin_name}"
else:
module_path = f"modules.commands.{plugin_name}"
# Check if module is already loaded
if module_path in sys.modules:
@@ -79,7 +126,8 @@ class PluginLoader:
metadata['name'] = command_class.__name__.lower().replace('command', '')
plugin_instance.name = metadata['name']
self.logger.info(f"Successfully loaded plugin: {metadata['name']} from {plugin_name}")
source = "alternatives" if from_alternatives else "default"
self.logger.info(f"Successfully loaded plugin: {metadata['name']} from {plugin_name} ({source})")
return plugin_instance
except Exception as e:
@@ -87,20 +135,86 @@ class PluginLoader:
return None
def load_all_plugins(self) -> Dict[str, BaseCommand]:
"""Load all discovered plugins"""
plugin_files = self.discover_plugins()
"""Load all discovered plugins, with alternative plugins taking priority when configured"""
# First, discover all default and alternative plugins
default_plugin_files = self.discover_plugins()
alternative_plugin_files = self.discover_alternative_plugins()
# Build a map of plugin names to their file names for default plugins
default_plugin_map = {} # plugin_name -> file_name
loaded_plugins = {}
for plugin_file in plugin_files:
plugin_instance = self.load_plugin(plugin_file)
# First pass: Load all default plugins and build the map
for plugin_file in default_plugin_files:
plugin_instance = self.load_plugin(plugin_file, from_alternatives=False)
if plugin_instance:
metadata = plugin_instance.get_metadata()
plugin_name = metadata['name']
default_plugin_map[plugin_name] = plugin_file
loaded_plugins[plugin_name] = plugin_instance
self.plugin_metadata[plugin_name] = metadata
# Second pass: Check for overrides and load alternative plugins
# Check config-based overrides first
for plugin_name, alternative_file in self.plugin_overrides.items():
if alternative_file in alternative_plugin_files:
# Load the alternative plugin
alt_instance = self.load_plugin(alternative_file, from_alternatives=True)
if alt_instance:
alt_metadata = alt_instance.get_metadata()
alt_plugin_name = alt_metadata['name']
# If the alternative plugin has a different name, log a warning
if alt_plugin_name != plugin_name:
self.logger.warning(
f"Alternative plugin {alternative_file} has name '{alt_plugin_name}' "
f"but is configured to override '{plugin_name}'. Using '{alt_plugin_name}'."
)
plugin_name = alt_plugin_name
# Replace the default plugin with the alternative
if plugin_name in loaded_plugins:
self.logger.info(f"Replacing default plugin '{plugin_name}' with alternative '{alternative_file}'")
loaded_plugins[plugin_name] = alt_instance
self.plugin_metadata[plugin_name] = alt_metadata
else:
self.logger.warning(f"Failed to load alternative plugin '{alternative_file}' for '{plugin_name}'")
else:
self.logger.warning(
f"Alternative plugin '{alternative_file}' not found in alternatives directory "
f"for override of '{plugin_name}'"
)
# Third pass: Load alternative plugins that aren't overriding anything
# (standalone alternative plugins)
for alt_file in alternative_plugin_files:
# Skip if this alternative is already loaded as an override
if alt_file in self.plugin_overrides.values():
continue
alt_instance = self.load_plugin(alt_file, from_alternatives=True)
if alt_instance:
alt_metadata = alt_instance.get_metadata()
alt_plugin_name = alt_metadata['name']
# Build keyword mappings
self._build_keyword_mappings(plugin_name, metadata)
# If an alternative plugin has the same name as a default plugin,
# it will replace it (unless already overridden by config)
if alt_plugin_name in loaded_plugins:
if alt_plugin_name not in self.plugin_overrides:
self.logger.info(
f"Alternative plugin '{alt_file}' replaces default plugin '{alt_plugin_name}' "
f"(same plugin name detected)"
)
else:
self.logger.info(f"Loading standalone alternative plugin: {alt_plugin_name} from {alt_file}")
loaded_plugins[alt_plugin_name] = alt_instance
self.plugin_metadata[alt_plugin_name] = alt_metadata
# Build keyword mappings for all loaded plugins
for plugin_name, plugin_instance in loaded_plugins.items():
metadata = self.plugin_metadata[plugin_name]
self._build_keyword_mappings(plugin_name, metadata)
self.loaded_plugins = loaded_plugins
self.logger.info(f"Loaded {len(loaded_plugins)} plugins: {list(loaded_plugins.keys())}")
@@ -164,8 +278,36 @@ class PluginLoader:
for keyword in keywords_to_remove:
del self.keyword_mappings[keyword]
# Reload the plugin
plugin_instance = self.load_plugin(plugin_name)
# Check if this plugin should be loaded from alternatives
from_alternatives = False
if plugin_name in self.plugin_overrides:
# This plugin is overridden, reload from alternatives
alternative_file = self.plugin_overrides[plugin_name]
plugin_instance = self.load_plugin(alternative_file, from_alternatives=True)
else:
# Try to find the plugin file name
# First check if it's in default plugins
default_plugins = self.discover_plugins()
plugin_file = None
for df in default_plugins:
test_instance = self.load_plugin(df, from_alternatives=False)
if test_instance and test_instance.get_metadata().get('name') == plugin_name:
plugin_file = df
break
if plugin_file:
plugin_instance = self.load_plugin(plugin_file, from_alternatives=False)
else:
# Try alternatives
alt_plugins = self.discover_alternative_plugins()
for alt in alt_plugins:
test_instance = self.load_plugin(alt, from_alternatives=True)
if test_instance and test_instance.get_metadata().get('name') == plugin_name:
plugin_instance = self.load_plugin(alt, from_alternatives=True)
break
else:
plugin_instance = None
if plugin_instance:
metadata = plugin_instance.get_metadata()
self.loaded_plugins[plugin_name] = plugin_instance