Bot to monitor the twitch chat

Hi
I’m making a bot to every time that a message is sent in the twitch chat, its should check to see if it doesn’t have a command on text (searching for !).

However my bot is not getting the twitch chat correctly, running the command search function only once, and not every time a message is sent in the chat.

Could you give me tips on how to improve this code?

import irc.bot
import requests

bot_username = 'The Big Boss'
channel_name = twitchchannel
access_token = oauth:mytoken
app_client_id = myappid
app_client_secret = mysecret

class TwitchBot(irc.bot.SingleServerIRCBot):
    def __init__(self, bot_username, channel_name, access_token, app_client_id, app_client_secret):
        self.bot_username = bot_username
        self.channel_name = channel_name
        self.access_token = access_token
        self.app_client_id = app_client_id
        self.app_client_secret = app_client_secret
        self.channel_id = ''
        server = 'irc.chat.twitch.tv'
        port = 6667

        #Authentication
        body = {
        'client_id': app_client_id,
        'client_secret': app_client_secret,
        "grant_type": 'client_credentials'
        }

        authentication = requests.post('https://id.twitch.tv/oauth2/token', body)
        keys = authentication.json()

        headers = {
            'Client-ID': app_client_id,
            'Authorization': 'Bearer ' + keys['access_token']

        }

        get_stream_info = requests.get('https://api.twitch.tv/helix/search/channels?query=' + channel_name, headers=headers)
        stream_info = get_stream_info.json()

        #get channel id
        for channel in stream_info["data"]:
            if channel['broadcaster_login'] == self.channel_name:
                self.channel_id = channel['id']
                print("id: ", self.channel_id)
                break

        #bot connection in chat
        print('Connecting to ' + server + ' on port ' + str(port) + '...')
        irc.bot.SingleServerIRCBot.__init__(self, [(server, port, access_token)], bot_username, bot_username)

    def on_welcome(self, c, e):
        print('Joining ' + self.channel_name)

        #You must request specific capabilities before you can use them
        c.cap('REQ', ':twitch.tv/membership')
        c.cap('REQ', ':twitch.tv/tags')
        c.cap('REQ', ':twitch.tv/commands')
        c.join(self.channel_name)
        print('Joined ' + self.channel_name)
        c.privmsg(self.channel_name, "Connected!")
        self.chatMonitoration(c, e)

    def chatMonitoration(self, c, e):
        print('chat monitoration started!')
        chatText = e.arguments

        #If a chat message starts with an exclamation point, try to run it as a command
        command_indicator = '!' 
        for text in chatText:
            if command_indicator in text:
                self.do_command(e, text)

    def do_command(self, e, command):
        c = self.connection
        print("command: ", command)

        #Get game status
        if command == "game":
            url = 'https://api.twitch.tv/kraken/channels/' + self.channel_id
            headers = {'Client-ID': self.client_id, 'Accept': 'application/vnd.twitchtv.v5+json'}
            r = requests.get(url, headers=headers).json()
            c.privmsg(self.channel_name, r['display_name'] + ' is currently playing ' + r['game'])

        #Get stream status
        elif command == "title":
            url = 'https://api.twitch.tv/kraken/channels/' + self.channel_id
            headers = {'Client-ID': self.client_id, 'Accept': 'application/vnd.twitchtv.v5+json'}
            r = requests.get(url, headers=headers).json()
            c.privmsg(self.channel_name, r['display_name'] + ' channel title is currently ' + r['status'])

       #Provide basic information to viewers

        elif command == "raffle":
            message = "This is an example bot, replace this text with your raffle text."
            c.privmsg(self.channel_name, message)

        elif command == "schedule":
            message = "This is an example bot, replace this text with your schedule text."
            c.privmsg(self.channel_name, message)

        #The command was not recognized
        else:
            c.privmsg(self.channel_name, "Did not understand command: " + command)

def main():
    """
    if bot_username and channel_name and access_token and app_client_id and app_client_secret:
        bot = TwitchBot(bot_username, channel_name, access_token, app_client_id, app_client_secret)
        bot.start()

    else: 
        print("Access information is invalid")
        sys.exit(1)
    """

    bot = TwitchBot(bot_username, channel_name, access_token, app_client_id, app_client_secret)
    bot.start()

if __name__ == "__main__":
    main()

You declared the bot username as “The Big Boss”

Twitch usernames can’t have spaces of capital letters in.
And then need to be a real user.

So sounds like you can’t connect at all?

You also need to generate a user access token to use as a “password” to login to chat with

You tried to generate a client_credentials token but that doesn’t represent a user. And can’t be used to login to Chat with.

See

You also used the search channels endpoint instead of any of the direct data endpoints.

And later in your file used kraken endpoints, kraken is deprecated and you need to migrate to the relevant helix endpoints. You can refer to the one page reference for the needed endpoints

1 Like

hello

I changed my code according to the links you sent me, but the bot is still unable to send messages in the chat and without reading the text of the messages sent in the twitch chat. When I run the script I get the following feedback on the python console: Welcome GLHF, and this is the only text identified by the bot.

import irc.bot

import requests

