Dockerfile_standalone_channelRewards

@@ -0,0 +1,11 @@
+FROM python:3.10.0a7-alpine3.13
+
+WORKDIR /Praxis
+
+COPY requirements_sa_command.txt requirements_sa_command.txt
+RUN apk add --update gcc libc-dev linux-headers && rm -rf /var/cache/apk/*
+RUN pip3 install -r requirements_sa_command.txt
+
+COPY . .
+
+CMD [ "python3", "standalone_channelrewards.py"]

channel_rewards/channelRewards_base.py

@@ -0,0 +1,56 @@
+from abc import ABCMeta, abstractmethod
+from enum import Enum, auto
+
+
+class AbstractChannelRewards(metaclass=ABCMeta):
+    """
+    This is the base class for channel points. In order to load a channel point redemption a few conditions must be met:
+    1) The class name MUST begin with 'ChannelPoint' i.e. CommandTTS, CommandBan, etc...
+    2) the class MUST extend AbstractCommand
+
+    Generally, it would be advisable to define the ChannelPointPrize redemption name as a variable of the
+    class and to then call super().__init__(command)
+    """
+
+    class ChannelRewardsType(Enum):
+        NONE = auto()
+        channelPoints = auto()
+        twitch_bits = auto()
+        twitch_subs = auto()
+
+    class ChannelRewardsSource(Enum):
+        default = 0
+        Praxis = 1
+        Twitch = 2
+        Discord = 3
+
+    def __init__(self, ChannelRewardName: str, n_args: int = 0, ChannelRewardType=ChannelRewardsType.NONE, helpText:list=["No Help"], isChannelRewardEnabled = True):
+        self.ChannelRewardName = ChannelRewardName
+        self.n_args = n_args
+        self.ChannelRewardType = ChannelRewardType
+        self.help = helpText
+        self.isChannelRewardEnabled = isChannelRewardEnabled
+
+    # no touch!
+    def get_args(self, text: str) -> list:
+        return text.split(" ")[0:self.n_args + 1]
+
+    # no touch!
+    def get_ChannelRewardName(self) -> str:
+        return self.ChannelRewardName
+
+    # no touch!
+    def get_ChannelRewardType(self):
+        return self.ChannelRewardType
+
+    # no touch!
+    def get_help(self):
+        return self.help
+
+    # no touch!
+    def is_ChannelReward_enabled(self):
+        return self.isChannelRewardEnabled
+
+    @abstractmethod
+    def do_ChannelReward(self, source, user, command, rewardPrompt, userInput, bonusData):
+        pass

channel_rewards/channelRewards_loader.py

@@ -0,0 +1,79 @@
+import importlib
+import importlib.util
+import inspect
+import os
+import sys
+from typing import Dict
+
+from channel_rewards.channelRewards_base import AbstractChannelRewards
+
+
+#New
+def load_rewards(channelRewardsType: AbstractChannelRewards.ChannelRewardsType) -> Dict[str, AbstractChannelRewards]:
+    print(" -Loading ", channelRewardsType ," ChannelRewards...\n")
+    ChannelRewards = compile_and_load(channelRewardsType)
+    return ChannelRewards
+
+#New
+def compile_and_load_file(path: str, channelRewardsType: AbstractChannelRewards.ChannelRewardsType):
+    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("ChannelReward"):
+            ChannelReward_inst = obj()
+            if channelRewardsType == ChannelReward_inst.get_ChannelRewardType():
+                print(" ---Successfully loaded %s: %s" % (channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
+                return ChannelReward_inst.get_ChannelRewardName(), ChannelReward_inst
+            elif channelRewardsType != ChannelReward_inst.get_ChannelRewardType():
+                print(" -%s ChannelRewardsType did not match: %s for: %s" % (ChannelReward_inst.get_ChannelRewardType(), channelRewardsType, ChannelReward_inst.get_ChannelRewardName()))
+    return "", None
+
+
+#New
+def compile_and_load(ChannelRewardType: AbstractChannelRewards.ChannelRewardsType) -> Dict[str, AbstractChannelRewards]:
+    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, reward = compile_and_load_file(name, ChannelRewardType)
+            if reward is not None and reward.ChannelRewardType is ChannelRewardType:
+                dic[name] = reward
+        break
+    return dic
+
+def get_base_dir() -> str:
+    cwd = os.getcwd()
+    split = os.path.split(cwd)
+    current = split[len(split) - 1]
+    if current == 'channel_rewards':
+        return check_dir(cwd)
+    elif current == 'Praxis_Bot' or current == 'Praxis':
+        return check_dir(os.path.join(cwd, "channel_rewards"))
+    else:
+        print("could not find working directory for Praxis_Bot/channel_rewards")
+        raise Exception
+
+
+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__":
+    rewards = load_rewards()
+

channel_rewards/implemented/ChannelReward_Hydration.py

@@ -0,0 +1,21 @@
+from abc import ABCMeta
+
+from channel_rewards.channelRewards_base import AbstractChannelRewards
+
+class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
+    """
+    this is the hydration reward.
+    """
+    ChannelRewardName = "Hydrate"
+
+    def __init__(self):
+        super().__init__(ChannelReward_Hydration_v2.ChannelRewardName, n_args=1, ChannelRewardType=AbstractChannelRewards.ChannelRewardsType.channelPoints)
+        self.help = ["This is a hydration channel point reward."]
+        self.isChannelRewardEnabled = True
+
+    def do_ChannelReward(self, source = AbstractChannelRewards.ChannelRewardsSource.default, user = "User",  rewardName = "", rewardPrompt = "", userInput = "", bonusData = None):
+
+        return None
+
+    def get_help(self):
+        return self.help

commands/command_base.py

@@ -53,5 +53,5 @@ class AbstractCommand(metaclass=ABCMeta):
         return self.isCommandEnabled
 
     @abstractmethod
-    def do_command(self, bot, user, command, rest, bonusData):
+    def do_command(self, source, user, command, rest, bonusData):
         pass

docker-compose.yaml

@@ -3,7 +3,13 @@ services:
   standalone_command:
     image: standalone_command
     ports:
-      - 5000:5000
+      - 6009:6009
+    environment:
+      - ISDOCKER=cat
+  standalone_channelrewards:
+    image: standalone_channelrewards
+    ports:
+      - 6969:6969
     environment:
       - ISDOCKER=cat
   standalone_twitchscript:

lights_module.py

@@ -197,4 +197,4 @@ if __name__ == "__main__":
     #testModule.discordCredential = credentials_manager.find_Discord_Credential(config.credentialsNickname)
 
     testModule.main()
-    testModule.raveMode()
+    #testModule.raveMode()

makedockerimages.bat

@@ -1,4 +1,5 @@
 docker build --file Dockerfile_standalone_command --tag standalone_command .
+docker build --file Dockerfile_standalone_channelRewards --tag standalone_channelrewards .
 docker build --file Dockerfile_standalone_DiscordScript --tag standalone_discordscript .
 docker build --file Dockerfile_standalone_TwitchScript --tag standalone_twitchscript .
 docker build --file Dockerfile_standalone_Twitch_Pubsub --tag standalone_twitch_pubsub .

rebuild_docker_and_run.bat

@@ -0,0 +1,8 @@
+cd "c:\praxis"
+docker-compose down
+docker build --file Dockerfile_standalone_command --tag standalone_command .
+docker build --file Dockerfile_standalone_channelRewards --tag standalone_channelrewards .
+docker build --file Dockerfile_standalone_DiscordScript --tag standalone_discordscript .
+docker build --file Dockerfile_standalone_TwitchScript --tag standalone_twitchscript .
+docker build --file Dockerfile_standalone_Twitch_Pubsub --tag standalone_twitch_pubsub .
+docker-compose up -d

standalone_channelrewards.py

@@ -0,0 +1,94 @@
+import flask
+from flask import request
+
+import channel_rewards.channelRewards_loader as rewards_loader
+from channel_rewards.channelRewards_base import AbstractChannelRewards
+
+api = flask.Flask(__name__)
+# enable/disable this to get web pages of crashes returned
+api.config["DEBUG"] = True
+
+loadedRewards = {}
+
+def init():
+    # todo load entire reward library and cache it here
+    print("init stuff")
+    loadedRewards[AbstractChannelRewards.ChannelRewardsType.channelPoints] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.channelPoints)
+    loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_bits] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_bits)
+    loadedRewards[AbstractChannelRewards.ChannelRewardsType.twitch_subs] = rewards_loader.load_rewards(AbstractChannelRewards.ChannelRewardsType.twitch_subs)
+
+
+def is_reward(reward_name, reward_type) -> bool:
+    #global loadedRewards
+    tempType = reward_type.replace('ChannelRewardsType.', '')
+    realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
+
+    for reward in loadedRewards[realTempType]:
+        print("found: ",reward,"type: ",type(reward))
+        if reward_name == reward:
+            print("Equal")
+            return True
+
+
+    if reward_name == "!echo":
+        return True
+    else:
+        return False
+
+
+def handle_reward(source, username, reward_name, reward_type, rewardPrompt, userInput, bonusData):
+    #reward:AbstractChannelRewards = loadedRewards[reward_name]
+    tempType = reward_type.replace('ChannelRewardsType.', '')
+    realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
+    reward:AbstractChannelRewards = loadedRewards[realTempType][reward_name]
+    if reward is not None:
+        reward_response = reward.do_ChannelReward(source, username, reward_name, rewardPrompt, userInput, bonusData)
+        return flask.make_response("{\"message\":\"%s\"}" % reward_response, 200, {"Content-Type": "application/json"})
+
+    #print("Doing a reward")
+
+
+@api.route('/api/v1/reward', methods=['GET'])
+def reward_check():
+    if 'reward_name' in request.args and 'reward_type' in request.args:
+        print("reward_name:", request.args['reward_name'],"reward_type:", request.args['reward_type'])
+        if is_reward(request.args['reward_name'], request.args['reward_type']):
+            print("about to send")
+            return flask.make_response('', 200)
+        else:
+            return flask.make_response('', 404)
+
+
+@api.route('/api/v1/exec_reward', methods=['GET'])
+def exec_reward():
+    if 'reward_name' not in request.args:
+        return flask.make_response('{\"text\":"Argument \'reward_name\' not in request"}', 400)
+    if 'reward_type' not in request.args:
+        return flask.make_response('{\"text\":"Argument \'reward_name\' not in request"}', 400)
+    if 'reward_prompt' not in request.args:
+        return flask.make_response('{\"text\":"Argument \'reward_prompt\' not in request"}', 400)
+    if 'user_input' not in request.args:
+        return flask.make_response('{\"text\":"Argument \'user_input\' not in request"}', 400)
+
+
+    if 'reward_source' not in request.args:
+        return flask.make_response('{\"text\":"Argument \'reward_source\' not in request"}', 400)
+
+    if 'user_name' not in request.args:
+        username = "User"
+    else:
+        username = request.args['user_name']
+
+    return handle_reward(
+        request.args['reward_source'],
+        username,
+        request.args['reward_name'],
+        request.args['reward_type'],
+        request.args['reward_prompt'],
+        request.args['user_input'],
+        request.args['bonus_data'])
+
+
+if __name__ == '__main__':
+    init()
+    api.run(host='0.0.0.0', port=6969)

standalone_command.py

@@ -55,7 +55,7 @@ def command_check():
             return flask.make_response('', 404)
 
 
-@api.route('/api/v1/exec', methods=['GET'])
+@api.route('/api/v1/exec_command', methods=['GET'])
 def exec_command():
     if 'command_name' not in request.args:
         return flask.make_response('{\"text\":"Argument \'command_name\' not in request"}', 400)
@@ -75,4 +75,4 @@ def exec_command():
 
 if __name__ == '__main__':
     init()
-    api.run(host='0.0.0.0')
+    api.run(host='0.0.0.0', port=6009)

standalone_discord_script.py

@@ -87,14 +87,14 @@ class Discord_Module(discord.Client):
     async def is_command(self, word: str) -> bool:
         # todo need to url-escape word
         clean_param = urlencode({'name': word})
-        url = "http://standalone_command:5000/api/v1/command?%s" % clean_param
+        url = "http://standalone_command:6009/api/v1/command?%s" % clean_param
         resp = requests.get(url)
         return resp.status_code == 200
 
     async def exec_command(self, realMessage: discord.Message, command: str, rest: str):
         # todo need to url-escape command and rest
         params = urlencode({'command_source': commands.command_base.AbstractCommand.CommandSource.Discord, 'user_name': realMessage.author.mention, 'command_name': command, 'rest': rest, 'bonus_data': realMessage})
-        url = "http://standalone_command:5000/api/v1/exec?%s" % params
+        url = "http://standalone_command:6009/api/v1/exec_command?%s" % params
         resp = requests.get(url)
         if resp.status_code == 200:
             print("Got the following message: %s" % resp.text)

standalone_twitch_pubsub.py

@@ -1,7 +1,15 @@
-import credentials
+import re
+from json import loads
+from urllib.parse import urlencode
 
+import requests
+
+import credentials
 import config
 
+from channel_rewards.channelRewards_base import AbstractChannelRewards
+import channel_rewards.channelRewards_base
+
 import twitchAPI
 from twitchAPI.pubsub import PubSub
 from twitchAPI.twitch import Twitch
@@ -10,6 +18,7 @@ from twitchAPI.oauth import UserAuthenticator
 from pprint import pprint
 from uuid import UUID
 
+from cooldowns import Cooldown_Module
 
 class Twitch_Pubsub():
     def __init__(self):
@@ -22,6 +31,9 @@ class Twitch_Pubsub():
         self.uuid_1 = None
         self.uuid_2 = None
 
+        self.cooldownModule: Cooldown_Module = Cooldown_Module()
+        self.cooldownModule.setupCooldown("twitchChat", 20, 32)
+
     def setup(self):
         self.twitch.authenticate_app(self.target_scope)
 
@@ -68,6 +80,75 @@ class Twitch_Pubsub():
         print("Channel Point Redemption")
         print('got callback for UUID ' + str(uuid))
         pprint(data)
+        #print("attempting to get data: ")
+        #print(data['data']['redemption']['user']['display_name'])
+        #print(data['data']['redemption']['reward']['title'])
+        #print(data['data']['redemption']['reward']['prompt'])
+        try:
+            userinput = data['data']['redemption']['user_input']
+        except:
+            userinput = ""
+        #print(userinput)
+        self.callback_EXEC(
+            data['data']['redemption']['user']['display_name'],
+            data['data']['redemption']['reward']['title'],
+            AbstractChannelRewards.ChannelRewardsType.channelPoints,
+            data['data']['redemption']['reward']['prompt'],
+            userinput,
+            data)
+
+    def callback_bits(self, uuid: UUID, data: dict) -> None:
+        print("Bits Redemption")
+        print('got callback for UUID ' + str(uuid))
+        pprint(data)
+
+    def callback_subs(self, uuid: UUID, data: dict) -> None:
+        print("Subs Redemption")
+        print('got callback for UUID ' + str(uuid))
+        pprint(data)
+
+    def callback_EXEC(self, sender, rewardName:str, rewardType, rewardPrompt, userInput, raw_data):
+        try:
+            is_actionable = self.is_reward(rewardName, rewardType)
+            if is_actionable:
+                print("Trying to do the thing")
+                if self.cooldownModule.isCooldownActive("twitchChat") == False:
+                        self.exec_reward(sender, rewardName, rewardType, rewardPrompt, userInput, raw_data)
+        except:
+            print("something went wrong with a reward")
+
+    def is_reward(self, rewardName, rewardType):
+        # todo need to url-escape word
+        clean_param = urlencode({'reward_name': rewardName, 'reward_type':rewardType})
+        print(rewardName, rewardType)
+        #standalone_channelrewards
+        url = "http://standalone_channelrewards:6969/api/v1/reward?%s" % clean_param
+        resp = requests.get(url)
+        return resp.status_code == 200
+
+    def exec_reward(self, sender, rewardName, rewardType, rewardPrompt, userInput, realMessage):
+        params = urlencode(
+            {'reward_source': channel_rewards.channelRewards_base.AbstractChannelRewards.ChannelRewardsSource.Twitch,
+            'user_name': sender,
+            'reward_name': rewardName,
+            'reward_type': rewardType,
+            'reward_prompt': rewardPrompt,
+            'user_input' : userInput,
+            'bonus_data': realMessage})
+
+        #standalone_channelrewards
+        url = "http://standalone_channelrewards:6969/api/v1/exec_reward?%s" % params
+        resp = requests.get(url)
+        if resp.status_code == 200:
+            print("Got the following message: %s" % resp.text)
+            data = loads(resp.text)
+            msg = data['message']
+            if msg is not None:
+                #self.send_message(msg) #Cant Send messages with this pubsub library afaik
+                pass
+        else:
+            # todo handle failed requests
+            pass
 
 if __name__ == "__main__":
     testModule = Twitch_Pubsub()

standalone_twitch_script.py

@@ -69,14 +69,14 @@ class Twitch_Module():
     def is_command(self, word: str) -> bool:
         # todo need to url-escape word
         clean_param = urlencode({'name': word})
-        url = "http://standalone_command:5000/api/v1/command?%s" % clean_param
+        url = "http://standalone_command:6009/api/v1/command?%s" % clean_param
         resp = requests.get(url)
         return resp.status_code == 200
 
     def exec_command(self, realMessage: twitch.chat.Message, command: str, rest: str):
         # todo need to url-escape command and rest
         params = urlencode({'command_source': commands.command_base.AbstractCommand.CommandSource.Twitch,'user_name': realMessage.sender, 'command_name': command, 'rest': rest, 'bonus_data': realMessage})
-        url = "http://standalone_command:5000/api/v1/exec?%s" % params
+        url = "http://standalone_command:6009/api/v1/exec_command?%s" % params
         resp = requests.get(url)
         if resp.status_code == 200:
             print("Got the following message: %s" % resp.text)