hash/FunguyBot
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

karma plugin added. various debug fixes to funguy.py

Browse Source
This commit is contained in:
Hash Borgir
2026-05-06 23:14:35 -05:00
parent c72ea72bae
commit abb4b5e245
5 changed files with 1114 additions and 253 deletions
Download Patch File Download Diff File Expand all files Collapse all files
funguy.py
+214 -165
View File
@@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
Funguy Bot Class - A modular Matrix bot with plugin support
Funguy Bot Class
"""
# Importing necessary libraries and modules
@@ -13,76 +13,78 @@ from dotenv import load_dotenv # Library for loading environment variables from
import time # Time-related functions
import sys # System-specific parameters and functions
import toml # Library for parsing TOML configuration files
import socket # For network diagnostics
# Importing FunguyConfig class from plugins.config module
from plugins.config import FunguyConfig
class FunguyBot:
"""
A bot class for managing plugins and handling commands in a Matrix chat environment.
Methods:
- __init__: Constructor method for initializing the bot.
- load_dotenv: Method to load environment variables from a .env file.
- setup_logging: Method to configure logging settings.
- load_plugins: Method to load plugins from the specified directory.
- reload_plugins: Method to reload all plugins.
- load_config: Method to load configuration settings.
- load_disabled_plugins: Method to load disabled plugins from configuration file.
- save_disabled_plugins: Method to save disabled plugins to configuration file.
- handle_commands: Method to handle incoming commands and dispatch them to appropriate plugins.
- rehash_config: Method to rehash the configuration settings.
- disable_plugin: Method to disable a plugin for a specific room.
- enable_plugin: Method to enable a plugin for a specific room.
- run: Method to initialize and run the bot.
Properties:
- PLUGINS_DIR: Directory where plugins are stored
- PLUGINS: Dictionary to store loaded plugins
- config: Configuration object
- bot: Bot object
- disabled_plugins: Dictionary to store disabled plugins for each room
"""
def __init__(self):
"""
Constructor method for FunguyBot class.
"""
print("[INIT] Starting FunguyBot initialization...")
# Setting up instance variables
self.PLUGINS_DIR = "plugins" # Directory where plugins are stored
self.PLUGINS = {} # Dictionary to store loaded plugins
self.config = None # Configuration object
self.bot = None # Bot object
self.disabled_plugins = {} # Dictionary to store disabled plugins for each room
print("[INIT] Loading environment variables...")
self.load_dotenv() # Loading environment variables from .env file
print("[INIT] Setting up logging...")
self.setup_logging() # Setting up logging configurations
print("[INIT] Loading plugins...")
self.load_plugins() # Loading plugins
print("[INIT] Loading config...")
self.load_config() # Loading bot configuration
print("[INIT] Loading disabled plugins...")
self.load_disabled_plugins() # Loading disabled plugins from configuration file
print("[INIT] FunguyBot initialization complete!")
def load_dotenv(self):
"""
Method to load environment variables from a .env file.
"""
load_dotenv()
print("[ENV] Environment variables loaded")
def setup_logging(self):
"""
Method to configure logging settings.
"""
# Basic configuration for logging messages to console
# Get log level from environment, default to INFO
log_level = os.getenv("LOG_LEVEL", "INFO").upper()
# Configure logging format
# Convert string to logging constant
level_map = {
"DEBUG": logging.DEBUG,
"INFO": logging.INFO,
"WARNING": logging.WARNING,
"ERROR": logging.ERROR,
"CRITICAL": logging.CRITICAL
}
level = level_map.get(log_level, logging.INFO)
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=getattr(logging, log_level, logging.INFO)
level=level
)
logging.getLogger().setLevel(level)
# Set specific loggers to appropriate levels
logging.getLogger().setLevel(getattr(logging, log_level, logging.INFO))
logging.getLogger("simplematrixbotlib").setLevel(logging.WARNING)
# Optionally silence noisy libraries
logging.getLogger("aiohttp").setLevel(logging.WARNING)
logging.getLogger("nio").setLevel(logging.WARNING)
logging.info(f"Logging configured with level: {log_level}")
@@ -92,20 +94,16 @@ class FunguyBot:
Method to load plugins from the specified directory.
"""
# Iterating through files in the plugins directory
plugin_count = 0
for plugin_file in os.listdir(self.PLUGINS_DIR):
if plugin_file.endswith(".py") and plugin_file != "__init__.py":
if plugin_file.endswith(".py"): # Checking if file is a Python file
plugin_name = os.path.splitext(plugin_file)[0] # Extracting plugin name
try:
# Importing plugin module dynamically
module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}")
self.PLUGINS[plugin_name] = module # Storing loaded plugin module
logging.info(f"Loaded plugin: {plugin_name}")
plugin_count += 1
logging.info(f"Loaded plugin: {plugin_name}") # Logging successful plugin loading
except Exception as e:
logging.error(f"Error loading plugin {plugin_name}: {e}", exc_info=True)
logging.info(f"Total plugins loaded: {plugin_count}")
logging.error(f"Error loading plugin {plugin_name}: {e}") # Logging error if plugin loading fails
def setup_plugins(self):
"""
@@ -115,45 +113,28 @@ class FunguyBot:
that plugins which register custom event listeners (e.g. on_custom_event
for RoomMemberEvent) receive a valid bot instance.
"""
setup_count = 0
for plugin_name, plugin_module in self.PLUGINS.items():
if hasattr(plugin_module, "setup") and callable(plugin_module.setup):
try:
plugin_module.setup(self.bot)
logging.info(f"Setup called for plugin: {plugin_name}")
setup_count += 1
except Exception as e:
logging.error(f"Error during setup of plugin {plugin_name}: {e}", exc_info=True)
logging.info(f"Setup completed for {setup_count} plugins")
logging.error(f"Error during setup of plugin {plugin_name}: {e}")
def reload_plugins(self):
"""
Method to reload all plugins.
"""
logging.info("Reloading plugins...")
# Clear loaded plugins dictionary
self.PLUGINS.clear()
self.PLUGINS = {} # Clearing loaded plugins dictionary
# Unloading modules from sys.modules
modules_to_remove = []
for module_name in sys.modules.keys():
if module_name.startswith(self.PLUGINS_DIR + "."):
modules_to_remove.append(module_name)
for module_name in modules_to_remove:
del sys.modules[module_name]
# Reload plugins
self.load_plugins()
for plugin_name in list(sys.modules.keys()):
if plugin_name.startswith(self.PLUGINS_DIR + "."):
del sys.modules[plugin_name] # Deleting plugin module from system modules
self.load_plugins() # Reloading plugins
# Re-run setup for any plugin that needs it (bot already exists at this point)
if self.bot is not None:
self.setup_plugins()
logging.info("Plugins reloaded successfully")
def load_config(self):
"""
Method to load configuration settings.
@@ -167,18 +148,11 @@ class FunguyBot:
"""
# Checking if configuration file exists
if os.path.exists('funguy.conf'):
try:
# Loading configuration data from TOML file
with open('funguy.conf', 'r') as f:
config_data = toml.load(f)
# Extracting disabled plugins from configuration data
self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {})
logging.info(f"Loaded disabled plugins configuration for {len(self.disabled_plugins)} rooms")
except Exception as e:
logging.error(f"Error loading disabled plugins configuration: {e}")
self.disabled_plugins = {}
else:
logging.info("No funguy.conf found, starting with empty disabled plugins list")
# Loading configuration data from TOML file
with open('funguy.conf', 'r') as f:
config_data = toml.load(f)
# Extracting disabled plugins from configuration data
self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {})
def save_disabled_plugins(self):
"""
@@ -187,25 +161,14 @@ class FunguyBot:
existing_config = {}
# Checking if configuration file exists
if os.path.exists('funguy.conf'):
try:
# Loading existing configuration data
with open('funguy.conf', 'r') as f:
existing_config = toml.load(f)
except Exception as e:
logging.error(f"Error reading funguy.conf: {e}")
# Loading existing configuration data
with open('funguy.conf', 'r') as f:
existing_config = toml.load(f)
# Updating configuration data with disabled plugins
if 'plugins' not in existing_config:
existing_config['plugins'] = {}
existing_config['plugins']['disabled'] = self.disabled_plugins
existing_config['plugins'] = {'disabled': self.disabled_plugins}
# Writing updated configuration data back to file
try:
with open('funguy.conf', 'w') as f:
toml.dump(existing_config, f)
logging.info("Saved disabled plugins configuration")
except Exception as e:
logging.error(f"Error saving disabled plugins configuration: {e}")
with open('funguy.conf', 'w') as f:
toml.dump(existing_config, f)
async def handle_commands(self, room, message):
"""
@@ -217,9 +180,9 @@ class FunguyBot:
if match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.reload_plugins() # Reloading plugins
await self.bot.api.send_text_message(room.room_id, "✅ Plugins reloaded successfully")
await self.bot.api.send_text_message(room.room_id, "Plugins reloaded successfully") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "❌ You are not authorized to reload plugins.")
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message
return
# Disable plugin command
@@ -227,13 +190,13 @@ class FunguyBot:
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() # Getting command arguments
if len(args) != 2: # Checking if correct number of arguments provided
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin> <room_id>")
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin> <room_id>") # Sending usage message
else:
plugin_name, room_id = args # Extracting plugin name and room ID
await self.disable_plugin(room_id, plugin_name) # Disabling plugin
await self.bot.api.send_text_message(room.room_id, f"✅ Plugin '{plugin_name}' disabled for room '{room_id}'")
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' disabled for room '{room_id}'") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "❌ You are not authorized to disable plugins.")
await self.bot.api.send_text_message(room.room_id, "You are not authorized to disable plugins.") # Sending unauthorized message
return
# Enable plugin command
@@ -241,85 +204,88 @@ class FunguyBot:
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() # Getting command arguments
if len(args) != 2: # Checking if correct number of arguments provided
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin> <room_id>")
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin> <room_id>") # Sending usage message
else:
plugin_name, room_id = args # Extracting plugin name and room ID
await self.enable_plugin(room_id, plugin_name) # Enabling plugin
await self.bot.api.send_text_message(room.room_id, f"✅ Plugin '{plugin_name}' enabled for room '{room_id}'")
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' enabled for room '{room_id}'") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "❌ You are not authorized to enable plugins.")
await self.bot.api.send_text_message(room.room_id, "You are not authorized to enable plugins.") # Sending unauthorized message
return
# List plugins command with descriptions (bold plugin names)
if match.is_not_from_this_bot() and match.prefix() and match.command("plugins"):
if self.PLUGINS:
# Build a list with plugin names and their descriptions
plugin_list = []
for plugin_name in sorted(self.PLUGINS.keys()):
plugin_module = self.PLUGINS[plugin_name]
# Try to get description from plugin
description = getattr(plugin_module, "__description__", None)
if not description:
# Try to get from docstring
docstring = getattr(plugin_module, "__doc__", None)
if docstring:
# Get first line of docstring
description = docstring.strip().split('\n')[0]
else:
description = "No description available"
# Make plugin name bold using HTML <strong> tag
plugin_list.append(f"<strong>[{plugin_name}.py]</strong>: {description}")
# Send as HTML message (bold will render)
response = "\n".join(plugin_list)
# Split into multiple messages if too long
if len(response) > 4000:
chunk = ""
for line in plugin_list:
if len(chunk) + len(line) + 1 > 4000:
await self.bot.api.send_markdown_message(room.room_id, chunk)
chunk = line + "\n"
else:
chunk += line + "\n"
if chunk:
await self.bot.api.send_markdown_message(room.room_id, chunk)
else:
await self.bot.api.send_markdown_message(room.room_id, response)
else:
await self.bot.api.send_text_message(room.room_id, "No plugins are currently loaded.")
return
# Rehash config command
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.rehash_config() # Rehashing configuration
await self.bot.api.send_text_message(room.room_id, "✅ Configuration rehashed successfully")
await self.bot.api.send_text_message(room.room_id, "Config rehashed") # Sending success message
else:
await self.bot.api.send_text_message(room.room_id, "❌ You are not authorized to rehash configuration.")
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message
return
# List plugins command
if match.is_not_from_this_bot() and match.prefix() and match.command("plugins"):
if self.PLUGINS:
plugin_list = "\n".join([f" • {name}" for name in sorted(self.PLUGINS.keys())])
await self.bot.api.send_markdown_message(
room.room_id,
f"**📦 Loaded Plugins ({len(self.PLUGINS)})**\n\n{plugin_list}"
)
else:
await self.bot.api.send_text_message(room.room_id, "No plugins are currently loaded.")
return
# Dispatching commands to plugins (only if plugin is not disabled for this room)
# Dispatching commands to plugins
for plugin_name, plugin_module in self.PLUGINS.items():
# Check if plugin is disabled for this room
if plugin_name in self.disabled_plugins.get(room.room_id, []):
continue
# Check if plugin has handle_command function
if hasattr(plugin_module, "handle_command") and callable(plugin_module.handle_command):
if plugin_name not in self.disabled_plugins.get(room.room_id, []):
try:
await plugin_module.handle_command(room, message, self.bot, self.config.prefix, self.config)
except Exception as e:
logging.error(f"Error in plugin {plugin_name}: {e}", exc_info=True)
# Don't break the loop, continue with other plugins
def rehash_config(self):
"""
Method to rehash the configuration settings.
"""
logging.info("Rehashing configuration...")
try:
# Reload environment variables
load_dotenv()
# Reload configuration
del self.config # Deleting current configuration object
self.config = FunguyConfig() # Creating new instance of FunguyConfig to load updated configuration
# Reload disabled plugins
self.load_disabled_plugins()
logging.info("Configuration rehashed successfully")
except Exception as e:
logging.error(f"Error rehashing configuration: {e}")
del self.config # Deleting current configuration object
self.config = FunguyConfig() # Creating new instance of FunguyConfig to load updated configuration
async def disable_plugin(self, room_id, plugin_name):
"""
Method to disable a plugin for a specific room.
"""
if plugin_name not in self.PLUGINS:
logging.warning(f"Attempted to disable non-existent plugin: {plugin_name}")
return
if room_id not in self.disabled_plugins:
self.disabled_plugins[room_id] = [] # Creating entry for room ID if not exist
if plugin_name not in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].append(plugin_name) # Adding plugin to list of disabled plugins for the room
self.save_disabled_plugins() # Saving disabled plugins to configuration file
logging.info(f"Plugin '{plugin_name}' disabled for room {room_id}")
async def enable_plugin(self, room_id, plugin_name):
"""
@@ -327,60 +293,143 @@ class FunguyBot:
"""
if room_id in self.disabled_plugins and plugin_name in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].remove(plugin_name) # Removing plugin from list of disabled plugins for the room
# Clean up empty room entries
if not self.disabled_plugins[room_id]:
del self.disabled_plugins[room_id]
self.save_disabled_plugins() # Saving disabled plugins to configuration file
logging.info(f"Plugin '{plugin_name}' enabled for room {room_id}")
def test_connectivity(self, hostname, port=443):
"""
Test network connectivity to Matrix server.
"""
logging.info(f"Testing connectivity to {hostname}:{port}...")
try:
# Test DNS resolution
ip_address = socket.gethostbyname(hostname)
logging.info(f"✓ DNS resolution successful: {hostname} -> {ip_address}")
# Test socket connection
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(10)
result = sock.connect_ex((hostname, port))
sock.close()
if result == 0:
logging.info(f"✓ Socket connection successful to {hostname}:{port}")
return True
else:
logging.error(f"✗ Socket connection failed to {hostname}:{port} (error code: {result})")
return False
except socket.gaierror as e:
logging.error(f"✗ DNS resolution failed for {hostname}: {e}")
return False
except Exception as e:
logging.error(f"✗ Connectivity test failed: {e}")
return False
def run(self):
"""
Method to initialize and run the bot.
"""
print("\n" + "="*60)
print("FUNGUY BOT - STARTING")
print("="*60 + "\n")
# Retrieving Matrix credentials from environment variables
MATRIX_URL = os.getenv("MATRIX_URL")
MATRIX_USER = os.getenv("MATRIX_USER")
MATRIX_PASS = os.getenv("MATRIX_PASS")
# Validate credentials
if not MATRIX_URL:
logging.error("MATRIX_URL not set in .env file")
return
if not MATRIX_USER:
logging.error("MATRIX_USER not set in .env file")
return
if not MATRIX_PASS:
logging.error("MATRIX_PASS not set in .env file")
return
logging.info(f"Matrix URL: {MATRIX_URL}")
logging.info(f"Matrix User: {MATRIX_USER}")
# Extract hostname from URL for connectivity test
hostname = MATRIX_URL.replace("https://", "").replace("http://", "").split("/")[0]
# Test connectivity before attempting to connect
logging.info("="*40)
logging.info("RUNNING NETWORK DIAGNOSTICS")
logging.info("="*40)
if not self.test_connectivity(hostname, 443):
logging.error("Connectivity test failed. Please check:")
logging.error(" 1. Your internet connection")
logging.error(" 2. Firewall settings (outbound port 443)")
logging.error(" 3. DNS resolution")
logging.error(f" 4. If {hostname} is accessible")
return
logging.info("="*40)
logging.info("ATTEMPTING MATRIX CONNECTION")
logging.info("="*40)
try:
# Retrieving Matrix credentials from environment variables
MATRIX_URL = os.getenv("MATRIX_URL")
MATRIX_USER = os.getenv("MATRIX_USER")
MATRIX_PASS = os.getenv("MATRIX_PASS")
# Validate credentials
if not all([MATRIX_URL, MATRIX_USER, MATRIX_PASS]):
logging.error("Missing Matrix credentials in .env file. Please check MATRIX_URL, MATRIX_USER, MATRIX_PASS")
return
logging.info(f"Creating credentials object for {MATRIX_USER}...")
creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS) # Creating credentials object
logging.info("✓ Credentials object created")
logging.info("Creating bot instance...")
self.bot = botlib.Bot(creds, self.config) # Creating bot instance
logging.info("✓ Bot instance created")
logging.info(f"Bot starting with user: {MATRIX_USER}")
logging.info(f"Connected to homeserver: {MATRIX_URL}")
# Check if async_client is available
if hasattr(self.bot, 'async_client'):
logging.info("✓ Async client available")
else:
logging.warning("⚠ Async client not yet available (will be created on login)")
logging.info("Calling setup_plugins()...")
# Call setup() on any plugin that defines it, now that self.bot exists.
# This is what registers custom event listeners such as the join-event
# listener in the welcome plugin.
self.setup_plugins()
logging.info("✓ Plugin setup complete")
# Defining listener for message events
@self.bot.listener.on_message_event
async def wrapper_handle_commands(room, message):
await self.handle_commands(room, message) # Calling handle_commands method for incoming messages
logging.info("Bot is ready and listening for commands...")
logging.info("="*40)
logging.info("BOT IS READY - ATTEMPTING TO CONNECT TO MATRIX")
logging.info("="*40)
logging.info(f"Connecting to {MATRIX_URL} as {MATRIX_USER}...")
logging.info("(This may take up to 30 seconds...)")
self.bot.run() # Running the bot
except Exception as e:
logging.error(f"Fatal error running bot: {e}", exc_info=True)
sys.exit(1)
logging.error(f"Fatal error during bot startup: {e}", exc_info=True)
logging.error("="*40)
logging.error("TROUBLESHOOTING SUGGESTIONS:")
logging.error("1. Check your internet connection")
logging.error("2. Verify MATRIX_URL is correct (should be https://matrix.org)")
logging.error("3. Verify MATRIX_USER and MATRIX_PASS are correct")
logging.error("4. Check if matrix.org is accessible from your network")
logging.error("5. Try increasing timeout in config")
logging.error("="*40)
raise
if __name__ == "__main__":
print("\n" + "="*60)
print("FUNGUY BOT LAUNCHER")
print("="*60)
try:
print("Creating bot instance...")
bot = FunguyBot() # Creating instance of FunguyBot
print("Bot instance created. Running bot...")
bot.run() # Running the bot
except KeyboardInterrupt:
logging.info("Bot stopped by user")
print("\n[!] Bot stopped by user")
sys.exit(0)
except Exception as e:
print(f"\n[!] Fatal error: {e}")
logging.error(f"Unhandled exception: {e}", exc_info=True)
sys.exit(1)