86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
import credentials
|
|
|
|
import config
|
|
|
|
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
|
|
|
|
|
|
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
|
|
|
|
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")
|
|
|
|
# TwitchAuthorizationException() gets raised by the listen_channel_points
|
|
def next(self):
|
|
user_id = self.twitch.get_users(logins=['my_username'])['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)
|
|
|
|
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() |