190 lines
7.1 KiB
Python
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
|