diff --git a/commands/command_base.py b/commands/command_base.py new file mode 100644 index 0000000..9bcbcb3 --- /dev/null +++ b/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 diff --git a/commands/implemented/command_test.py b/commands/implemented/command_test.py new file mode 100644 index 0000000..d6fb5a5 --- /dev/null +++ b/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") diff --git a/commands/implemented/command_tts.py b/commands/implemented/command_tts.py new file mode 100644 index 0000000..4392cc5 --- /dev/null +++ b/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 diff --git a/commands/loader.py b/commands/loader.py new file mode 100644 index 0000000..cab69b1 --- /dev/null +++ b/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() + diff --git a/twitch_script.py b/twitch_script.py index fa5f011..a9f2ea8 100644 --- a/twitch_script.py +++ b/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: