rollbot3/src/command_system.py

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 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)