rollbot3/lib/rollbot/plugins.py

import logging
import inspect
from functools import reduce

from .messaging import RollbotResponse


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 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)
        converters = []
        for p in sig.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.startswith("data") or p.endswith("data") or p in ("group_singleton", "singleton"):
                annot = fn.__annotations__.get(p, p)
                converters.append(lambda cmd, db, msg, sing_cls=annot: sing_cls.get_or_create(db, msg.group_id))
            else:
                raise ValueError(f"Illegal argument name {p} in decorated plugin {command_name}")

        def on_command_standin(self, db, msg):
            res = fn(*[c(self, db, msg) for c in converters])
            if isinstance(res, RollbotResponse):
                return res
            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)