import json

server = 'irc.chat.twitch.tv'

port = 6667

bot_username = 'mytwitchusername'

access_token = 'oauth:mytoken'

channel_name = 'mytwitchchannelname"

app_client_id = 'myappid'

app_client_secret = 'myclientsecret'

class TwitchBot(irc.bot.SingleServerIRCBot):

    def __init__(self, server, port, bot_username, access_token, channel_name):

        self.server = server

        self.port = port

        self.bot_username = bot_username

        self.access_token = access_token

        self.channel_name = channel_name

        self.app_client_id = app_client_id

        self.app_client_secret = app_client_secret

        self.chat_connection(server, port, bot_username, access_token, channel_name)

    

    def chat_connection(self, server, port, bot_username, access_token, channel_name):

        irc.bot.SingleServerIRCBot.__init__(self, [(server, port, access_token)], bot_username, bot_username)

        print('Connected to ' + server + ' on port ' + str(port) + '...')

    def string_to_list(self, strings):

        return strings[0].split()

    def get_channel_id(self, app_client_id, app_client_secret):

        body = {

        'client_id': app_client_id,

        'client_secret': app_client_secret,

        "grant_type": 'client_credentials'

        }

        authentication = requests.post('https://id.twitch.tv/oauth2/token', body)

        keys = authentication.json()

        headers = {

            'Client-ID': app_client_id,

            'Authorization': 'Bearer ' + keys['access_token']

        }

        get_stream_info = requests.get('https://api.twitch.tv/helix/search/channels?query=' + channel_name, headers=headers)

        stream_info = get_stream_info.json()

        for channel in stream_info["data"]:

            if channel['broadcaster_login'] == self.channel_name:

                return channel['id']

    def on_welcome(self, connection, event):

        print(bot_username + ' joining in ' + self.channel_name)

        #You must request specific capabilities before you can use them

        connection.cap('REQ', ':twitch.tv/membership')

        connection.cap('REQ', ':twitch.tv/tags')

        connection.cap('REQ', ':twitch.tv/commands')

        connection.join(self.channel_name)

        print(bot_username + ' joined in ' + self.channel_name)

        connection.privmsg(event.target, bot_username + ' joined in ' + self.channel_name)

        self.chatMonitoration(connection, event)

    def chatMonitoration(self, connection, event):

        print(bot_username + ' is monitoring the chat')

        connection.privmsg(event.target, bot_username + ' is monitoring the chat')

        chatText = self.string_to_list(event.arguments)

        #If a chat message starts with an exclamation point, try to run it as a command

        print("searching for commands in text chat: ", chatText)

        command_indicator = '!'

        for text in chatText:

            if command_indicator in text:

                if text[0] == command_indicator:

                    self.do_command(event, text, app_client_id, app_client_secret)

    def do_command(self, event, command, app_client_id, app_client_secret):

        channel_id = self.get_channel_id(app_client_id, app_client_secret)

        

        print("command: ", command)

        match command:

            case "!game":

                url = 'https://api.twitch.tv/helix/channels/' + channel_id

                headers = {'Client-ID': self.client_id, 'Accept': 'application/vnd.twitchtv.v5+json'}

                r = requests.get(url, headers=headers).json()

                self.connection.privmsg(self.channel_name, r['display_name'] + ' is currently playing ' + r['game'])

            case "!title":

                url = 'https://api.twitch.tv/helix/channels/' + channel_id

                headers = {'Client-ID': self.client_id, 'Accept': 'application/vnd.twitchtv.v5+json'}

                r = requests.get(url, headers=headers).json()

                self.connection.privmsg(self.channel_name, r['display_name'] + ' channel title is currently ' + r['status'])

            case "!raffle":

                message = "This is an example bot, replace this text with your raffle text."

                self.connection.privmsg(self.channel_name, message)

            case "!schedule":

                message = "This is an example bot, replace this text with your schedule text."

                self.connection.privmsg(self.channel_name, message)

            case _:

                print(command + " is a invalid command, try again")

def main():

    """

    if bot_username and channel_name and access_token and app_client_id and app_client_secret:

        bot = TwitchBot(bot_username, channel_name, access_token, app_client_id, app_client_secret)

        bot.start()

    else: 

        print("Access information is invalid")

        sys.exit(1)

    """

    bot = TwitchBot(server, port, bot_username, access_token, channel_name)

    bot.start()

if __name__ == "__main__":

    main()

Channel names need to be preceeded by a #

So you joined a room that didn’t exist.

channel_name = 'mytwitchchannelname"

Should be

channel_name = '#mytwitchchannelname"

1 Like

Yes, the channel name was one of the problems, thank you.
Now the bot is able to connect to the chat and send messages, but still does not pick up the chat messages sent in chat after connecting, considering only the “Welcome GLHF” text.

Sounds like your bot is not processing incoming messages from the socket correctly.

Since there are more messages after the Welcome GLHF message.

1 Like

for some reason the problem was solved by changing the name of the chatMonitoration function to on_pubmsg.

Thanks !

This topic was automatically closed 30 days after the last reply. New replies are no longer allowed.