import importlib import os import time from pathlib import Path import irc.client from irc.client import NickMask from edmond.log import Logger class Bot(irc.client.SimpleIRCClient, Logger): def __init__(self, config, logger): super().__init__() self.config = config self.logger = logger self.plugins = [] self.values = {} @property def nick(self): return self.config["nick"] @property def names(self): return (self.config["nick"], *self.config["alternative_nicks"]) def on_welcome(self, connection, event): self.log_i(f"Connected to server {event.source}.") self.run_plugin_callbacks(event) for channel in self.config["channels"]: connection.join(channel) def on_join(self, connection, event): self.log_i(f"Joined {event.target}.") self.run_plugin_callbacks(event) def on_part(self, connection, event): self.log_i(f"Left {event.target} (args: {event.arguments[0]}).") self.run_plugin_callbacks(event) def on_pubmsg(self, connection, event): channel = event.target nick = NickMask(event.source).nick message = event.arguments[0] self.log_d(f"Message in {channel} from {nick}: {message}") self.run_plugin_callbacks(event) def on_privmsg(self, connection, event): nick = NickMask(event.source).nick target = event.target message = event.arguments[0] self.log_d(f"Private message from {nick} to {target}: {message}") self.run_plugin_callbacks(event) def run(self): """Connect the bot to server, join channels and start responding.""" self.log_i("Starting Edmond.") self.load_plugins() self.connect(self.config["host"], self.config["port"], self.nick) try: self.start() except KeyboardInterrupt: self.log_i("Stopping Edmond.") def load_plugins(self): """Load all installed plugins.""" plugin_files = os.listdir(Path(__file__).parent / "plugins") plugin_names = map( lambda f: os.path.splitext(f)[0], filter(lambda f: f.endswith(".py"), plugin_files) ) for plugin_name in plugin_names: module = importlib.import_module(f"edmond.plugins.{plugin_name}") are_dependencies_ok = getattr(module, "DEPENDENCIES_FOUND", True) if not are_dependencies_ok: self.log_e(f"Dependencies not found for plugin {plugin_name}.") continue class_name = plugin_name.capitalize() + "Plugin" plugin_class = getattr(module, class_name) self.plugins.append(plugin_class(self)) self.values[plugin_name] = {} self.log_d(f"Loaded {class_name}.") def say(self, target, message): """Send message to target after a slight delay.""" time.sleep(self.config["speak_delay"]) self.log_d(f"Sending to {target}: {message}") if message.startswith("/me "): self.connection.action(target, message[4:]) else: self.connection.privmsg(target, message) def run_plugin_callbacks(self, event): etype = event.type for plugin in filter(lambda p: p.is_ready, self.plugins): callbacks = plugin.callbacks if etype not in callbacks: continue callbacks[etype](event)