A hacked-together Python Twitch chatbot using Twitchio

A lot of the code is taken from Twitchio’s extensive documentation (https://twitchio.dev/) and from their active Discord. The code below has some key pieces that are there to remind me how to do certain things, like send a single message from the bot, or send multiple messages. I’m sure there are better ways of doing the things coded here, but this has been working for me as the backbone of my personal Twitch chatbots for a couple of years. Maybe it’ll be helpful to you in your chatbot endeavors.

Getting the OAUTH tokens always confuses me, and makes me feel like I’m selling my soul and information to Big Data or clandestine hackers (or very overt hackers that I’m just willingly giving my info to). I used TwitchTokenGenerator to get my tokens. The Internet told me it was quality. 🙂 You can find it here: https://twitchtokengenerator.com/

I originally built my chatbots for the dnbradio.com chatroom IRC on Quakenet. Twitch chat is based on IRC, so there are some opportunities to branch out from here to other chat-based-shenanigans. Again, this is just the barebones version of the code. My current Twitch chatbot, HTPbot, has its own channel. The bot is filled with a bunch of convoluted and wonky code that makes the bot idiosyncratic, interactive, and fun. It integrates with WLED addressable lighting installations, communicates with out on-stream physical robot, and hosts a few games (like !hackingrules – the game about hacking rules).

There are three python scripts: 1. the main bot (bot_main.py) that handles the online communication and overall structure; 2. the script that handles the commands (cog_commands.py), like !help or !eatpoop; 3. the script that handles things like cheers, subscriptions, and stuff like that (cog_pubsub.py).

Thanks to the big work done by the Twitchio developers and community!

---

bot_main.py

# This is the main bot guts.
# It handles signing in to twitch, getting ready, sending messages, and redirecting commands to the cogs.
# cog_commands.py ==> handles general chatbot commands, like !help, !shoutout, etc.
# cog_pubsub.py ==> deals with subscriptions, channel points, and stuff like that

from twitchio.ext import commands, routines, eventsub
from random import choices
import asyncio

# you need the following Twitch API Scopes (https://dev.twitch.tv/docs/authentication/scopes/)
# bits:read
# channel:read:redemptions
# channel:manage:redemptions
ACCESS_TOKEN = "" 

prefix_char = "!"  # the prefix for commands

botnick = "Your Bot Name Here"
initial_channels = ['<your_channel_name>']


class Bot(commands.Bot):

    def __init__(self):
        super().__init__(token=ACCESS_TOKEN, prefix=prefix_char, initial_channels=initial_channels)
        self.esclient = eventsub.EventSubWSClient(self)

    async def event_ready(self):
        print(f'Logged in as | {self.nick}')
        print(f'User id is   | {self.user_id}')
        self.RT_info_reminder.start()
        print(f"Ready!")
        print()

    async def event_message(self, message):
        # Messages with echo set to True are messages sent by the bot.
        if message.echo:
            print(f"    {botnick}: {message.content}")
            return

        # Print the contents of our message to console...
        print(f"{message.author.name}: {message.content}")

        # Since we have commands and are overriding the default `event_message`
        # We must let the bot know we want to handle and invoke our commands.
        await self.handle_commands(message)

    async def send_multiple(self, temp_list, sleep_time=2):  # code to send multiple messages from a list
        channel = self.bot.get_channel(self.channel)
        for t in temp_list:
            await channel.send(t)
            await asyncio.sleep(sleep_time)
    
    async def send_message(self, MESSAGE):  # code to send multiple messages from a list
        channel = self.bot.get_channel(self.channel)
        await channel.send(MESSAGE)

    # This error replaces the one in ext.commands.bot.event_command_errors()
    # Comment it out for the real error to show up
    async def event_command_error(self, context: commands.Context, error: Exception):
        pass

    # ========== put COMMANDS here ========================================

    

    # ========== routines are below ========================================

    @routines.routine(minutes=14)
    async def junk_reminder(self):  # sends random 1s and 0s to the first channel in the list
        zero_ones = ["0", "1"]
        random_junk = ''.join(choices(zero_ones, k=3))
        channel = self.get_channel(initial_channels[0])
        await channel.send(f"Try {prefix_char}info. #.{random_junk}") # The random junk helps Twitch not think it's a duplicate message

    @RT_info_reminder.before_routine
    async def junk_reminder_before(self): # sends a welcome message before launching into the junk_reminder routine above
        channel = self.get_channel(initial_channels[0])
        await channel.send(f"Remember, there will be junk!")


    # ============================

    
bot = Bot()
initial_extensions = ["cog_commands", "cog_pubsub"] # the cogs handle stuff like commands, bits, subscriptions, etc. 
for extension in initial_extensions:
    bot.load_module(extension)

bot.run()
# bot.run() is blocking and will stop execution of any below code here until stopped or closed.

---

cog_commands.py

from twitchio.ext import commands
import asyncio

prefix_char = "!"  # the prefix for commands


class ChatMethodCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.channel_name = "your_channel_name"  # the channel id
       

    # ============= put commands below here ======================

    # === send single message ====================================

    @commands.command()
    async def test(self, ctx: commands.Context):
        '''
        Test the system
        '''
        await ctx.send(f'Testing of the Mugtion Testing system complete @{ctx.author.name}!')

    @commands.command(name="commands", aliases=["help", "imlost"])
    async def commands_command(self, ctx: commands.Context):
        '''
        List all the commands available
        '''
        commands_list = []
        for k in self._commands:
            commands_list.append(k)
        commands_str = ', '.join(commands_list)
        await ctx.send(commands_str)
 

    # === send multiple messages ================================

    # general function that sends multiple messages, useful for lists of messages
    async def send_multiple(self, temp_list, sleep_time=2):
        channel = self.bot.get_channel(self.channel_name)
        for t in temp_list:
            await channel.send(t)
            await asyncio.sleep(sleep_time)

    @commands.command()
    async def info(self, ctx: commands.Context):
        '''
        An example of sending multiple messages
        '''
        temp_list = ["Line one of test message", "This is another line of the test message"]
        task = asyncio.create_task(self.send_multiple(temp_list))
    
 
    # =========================================

def prepare(bot):
    bot.add_cog(ChatMethodCog(bot))

---

cog_pubsub.py

from twitchio.ext import pubsub, commands
from random import choice
import asyncio

USERS_OAUTH_TOKEN = ""
USERS_CHANNEL_ID = ""

prefix_char = "!"

class PubSubCog(commands.Cog):
    def __init__(self, bot):
        self.bot = bot
        self.bot.pubsub = pubsub.PubSubPool(bot)
        
        self.channel = "YOUR CHANNEL NAME"
        
    @commands.Cog.event()
    async def event_ready(self):
        topics = [
            pubsub.channel_points(USERS_OAUTH_TOKEN)[USERS_CHANNEL_ID],
            pubsub.bits(USERS_OAUTH_TOKEN)[USERS_CHANNEL_ID],]
        await self.bot.pubsub.subscribe_topics(topics)

    async def send_multiple(self, temp_list, sleep_time=2):  # code to send multiple messages from a list
        channel = self.bot.get_channel(self.channel)
        for t in temp_list:
            await channel.send(t)
            await asyncio.sleep(sleep_time)
    
    async def send_message(self, MESSAGE):  # code to send a single messsage
        channel = self.bot.get_channel(self.channel)
        await channel.send(MESSAGE)
    

    # ==== CHANNEL POINTS =============
    @commands.Cog.event()
    async def event_pubsub_channel_points(self, event: pubsub.PubSubChannelPointsMessage):
        await self.event_pubsub_channel_points2(event)

    async def event_pubsub_channel_points2(self, event: pubsub.PubSubChannelPointsMessage):
        # You could do this direct in the event if you wanted to
        MESSAGE = f"{event.user.name} used '{event.reward.title}' for {str(event.reward.cost)} Channel Points."
        print(MESSAGE)

        # =========== put channel points reward logic below here ==================
        if event.reward.title == "Get a new DJ name":
            dj1 = ["DJ", "Captain", "Flippin"]
            dj2 = ["Twinkie", "Dingus", "MadDog"]
            
            dj_name_1 = choice(dj1)
            dj_name_2 = choice(dj2)
            dj_name = f"{dj_name_1} {dj_name_2}"
            MESSAGE = f"{event.user.name} new DJ Name is: {dj_name}"
        
            await self.send_message(MESSAGE)
   


    # === BITS ====================

    @commands.Cog.event()
    async def event_pubsub_bits(self, event: pubsub.PubSubBitsMessage):
        # do stuff on bit redemptions
        print(f"{event.message.content = }")
        print(f"{event.user.name} redeemed {event.bits_used} bits")
        MESSAGE = f"Thanks @{event.user.name} for the bits! You're real swell!"
        
        self.send_message(MESSAGE)

    # === SUBSCRIPTIONS ==================

    @commands.Cog.event()
    async def event_pubsub_subscription(self, event: pubsub.channel_subscriptions):
        print(f"{event.user.name} subscribed!")

def prepare(bot):
    bot.add_cog(PubSubCog(bot))