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 def help_msg(self): return f"No help message provided for {self.command} command!" @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) def with_help(help_msg): def getter(self): return help_msg def decorator(cls): setattr(cls, "help_msg", getter) return cls return decorator