rollbot3/lib/rollbot/plugins.py

import logging
import inspect
from functools import reduce

from .messaging import RollbotResponse, RollbotMessage, RollbotFailure


class RollbotPlugin:
    def __init__(self, command, bot, logger=logging.getLogger(__name__)):
        self.command = command
        self.bot = bot
        self.logger = logger
        self.logger.info(f"Intializing {type(self).__name__} matching {command}")

    def on_start(self, db):
        self.logger.info(f"No on_start initialization of {type(self).__name__}")

    def on_shutdown(self, db):
        self.logger.info(f"No on_shutdown de-initialization of {type(self).__name__}")

    def on_command(self, db, message):
        raise NotImplementedError

    @staticmethod
    def find_all_plugins():
        def get_subclasses(clazz):
            s = set(clazz.__subclasses__())
            return s | reduce(set.union, (get_subclasses(sc) for sc in s), set())
        return get_subclasses(RollbotPlugin)


def get_converters(parameters, annotations):
    converters = []
    for p in parameters:
        if p in ("msg", "message", "_msg"):
            converters.append(lambda cmd, db, msg: msg)
        elif p in ("db", "database"):
            converters.append(lambda cmd, db, msg: db)
        elif p in ("log", "logger"):
            converters.append(lambda cmd, db, msg: cmd.logger)
        elif p in ("bot", "rollbot"):
            converters.append(lambda cmd, db, msg: cmd.bot)
        elif p in ("subc", "subcommand"):
            converters.append(lambda cmd, db, msg: RollbotMessage.from_subcommand(msg))
        elif p.startswith("data") or p.endswith("data"):
            annot = annotations.get(p, p)
            converters.append(lambda cmd, db, msg, sing_cls=annot: sing_cls.get_or_create(db, msg))
        else:
            raise ValueError(p)
    return converters


def as_plugin(command):
    if isinstance(command, str):
        command_name = command
    else:
        command_name = command.__name__

    def init_standin(self, bot, logger=logging.getLogger(__name__)):
        RollbotPlugin.__init__(self, command_name, bot, logger=logger)

    def decorator(fn):
        sig = inspect.signature(fn)
        try:
            converters = get_converters(sig.parameters, fn.__annotations__)
        except ValueError as ve:
            raise ValueError(f"Illegal argument name {str(ve)} in decorated plugin {command_name}")

        def on_command_standin(self, db, msg):
            res = fn(*[c(self, db, msg) for c in converters])
            if res is None:
                return RollbotResponse(msg, respond=False)
            elif isinstance(res, RollbotResponse):
                return res
            elif isinstance(res, RollbotFailure):
                return RollbotResponse(msg, failure=res, debugging=res.get_debugging())
            else:
                return RollbotResponse(msg, txt=str(res))

        return type(
            f"AutoGenerated`{command_name}`Command",
            (RollbotPlugin,),
            dict(
                __init__=init_standin,
                on_command=on_command_standin,
            )
        )

    if isinstance(command, str):
        return decorator
    else:
        return decorator(command)