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

Working Bits Pubsubs

Browse Source
This commit is contained in:
Alex Orid 2021-04-29 15:35:50 -04:00
parent 834bae4773
commit 25fee4cb5c
6 changed files with 115 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
channel_rewards/implemented/ChannelReward_Hydration.py
View File

@ -8,6 +8,8 @@ import requests
import threading
import random
class ChannelReward_Hydration(AbstractChannelRewards, metaclass=ABCMeta):
"""
this is the hydration reward.
@ -30,7 +32,8 @@ class ChannelReward_Hydration(AbstractChannelRewards, metaclass=ABCMeta):
self.threads.append(thread_)
thread_.start()
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.send_TTS, args=(user, rewardPrompt))
prompt_ = self.get_Phrase(rewardPrompt)
thread_ = threading.Thread(target=self.send_TTS, args=(user, prompt_))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
@ -72,5 +75,13 @@ class ChannelReward_Hydration(AbstractChannelRewards, metaclass=ABCMeta):
# todo handle failed requests
pass
def get_Phrase(self, defaultRewardPrompt,
phrases = ["I demand you drink at once!", "I reccomend you drink some water uwu!", "Get a bucket and a mop cuz you need hydration!", "Hi, I think you look dehydrated, so go get some water!"]):
phrases.append(defaultRewardPrompt)
totalPhrases = len(phrases) - 1
targetPhrase = phrases[random.randint(0,totalPhrases)]
return targetPhrase
def get_help(self):
return self.help

13
channel_rewards/implemented/ChannelReward_RubiksCube.py
View File

@ -8,6 +8,8 @@ import requests
import threading
import random
class ChannelReward_RubiksCube(AbstractChannelRewards, metaclass=ABCMeta):
"""
this is the hydration reward.
@ -31,7 +33,8 @@ class ChannelReward_RubiksCube(AbstractChannelRewards, metaclass=ABCMeta):
self.threads.append(thread_)
thread_.start()
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.send_TTS, args=(user, rewardPrompt))
prompt_ = self.get_Phrase(rewardPrompt)
thread_ = threading.Thread(target=self.send_TTS, args=(user, prompt_))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
@ -73,5 +76,13 @@ class ChannelReward_RubiksCube(AbstractChannelRewards, metaclass=ABCMeta):
# todo handle failed requests
pass
def get_Phrase(self, defaultRewardPrompt,
phrases = [""]):
phrases.append(defaultRewardPrompt)
totalPhrases = len(phrases) - 1
targetPhrase = phrases[random.randint(0,totalPhrases)]
return targetPhrase
def get_help(self):
return self.help

37
channel_rewards/implemented/ChannelReward_twitchBits.py
View File

@ -8,6 +8,8 @@ import requests
import threading
import random
class ChannelReward_TwitchBits(AbstractChannelRewards, metaclass=ABCMeta):
"""
this is the hydration reward.
@ -16,7 +18,7 @@ class ChannelReward_TwitchBits(AbstractChannelRewards, metaclass=ABCMeta):
def __init__(self):
super().__init__(ChannelReward_TwitchBits.ChannelRewardName, n_args=1, ChannelRewardType=AbstractChannelRewards.ChannelRewardsType.twitch_bits)
self.help = ["This is a hydration channel point reward."]
self.help = ["This is a TwitchBits reward."]
self.isChannelRewardEnabled = True
self.threads = []
@ -24,13 +26,13 @@ class ChannelReward_TwitchBits(AbstractChannelRewards, metaclass=ABCMeta):
#print("sending:",user, 16, "!lights hydration")
try:
#if self.is_ChannelReward_enabled:
#thread_ = threading.Thread(target=self.dothething, args=(user, 16, "!lights hydration", ""))
#thread_.daemon = True
#self.threads.append(thread_)
#thread_.start()
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.dothething, args=(user, 16, "!lights hydration", ""))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.send_TTS, args=(user, rewardPrompt))
thread_ = threading.Thread(target=self.send_TTS, args=("", user + " showered the stream with " + userInput + " bits and says: " + rewardPrompt))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
@ -72,5 +74,26 @@ class ChannelReward_TwitchBits(AbstractChannelRewards, metaclass=ABCMeta):
# todo handle failed requests
pass
def get_bitsAmount(self, dataToParse:dict):
try:
getData = str(dataToParse['data']['message']['data']['bits_used'])
except:
getData = "Error"
return getData
def get_bitsMessage(self, dataToParse:dict):
try:
getData = str(dataToParse['data']['message']['data']['chat_message'])
except:
getData = "Error"
return getData
def get_Phrase(self, defaultRewardPrompt,
phrases = [""]):
phrases.append(defaultRewardPrompt)
totalPhrases = len(phrases) - 1
targetPhrase = phrases[random.randint(0,totalPhrases)]
return targetPhrase
def get_help(self):
return self.help

17
channel_rewards/implemented/ChannelReward_twitchSubs.py
View File

@ -8,6 +8,8 @@ import requests
import threading
import random
class ChannelReward_TwitchSubs(AbstractChannelRewards, metaclass=ABCMeta):
"""
this is the TwitchSubs reward.
@ -16,7 +18,7 @@ class ChannelReward_TwitchSubs(AbstractChannelRewards, metaclass=ABCMeta):
def __init__(self):
super().__init__(ChannelReward_TwitchSubs.ChannelRewardName, n_args=1, ChannelRewardType=AbstractChannelRewards.ChannelRewardsType.twitch_subs)
self.help = ["This is a hydration channel point reward."]
self.help = ["This is a TwitchSub reward."]
self.isChannelRewardEnabled = True
self.threads = []
@ -25,12 +27,13 @@ class ChannelReward_TwitchSubs(AbstractChannelRewards, metaclass=ABCMeta):
#print("sending:",user, 16, "!lights hydration")
try:
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.dothething, args=(user, 16, "!lights hydration", ""))
thread_ = threading.Thread(target=self.dothething, args=(user, 16, "!lights ravemode", ""))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
if self.is_ChannelReward_enabled:
thread_ = threading.Thread(target=self.send_TTS, args=(user, rewardPrompt))
prompt_ = self.get_Phrase("a funny")
thread_ = threading.Thread(target=self.send_TTS, args=("", user + " is " + prompt_ + " nerd! Thank you for your support!"))
thread_.daemon = True
self.threads.append(thread_)
thread_.start()
@ -72,5 +75,13 @@ class ChannelReward_TwitchSubs(AbstractChannelRewards, metaclass=ABCMeta):
# todo handle failed requests
pass
def get_Phrase(self, defaultRewardPrompt,
phrases = ["an amazing", "a special", "an inquisitive", "a thoughtful", "a curious", "a hyper focusing", "a trend setting", "an artistic"]):
phrases.append(defaultRewardPrompt)
totalPhrases = len(phrases) - 1
targetPhrase = phrases[random.randint(0,totalPhrases)]
return targetPhrase
def get_help(self):
return self.help

55
standalone_twitch_pubsub.py
View File

