Merge pull request 'command-loader' (#6) from command-loader into master

Reviewed-on: #6

commands/command_base.py

@@ -0,0 +1,35 @@
+from abc import ABCMeta, abstractmethod
+from enum import Enum, auto
+
+
+class AbstractCommand(metaclass=ABCMeta):
+    """
+    This is the base class for commands. In order to load a command a few conditions must be met:
+    1) The class name MUST begin with 'Command' i.e. CommandTTS, CommandBan, etc...
+    2) the class MUST extend AbstractCommand
+
+    Generally, it would be advisable to define the command (something like !so, !tts, !songrequest) as a variable of the
+    class and to then call super().__init__(command)
+    """
+
+    class CommandType(Enum):
+        NONE = auto()
+        TWITCH = auto()
+        DISCORD = auto()
+
+    def __init__(self, command: str, n_args: int = 0, command_type=CommandType.NONE):
+        self.command = command
+        self.n_args = n_args
+        self.command_type = command_type
+
+    # no touch!
+    def get_args(self, text: str) -> list:
+        return text.split(" ")[0:self.n_args + 1]
+
+    # no touch!
+    def get_command(self) -> str:
+        return self.command
+
+    @abstractmethod
+    def do_command(self, bot, twitch_message):
+        pass

commands/implemented/command_test.py

@@ -0,0 +1,17 @@
+from abc import ABCMeta
+
+from commands.command_base import AbstractCommand
+
+
+class CommandTest(AbstractCommand, metaclass=ABCMeta):
+    """
+    this is a test command. and a poor excuse for a git commit.
+    """
+    command = "!test"
+
+    def __init__(self):
+        super().__init__(CommandTest.command, command_type=AbstractCommand.CommandType.TWITCH)
+
+    def do_command(self, bot, twitch_message):
+        print("!test Detected")
+        twitch_message.chat.send("test acknowledged")

commands/implemented/command_tts.py

@@ -0,0 +1,20 @@
+from abc import ABCMeta
+
+from commands.command_base import AbstractCommand
+
+
+
+class CommandTTS(AbstractCommand, metaclass=ABCMeta):
+    command = "!tts"
+
+    def __init__(self):
+        super().__init__(CommandTTS.command, n_args=1, command_type=AbstractCommand.CommandType.TWITCH)
+
+    def do_command(self, bot, twitch_message):
+        args = self.get_args(twitch_message.text)
+        if args[1] == "start":
+            bot.send_message("tts activated on #%s" % twitch_message.channel)
+            bot.tts_enabled = True
+        elif args[1] == "stop":
+            bot.send_message("tts deactivated")
+            bot.tts_enabled = False

commands/loader.py

@@ -0,0 +1,74 @@
+import importlib
+import importlib.util
+import inspect
+import os
+import sys
+from typing import Dict
+
+from commands.command_base import AbstractCommand
+
+
+def load_commands() -> Dict[str, AbstractCommand]:
+    commands = compile_and_load()
+    return commands
+
+
+def compile_and_load_file(path: str) -> (str, AbstractCommand):
+    module_name = os.path.split(path)[1].replace(".py", "")
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.load_module(module_name)
+
+    for name, obj in inspect.getmembers(module):
+        if inspect.isclass(obj) and name.startswith("Command"):
+            command_inst = obj()
+            print("Successfully loaded %s: %s" % (name, command_inst.get_command()))
+            return command_inst.get_command(), command_inst
+    return "", None
+
+
+def compile_and_load() -> Dict[str, AbstractCommand]:
+    dic = {}
+    implementations = get_implementations_dir()
+    for dirName, subdirList, fileList in os.walk(implementations):
+        for file in fileList:
+            name = os.path.join(dirName, file)
+            print("compiling %s" % name)
+            name, command = compile_and_load_file(name)
+            if command is not None:
+                dic[name] = command
+        break
+    return dic
+
+
+def get_base_dir() -> str:
+    cwd = os.getcwd()
+    split = os.path.split(cwd)
+    current = split[len(split) - 1]
+    if current == 'commands':
+        return check_dir(cwd)
+    elif current == 'Praxis_Bot' or current == 'Praxis':
+        return check_dir(os.path.join(cwd, "commands"))
+    else:
+        print("could not find working directory for Praxis_Bot/commands")
+        raise
+
+
+def get_implementations_dir() -> str:
+    return check_dir(os.path.join(get_base_dir(), "implemented"))
+
+
+def get_compiled_dir() -> str:
+    return check_dir(os.path.join(get_base_dir(), "compiled"))
+
+
+def check_dir(path: str) -> str:
+    if not os.path.exists(path):
+        os.mkdir(path, 0x777)
+    return path
+
+
+if __name__ == "__main__":
+    cmds = load_commands()
+

twitch_script.py

@@ -7,8 +7,11 @@ import twitch.chat
 import config as config
 import db
 import tts
+import commands.loader as command_loader
 
 import credentials
+from commands.command_base import AbstractCommand
+
 
 class Twitch_Module():
     def __init__(self):
@@ -18,7 +21,7 @@ class Twitch_Module():
 
         self.db_manager: db.db_module = db.db_module()
         self.chat: twitch.Chat
-        
+        self.commands = command_loader.load_commands()
         self.tts_enabled: bool = False
         self.tts_whitelist_enabled: bool = False
         self.links_allowed: bool = True
@@ -66,22 +69,34 @@ class Twitch_Module():
                 self.tts_message(message)
 
     def eval_commands(self, message: twitch.chat.Message):
-        containsURL: bool = self.contains_url(message)
+        # containsURL: bool = self.contains_url(message)
+        try:
+            first_space_idx = message.text.index(' ')
+            command_text = ' '
+            if first_space_idx > -1:
+                command_text = message.text[0:first_space_idx]
+            else:
+                command_text = message.text
 
-        if message.text.startswith('!tts start'):
-            print("tts activated on #" + message.channel)
-            self.send_message("tts activated")
-            self.tts_enabled = True
-
-        if message.text.startswith('!tts stop'):
-            print("tts deactivated on #" + message.channel)
-            self.send_message("tts deactivated")
-            self.tts_enabled = True
-
-        if message.text.startswith('!test'):
-            print("!test Detected")
-            message.chat.send("test acknowledged")
-            # message.chat.send(f'@{message.user().display_name}, you have {message.user().view_count} views.')
+            command = self.commands[command_text]
+            if command is not None and command.command_type is AbstractCommand.CommandType.TWITCH:
+                command.do_command(self, message)
+        except Exception as e:
+            pass  # we don't care
+        # if message.text.startswith('!tts start'):
+        #     print("tts activated on #" + message.channel)
+        #     self.send_message("tts activated")
+        #     self.tts_enabled = True
+        #
+        # if message.text.startswith('!tts stop'):
+        #     print("tts deactivated on #" + message.channel)
+        #     self.send_message("tts deactivated")
+        #     self.tts_enabled = False
+        #
+        # if message.text.startswith('!test'):
+        #     print("!test Detected")
+        #     message.chat.send("test acknowledged")
+        #     # message.chat.send(f'@{message.user().display_name}, you have {message.user().view_count} views.')
 
         if message.text.startswith('!roll'):
             try: