renaming and fixing parameters
This commit is contained in:
parent
2c7a2b3ef0
commit
65a723b5df
@ -8,4 +8,4 @@ RUN pip3 install -r requirements_sa_command.txt
|
||||
|
||||
COPY . .
|
||||
|
||||
CMD [ "python3", "standalone_channelpoints.py"]
|
||||
CMD [ "python3", "standalone_channelrewards.py"]
|
||||
@ -1,24 +0,0 @@
|
||||
from abc import ABCMeta
|
||||
|
||||
from channel_points.channelPoints_base import AbstractChannelPoints
|
||||
|
||||
import utilities_script as utility
|
||||
|
||||
class ChannelPointReward_Hydration_v2(AbstractChannelPoints, metaclass=ABCMeta):
|
||||
"""
|
||||
this is the hydration reward.
|
||||
"""
|
||||
ChannelPointRewardName = "hydration"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(ChannelPointReward_Hydration_v2.ChannelPointRewardName, n_args=1, command_type=AbstractChannelPoints.ChannelPointsType.Ver2)
|
||||
self.help = ["This is a hydration channel point reward."]
|
||||
self.isChannelPointRewardEnabled = True
|
||||
|
||||
def do_ChannelPointReward(self, source = AbstractChannelPoints.ChannelPointsSource.default, user = "User", command = "", rest = "", bonusData = None):
|
||||
returnString = user + " sent: [ " + command + " ] with: " + rest
|
||||
#print(returnString)
|
||||
return returnString
|
||||
|
||||
def get_help(self):
|
||||
return self.help
|
||||
@ -1,25 +0,0 @@
|
||||
from abc import ABCMeta
|
||||
|
||||
from commands.command_base import AbstractCommand
|
||||
|
||||
import utilities_script as utility
|
||||
|
||||
class Command_test_v2(AbstractCommand, metaclass=ABCMeta):
|
||||
"""
|
||||
this is the test command.
|
||||
"""
|
||||
command = "testerino"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(Command_test_v2.command, n_args=1, command_type=AbstractCommand.CommandType.Ver2)
|
||||
self.help = ["This is a test command.",
|
||||
"\nExample:","testerino"]
|
||||
self.isCommandEnabled = True
|
||||
|
||||
def do_command(self, source = AbstractCommand.CommandSource.default, user = "User", command = "", rest = "", bonusData = None):
|
||||
returnString = user + " sent: [ " + command + " ] with: " + rest
|
||||
#print(returnString)
|
||||
return returnString
|
||||
|
||||
def get_help(self):
|
||||
return self.help
|
||||
@ -2,7 +2,7 @@ from abc import ABCMeta, abstractmethod
|
||||
from enum import Enum, auto
|
||||
|
||||
|
||||
class AbstractChannelPoints(metaclass=ABCMeta):
|
||||
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...
|
||||
@ -12,43 +12,45 @@ class AbstractChannelPoints(metaclass=ABCMeta):
|
||||
class and to then call super().__init__(command)
|
||||
"""
|
||||
|
||||
class ChannelPointsType(Enum):
|
||||
class ChannelRewardsType(Enum):
|
||||
NONE = auto()
|
||||
Ver2 = auto()
|
||||
channelPoints = auto()
|
||||
twitch_bits = auto()
|
||||
twitch_subs = auto()
|
||||
|
||||
class ChannelPointsSource(Enum):
|
||||
class ChannelRewardsSource(Enum):
|
||||
default = 0
|
||||
Praxis = 1
|
||||
Twitch = 2
|
||||
Discord = 3
|
||||
|
||||
def __init__(self, ChannelPointRewardName: str, n_args: int = 0, channelPointReward_type=ChannelPointsType.NONE, helpText:list=["No Help"], isChannelPointRewardEnabled = True):
|
||||
self.ChannelPointRewardName = ChannelPointRewardName
|
||||
def __init__(self, ChannelRewardName: str, n_args: int = 0, ChannelReward_type=ChannelRewardsType.NONE, helpText:list=["No Help"], isChannelRewardEnabled = True):
|
||||
self.ChannelRewardName = ChannelRewardName
|
||||
self.n_args = n_args
|
||||
self.ChannelPointRewardType = channelPointReward_type
|
||||
self.ChannelRewardType = ChannelReward_type
|
||||
self.help = helpText
|
||||
self.isChannelPointRewardEnabled = isChannelPointRewardEnabled
|
||||
self.isChannelRewardEnabled = isChannelRewardEnabled
|
||||
|
||||
# no touch!
|
||||
def get_args(self, text: str) -> list:
|
||||
return text.split(" ")[0:self.n_args + 1]
|
||||
|
||||
# no touch!
|
||||
def get_ChannelPointRewardName(self) -> str:
|
||||
return self.ChannelPointRewardName
|
||||
def get_ChannelRewardName(self) -> str:
|
||||
return self.ChannelRewardName
|
||||
|
||||
# no touch!
|
||||
def get_ChannelPointRewardType(self):
|
||||
return self.ChannelPointRewardType
|
||||
def get_ChannelRewardType(self):
|
||||
return self.ChannelRewardType
|
||||
|
||||
# no touch!
|
||||
def get_help(self):
|
||||
return self.help
|
||||
|
||||
# no touch!
|
||||
def is_ChannelPointReward_enabled(self):
|
||||
return self.isChannelPointRewardEnabled
|
||||
def is_ChannelReward_enabled(self):
|
||||
return self.isChannelRewardEnabled
|
||||
|
||||
@abstractmethod
|
||||
def do_ChannelPointReward(self, bot, user, command, rest, bonusData):
|
||||
def do_ChannelReward(self, source, user, command, rest, bonusData):
|
||||
pass
|
||||
@ -5,17 +5,17 @@ import os
|
||||
import sys
|
||||
from typing import Dict
|
||||
|
||||
from channel_points.channelPoints_base import AbstractChannelPoints
|
||||
from channel_rewards.channelRewards_base import AbstractChannelRewards
|
||||
|
||||
|
||||
#New
|
||||
def load_rewards(channelPointsType: AbstractChannelPoints.ChannelPointsType) -> Dict[str, AbstractChannelPoints]:
|
||||
print(" -Loading ", channelPointsType ," ChannelPointRewards...\n")
|
||||
channelPointRewards = compile_and_load(channelPointsType)
|
||||
return channelPointRewards
|
||||
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, channelPointsType: AbstractChannelPoints.ChannelPointsType):
|
||||
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)
|
||||
@ -23,26 +23,26 @@ def compile_and_load_file(path: str, channelPointsType: AbstractChannelPoints.Ch
|
||||
spec.loader.load_module(module_name)
|
||||
|
||||
for name, obj in inspect.getmembers(module):
|
||||
if inspect.isclass(obj) and name.startswith("ChannelPointReward"):
|
||||
channelPointReward_inst = obj()
|
||||
if channelPointsType == channelPointReward_inst.get_ChannelPointRewardType():
|
||||
print(" ---Successfully loaded %s: %s" % (channelPointsType, channelPointReward_inst.get_ChannelPointRewardType()))
|
||||
return channelPointReward_inst.get_ChannelPointRewardType(), channelPointReward_inst
|
||||
elif channelPointsType != channelPointReward_inst.get_ChannelPointRewardType():
|
||||
print(" -%s ChannelPointsType did not match: %s for: %s" % (channelPointReward_inst.get_ChannelPointRewardType(), channelPointsType, channelPointReward_inst.get_ChannelPointRewardName()))
|
||||
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_ChannelRewardType()))
|
||||
return ChannelReward_inst.get_ChannelRewardType(), 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(ChannelPointsRewardType: AbstractChannelPoints.ChannelPointsType) -> Dict[str, AbstractChannelPoints]:
|
||||
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, ChannelPointsRewardType)
|
||||
if reward is not None and reward.ChannelPointRewardType is ChannelPointsRewardType:
|
||||
name, reward = compile_and_load_file(name, ChannelRewardType)
|
||||
if reward is not None and reward.ChannelRewardType is ChannelRewardType:
|
||||
dic[name] = reward
|
||||
break
|
||||
return dic
|
||||
22
channel_rewards/implemented/ChannelReward_Hydration.py
Normal file
22
channel_rewards/implemented/ChannelReward_Hydration.py
Normal file
@ -0,0 +1,22 @@
|
||||
from abc import ABCMeta
|
||||
|
||||
from channel_rewards.channelRewards_base import AbstractChannelRewards
|
||||
|
||||
class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
|
||||
"""
|
||||
this is the hydration reward.
|
||||
"""
|
||||
ChannelRewardName = "hydration"
|
||||
|
||||
def __init__(self):
|
||||
super().__init__(ChannelReward_Hydration_v2.ChannelRewardName, n_args=1, command_type=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", command = "", rest = "", bonusData = None):
|
||||
returnString = user + " sent: [ " + command + " ] with: " + rest
|
||||
#print(returnString)
|
||||
return returnString
|
||||
|
||||
def get_help(self):
|
||||
return self.help
|
||||
@ -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
|
||||
@ -6,8 +6,8 @@ services:
|
||||
- 6009:6009
|
||||
environment:
|
||||
- ISDOCKER=cat
|
||||
standalone_channelpoints:
|
||||
image: standalone_channelpoints
|
||||
standalone_channelrewards:
|
||||
image: standalone_channelrewards
|
||||
ports:
|
||||
- 6969:6969
|
||||
environment:
|
||||
|
||||
@ -1,5 +1,5 @@
|
||||
docker build --file Dockerfile_standalone_command --tag standalone_command .
|
||||
docker build --file Dockerfile_standalone_channelpoints --tag standalone_channelpoints .
|
||||
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 .
|
||||
@ -1,78 +0,0 @@
|
||||
import flask
|
||||
from flask import request
|
||||
|
||||
import commands.loader as command_loader
|
||||
from commands.command_base import AbstractCommand
|
||||
|
||||
api = flask.Flask(__name__)
|
||||
# enable/disable this to get web pages of crashes returned
|
||||
api.config["DEBUG"] = True
|
||||
|
||||
loadedCommands = {}
|
||||
|
||||
def init():
|
||||
# todo load entire command library and cache it here
|
||||
load_commands()
|
||||
|
||||
|
||||
def load_commands():
|
||||
global loadedCommands
|
||||
loadedCommands = command_loader.load_commands(AbstractCommand.CommandType.Ver2)
|
||||
|
||||
|
||||
def is_command(command: str) -> bool:
|
||||
#print(command)
|
||||
for cmd in loadedCommands:
|
||||
#print(cmd)
|
||||
if command == cmd:
|
||||
return True
|
||||
|
||||
if command == "!echo":
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def handle_command(source, username, command, rest, bonusData):
|
||||
if command == "!echo":
|
||||
message = "Got payload [%s]" % rest
|
||||
#print(message)
|
||||
return flask.make_response("{\"message\":\"%s\"}" % message, 200, {"Content-Type": "application/json"})
|
||||
|
||||
cmd:AbstractCommand = loadedCommands[command]
|
||||
if cmd is not None:
|
||||
cmd_response = cmd.do_command(source, username, command, rest, bonusData)
|
||||
return flask.make_response("{\"message\":\"%s\"}" % cmd_response, 200, {"Content-Type": "application/json"})
|
||||
|
||||
#print("Doing a command")
|
||||
|
||||
|
||||
@api.route('/api/v1/command', methods=['GET'])
|
||||
def command_check():
|
||||
if 'name' in request.args:
|
||||
if is_command(request.args['name']):
|
||||
return flask.make_response('', 200)
|
||||
else:
|
||||
return flask.make_response('', 404)
|
||||
|
||||
|
||||
@api.route('/api/v1/exec', methods=['GET'])
|
||||
def exec_command():
|
||||
if 'command_name' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'command_name\' not in request"}', 400)
|
||||
if 'rest' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'rest\' not in request"}', 400)
|
||||
|
||||
if 'command_source' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'command_source\' not in request"}', 400)
|
||||
|
||||
if 'user_name' not in request.args:
|
||||
username = "User"
|
||||
else:
|
||||
username = request.args['user_name']
|
||||
|
||||
return handle_command(request.args['command_source'], username, request.args['command_name'], request.args['rest'], request.args['bonus_data'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
init()
|
||||
api.run(host='0.0.0.0', port=6969)
|
||||
78
standalone_channelrewards.py
Normal file
78
standalone_channelrewards.py
Normal file
@ -0,0 +1,78 @@
|
||||
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
|
||||
load_rewards()
|
||||
|
||||
|
||||
def load_rewards():
|
||||
global loadedRewards
|
||||
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: str, reward_type: str) -> bool:
|
||||
#print(reward)
|
||||
for reward in loadedRewards[reward_type]:
|
||||
#print(reward)
|
||||
if reward_name == reward:
|
||||
return True
|
||||
|
||||
if reward_name == "!echo":
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
|
||||
def handle_reward(source, username, reward_name, reward_type, rest, bonusData):
|
||||
#reward:AbstractChannelRewards = loadedRewards[reward_name]
|
||||
reward:AbstractChannelRewards = loadedRewards[reward_type][reward_name]
|
||||
if reward is not None:
|
||||
reward_response = reward.do_ChannelReward(source, username, reward_name, rest, 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:
|
||||
if is_reward(request.args['reward_name'], request.args['reward_type']):
|
||||
return flask.make_response('', 200)
|
||||
else:
|
||||
return flask.make_response('', 404)
|
||||
|
||||
|
||||
@api.route('/api/v1/exec', 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 'rest' not in request.args:
|
||||
return flask.make_response('{\"text\":"Argument \'rest\' 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['rest'], request.args['bonus_data'])
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
init()
|
||||
api.run(host='0.0.0.0', port=6969)
|
||||
Loading…
Reference in New Issue
Block a user