This one was fun, first time playing with IRC in Python.

The idea is simple, StreamElements gives you points when you're a part of the Twitch Chat. As Twitch's chat is based on IRC, staying in the channel should be enough.

Previously, we had this endpoint, but it's been deprecated in favour of this one (which requires OAuth from the streamer in question). This endpoint gives us a list of everyone connected to the channel, and is presumably what SE uses to allot points. It was nice to have it while testing!

I was inspired by jschlenker/twitch-multistream-chat, it wasn't working at the time and the developer didn't abstract away much on the IRC Side, prompting a rewrite instead of a monkey-patch.

First steps with IRC

First, let's connect to Twitch's IRC Server!

bot = bottom.Client(host="irc.chat.twitch.tv", port=6697, ssl=True)

Bottom felt comfortable to use as a beginner because it made IRC a lot simpler and completely abstracted away sockets. Instead, we have a nify decorator-based dispatch system (similar to discord.py and fastapi events).

Authenticating

@bot.on("CLIENT_CONNECT")
async def connect(**kwargs):
    bot.send("PASS", password=oauth_token)
    bot.send("NICK", nick=bot_username)
    bot.send_raw("CAP REQ :twitch.tv/membership")

The authentication flow requires an OAuth token (which the user supplies in the config) and our username. The CAP REQ is akin to defining our scopes, Twitch calls these Twitch-specific Capabilities.

We just want to join the room (the streamer's chat), so we request the :twitch.tv/membership capability.

Joining a Channel

Once we're connected to the server, we can join a room.

bot.send("JOIN", channel=f"#{room_name}")

room_name would be the streamer's name.

Straightforward so far!

KeepAlive

Now, we just need to keep the connection alive.

This is a simple ping-pong sequence,

@bot.on("PING")
def keepalive(message, **kwargs):
    bot.send("PONG", message=message)

The docs for which are here.

That's it! We're sitting in the chat and AFKing points. Gonna use them to buy silly things later.

Managing multiple streamers

Now, we only want to be in streamer's chats when they're streaming. We want to leave when they stop, and join when they start!

Twitch API for Streamer Status

We'll need to fiddle with OAuth: exchanging the Client ID and Secret (from the config) for an access token - This access token allows us to make API Requests on the User's behalf.

async def retrieve_access_token(
    client_id: str, client_secret: str
):
    """Retrieves the access token by sending a client credentials request to Twitch."""

    body = {
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "client_credentials",
    }

    async with aiohttp.ClientSession() as session:
        async with session.post("https://id.twitch.tv/oauth2/token", json=body) as response:
            response_data = await response.json()

    access_token = response_data.get("access_token")
    
    return access_token

Most pleasant OAuth devex (starkly contrasted by Michaelsoft 😬).

Using the Access Token to find Online Streamers

The fun part! We get to use our shiny access token to request API Resources. Right now, we want to check if a streamer is online:

async def retrieve_streaming_status(
    client_id: str,
    access_token: str,
    streamer_name: str,
):
    """Retrieves streamer status, ie, whether they're streaming or not at the moment."""

    headers = {"Client-ID": client_id, "Authorization": "Bearer " + access_token}

    async with aiohttp.ClientSession() as session:
        async with session.get(
            f"https://api.twitch.tv/helix/streams?user_login={streamer_name}",
            headers=headers,
        ) as response:
            response_data = await response.json()

    return {streamer_name: len(response_data["data"]) == 1}

The Client ID and Access Token from earlier are used to retrieve a list of running streams from the streamer in question. The Helix API gives us a list of these streams and their URLs. If the streamer is offline, the list returned is empty.

The datastructure is a bit weird in isolation, but we'll be deploying this using asyncio.gather - which runs coros concurrently.

This is what it looks like put together:

async def get_alive_streamers(
    streamers: List[str]
):
    """Gets all alive streamers"""

    access_token = await retrieve_access_token(
        config["client_id"], config["client_secret"]
    )

    data = await asyncio.gather(
        *[
            retrieve_streaming_status(
                config["client_id"],
                access_token=access_token,
                streamer_name=streamer,
            )
            for streamer in streamers
        ]
    )

    result = [
        k for k, v in dict(ChainMap(*data)).items() if v is True
    ]  # A list of all the streamers whose values are True, ie, a list of channels that are streaming

    return result

The ChainMap combines multiple dictionaries into one, for example:

dict(ChainMap({'a': 1}, {'b': 2}))
> {'b': 2, 'a': 1}

In our case, that becomes {streamer_one: True, streamer_two: False}, for whether or not they're streaming currently.

We'll then filter for the value and only retain streamers that are currently online, and return them as a list.

The get_alive_streamers (should've called it get_online_streamers instead lol) (you can make a PR for this) function is run periodically.

We now have a list of previously online streamers, and currently online streamers. Taking both as sets, we can find their difference to find who went offline, and who just started streaming. Leaving and joining their chats respectively.

That's it! Automatically join streamers' chats as they go live, and leave old chats after the stream stops.

Twitch's API Documentation was really handy, and numberoverzero/bottom, the Async IRC Client used for this project, was a pleasure to work with!

Thanks for reading :)

Deployment Data

{
  "Repository": "https://github.com/TheOnlyWayUp/Twitch-Chat-Joiner"
}