155 lines
6.1 KiB
Python
Executable File
155 lines
6.1 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
"""
|
|
Funguy Bot Class
|
|
"""
|
|
|
|
import os
|
|
import logging
|
|
import importlib
|
|
import simplematrixbotlib as botlib
|
|
from dotenv import load_dotenv
|
|
import time
|
|
import sys
|
|
import toml
|
|
|
|
from plugins.config import FunguyConfig
|
|
|
|
class FunguyBot:
|
|
|
|
def __init__(self):
|
|
self.PLUGINS_DIR = "plugins"
|
|
self.PLUGINS = {}
|
|
self.config = None
|
|
self.bot = None
|
|
self.disabled_plugins = {} # Dictionary to store disabled plugins for each room
|
|
self.load_dotenv()
|
|
self.setup_logging()
|
|
self.load_plugins()
|
|
self.load_config()
|
|
self.load_disabled_plugins()
|
|
|
|
def load_dotenv(self):
|
|
load_dotenv()
|
|
|
|
def setup_logging(self):
|
|
logging.basicConfig(format='%(asctime)s - %(levelname)s - %(message)s', level=logging.INFO)
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
|
|
def load_plugins(self):
|
|
for plugin_file in os.listdir(self.PLUGINS_DIR):
|
|
if plugin_file.endswith(".py"):
|
|
plugin_name = os.path.splitext(plugin_file)[0]
|
|
try:
|
|
module = importlib.import_module(f"{self.PLUGINS_DIR}.{plugin_name}")
|
|
self.PLUGINS[plugin_name] = module
|
|
logging.info(f"Loaded plugin: {plugin_name}")
|
|
except Exception as e:
|
|
logging.error(f"Error loading plugin {plugin_name}: {e}")
|
|
|
|
def reload_plugins(self):
|
|
self.PLUGINS = {}
|
|
# Unload modules from sys.modules
|
|
for plugin_name in list(sys.modules.keys()):
|
|
if plugin_name.startswith(self.PLUGINS_DIR + "."):
|
|
del sys.modules[plugin_name]
|
|
self.load_plugins()
|
|
|
|
def load_config(self):
|
|
self.config = FunguyConfig()
|
|
|
|
def load_disabled_plugins(self):
|
|
if os.path.exists('funguy.conf'):
|
|
with open('funguy.conf', 'r') as f:
|
|
config_data = toml.load(f)
|
|
self.disabled_plugins = config_data.get('plugins', {}).get('disabled', {})
|
|
|
|
def save_disabled_plugins(self):
|
|
existing_config = {}
|
|
if os.path.exists('funguy.conf'):
|
|
with open('funguy.conf', 'r') as f:
|
|
existing_config = toml.load(f)
|
|
existing_config['plugins'] = {'disabled': self.disabled_plugins}
|
|
with open('funguy.conf', 'w') as f:
|
|
toml.dump(existing_config, f)
|
|
|
|
async def handle_commands(self, room, message):
|
|
match = botlib.MessageMatch(room, message, self.bot, self.config.prefix)
|
|
if match.is_not_from_this_bot() and match.prefix() and match.command("reload"):
|
|
if str(message.sender) == self.config.admin_user:
|
|
self.reload_plugins()
|
|
await self.bot.api.send_text_message(room.room_id, "Plugins reloaded successfully")
|
|
else:
|
|
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
|
|
|
|
if match.is_not_from_this_bot() and match.prefix() and match.command("disable"):
|
|
if str(message.sender) == self.config.admin_user:
|
|
args = match.args()
|
|
if len(args) != 2:
|
|
await self.bot.api.send_text_message(room.room_id, "Usage: !disable <plugin> <room_id>")
|
|
else:
|
|
plugin_name, room_id = args
|
|
await self.disable_plugin(room_id, plugin_name)
|
|
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' disabled for room '{room_id}'")
|
|
else:
|
|
await self.bot.api.send_text_message(room.room_id, "You are not authorized to disable plugins.")
|
|
|
|
if match.is_not_from_this_bot() and match.prefix() and match.command("enable"):
|
|
if str(message.sender) == self.config.admin_user:
|
|
args = match.args()
|
|
if len(args) != 2:
|
|
await self.bot.api.send_text_message(room.room_id, "Usage: !enable <plugin> <room_id>")
|
|
else:
|
|
plugin_name, room_id = args
|
|
await self.enable_plugin(room_id, plugin_name)
|
|
await self.bot.api.send_text_message(room.room_id, f"Plugin '{plugin_name}' enabled for room '{room_id}'")
|
|
else:
|
|
await self.bot.api.send_text_message(room.room_id, "You are not authorized to enable plugins.")
|
|
|
|
for plugin_name, plugin_module in self.PLUGINS.items():
|
|
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)
|
|
|
|
if match.is_not_from_this_bot() and match.prefix() and match.command("rehash"):
|
|
if str(message.sender) == self.config.admin_user:
|
|
self.rehash_config()
|
|
await self.bot.api.send_text_message(room.room_id, "Config rehashed")
|
|
else:
|
|
await self.bot.api.send_text_message(room.room_id, "You are not authorized to reload plugins.")
|
|
|
|
def rehash_config(self):
|
|
del self.config
|
|
self.config = FunguyConfig()
|
|
|
|
async def disable_plugin(self, room_id, plugin_name):
|
|
if room_id not in self.disabled_plugins:
|
|
self.disabled_plugins[room_id] = []
|
|
if plugin_name not in self.disabled_plugins[room_id]:
|
|
self.disabled_plugins[room_id].append(plugin_name)
|
|
self.save_disabled_plugins()
|
|
|
|
async def enable_plugin(self, room_id, plugin_name):
|
|
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.save_disabled_plugins()
|
|
|
|
def run(self):
|
|
MATRIX_URL = os.getenv("MATRIX_URL")
|
|
MATRIX_USER = os.getenv("MATRIX_USER")
|
|
MATRIX_PASS = os.getenv("MATRIX_PASS")
|
|
creds = botlib.Creds(MATRIX_URL, MATRIX_USER, MATRIX_PASS)
|
|
self.bot = botlib.Bot(creds, self.config)
|
|
|
|
@self.bot.listener.on_message_event
|
|
async def wrapper_handle_commands(room, message):
|
|
await self.handle_commands(room, message)
|
|
|
|
self.bot.run()
|
|
|
|
if __name__ == "__main__":
|
|
bot = FunguyBot()
|
|
bot.run()
|
|
|
|
|
|
|