Command system baseline

src/command_system.py

@@ -0,0 +1,169 @@
+import logging
+from dataclasses import dataclass
+from enum import Enum, auto
+
+from sqlalchemy.ext.declarative import declarative_base
+
+from config import GLOBAL_ADMINS, GROUP_ADMINS
+
+
+BANGS = ('!',)
+
+ModelBase = declarative_base()
+
+
+def _pop_arg(text):
+    if text is None:
+        return None, None
+    parts = text.split(maxsplit=1)
+    if len(parts) == 1:
+        return parts[0], None
+    return parts[0], parts[1].strip()
+
+
+@dataclass
+class RollbotMessage:
+    src: str
+    name: str
+    sender_id: str
+    group_id: str
+    message_id: str
+    message_txt: str
+
+    def __post_init__(self):
+        self.is_command = False
+        if len(self.message_txt) > 0 and self.message_txt[0] in BANGS:
+            cmd, raw = _pop_arg(self.message_txt[1:].strip())
+            if cmd is not None:
+                self.is_command = True
+                self.command = cmd.lower()
+                self.raw_args = raw
+
+        self.from_admin = self.sender_id is not None and \
+            self.sender_id in GLOBAL_ADMINS or (
+                self.group_id in GROUP_ADMINS and
+                self.sender_id in GROUP_ADMINS[self.group_id])
+
+    @staticmethod
+    def from_groupme(msg):
+        return RollbotMessage("GROUPME", msg["name"], msg["sender_id"], msg["group_id"], msg["id"], msg["text"].strip())
+
+    @staticmethod
+    def from_web(content):
+        content = content.strip()
+        if len(content) > 0 and content[0] not in BANGS:
+            content = BANGS[0] + content
+        # TODO should still assign an id...
+        return RollbotMessage("WEB_FRONTEND", "user", None, None, None, content)
+
+    def args(self, normalize=True):
+        arg, rest = _pop_arg(self.raw_args)
+        while arg is not None:
+            yield arg.lower() if normalize else arg
+            arg, rest = _pop_arg(rest)
+
+
+class RollbotFailure(Enum):
+    INVALID_COMMAND = auto()
+    MISSING_SUBCOMMAND = auto()
+    INVALID_SUBCOMMAND = auto()
+    INVALID_ARGUMENTS = auto()
+    SERVICE_DOWN = auto()
+    PERMISSIONS = auto()
+
+
+_RESPONSE_TEMPLATE = """Response{
+    Original Message: %s,
+    Text Response: %s,
+    Image Response: %s,
+    Respond: %s,
+    Failure Reason: %s,
+    Failure Notes: %s
+}"""
+
+
+@dataclass
+class RollbotResponse:
+    msg: RollbotMessage
+    txt: str = None
+    img: str = None
+    respond: bool = True
+    failure: RollbotFailure = None
+    debugging: dict = None
+
+    def __post_init__(self):
+        self.info = _RESPONSE_TEMPLATE % (self.msg, self.txt, self.img, self.respond, self.failure, self.debugging)
+        self.is_success = self.failure is None
+        if self.failure is None:
+            self.failure_msg = None
+        elif self.failure == RollbotFailure.INVALID_COMMAND:
+            self.failure_msg = "Sorry - I don't think I understand the command '!%s'... " % self.msg.command \
+                    + "I'll try to figure it out and get back to you!"
+        elif self.failure == RollbotFailure.MISSING_SUBCOMMAND:
+            self.failure_msg = "Sorry - !%s requires a sub-command." % self.msg.command
+        elif self.failure == RollbotFailure.INVALID_SUBCOMMAND:
+            self.failure_msg = "Sorry - the sub-command you used for %s was not valid." % self.msg.command
+        elif self.failure == RollbotFailure.INVALID_ARGUMENTS:
+            self.failure_msg = "Sorry - %s cannot use those arguments!" % self.msg.command
+        elif self.failure == RollbotFailure.SERVICE_DOWN:
+            self.failure_msg = "Sorry - %s relies on a service I couldn't reach!" % self.msg.command
+        elif self.failure == RollbotFailure.PERMISSIONS:
+            self.failure_msg = "Sorry - you don't have permission to use that command or sub-command in this chat!"
+
+        if self.debugging is not None and "explain" in self.debugging:
+            self.failure_msg += " " + self.debugging["explain"]
+
+
+class RollbotPlugin:
+    def __init__(self, command, logger=logging.getLogger(__name__)):
+        self.command = command
+        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 wrap_simple(fn):
+    def wrapper(db, message):
+        return RollbotResponse(message, txt=fn(db, message.raw_args))
+    wrapper.__name__ = fn.__name__
+    return wrapper
+
+
+def wrap_simple_no_db(fn):
+    def wrapper(db, message):
+        return RollbotResponse(message, txt=fn(message.raw_args))
+    wrapper.__name__ = fn.__name__
+    return wrapper
+
+
+def as_plugin(command):
+    if isinstance(command, str):
+        command_name = command
+    else:
+        command_name = command.__name__
+
+    def init_standin(self, logger=logging.getLogger(__name__)):
+        RollbotPlugin.__init__(self, command_name, logger)
+
+    def decorator(fn):
+        return type(
+            f"AutoGenerated`{fn.__name__}`Command",
+            (RollbotPlugin,),
+            dict(
+                __init__=init_standin,
+                on_command=(lambda _, *a: fn(*a))
+            )
+        )
+
+    if isinstance(command, str):
+        return decorator
+    else:
+        return decorator(command)