diff --git a/edmond/bot.py b/edmond/bot.py index b8c1220..67a236a 100644 --- a/edmond/bot.py +++ b/edmond/bot.py @@ -4,46 +4,48 @@ import os import time import signal from pathlib import Path +from typing import Any, Iterable, Optional import irc.client # type: ignore -from irc.client import NickMask +from irc.client import Connection, Event, NickMask from edmond.log import Logger +from edmond.plugin import Plugin class Bot(irc.client.SimpleIRCClient, Logger): CHANNELS_RUNTIME_KEY = "_channels" - def __init__(self, config, logger): + def __init__(self, config: dict, logger): super().__init__() - self.config = config + self.config: dict = config self.logger = logger - self.plugins = [] - self.values = {} - self.storage = self.__get_storage() - self.done = False + 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): + 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): + def names(self) -> Iterable[str]: """Collection of names the bot should identify with.""" return (self.nick, *self.config["alternative_nicks"]) @property - def channels(self): + 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): + def __get_storage(self) -> dict: """Load data from storage.""" try: with open(self.config["storage_file"], "rt") as storage_file: @@ -58,7 +60,7 @@ class Bot(irc.client.SimpleIRCClient, Logger): ) return {} - def __save_storage(self): + def __save_storage(self) -> None: """Save storage data to disk.""" try: with open(self.config["storage_file"], "wt") as storage_file: @@ -67,28 +69,28 @@ class Bot(irc.client.SimpleIRCClient, Logger): except OSError as exc: self.log_e(f"Could not save storage file: {exc}") - def on_welcome(self, connection, event): + 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, event): + 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, 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, event): + def on_pubmsg(self, connection: Connection, event: Event): """Handle a message received in a channel.""" channel = event.target nick = NickMask(event.source).nick @@ -96,7 +98,7 @@ class Bot(irc.client.SimpleIRCClient, Logger): self.log_d(f"Message in {channel} from {nick}: {message}") self.run_plugin_callbacks(event) - def on_privmsg(self, connection, 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 @@ -104,7 +106,7 @@ class Bot(irc.client.SimpleIRCClient, Logger): self.log_d(f"Private message from {nick} to {target}: {message}") self.run_plugin_callbacks(event) - def on_ping(self, connection, 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) @@ -149,11 +151,12 @@ class Bot(irc.client.SimpleIRCClient, Logger): self.values[plugin_name] = {} self.log_d(f"Loaded {class_name}.") - def get_plugin(self, name): + def get_plugin(self, name: str) -> Optional[Plugin]: """Get a loaded plugin by its name (e.g. 'mood'), or None.""" - return next(filter(lambda p: p.name == name, self.plugins), None) + matching_plugins = filter(lambda plugin: plugin.name == name, self.plugins) + return next(matching_plugins, None) - def say(self, target, message): + 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"]) @@ -166,11 +169,11 @@ class Bot(irc.client.SimpleIRCClient, Logger): except irc.client.MessageTooLong: self.log_e("Could not send, message is too long.") - def run_plugin_callbacks(self, event): + def run_plugin_callbacks(self, event: Event) -> None: """Run appropriate callbacks for each plugin.""" etype = event.type - plugins = filter(lambda p: p.is_ready, self.plugins) - plugins = sorted(plugins, key=lambda p: p.priority, reverse=True) + 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: @@ -183,7 +186,7 @@ class Bot(irc.client.SimpleIRCClient, Logger): self.cleanup() exit("Exiting after received SIGTERM.") - def cleanup(self): + def cleanup(self) -> None: """Save the storage file and close the connection. Run only once.""" if self.done: return diff --git a/edmond/config.py b/edmond/config.py index 7ab002e..ffd7076 100644 --- a/edmond/config.py +++ b/edmond/config.py @@ -1,7 +1,7 @@ import json -def load_config(config_path, logger=None): +def load_config(config_path: str, logger=None) -> dict: try: with open(config_path, "rt") as config_file: return json.load(config_file) diff --git a/edmond/plugin.py b/edmond/plugin.py index 72fdbaa..967d10b 100644 --- a/edmond/plugin.py +++ b/edmond/plugin.py @@ -1,7 +1,26 @@ from dataclasses import dataclass from pathlib import Path +from typing import Any, Optional -from irc.client import NickMask +from irc.client import NickMask # type: ignore + + +@dataclass +class Question: + preamble: str + content: str + + +@dataclass +class Command: + # Identifier as set in config. Set even when an alias has been used. + ident: str + # Content of the command when it has been parsed, empty str otherwise. + content: str + # Raw command content (minus name and suffix), always set. + raw: str + # Identifier matched, possibly an alias. Set only when matched. + matched: str = "" class Plugin: @@ -40,14 +59,16 @@ class Plugin: - -10: lowest, e.g. for random reactions usually with a very low rate """ - REQUIRED_CONFIGS = [] + REQUIRED_CONFIGS: list[str] = [] def __init__(self, bot): - self.bot = bot - self.name = self.__class__.__name__.lower()[:-6] # Remove "Plugin". - self.priority = 0 - self.config = self.__get_config() - self.is_ready = self.__check_config() + from edmond.bot import Bot + self.bot: Bot = bot + # self.name is the plugin name, lowercased, without the Plugin suffix. + self.name: str = self.__class__.__name__.lower()[:-6] + self.priority: int = 0 + self.config: dict = self.__get_config() + self.is_ready: bool = self.__check_config() @property def callbacks(self): @@ -58,7 +79,7 @@ class Plugin: if cb.startswith("on_") and callable(getattr(self, cb)) } - def __get_config(self): + def __get_config(self) -> dict: """Return the plugin section from the bot config plus common values.""" plugins_configs = self.bot.config["plugins"] config = plugins_configs["common"].copy() @@ -69,12 +90,14 @@ class Plugin: res_path = Path(res_dir) / res_path try: with open(res_path, "rt") as res_file: - config[key] = [l.strip() for l in res_file.readlines()] + config[key] = [ + line.strip() for line in res_file.readlines() + ] except OSError: self.bot.log_e(f"Could not load resource at {res_path}.") return config - def __check_config(self): + def __check_config(self) -> bool: """Return True if the plugin config is properly setup.""" missing = False for key in self.REQUIRED_CONFIGS: @@ -83,7 +106,9 @@ class Plugin: missing = True return not missing - def get_runtime_value(self, key, default=None, ns=None): + def get_runtime_value( + self, key: str, default: Any = None, ns: str = None + ) -> Any: """Get a value from the plugin runtime dict. This will get the value from the plugin namespace, but it is possible @@ -93,11 +118,11 @@ class Plugin: ns = self.name return self.bot.values[ns].get(key, default) - def set_runtime_value(self, key, value): + def set_runtime_value(self, key: str, value: Any) -> Any: """Set a value in the plugin runtime dict.""" self.bot.values[self.name][key] = value - def get_storage_value(self, key, default=None, ns=None): + def get_storage_value(self, key: str, default=None, ns: str = None) -> Any: """Get a value from the plugin persistent storage. This will get the value from the plugin namespace, but it is possible @@ -107,7 +132,7 @@ class Plugin: ns = self.name return self.bot.storage.get(ns, {}).get(key, default) - def set_storage_value(self, key, value, ns=None): + def set_storage_value(self, key: str, value: Any, ns: str = None) -> None: """Set a value in the plugin persistent storage.""" name = ns or self.name if name not in self.bot.storage: @@ -115,7 +140,7 @@ class Plugin: else: self.bot.storage[name][key] = value - def append_storage_list_value(self, key, value): + def append_storage_list_value(self, key: str, value: Any) -> None: """Append a value to a list in the plugin persistent storage.""" if self.name not in self.bot.storage: self.bot.storage[self.name] = {key: [value]} @@ -124,7 +149,7 @@ class Plugin: else: self.bot.storage[self.name][key].append(value) - def remove_storage_list_value(self, key, value): + def remove_storage_list_value(self, key: str, value: Any) -> None: """Remove a value from a persistent storage list.""" if ( self.name in self.bot.storage @@ -132,7 +157,7 @@ class Plugin: ): self.bot.storage[self.name][key].remove(value) - def should_read_message(self, message): + def should_read_message(self, message: str) -> Optional[str]: """Return a message content if it has been addressed to me, else None. If the message starts with one of the bot's names, the rest of the @@ -159,8 +184,9 @@ class Plugin: content = first_word_and_rest[1].strip() self.bot.log_i(f"Reading message from {self.name}: {content}") return content + return None - def should_answer_question(self, message): + def should_answer_question(self, message: str) -> bool: """Store Question in object and return True if I should answer it. To answer a question, the message must start with one of the bot's @@ -188,19 +214,18 @@ class Plugin: return True return False - def __save_question(self, question, matched, preamble): + def __save_question(self, question: str, matched: str, preamble: str): content = question[len(matched) :].strip() content = content.rstrip("?").rstrip() - question = Question(preamble, content) - self.question = question - self.bot.log_i(f"Answering from plugin {self.name}: {question}") + self.question = Question(preamble, content) + self.bot.log_i(f"Answering from plugin {self.name}: {self.question}") def should_handle_command( self, - message, - no_content=False, - exclude_conditions=None, - ): + message: str, + no_content: bool = False, + exclude_conditions: Optional[dict] = None, + ) -> bool: """Store Command in object and return True if I should handle it. If no_content is True, the command does not parse command contents and @@ -254,7 +279,9 @@ class Plugin: return True return False - def __parse_command(self, message, no_content=False): + def __parse_command( + self, message: str, no_content: bool = False + ) -> Optional[Command]: """Return a parsed Command if this message is a command, else None. The command raw field is always set. The ident and content fields are @@ -274,13 +301,14 @@ class Plugin: ident = words[1] content = " ".join(words[2:-1]) return Command(ident, content, raw) + return None - def __save_command(self, command): + def __save_command(self, command: Command): """Save command in instance for further processing by the plugin.""" self.command = command self.bot.log_i(f"Processing command from p. {self.name}: {command}") - def respects_handling_conditions(self, exclude_conditions=None): + def respects_handling_conditions(self, exclude_conditions: Optional[dict] = None): """Check if handling conditions are valid. Handling conditions can be specified by each plugin to create states in @@ -314,21 +342,3 @@ class Plugin: event.target = NickMask(event.source).nick return on_pubmsg(event) return False - - -@dataclass -class Question: - preamble: str - content: str - - -@dataclass -class Command: - # Identifier as set in config. Set even when an alias has been used. - ident: str - # Content of the command when it has been parsed, empty str otherwise. - content: str - # Raw command content (minus name and suffix), always set. - raw: str - # Identifier matched, possibly an alias. Set only when matched. - matched: str = "" diff --git a/edmond/utils.py b/edmond/utils.py index 1f41f49..410b9a2 100644 --- a/edmond/utils.py +++ b/edmond/utils.py @@ -1,13 +1,15 @@ import random +from typing import Optional import requests -def http_get(url): +def http_get(url: str) -> Optional[str]: response = requests.get(url) if response.status_code == 200: return response.text + return None -def proc(proba_percentage): +def proc(proba_percentage: int) -> bool: return random.random() < (proba_percentage / 100.0)