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 from twitchAPI.types import AuthScope from twitchAPI.oauth import UserAuthenticator from pprint import pprint from uuid import UUID from cooldowns import Cooldown_Module import os import praxis_logging praxis_logger_obj = praxis_logging.praxis_logger() praxis_logger_obj.init(os.path.basename(__file__)) praxis_logger_obj.log("\n -Starting Logs: " + os.path.basename(__file__)) class Twitch_Pubsub(): def __init__(self): super().__init__() self.credential : credentials.Twitch_Credential() self.twitch : Twitch() self.pubsub: PubSub() self.target_scope = [AuthScope.WHISPERS_READ, AuthScope.CHANNEL_READ_REDEMPTIONS] 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) self.twitch.set_user_authentication(self.credential.pubsub_AccessToken, self.target_scope, self.credential.pubsub_RefreshToken) def get_tokens(self): self.twitch.authenticate_app(self.target_scope) for scope_ in self.target_scope: print(scope_) auth = UserAuthenticator(self.twitch, self.target_scope, force_verify=True) token, refresh_token = auth.authenticate() if token is not None: print("found token") if refresh_token is not None: print("found refresh_token") print(token) print(refresh_token) self.twitch.set_user_authentication(token, self.target_scope, refresh_token) def start(self): self.pubsub = PubSub(self.twitch) #self.pubsub.ping_frequency = 30 self.pubsub.start() print("started") def next(self): 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) #input('press ENTER to close...') def stop(self): self.pubsub.unlisten(self.uuid_1) self.pubsub.unlisten(self.uuid_2) self.pubsub.stop() def callback_whisper(self, uuid: UUID, data: dict) -> None: print('got callback for UUID ' + str(uuid)) pprint(data) def callback_channelPoints(self, uuid: UUID, data: dict) -> None: 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 = "" praxis_logger_obj.log("\n\n") praxis_logger_obj.log(data['data']['redemption']['user']['display_name']) praxis_logger_obj.log(data['data']['redemption']['reward']['title']) praxis_logger_obj.log(AbstractChannelRewards.ChannelRewardsType.channelPoints) praxis_logger_obj.log(data['data']['redemption']['reward']['prompt']) praxis_logger_obj.log(userinput) praxis_logger_obj.log(data) 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: praxis_logger_obj.log("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() credentials_manager = credentials.Credentials_Module() credentials_manager.load_credentials() testModule.credential = credentials_manager.find_Twitch_Credential(config.credentialsNickname) testModule.twitch = Twitch(testModule.credential.pubsub_client_id, testModule.credential.pubsub_secret) #pprint(testModule.twitch.get_users(logins=['thecuriousnerd'])) #testModule.get_tokens() testModule.setup() testModule.start() testModule.next() #testModule.stop()