Edm0nd/edmond/plugin.py

190 lines
7.1 KiB
Python

from dataclasses import dataclass
from pathlib import Path
class Plugin:
REQUIRED_CONFIGS = []
def __init__(self, bot):
self.bot = bot
self.name = self.__class__.__name__.lower()[:-6] # Remove "Plugin".
self.config = self.__get_config()
self.is_ready = 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):
"""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.
resources_dir = self.bot.config["resources_dir"]
for key, resource_path in config.get("resources", {}).items():
resource_path = Path(resources_dir) / resource_path
try:
with open(resource_path, "rt") as resource_file:
config[key] = [l.strip() for l in resource_file.readlines()]
except OSError:
self.bot.log_e(f"Could not load resource at {resource_path}.")
return config
def __check_config(self):
"""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} configuration.")
missing = True
return not missing
def get_runtime_value(self, key, default=None, ns=None):
"""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, value):
"""Set a value in the plugin runtime dict."""
self.bot.values[self.name][key] = value
def get_storage_value(self, key, default=None):
"""Get a value from the plugin persistent storage."""
return self.bot.storage.get(self.name, {}).get(key, default)
def set_storage_value(self, key, value):
"""Set a value in the plugin persistent storage."""
if self.name not in self.bot.storage:
self.bot.storage[self.name] = {key: value}
else:
self.bot.storage[self.name][key] = value
def append_storage_list_value(self, key, value):
"""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)
def remove_storage_list_value(self, key, value):
"""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)
def should_answer_question(self, message):
"""Store Question in object and return True if it should answer it."""
words = message.split()
# Is the message addressed to me?
if 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 q in self.config.get("questions", []):
if question.startswith(q):
self.question = Question(q, question[len(q):].strip())
self.bot.log_d(f"Answering question from plugin {self.name}.")
return True
return False
def should_handle_command(
self,
message,
no_content=False,
exclude_conditions=None,
):
"""Store Command in object and return True if it 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?
command = self.parse_command(message, no_content=no_content)
if not command:
return False
# Is it a command I can handle?
commands = self.config.get("commands", [])
if (
any(command.ident == c for c in commands) or
(no_content and any(command.ident.startswith(c) for c in commands))
):
self.command = command
self.bot.log_d(f"Processing command from plugin {self.name}.")
return True
return False
def parse_command(self, message, no_content=False):
"""Return a command ID if this message is a command."""
words = message.split()
command_suffix = self.config["command_suffix"]
if words[0].lower() in self.bot.names and words[-1] == command_suffix:
if no_content:
ident = " ".join(words[1:-1])
content = ""
else:
ident = words[1]
content = " ".join(words[2:-1])
return Command(ident, content)
def __respects_handling_conditions(self, exclude_conditions=None):
"""Check if question conditions are valid."""
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:
self.bot.log_d(
f"Handling condition {plugin_ns}.{condition_key} false."
)
return False
return True
def signal_failure(self, target):
"""Signal a plugin failure to target."""
self.bot.say(target, self.bot.config["error_message"])
@dataclass
class Question:
preambule: str
content: str
@dataclass
class Command:
ident: str
content: str