dece/Edm0nd
1
0
Fork 0
Code Issues 2 Pull requests Releases Activity

add some typings to Bot, Plugin and more

Browse source
This commit is contained in:
dece 2022-08-15 13:19:48 +02:00
parent bbacc2fbef
commit 24bf8d02d0
4 changed files with 90 additions and 75 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

53
edmond/bot.py
View file

@ -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

2
edmond/config.py
View file

@ -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)

104
edmond/plugin.py
View file

@ -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 = ""

6
edmond/utils.py
View file

@ -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)