alex_orid/Praxis_Bot
2
0
Fork 0
Code Issues 1 Pull Requests Releases Wiki Activity

channel rewards service #34

Merged
alex_orid merged 13 commits from channelpoints-service into v2.0 2021-04-27 18:15:04 +00:00
Conversation 0 Commits 13 Files Changed 14 +47 -26
3 changed files with 47 additions and 26 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit a29f71e268 - Show all commits

12
channel_rewards/implemented/ChannelReward_Hydration.py
View File

@ -6,16 +6,18 @@ class ChannelReward_Hydration_v2(AbstractChannelRewards, metaclass=ABCMeta):
"""
this is the hydration reward.
"""
ChannelRewardName = "hydration"
ChannelRewardName = "Hydrate"
def __init__(self):
super().__init__(ChannelReward_Hydration_v2.ChannelRewardName, n_args=1, command_type=AbstractChannelRewards.ChannelRewardsType.channelPoints)
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", command = "", rest = "", bonusData = None):
returnString = user + " sent: [ " + command + " ] with: " + rest
#print(returnString)
def do_ChannelReward(self, source = AbstractChannelRewards.ChannelRewardsSource.default, user = "User", rewardName = "", rest = "", bonusData = None):
returnString = user + " sent: [ " + rewardName + " ] with: " + rest
print(returnString)
for x in range(10):
print("This is wet")
return returnString
def get_help(self):

16
standalone_channelrewards.py
View File

@ -22,18 +22,22 @@ def load_rewards():
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
def is_reward(reward_name, reward_type) -> bool:
tempType = reward_type.replace('ChannelRewardsType.', '')
realTempType = AbstractChannelRewards.ChannelRewardsType.__dict__[tempType]
rewardList = loadedRewards[realTempType]
print("testing mcgoo")
for reward in rewardList:
print(reward)
print("testing oogcm")
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]

45
standalone_twitch_pubsub.py
View File

@ -7,6 +7,7 @@ import requests
import credentials
import config
from channel_rewards.channelRewards_base import AbstractChannelRewards
import channel_rewards.channelRewards_base
import twitchAPI
@ -79,11 +80,22 @@ class Twitch_Pubsub():
print("Channel Point Redemption")
print('got callback for UUID ' + str(uuid))
pprint(data)
#self.callback_EXEC(
# "sender",
# "rewardName",
# channel_rewards.channelRewards_base.AbstractChannelRewards.ChannelRewardsType.channelPoints,
# 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")
@ -95,33 +107,36 @@ class Twitch_Pubsub():
print('got callback for UUID ' + str(uuid))
pprint(data)
def callback_EXEC(self, sender, rewardName:str, rewardType, raw_data):
def callback_EXEC(self, sender, rewardName:str, rewardType, rewardPrompt, userInput, raw_data):
try:
is_actionable = self.is_reward(rewardName, rewardType)
if is_actionable:
if self.cooldownModule.isCooldownActive("twitchChat") == False:
self.exec_reward(sender, rewardName, rewardType, "", raw_data)
self.exec_reward(sender, rewardName, rewardType, rewardPrompt, userInput, raw_data)
except:
print("something went wrong with a reward")
def is_reward(self, rewardName:str, rewardType):
def is_reward(self, rewardName, rewardType):
# todo need to url-escape word
clean_param = urlencode({'reward_name': rewardName, 'reward_type':rewardType})
url = "http://channelrewards:6969/api/v1/reward?%s" % clean_param
print(rewardName, rewardType)
#standalone_channelrewards
url = "http://localhost:6969/api/v1/reward?%s" % clean_param
resp = requests.get(url)
return resp.status_code == 200
def exec_reward(self, sender, reward, rewardType, rest, realMessage):
def exec_reward(self, sender, rewardName, rewardType, rewardPrompt, userInput, realMessage):
params = urlencode(
{'command_source': channel_rewards.channelRewards_base.AbstractChannelRewards.ChannelRewardsSource.Twitch,
{'reward_source': channel_rewards.channelRewards_base.AbstractChannelRewards.ChannelRewardsSource.Twitch,
'user_name': sender,
'reward_name': reward,
'reward_name': rewardName,
'reward_type': rewardType,
'rest': rest,
'reward_prompt': rewardPrompt,
'user_input' : userInput,
'bonus_data': realMessage})
url = "http://channelrewards:6969/api/v1/exec_reward?%s" % params
#standalone_channelrewards
url = "http://localhost:6969/api/v1/exec_reward?%s" % params
resp = requests.get(url)
if resp.status_code == 200:
print("Got the following message: %s" % resp.text)