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

Funguy Class updated

Browse Source
This commit is contained in:
Hash Borgir 2024-03-03 15:17:40 -07:00
parent 41c59618cc
commit 387606426c
2 changed files with 150 additions and 69 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
funguy.conf
View File

@ -12,5 +12,5 @@ prefix = "!"
config_file = "funguy.conf" config_file = "funguy.conf"
[plugins.disabled] [plugins.disabled]
"!uFhErnfpYhhlauJsNK:matrix.org" = [ "youtube-preview",] "!uFhErnfpYhhlauJsNK:matrix.org" = [ "youtube-preview", "ai-tech", "ai-music", "proxy",]
"!vYcfWXpPvxeQvhlFdV:matrix.org" = [] "!vYcfWXpPvxeQvhlFdV:matrix.org" = []

217
funguy.py
View File

@ -4,151 +4,232 @@
Funguy Bot Class Funguy Bot Class
""" """
import os # Importing necessary libraries and modules
import logging import os # Operating System functions
import importlib import logging # Logging library for logging messages
import simplematrixbotlib as botlib import importlib # Library for dynamically importing modules
from dotenv import load_dotenv import simplematrixbotlib as botlib # Library for interacting with Matrix chat
import time from dotenv import load_dotenv # Library for loading environment variables from a .env file
import sys import time # Time-related functions
import toml import sys # System-specific parameters and functions
import toml # Library for parsing TOML configuration files
# Importing FunguyConfig class from plugins.config module
from plugins.config import FunguyConfig from plugins.config import FunguyConfig
class FunguyBot: 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): def __init__(self):
self.PLUGINS_DIR = "plugins" """
self.PLUGINS = {} Constructor method for FunguyBot class.
self.config = None """
self.bot = None # 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 self.disabled_plugins = {} # Dictionary to store disabled plugins for each room
self.load_dotenv() self.load_dotenv() # Loading environment variables from .env file
self.setup_logging() self.setup_logging() # Setting up logging configurations
self.load_plugins() self.load_plugins() # Loading plugins
self.load_config() self.load_config() # Loading bot configuration
self.load_disabled_plugins() self.load_disabled_plugins() # Loading disabled plugins from configuration file
def load_dotenv(self): def load_dotenv(self):
"""
Method to load environment variables from a .env file.
"""
load_dotenv() load_dotenv()
def setup_logging(self): def setup_logging(self):
"""
Method to configure logging settings.
"""
# Basic configuration for logging messages to console
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO) logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
logging.getLogger().setLevel(logging.INFO) logging.getLogger().setLevel(logging.INFO)
def load_plugins(self): 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): for plugin_file in os.listdir(self.PLUGINS_DIR):
if plugin_file.endswith(".py"): if plugin_file.endswith(".py"): # Checking if file is a Python file
plugin_name = os.path.splitext(plugin_file)[0] plugin_name = os.path.splitext(plugin_file)[0] # Extracting plugin name
try: try:
# Importing plugin module dynamically
module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}") module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}")
self.PLUGINS[plugin_name] = module self.PLUGINS[plugin_name] = module # Storing loaded plugin module
logging.info(f"Loaded plugin: {plugin_name}") logging.info(f"Loaded plugin: {plugin_name}") # Logging successful plugin loading
except Exception as e: except Exception as e:
logging.error(f"Error loading plugin {plugin_name}: {e}") logging.error(f"Error loading plugin {plugin_name}: {e}") # Logging error if plugin loading fails
def reload_plugins(self): def reload_plugins(self):
self.PLUGINS = {} """
# Unload modules from sys.modules Method to reload all plugins.
"""
self.PLUGINS = {} # Clearing loaded plugins dictionary
# Unloading modules from sys.modules
for plugin_name in list(sys.modules.keys()): for plugin_name in list(sys.modules.keys()):
if plugin_name.startswith(self.PLUGINS_DIR + "."): if plugin_name.startswith(self.PLUGINS_DIR + "."):
del sys.modules[plugin_name] del sys.modules[plugin_name] # Deleting plugin module from system modules
self.load_plugins() self.load_plugins() # Reloading plugins
def load_config(self): def load_config(self):
self.config = FunguyConfig() """
Method to load configuration settings.
"""
self.config = FunguyConfig() # Creating instance of FunguyConfig to load configuration
def load_disabled_plugins(self): def load_disabled_plugins(self):
"""
Method to load disabled plugins from configuration file.
"""
# Checking if configuration file exists
if os.path.exists('funguy.conf'): if os.path.exists('funguy.conf'):
# Loading configuration data from TOML file
with open('funguy.conf', 'r') as f: with open('funguy.conf', 'r') as f:
config_data = toml.load(f) config_data = toml.load(f)
# Extracting disabled plugins from configuration data
self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {}) self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {})
def save_disabled_plugins(self): def save_disabled_plugins(self):
"""
Method to save disabled plugins to configuration file.
"""
existing_config = {} existing_config = {}
# Checking if configuration file exists
if os.path.exists('funguy.conf'): if os.path.exists('funguy.conf'):
# Loading existing configuration data
with open('funguy.conf', 'r') as f: with open('funguy.conf', 'r') as f:
existing_config = toml.load(f) existing_config = toml.load(f)
# Updating configuration data with disabled plugins
existing_config['plugins'] = {'disabled': self.disabled_plugins} existing_config['plugins'] = {'disabled': self.disabled_plugins}
# Writing updated configuration data back to file
with open('funguy.conf', 'w') as f: with open('funguy.conf', 'w') as f:
toml.dump(existing_config, f) toml.dump(existing_config, f)
async def handle_commands(self, room, message): async def handle_commands(self, room, message):
match = botlib.MessageMatch(room, message, self.bot, self.config.prefix) """
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 match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
if str(message.sender) == self.config.admin_user: if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.reload_plugins() 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: 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
# Disable plugin command
if match.is_not_from_this_bot() and match.prefix() and match.command("disable"): if match.is_not_from_this_bot() and match.prefix() and match.command("disable"):
if str(message.sender) == self.config.admin_user: if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() args = match.args() # Getting command arguments
if len(args) != 2: 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: else:
plugin_name, room_id = args plugin_name, room_id = args # Extracting plugin name and room ID
await self.disable_plugin(room_id, plugin_name) 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: 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
# Enable plugin command
if match.is_not_from_this_bot() and match.prefix() and match.command("enable"): if match.is_not_from_this_bot() and match.prefix() and match.command("enable"):
if str(message.sender) == self.config.admin_user: if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
args = match.args() args = match.args() # Getting command arguments
if len(args) != 2: 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: else:
plugin_name, room_id = args plugin_name, room_id = args # Extracting plugin name and room ID
await self.enable_plugin(room_id, plugin_name) 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: 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
# Dispatching commands to plugins
for plugin_name, plugin_module in self.PLUGINS.items(): for plugin_name, plugin_module in self.PLUGINS.items():
if plugin_name not in self.disabled_plugins.get(room.room_id, []): if plugin_name not in self.disabled_plugins.get(room.room_id, []):
await plugin_module.handle_command(room, message, self.bot, self.config.prefix, self.config) await plugin_module.handle_command(room, message, self.bot, self.config.prefix, self.config)
# Rehash config command
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"): if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
if str(message.sender) == self.config.admin_user: if str(message.sender) == self.config.admin_user: # Checking if sender is admin user
self.rehash_config() self.rehash_config() # Rehashing configuration
await self.bot.api.send_text_message(room.room_id, "Config rehashed") await self.bot.api.send_text_message(room.room_id, "Config rehashed") # Sending success message
else: 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
def rehash_config(self): def rehash_config(self):
del self.config """
self.config = FunguyConfig() 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): 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: if room_id not in self.disabled_plugins:
self.disabled_plugins[room_id] = [] self.disabled_plugins[room_id] = [] # Creating entry for room ID if not exist
if plugin_name not in self.disabled_plugins[room_id]: if plugin_name not in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].append(plugin_name) self.disabled_plugins[room_id].append(plugin_name) # Adding plugin to list of disabled plugins for the room
self.save_disabled_plugins() self.save_disabled_plugins() # Saving disabled plugins to configuration file
async def enable_plugin(self, room_id, plugin_name): 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]: if room_id in self.disabled_plugins and plugin_name in self.disabled_plugins[room_id]:
self.disabled_plugins[room_id].remove(plugin_name) self.disabled_plugins[room_id].remove(plugin_name) # Removing plugin from list of disabled plugins for the room
self.save_disabled_plugins() self.save_disabled_plugins() # Saving disabled plugins to configuration file
def run(self): def run(self):
"""
Method to initialize and run the bot.
"""
# Retrieving Matrix credentials from environment variables
MATRIX_URL = os.getenv("MATRIX_URL") MATRIX_URL = os.getenv("MATRIX_URL")
MATRIX_USER = os.getenv("MATRIX_USER") MATRIX_USER = os.getenv("MATRIX_USER")
MATRIX_PASS = os.getenv("MATRIX_PASS") MATRIX_PASS = os.getenv("MATRIX_PASS")
creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS) creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS) # Creating credentials object
self.bot = botlib.Bot(creds, self.config) self.bot = botlib.Bot(creds, self.config) # Creating bot instance
# Defining listener for message events
@self.bot.listener.on_message_event @self.bot.listener.on_message_event
async def wrapper_handle_commands(room, message): async def wrapper_handle_commands(room, message):
await self.handle_commands(room, message) await self.handle_commands(room, message) # Calling handle_commands method for incoming messages
self.bot.run() self.bot.run() # Running the bot
if __name__ == "__main__": if __name__ == "__main__":
bot = FunguyBot() bot = FunguyBot() # Creating instance of FunguyBot
bot.run() bot.run() # Running the bot