Source code for errbot

import argparse
import inspect
import logging
import re
import shlex
import sys
from functools import wraps
from typing import Any, Callable, List, Optional, Tuple

from .backends.base import AWAY, DND, OFFLINE, ONLINE, Message  # noqa
from .botplugin import (  # noqa
    BotPlugin,
    Command,
    CommandError,
    SeparatorArgParser,
    ShlexArgParser,
    ValidationException,
)
from .core_plugins.wsview import WebView, route
from .flow import FLOW_END, BotFlow, Flow, FlowRoot

__all__ = [
    "BotPlugin",
    "CommandError",
    "Command",
    "webhook",
    "webroute",
    "cmdfilter",
    "botcmd",
    "re_botcmd",
    "arg_botcmd",
    "botflow",
    "botmatch",
    "BotFlow",
    "FlowRoot",
    "Flow",
    "FLOW_END",
]

log = logging.getLogger(__name__)

webroute = (
    route  # this allows plugins to expose dynamic webpages on Errbot embedded webserver
)

# Some clients automatically convert consecutive dashes into a fancy
# hyphen, which breaks long-form arguments. Undo this conversion to
# provide a better user experience.
# Same happens with quotations marks, which are required for parsing
# complex strings in arguments
# Map of characters to sanitized equivalents
ARG_BOTCMD_CHARACTER_REPLACEMENTS = {"—": "--", "“": '"', "”": '"', "’": "'", "‘": "'"}


class ArgumentParseError(Exception):
    """Raised when ArgumentParser couldn't parse given arguments."""


class HelpRequested(Exception):
    """Signals that -h/--help was used and help should be displayed to the user."""


class ArgumentParser(argparse.ArgumentParser):
    """
    The python argparse.ArgumentParser, adapted for use within Err.
    """

    def error(self, message):
        raise ArgumentParseError(message)

    def print_help(self, file=None):
        # Implementation note: Only easy way to do this appears to be
        #   through raising an exception which we can catch later in
        #   a place where we have the ability to return a message to
        #   the user.
        raise HelpRequested()


def _tag_botcmd(
    func,
    hidden=None,
    name=None,
    split_args_with="",
    admin_only=False,
    historize=True,
    template=None,
    flow_only=False,
    _re=False,
    syntax=None,  # botcmd_only
    pattern=None,  # re_cmd only
    flags=0,  # re_cmd only
    matchall=False,  # re_cmd_only
    prefixed=True,  # re_cmd_only
    _arg=False,
    command_parser=None,  # arg_cmd only
    re_cmd_name_help=None,
):  # re_cmd_only
    """
    Mark a method as a bot command.
    """
    if not hasattr(func, "_err_command"):  # don't override generated functions
        func._err_command = True
        func._err_command_name = name or func.__name__
        func._err_command_split_args_with = split_args_with
        func._err_command_admin_only = admin_only
        func._err_command_historize = historize
        func._err_command_template = template
        func._err_command_syntax = syntax
        func._err_command_flow_only = flow_only
        func._err_command_hidden = hidden if hidden is not None else flow_only

        # re_cmd
        func._err_re_command = _re
        if _re:
            func._err_command_re_pattern = re.compile(pattern, flags=flags)
            func._err_command_matchall = matchall
            func._err_command_prefix_required = prefixed
            func._err_command_syntax = pattern
            func._err_command_re_name_help = re_cmd_name_help

        # arg_cmd
        func._err_arg_command = _arg
        if _arg:
            func._err_command_parser = command_parser
            # func._err_command_syntax is set at wrapping time.
    return func


