dece
e601a77f72
Plugins may want to write several values at once and skip the first writes to only save the values at the last call.
395 lines
16 KiB
Python
395 lines
16 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 if `is_ready` is True, else accessing its
|
|
content is undefined behaviour.
|
|
|
|
Plugins may use two types of special values, stored in two dicts: runtime
|
|
values and storage values. Runtime values can be accessed by other plugins
|
|
to get information about the runtime state of the bot. Currently it is used
|
|
to store if the bot is asleep or awake by the sleep plugin, and its current
|
|
mood by the mood plugin. They are lost when the bot shuts down. Storage
|
|
values on the other hand are written in a storage file everytime the object
|
|
is modified during runtime and also when the bot shuts down. Storage values
|
|
are stored as JSON, so values must be JSON-encodable. Storage values should
|
|
be used everytime information needs to persist over restarts instead of
|
|
external files. Access here means read and write, for both runtime and
|
|
storage values.
|
|
|
|
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: Optional[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`.
|
|
"""
|
|
name = ns or self.name
|
|
return self.bot.values[name].get(key, default)
|
|
|
|
def set_runtime_value(
|
|
self,
|
|
key: str,
|
|
value: Any,
|
|
ns: Optional[str] = None,
|
|
) -> None:
|
|
"""Set a value in the plugin runtime dict."""
|
|
name = ns or self.name
|
|
self.bot.values[name][key] = value
|
|
|
|
def get_storage_value(
|
|
self,
|
|
key: str,
|
|
default: Any = None,
|
|
ns: Optional[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`.
|
|
"""
|
|
name = ns or self.name
|
|
return self.bot.storage.get(name, {}).get(key, default)
|
|
|
|
def set_storage_value(
|
|
self,
|
|
key: str,
|
|
value: Any,
|
|
ns: Optional[str] = None,
|
|
skip_save: bool = False,
|
|
) -> 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
|
|
if not skip_save:
|
|
self.bot.save_storage()
|
|
|
|
def append_storage_list_value(
|
|
self,
|
|
key: str,
|
|
value: Any,
|
|
ns: str = None,
|
|
skip_save: bool = False,
|
|
) -> None:
|
|
"""Append a value to a list in the plugin persistent storage."""
|
|
name = ns or self.name
|
|
if name not in self.bot.storage:
|
|
self.bot.storage[name] = {key: [value]}
|
|
elif key not in self.bot.storage[name]:
|
|
self.bot.storage[name][key] = [value]
|
|
else:
|
|
self.bot.storage[name][key].append(value)
|
|
if not skip_save:
|
|
self.bot.save_storage()
|
|
|
|
def remove_storage_list_value(
|
|
self,
|
|
key: str,
|
|
value: Any,
|
|
ns: Optional[str] = None,
|
|
skip_save: bool = False,
|
|
) -> None:
|
|
"""Remove a value from a persistent storage list."""
|
|
name = ns or self.name
|
|
if name in self.bot.storage and key in self.bot.storage[name]:
|
|
self.bot.storage[name][key].remove(value)
|
|
if not skip_save:
|
|
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
|