@ -32,13 +32,15 @@ class Twitch_Pubsub():
self.credential : credentials.Twitch_Credential()
self.twitch : Twitch()
self.pubsub: PubSub()
self.target_scope = [AuthScope.WHISPERS_READ, AuthScope.CHANNEL_READ_REDEMPTIONS]
self.target_scope = [AuthScope.WHISPERS_READ, AuthScope.CHANNEL_READ_REDEMPTIONS, AuthScope.BITS_READ, AuthScope.CHANNEL_READ_SUBSCRIPTIONS]
self.uuid_1 = None
self.uuid_2 = None
self.uuid_whisper = None
self.uuid_channelPoints = None
self.uuid_bits = None
self.uuid_subs = None
self.cooldownModule: Cooldown_Module = Cooldown_Module()
self.cooldownModule.setupCooldown("twitchChat", 20, 32)
self.cooldownModule.setupCooldown("twitchpubsub", 20, 32)
def setup(self):
self.twitch.authenticate_app(self.target_scope)
@ -69,13 +71,17 @@ class Twitch_Pubsub():
user_id = self.twitch.get_users(logins=[config.autoJoin_TwitchChannel])['data'][0]['id']
if user_id is not None: print("found user_id")
print(user_id)
self.uuid_1 = self.pubsub.listen_whispers(user_id, self.callback_whisper)
self.uuid_2 = self.pubsub.listen_channel_points(user_id, self.callback_channelPoints)
self.uuid_whisper = self.pubsub.listen_whispers(user_id, self.callback_whisper)
self.uuid_channelPoints = self.pubsub.listen_channel_points(user_id, self.callback_channelPoints)
self.uuid_bits = self.pubsub.listen_bits(user_id, self.callback_bits)
self.uuid_subs = self.pubsub.listen_channel_subscriptions(user_id, self.callback_subs)
#input('press ENTER to close...')
def stop(self):
self.pubsub.unlisten(self.uuid_1)
self.pubsub.unlisten(self.uuid_2)
self.pubsub.unlisten(self.uuid_whisper)
self.pubsub.unlisten(self.uuid_channelPoints)
self.pubsub.unlisten(self.uuid_bits)
self.pubsub.unlisten(self.uuid_subs)
self.pubsub.stop()
def callback_whisper(self, uuid: UUID, data: dict) -> None:
@ -109,22 +115,53 @@ class Twitch_Pubsub():
userinput,
data)
def callback_bits(self, uuid: UUID, data: dict) -> None:
print("Bits Redemption")
print('got callback for UUID ' + str(uuid))
pprint(data)
praxis_logger_obj.log(data['data']['user_name'])
try:
userinput = data['data']['chat_message']
praxis_logger_obj.log(data['data']['chat_message'])
except:
userinput = ""
praxis_logger_obj.log(data)
self.callback_EXEC(
data['data']['user_name'],
"TwitchBits",
AbstractChannelRewards.ChannelRewardsType.twitch_bits,
userinput,
data['data']['bits_used'],
data)
def callback_subs(self, uuid: UUID, data: dict) -> None:
print("Subs Redemption")
print('got callback for UUID ' + str(uuid))
pprint(data)
try:
userinput = data['data']['message']['sub_message']['message']
except:
userinput = ""
self.callback_EXEC(
data['data']['message']['display_name'],
"TwitchBits",
AbstractChannelRewards.ChannelRewardsType.twitch_subs,
userinput,
"",
data)
def callback_EXEC(self, sender, rewardName:str, rewardType, rewardPrompt, userInput, raw_data):
try:
is_actionable = self.is_reward(rewardName, rewardType)
if is_actionable:
praxis_logger_obj.log("Trying to do the thing")
if self.cooldownModule.isCooldownActive("twitchChat") == False:
if self.cooldownModule.isCooldownActive("twitchpubsub") == False:
self.exec_reward(sender, rewardName, rewardType, rewardPrompt, userInput, raw_data)
except:
print("something went wrong with a reward")

2
twitch_generate_credentials.py
View File

@ -16,7 +16,7 @@ class Twitch_Credential_Maker():
super().__init__()
self.credential : credentials.Twitch_Credential()
self.twitch : Twitch()
self.target_scope = [AuthScope.WHISPERS_READ, AuthScope.CHANNEL_READ_REDEMPTIONS]
self.target_scope = [AuthScope.WHISPERS_READ, AuthScope.CHANNEL_READ_REDEMPTIONS, AuthScope.BITS_READ, AuthScope.CHANNEL_READ_SUBSCRIPTIONS]
def get_tokens(self):
self.twitch.authenticate_app(self.target_scope)