[docs]def botcmd( *args, hidden: bool = None, name: str = None, split_args_with: str = "", admin_only: bool = False, historize: bool = True, template: str = None, flow_only: bool = False, syntax: str = None, ) -> Callable[[BotPlugin, Message, Any], Any]: """ Decorator for bot command functions :param hidden: Prevents the command from being shown by the built-in help command when `True`. :param name: The name to give to the command. Defaults to name of the function itself. :param split_args_with: Automatically split arguments on the given separator. Behaviour of this argument is identical to :func:`str.split()` :param admin_only: Only allow the command to be executed by admins when `True`. :param historize: Store the command in the history list (`!history`). This is enabled by default. :param template: The markdown template to use. :param syntax: The argument syntax you expect for example: '[name] <mandatory>'. :param flow_only: Flag this command to be available only when it is part of a flow. If True and hidden is None, it will switch hidden to True. This decorator should be applied to methods of :class:`~errbot.botplugin.BotPlugin` classes to turn them into commands that can be given to the bot. These methods are expected to have a signature like the following:: @botcmd def some_command(self, msg, args): pass The given `msg` will be the full message object that was received, which includes data like sender, receiver, the plain-text and html body (if applicable), etc. `args` will be a string or list (depending on your value of `split_args_with`) of parameters that were given to the command by the user. """ def decorator(func): return _tag_botcmd( func, _re=False, _arg=False, hidden=hidden, name=name or func.__name__, split_args_with=split_args_with, admin_only=admin_only, historize=historize, template=template, syntax=syntax, flow_only=flow_only, ) return decorator(args[0]) if args else decorator
[docs]def re_botcmd( *args, hidden: bool = None, name: str = None, admin_only: bool = False, historize: bool = True, template: str = None, pattern: str = None, flags: int = 0, matchall: bool = False, prefixed: bool = True, flow_only: bool = False, re_cmd_name_help: str = None, ) -> Callable[[BotPlugin, Message, Any], Any]: """ Decorator for regex-based bot command functions :param pattern: The regular expression a message should match against in order to trigger the command. :param flags: The `flags` parameter which should be passed to :func:`re.compile()`. This allows the expression's behaviour to be modified, such as making it case-insensitive for example. :param matchall: By default, only the first match of the regular expression is returned (as a `re.MatchObject`). When *matchall* is `True`, all non-overlapping matches are returned (as a list of `re.MatchObject` items). :param prefixed: Requires user input to start with a bot prefix in order for the pattern to be applied when `True` (the default). :param hidden: Prevents the command from being shown by the built-in help command when `True`. :param name: The name to give to the command. Defaults to name of the function itself. :param admin_only: Only allow the command to be executed by admins when `True`. :param historize: Store the command in the history list (`!history`). This is enabled by default. :param template: The template to use when using markdown output :param flow_only: Flag this command to be available only when it is part of a flow. If True and hidden is None, it will switch hidden to True. This decorator should be applied to methods of :class:`~errbot.botplugin.BotPlugin` classes to turn them into commands that can be given to the bot. These methods are expected to have a signature like the following:: @re_botcmd(pattern=r'^some command$') def some_command(self, msg, match): pass The given `msg` will be the full message object that was received, which includes data like sender, receiver, the plain-text and html body (if applicable), etc. `match` will be a :class:`re.MatchObject` containing the result of applying the regular expression on the user's input. """ def decorator(func): return _tag_botcmd( func, _re=True, _arg=False, hidden=hidden, name=name or func.__name__, admin_only=admin_only, historize=historize, template=template, pattern=pattern, flags=flags, matchall=matchall, prefixed=prefixed, flow_only=flow_only, re_cmd_name_help=re_cmd_name_help, ) return decorator(args[0]) if args else decorator
[docs]def botmatch(*args, **kwargs): """ Decorator for regex-based message match. :param *args: The regular expression a message should match against in order to trigger the command. :param flags: The `flags` parameter which should be passed to :func:`re.compile()`. This allows the expression's behaviour to be modified, such as making it case-insensitive for example. :param matchall: By default, only the first match of the regular expression is returned (as a `re.MatchObject`). When *matchall* is `True`, all non-overlapping matches are returned (as a list of `re.MatchObject` items). :param hidden: Prevents the command from being shown by the built-in help command when `True`. :param name: The name to give to the command. Defaults to name of the function itself. :param admin_only: Only allow the command to be executed by admins when `True`. :param historize: Store the command in the history list (`!history`). This is enabled by default. :param template: The template to use when using Markdown output. :param flow_only: Flag this command to be available only when it is part of a flow. If True and hidden is None, it will switch hidden to True. For example:: @botmatch(r'^(?:Yes|No)$') def yes_or_no(self, msg, match): pass """ def decorator(func, pattern): return _tag_botcmd( func, _re=True, _arg=False, prefixed=False, hidden=kwargs.get("hidden", None), name=kwargs.get("name", func.__name__), admin_only=kwargs.get("admin_only", False), flow_only=kwargs.get("flow_only", False), historize=kwargs.get("historize", True), template=kwargs.get("template", None), pattern=pattern, flags=kwargs.get("flags", 0), matchall=kwargs.get("matchall", False), ) if len(args) == 2: return decorator(*args) if len(args) == 1: return lambda f: decorator(f, args[0]) raise ValueError( "botmatch: You need to pass the pattern as parameter to the decorator." )
[docs]def arg_botcmd( *args, hidden: bool = None, name: str = None, admin_only: bool = False, historize: bool = True, template: str = None, flow_only: bool = False, unpack_args: bool = True, **kwargs, ) -> Callable[[BotPlugin, Message, Any], Any]: """ Decorator for argparse-based bot command functions https://docs.python.org/3/library/argparse.html This decorator creates an argparse.ArgumentParser and uses it to parse the commands arguments. This decorator can be used multiple times to specify multiple arguments. Any valid argparse.add_argument() parameters can be passed into the decorator. Each time this decorator is used it adds a new argparse argument to the command. :param hidden: Prevents the command from being shown by the built-in help command when `True`. :param name: The name to give to the command. Defaults to name of the function itself. :param admin_only: Only allow the command to be executed by admins when `True`. :param historize: Store the command in the history list (`!history`). This is enabled by default. :param template: The template to use when using markdown output :param flow_only: Flag this command to be available only when it is part of a flow. If True and hidden is None, it will switch hidden to True. :param unpack_args: Should the argparser arguments be "unpacked" and passed on the the bot command individually? If this is True (the default) you must define all arguments in the function separately. If this is False you must define a single argument `args` (or whichever name you prefer) to receive the result of `ArgumentParser.parse_args()`. This decorator should be applied to methods of :class:`~errbot.botplugin.BotPlugin` classes to turn them into commands that can be given to the bot. The methods will be called with the original msg and the argparse parsed arguments. These methods are expected to have a signature like the following (assuming `unpack_args=True`):: @arg_botcmd('value', type=str) @arg_botcmd('--repeat-count', dest='repeat', type=int, default=2) def repeat_the_value(self, msg, value=None, repeat=None): return value * repeat The given `msg` will be the full message object that was received, which includes data like sender, receiver, the plain-text and html body (if applicable), etc. `value` will hold the value passed in place of the `value` argument and `repeat` will hold the value passed in place of the `--repeat-count` argument. If you don't like this automatic *"unpacking"* of the arguments, you can use `unpack_args=False` like this:: @arg_botcmd('value', type=str) @arg_botcmd('--repeat-count', dest='repeat', type=int, default=2, unpack_args=False) def repeat_the_value(self, msg, args): return arg.value * args.repeat .. note:: The `unpack_args=False` only needs to be specified once, on the bottom `@args_botcmd` statement. """ argparse_args = args if len(args) >= 1 and callable(args[0]): argparse_args = args[1:] def decorator(func): if not hasattr(func, "_err_command"): err_command_parser = ArgumentParser( prog=name or func.__name__, description=func.__doc__, ) @wraps(func) def wrapper(self, msg, args): # Attempt to sanitize arguments of bad characters try: sanitizer_re = re.compile( "|".join( re.escape(ii) for ii in ARG_BOTCMD_CHARACTER_REPLACEMENTS ) ) args = sanitizer_re.sub( lambda mm: ARG_BOTCMD_CHARACTER_REPLACEMENTS[mm.group()], args ) args = shlex.split(args) parsed_args = err_command_parser.parse_args(args) except ArgumentParseError as e: yield f"I couldn't parse the arguments; {e}" yield err_command_parser.format_usage() return except HelpRequested: yield err_command_parser.format_help() return except ValueError as ve: yield f"I couldn't parse this command; {ve}" yield err_command_parser.format_help() return if unpack_args: func_args = [] func_kwargs = vars(parsed_args) else: func_args = [parsed_args] func_kwargs = {} if inspect.isgeneratorfunction(func): for reply in func(self, msg, *func_args, **func_kwargs): yield reply else: yield func(self, msg, *func_args, **func_kwargs) _tag_botcmd( wrapper, _re=False, _arg=True, hidden=hidden, name=name or wrapper.__name__, admin_only=admin_only, historize=historize, template=template, flow_only=flow_only, command_parser=err_command_parser, ) else: # the function has already been wrapped # alias it so we can update it's arguments below wrapper = func update_wrapper(wrapper, argparse_args, kwargs) return wrapper return decorator(args[0]) if callable(args[0]) else decorator
def update_wrapper(wrapper, argparse_args, kwargs): wrapper._err_command_parser.add_argument(*argparse_args, **kwargs) wrapper.__doc__ = wrapper._err_command_parser.format_help() fmt = wrapper._err_command_parser.format_usage() wrapper._err_command_syntax = fmt[ len("usage: ") + len(wrapper._err_command_parser.prog) + 1 : -1 ] def _tag_webhook( func: Callable, uri_rule: str, methods: Tuple[str], form_param: Optional[str], raw: bool, ) -> Callable: log.info(f"webhooks: Flag to bind {uri_rule} to {getattr(func, '__name__', func)}") func._err_webhook_uri_rule = uri_rule func._err_webhook_methods = methods func._err_webhook_form_param = form_param func._err_webhook_raw = raw return func def _uri_from_func(func: Callable) -> str: return r"/" + func.__name__
[docs]def webhook( *args, methods: Tuple[str] = ("POST", "GET"), form_param: str = None, raw: bool = False, ) -> Callable[[BotPlugin, Any], str]: """ Decorator for webhooks :param uri_rule: The URL to use for this webhook, as per Flask request routing syntax. For more information, see: * http://flask.pocoo.org/docs/1.0/quickstart/#routing * http://flask.pocoo.org/docs/1.0/api/#flask.Flask.route :param methods: A tuple of allowed HTTP methods. By default, only GET and POST are allowed. :param form_param: The key who's contents will be passed to your method's `payload` parameter. This is used for example when using the `application/x-www-form-urlencoded` mimetype. :param raw: When set to true, this overrides the request decoding (including form_param) and passes the raw http request to your method's `payload` parameter. The value of payload will be a Flask `Request <http://flask.pocoo.org/docs/1.0/api/#flask.Request>`_. This decorator should be applied to methods of :class:`~errbot.botplugin.BotPlugin` classes to turn them into webhooks which can be reached on Err's built-in webserver. The bundled *Webserver* plugin needs to be configured before these URL's become reachable. Methods with this decorator are expected to have a signature like the following:: @webhook def a_webhook(self, payload): pass """ if not args: # default uri_rule but with kwargs. return lambda func: _tag_webhook( func, _uri_from_func(func), methods=methods, form_param=form_param, raw=raw ) if isinstance(args[0], str): # first param is uri_rule. return lambda func: _tag_webhook( func, args[0] if args[0] == "/" else args[0].rstrip("/"), # trailing / is also be stripped on incoming. methods=methods, form_param=form_param, raw=raw, ) return _tag_webhook( args[0], # naked decorator so the first parameter is a function. _uri_from_func(args[0]), methods=methods, form_param=form_param, raw=raw, )
[docs]def cmdfilter(*args, **kwargs): """ Decorator for command filters. This decorator should be applied to methods of :class:`~errbot.botplugin.BotPlugin` classes to turn them into command filters. These filters are executed just before the execution of a command and provide the means to add features such as custom security, logging, auditing, etc. These methods are expected to have a signature and tuple response like the following:: @cmdfilter def some_filter(self, msg, cmd, args, dry_run): \"\"\" :param msg: The original chat message. :param cmd: The command name itself. :param args: Arguments passed to the command. :param dry_run: True when this is a dry-run. Dry-runs are performed by certain commands (such as !help) to check whether a user is allowed to perform that command if they were to issue it. If dry_run is True then the plugin shouldn't actually do anything beyond returning whether the command is authorized or not. \"\"\" # If wishing to block the incoming command: return None, None, None # Otherwise pass data through to the (potential) next filter: return msg, cmd, args Note that a cmdfilter plugin *could* modify `cmd` or `args` above and send that through in order to make it appear as if the user issued a different command. """ def decorate(func): if not hasattr( func, "_err_command_filter" ): # don't override generated functions func._err_command_filter = True func.catch_unprocessed = kwargs.get("catch_unprocessed", False) return func if len(args): return decorate(args[0]) return lambda func: decorate(func)
[docs]def botflow(*args, **kwargs): """ Decorator for flow of commands. TODO(gbin): example / docs """ def decorate(func): if not hasattr(func, "_err_flow"): # don't override generated functions func._err_flow = True return func if len(args): return decorate(args[0]) return lambda func: decorate(func)