Advertisement
Not a member of Pastebin yet?
Sign Up,
it unlocks many cool features!
- import requests
- import socket
- import string
- import time
- # EXAMPLE USAGE
- #
- # from TwitchBot import TwitchBot
- #
- # ChatBot = TwitchBot("yourchannelname", "yourbotname", "yourbotpassword")
- #
- # while True:
- # ChatBot.update()
- # print(ChatBot.user + ": " + ChatBot.message)
- class TwitchBot:
- def __init__(self,channelName="",botName="",botPassword="",host="irc.twitch.tv",port=6667):
- # Setting up class variables
- self.channelName = channelName
- self.botName = botName
- self.botPassword = botPassword
- self.host = host
- self.port = port
- self.message = ""
- self.user = ""
- self.modList = []
- self.usrList = []
- # Setting up the connection
- self.socket = self.openSocket(self.channelName,self.botName,self.botPassword,self.host,self.port)
- if self.joinRoom(self.socket) == True:
- print("Successful connection to " + "www.twitch.tv/" + self.channelName + " as " + self.botName)
- def openSocket(self,CHAN,USER,PASS,HOST,PORT):
- s = socket.socket()
- s.connect((HOST, PORT))
- message = "PASS " + PASS + "\r\n"
- s.send(message.encode('utf-8'))
- message = "NICK " + USER + "\r\n"
- s.send(message.encode('utf-8'))
- message = "JOIN #" + CHAN + "\r\n"
- s.send(message.encode('utf-8'))
- return s
- def joinRoom(self, s):
- readbuffer = ""
- Loading = True
- while Loading:
- readbuffer = readbuffer + s.recv(1024).decode()
- temp = readbuffer.split('\n')
- readbuffer = temp.pop()
- for line in temp:
- Loading = self.loadingComplete(line)
- return True
- def loadingComplete(self, line):
- if("End of /NAMES list" in line):
- return False;
- else: return True
- def getUser(self, line):
- separate = line.split(":", 2)
- user = separate[1].split("!", 1)[0]
- return user
- def getMessage(self, line):
- separate = line.split(":", 2)
- if(separate[0] == 'PING '):
- message = self.getPing(line)
- else:
- message = separate[2]
- return message
- def getPing(self, line):
- separate = line.split(":", 2)
- message = separate[1]
- return message
- def sendMessage(self, message):
- messageTemp = "PRIVMSG #" + self.channelName + " :" + message + "\r\n"
- self.socket.send(messageTemp.encode('utf-8'))
- def update(self):
- # Read the next chat message, if any
- readbuffer = ""
- try:
- readbuffer = readbuffer + s.recv(1024).decode("utf-8")
- except:
- self.socket = self.openSocket(self.channelName,self.botName,self.botPassword,self.host,self.port)
- self.joinRoom(self.socket)
- readbuffer = readbuffer + self.socket.recv(1024).decode("utf-8")
- temp = readbuffer.split('\n')
- readbuffer = temp.pop()
- for line in temp:
- if line.startswith("PING "):
- strSend = "PONG :tmi.twitch.tv\r\n".encode('utf-8')
- self.socket.send(strSend)
- lastMsg = self.message
- usr = self.getUser(line)
- msg = self.getMessage(line)
- # Do some maintenance on the data
- self.message = msg.strip("\n")
- self.message = self.message.strip("\r")
- self.user = usr.strip("\n")
- self.user = self.user.strip("\r")
- # Everything below here is optional. Used only if you want the ability to grab the list of viewers/mods
- def updateLists(self):
- response = requests.get("https://tmi.twitch.tv/group/user/" + self.channelName + "/chatters")
- findStr = response.text
- splitList = []
- modListTmp = []
- usrListTmp = []
- # Sloppy web scraping
- for i in range(0,5):
- begin = findStr.find("[")
- end = findStr.find("]")
- subStr = findStr[begin+1:end]
- subStrTokens = subStr.split("\"")
- for val in subStrTokens:
- subStrTokens[val.find(val)] = val.strip(",")
- if i == 0:
- for token in subStrTokens:
- if token != "" and token[0].isalnum():
- modListTmp.append(token)
- if i == 4:
- for token in subStrTokens:
- try:
- if token[0].isalnum():
- usrListTmp.append(token)
- except: pass
- findStr = findStr[end+1:]
- # Update the lists all at once
- self.usrList = usrListTmp[:]
- self.modList = modListTmp[:]
- def getUsers(self):
- self.updateLists()
- return self.usrList
- def getMods(self):
- self.updateLists()
- return self.modList
- def getNumViewers(self):
- self.updateLists()
- return len(self.usrList) + len(self.modList)
Advertisement
Add Comment
Please, Sign In to add comment
Advertisement