351 lines
14 KiB
Python
351 lines
14 KiB
Python
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
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:
|
|
"""Base class for the bot plugins.
|
|
|
|
This class provides a lot of tools to facilitate the implementation of
|
|
plugins. Callbacks for all events handled by the IRC library can be
|
|
implemented here (if the Bot class supports it). Most bot features are put
|
|
in plugins to avoid cluttering the Bot class, and no plugins should be
|
|
required for the bot to run; plugins can depend on each other though.
|
|
|
|
A plugin loads his basic information from the config file. It can use the
|
|
bot's runtime values facilities to make runtime values available to other
|
|
plugins. It can also save data using the Bot's storage feature to be
|
|
available after a restart.
|
|
|
|
Initialisation should be very fast, no network connections or anything.
|
|
They are initialised before connecting to the server, so their `is_ready`
|
|
flag is set at that point. The loading order is more or less random, so a
|
|
plugin cannot assume another has been loaded during initialisation. If it
|
|
wants to interact with another plugin, the earliest point to do that is in
|
|
the on_welcome callback which is called after connecting to a server, and
|
|
can disable itself by setting its own `is_ready` flag to false.
|
|
|
|
A plugin can access its config once the base `__init__` has been called.
|
|
The configuration is valid only is `is_ready` is True, else accessing its
|
|
content is undefined behaviour.
|
|
|
|
Plugins can have priorities and calling their callbacks will respect it.
|
|
For now these levels are used:
|
|
- 0: default
|
|
- -3: low, misc parsing of messages, answer to various messages
|
|
- -6: handling of unknown commands
|
|
- -7: handling of unknown questions
|
|
- -8: handling any message that might match something
|
|
- -10: lowest, e.g. for random reactions usually with a very low rate
|
|
"""
|
|
|
|
REQUIRED_CONFIGS: list[str] = []
|
|
|
|
def __init__(self, bot):
|
|
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):
|
|
"""List of callback types available for this plugin."""
|
|
return {
|
|
cb[3:]: getattr(self, cb)
|
|
for cb in dir(self)
|
|
if cb.startswith("on_") and callable(getattr(self, cb))
|
|
}
|
|
|
|
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()
|
|
config.update(plugins_configs.get(self.name, {}))
|
|
# Load resources as config values.
|
|
res_dir = self.bot.config["resources_dir"]
|
|
for key, res_path in config.get("resources", {}).items():
|
|
res_path = Path(res_dir) / res_path
|
|
try:
|
|
with open(res_path, "rt") as res_file:
|
|
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) -> bool:
|
|
"""Return True if the plugin config is properly setup."""
|
|
missing = False
|
|
for key in self.REQUIRED_CONFIGS:
|
|
if key not in self.config:
|
|
self.bot.log_e(f"Missing '{key}' in {self.name} config.")
|
|
missing = True
|
|
return not missing
|
|
|
|
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
|
|
to get runtime values from other plugins using their name as `ns`.
|
|
"""
|
|
if ns is None:
|
|
ns = self.name
|
|
return self.bot.values[ns].get(key, default)
|
|
|
|
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: 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
|
|
to get storage values from other plugins using their name as `ns`.
|
|
"""
|
|
if ns is None:
|
|
ns = self.name
|
|
return self.bot.storage.get(ns, {}).get(key, default)
|
|
|
|
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:
|
|
self.bot.storage[name] = {key: value}
|
|
else:
|
|
self.bot.storage[name][key] = value
|
|
self.bot.save_storage()
|
|
|
|
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]}
|
|
elif key not in self.bot.storage[self.name]:
|
|
self.bot.storage[self.name][key] = [value]
|
|
else:
|
|
self.bot.storage[self.name][key].append(value)
|
|
self.bot.save_storage()
|
|
|
|
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
|
|
and key in self.bot.storage[self.name]
|
|
):
|
|
self.bot.storage[self.name][key].remove(value)
|
|
self.bot.save_storage()
|
|
|
|
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
|
|
message is returned. It might be an empty str if the message was only
|
|
someone saying a bot's name. In other cases, return None.
|
|
|
|
Note that you do not need to use this method for command/question
|
|
plugins, the next methods already check for this and do more precise
|
|
parsing of the message. This is rather for plugins reacting to a
|
|
message addressed to the bot that are neither commands or questions.
|
|
"""
|
|
first_word_and_rest = message.split(maxsplit=1)
|
|
num_parts = len(first_word_and_rest)
|
|
if num_parts == 0:
|
|
return None
|
|
# Are the config conditions to answer a question valid?
|
|
elif not self.respects_handling_conditions():
|
|
return None
|
|
# Is the message addressed to me?
|
|
elif first_word_and_rest[0] in self.bot.names:
|
|
if num_parts == 1:
|
|
content = ""
|
|
else:
|
|
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: 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
|
|
names and optionally end with one or more question marks (they are
|
|
discarded). The bot checks that answering conditions are respected, and
|
|
they cannot be bypassed as with commands. A Question created is checked
|
|
from the plugin's registered questions and stored in the instance.
|
|
"""
|
|
words = message.split()
|
|
# Is the message addressed to me?
|
|
if len(words) == 0 or words[0].lower() not in self.bot.names:
|
|
return False
|
|
|
|
# Are the config conditions to answer a question valid?
|
|
if not self.respects_handling_conditions():
|
|
return False
|
|
|
|
# Is it a question I can answer?
|
|
question = message[len(words[0]) :].strip()
|
|
for preamble in self.config.get("questions", []):
|
|
aliases = self.config.get("aliases", {}).get(preamble, [])
|
|
for q in (preamble, *aliases):
|
|
if question.startswith(q):
|
|
self.__save_question(question, q, preamble)
|
|
return True
|
|
return False
|
|
|
|
def __save_question(self, question: str, matched: str, preamble: str):
|
|
content = question[len(matched) :].strip()
|
|
content = content.rstrip("?").rstrip()
|
|
self.question = Question(preamble, content)
|
|
self.bot.log_i(f"Answering from plugin {self.name}: {self.question}")
|
|
|
|
def should_handle_command(
|
|
self,
|
|
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
|
|
put all the command message (without suffix) to the command identifier.
|
|
This is useful for commands that have multiple words in their
|
|
identifier and do not have any content, or when the content parsing
|
|
should be done by the plugin itself.
|
|
|
|
If exclude_conditions is a collection of plugin namespaces, the
|
|
conditions of these plugins are not tested. A basic scenario is the
|
|
wakeup command from sleep plugin that requires excluding its own
|
|
condition (i.e. do not handle commands when sleeping).
|
|
"""
|
|
# Are the config conditions to handle a command valid?
|
|
if not self.respects_handling_conditions(exclude_conditions):
|
|
return False
|
|
|
|
# Is it a valid command?
|
|
parsed_command = self.__parse_command(message, no_content=no_content)
|
|
if not parsed_command:
|
|
return False
|
|
|
|
# Is it a command I can handle?
|
|
available_commands = self.config.get("commands", [])
|
|
aliases = self.config.get("aliases", {})
|
|
for ident in available_commands:
|
|
# Match commands differently according to no_content. If no_content
|
|
# is True, check the parsed command (pc) raw data as a string that
|
|
# may contain the available identifier (ai) at its beginning.
|
|
# If no_content is False (default), simply compare the parsed
|
|
# identifier with available identifiers.
|
|
if no_content:
|
|
matches = lambda pc, ai: (
|
|
pc.raw == ai or pc.raw.startswith(ai + " ")
|
|
)
|
|
else:
|
|
matches = lambda pc, ai: pc.ident == ai
|
|
# First case: the command identifier has been used.
|
|
if matches(parsed_command, ident):
|
|
parsed_command.ident = ident
|
|
parsed_command.matched = ident
|
|
self.__save_command(parsed_command)
|
|
return True
|
|
# Second case: an alias of the identifier has been used.
|
|
ident_aliases = aliases.get(ident, [])
|
|
for alias in ident_aliases:
|
|
if matches(parsed_command, alias):
|
|
parsed_command.ident = ident
|
|
parsed_command.matched = alias
|
|
self.__save_command(parsed_command)
|
|
return True
|
|
return 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
|
|
empty when no_content is True. The match field is never set by this
|
|
method.
|
|
"""
|
|
words = message.split()
|
|
if len(words) == 0:
|
|
return None
|
|
command_suffix = self.config["command_suffix"]
|
|
if words[0].lower() in self.bot.names and words[-1] == command_suffix:
|
|
raw = " ".join(words[1:-1])
|
|
if no_content:
|
|
ident = ""
|
|
content = ""
|
|
else:
|
|
ident = words[1]
|
|
content = " ".join(words[2:-1])
|
|
return Command(ident, content, raw)
|
|
return None
|
|
|
|
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: Optional[dict] = None
|
|
):
|
|
"""Check if handling conditions are valid.
|
|
|
|
Handling conditions can be specified by each plugin to create states in
|
|
which handling messages can be disabled. It is currently only used by
|
|
the sleep plugin (sleeping must disable most interactions) but every
|
|
plugin is free to introduce their own condition.
|
|
"""
|
|
conditions = self.config.get("handling_conditions", {})
|
|
for plugin_ns, plugin_conditions in conditions.items():
|
|
if exclude_conditions and plugin_ns in exclude_conditions:
|
|
continue
|
|
for condition_key, condition_value in plugin_conditions.items():
|
|
value = self.get_runtime_value(condition_key, ns=plugin_ns)
|
|
if condition_value != value:
|
|
return False
|
|
return True
|
|
|
|
def signal_failure(self, target):
|
|
"""Signal a plugin failure to target."""
|
|
self.bot.say(target, self.bot.config["error_message"])
|
|
|
|
def on_privmsg(self, event):
|
|
"""Default behaviour on privmsg is to answer like a pubmsg.
|
|
|
|
Because most plugins assume messages are sent publicly, they send their
|
|
replies to `event.target` so we replace it with the event source nick.
|
|
If it causes an issue with a plugin requiring the original target even
|
|
on private message, override this method.
|
|
"""
|
|
if on_pubmsg := getattr(self, "on_pubmsg", None):
|
|
event.target = NickMask(event.source).nick
|
|
return on_pubmsg(event)
|
|
return False
|