Edm0nd/edmond/bot.py

import importlib
import json
import os
import time
import signal
import sys
import traceback
from pathlib import Path
from typing import Any, Iterable, Optional

import irc.client
from irc.client import Connection, Event, NickMask

from edmond.log import Logger
from edmond.plugin import Plugin


class Bot(irc.client.SimpleIRCClient, Logger):
    """Main class for the IRC bot: handles connection and manages available plugins."""

    CHANNELS_RUNTIME_KEY = "_channels"

    def __init__(self, config: dict, logger):
        super().__init__()
        self.config: dict = config
        self.logger = logger
        self.plugins: list[Plugin] = []
        self.values: dict[str, Any] = {}
        self.storage: dict[str, Any] = self.get_storage()
        self.done: bool = False

    @property
    def nick(self) -> str:
        """Nickname validated by the server, or the configured nick."""
        if self.connection.is_connected():
            return self.connection.get_nickname()
        return self.config["nick"]

    @property
    def names(self) -> Iterable[str]:
        """Collection of names the bot should identify with."""
        return (self.nick, *self.config["alternative_nicks"])

    @property
    def channels(self) -> list[str]:
        """List of joined channels."""
        if self.CHANNELS_RUNTIME_KEY not in self.values:
            self.values[self.CHANNELS_RUNTIME_KEY] = []
        return self.values[self.CHANNELS_RUNTIME_KEY]

    def get_storage(self) -> dict:
        """Load data from storage."""
        try:
            with open(self.config["storage_file"], "rt") as storage_file:
                storage = json.load(storage_file)
                self.log_d("Loaded storage file.")
                return storage
        except (OSError, json.decoder.JSONDecodeError) as exc:
            self.log_e(f"Could not load storage file: {exc}")
            self.log_w(
                "If it's not the first time Edm0nd is run, you may lose"
                " data when closing the program."
            )
            return {}

    def save_storage(self) -> None:
        """Save storage data to disk."""
        try:
            with open(self.config["storage_file"], "wt") as storage_file:
                json.dump(self.storage, storage_file, indent=2, sort_keys=True)
                self.log_d("Saved storage file.")
        except OSError as exc:
            self.log_e(f"Could not save storage file: {exc}")

    def on_welcome(self, connection: Connection, event: Event):
        """Handle a successful connection to a server."""
        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: Connection, event: Event):
        """Handle someone, possibly the bot, joining a channel."""
        if event.source.nick == self.nick:
            self.log_i(f"Joined {event.target}.")
            self.channels.append(event.target)
        self.run_plugin_callbacks(event)

    def on_part(self, connection: Connection, event: Event):
        """Handle someone, possibly the bot, leaving a channel."""
        if event.source.nick == self.nick:
            self.log_i(f"Left {event.target} (args: {event.arguments[0]}).")
            self.channels.remove(event.target)
        self.run_plugin_callbacks(event)

    def on_pubmsg(self, connection: Connection, event: Event):
        """Handle a message received in a channel."""
        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: Connection, event: Event):
        """Handle a message received privately, usually like a channel msg."""
        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 on_ping(self, connection: Connection, event: Event):
        """Handle a ping; can be used as a random event timer."""
        self.log_d(f"Received ping from {event.target}.")
        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.log_i("Connecting to server.")
        self.connect(self.config["host"], self.config["port"], self.nick)
        signal.signal(signal.SIGTERM, self.handle_sigterm)
        try:
            self.start()
        except KeyboardInterrupt:
            self.log_i("Caught keyboard interrupt.")
        except Exception as exc:
            self.log_c(f"Caught unhandled exception: {exc}")
            _, _, exc_traceback = sys.exc_info()
            for line in traceback.format_tb(exc_traceback):
                self.log_d(line.rstrip())
        finally:
            self.cleanup()

    def load_plugins(self):
        """Load all installed plugins."""
        self.log_i("Loading 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") and f != "__init__.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
            # Get plugin class name from its module name.
            class_name = (
                "".join(map(lambda w: w.capitalize(), plugin_name.split("_")))
                + "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 get_plugin(self, name: str) -> Optional[Plugin]:
        """Get a loaded plugin by its name (e.g. 'mood'), or None."""
        matching_plugins = filter(lambda plugin: plugin.name == name, self.plugins)
        return next(matching_plugins, None)

    def say(self, target: str, message: str) -> None:
        """Send message to target after a slight delay."""
        message = message.replace("\n", " ").replace("\r", " ")
        time.sleep(self.config["speak_delay"])
        self.log_d(f"Sending to {target}: {message}")
        try:
            if message.startswith("/me "):
                self.connection.action(target, message[4:])
            else:
                self.connection.privmsg(target, message)
        except irc.client.MessageTooLong:
            self.log_e("Could not send, message is too long.")

    def run_plugin_callbacks(self, event: Event) -> None:
        """Run appropriate callbacks for each plugin."""
        etype = event.type
        ready_plugins = filter(lambda p: p.is_ready, self.plugins)
        plugins = sorted(ready_plugins, key=lambda p: p.priority, reverse=True)
        for plugin in plugins:
            callbacks = plugin.callbacks
            if etype not in callbacks:
                continue
            if callbacks[etype](event):
                break

    def handle_sigterm(self, *args):
        """Handle SIGTERM (keyboard interrupt, systemd stop, etc)."""
        self.cleanup()
        exit("Exiting after received SIGTERM.")

    def cleanup(self) -> None:
        """Save the storage file and close the connection. Run only once."""
        if self.done:
            return
        self.log_i("Stopping Edmond.")
        self.save_storage()
        if self.connection.is_connected():
            self.connection.close()
        self.done = True