Praxis_Bot/standalone_twitch_pubsub.py

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

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 = ""
        #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://localhost: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://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)
            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()