#!/usr/bin/env python3 """ Funguy Bot Class """ # Importing necessary libraries and modules import os # Operating System functions import logging # Logging library for logging messages import importlib # Library for dynamically importing modules import simplematrixbotlib as botlib # Library for interacting with Matrix chat from dotenv import load_dotenv # Library for loading environment variables from a .env file 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. """ 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. """ # Get log level from environment, default to INFO log_level = os.getenv("LOG_LEVEL", "INFO").upper() # 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=level ) logging.getLogger().setLevel(level) # 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}") def load_plugins(self): """ Method to load plugins from the specified directory. """ # Iterating through files in the plugins directory for plugin_file in os.listdir(self.PLUGINS_DIR): 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}") # Logging successful plugin loading except Exception as e: logging.error(f"Error loading plugin {plugin_name}: {e}") # Logging error if plugin loading fails def setup_plugins(self): """ Method to call setup(bot) on any plugin that defines it. This must be called AFTER self.bot is created (i.e. inside run()), so that plugins which register custom event listeners (e.g. on_custom_event for RoomMemberEvent) receive a valid bot instance. """ 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}") except Exception as e: logging.error(f"Error during setup of plugin {plugin_name}: {e}") def reload_plugins(self): """ Method to reload all plugins. """ self.PLUGINS = {} # Clearing loaded plugins dictionary # Unloading modules from sys.modules 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() def load_config(self): """ Method to load configuration settings. """ self.config = FunguyConfig() # Creating instance of FunguyConfig to load configuration logging.info("Configuration loaded") def load_disabled_plugins(self): """ Method to load disabled plugins from configuration file. """ # Checking if configuration file exists if os.path.exists('funguy.conf'): # 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): """ Method to save disabled plugins to configuration file. """ existing_config = {} # Checking if configuration file exists if os.path.exists('funguy.conf'): # Loading existing configuration data with open('funguy.conf', 'r') as f: existing_config = toml.load(f) # Updating configuration data with disabled plugins existing_config['plugins'] = {'disabled': self.disabled_plugins} # Writing updated configuration data back to file with open('funguy.conf', 'w') as f: toml.dump(existing_config, f) async def handle_commands(self, room, message): """ Method to handle incoming commands and dispatch them to appropriate plugins. """ match = botlib.MessageMatch(room, message, self.bot, self.config.prefix) # Matching message against bot's prefix # Reloading plugins command 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") # Sending success message else: await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message return # Disable plugin command if match.is_not_from_this_bot() and match.prefix() and match.command("disable"): 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 ") # 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}'") # Sending success message else: await self.bot.api.send_text_message(room.room_id, "You are not authorized to disable plugins.") # Sending unauthorized message return # Enable plugin command if match.is_not_from_this_bot() and match.prefix() and match.command("enable"): 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 ") # 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}'") # Sending success message else: 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 tag plugin_list.append(f"[{plugin_name}.py]: {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, "Config rehashed") # Sending success message else: await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.") # Sending unauthorized message return # Dispatching commands to plugins for plugin_name, plugin_module in self.PLUGINS.items(): 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) def rehash_config(self): """ Method to rehash the configuration settings. """ 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 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 async def enable_plugin(self, room_id, plugin_name): """ Method to enable a plugin for a specific room. """ 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 self.save_disabled_plugins() # Saving disabled plugins to configuration file 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: 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") # 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. self.setup_plugins() logging.info("✓ Plugin setup complete") # ----- NEW: Expose plugins dictionary on bot object ----- self.bot.plugins = self.PLUGINS logging.info("✓ Plugin dictionary exposed on bot.plugins") # -------------------------------------------------------- # 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("="*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 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: 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)