chat_bot.py

#!/usr/bin/env python3

# Run with:
#   python chat_bot.py [Twitch or Youtube Livestream URL] [Twitch or Youtube Livestream URL] ...

import argparse
import asyncio
import json
import sys
from urllib.parse import urlparse, parse_qs

from googleapiclient.discovery import build
from googleapiclient.errors import HttpError

import irc.client
import irc.client_aio

import keyboard



# Coroutine which consumes messages from the queue.
#
# The `message_queue` parameter is an asyncio.Queue object.
async def chat_listener_task(message_queue):
    while True:
        item = await message_queue.get()
        print(f"Received Chat message: {item}")
        message_queue.task_done()



# Coroutine for adding YouTube chat messages to the queue.
async def list_youtube_stream_chats_task(youtube, live_chat_id, message_queue):
    request = youtube.liveChatMessages().list(
        liveChatId=live_chat_id,
        part="snippet",
        maxResults=2000
    )
    response = request.execute()
    while True:
        for item in response["items"]:
            message = item["snippet"]["displayMessage"]
            timestamp = item["snippet"]["publishedAt"]

            await message_queue.put(message)

        # YouTube API's refresh rate is a few seconds.
        # Receive HTTP 403 if request more frequently than this.
        await asyncio.sleep(5)

        # SEE: https://developers.google.com/resources/api-libraries/documentation/youtube/v3/python/latest/youtube_v3.liveChatMessages.html
        request = youtube.liveChatMessages().list_next(request, response)
        response = request.execute()



class TwitchChatListener:
    def __init__(self, channel_name, message_queue):
        self.channel_name = channel_name
        self.message_queue = message_queue

    def on_connect(self, connection, event):
        connection.join('#' + self.channel_name)

    def on_pubmsg(self, connection, event):
        message = event.arguments[0]
        self.message_queue.put_nowait(message)



if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Livestream Chat Reader")
    parser.add_argument("urls", nargs="+", metavar="URL", help="List of URLs of livestreams (Twitch or YouTube)")
    args = parser.parse_args()

    with open("config.json", "r") as f:
        config_json = f.read()

    try:
        config = json.loads(config_json)
    except json.JSONDecodeError as e:
        print(f"Error parsing JSON: {str(e)}")

    # Twitch bot credentials.
    # SEE: https://dev.twitch.tv/docs/irc/authenticate-bot/
    twitch_username = config["twitch"]["username"]
    twitch_token = config["twitch"]["token"]
    # YouTube API key.
    # SEE: https://developers.google.com/youtube/v3/quickstart/python#step_1_set_up_your_project_and_credentials
    youtube_api_key = config["youtube"]["api_key"]

    loop = asyncio.get_event_loop()

    message_queue = asyncio.Queue()

    # Listen to the chats in each of the given URLs.
    for url in args.urls:
        url_parts = urlparse(url)

        if "twitch.tv" in url_parts.netloc:
            channel_name = url_parts.path.lstrip("/")

            irc_client = irc.client_aio.AioReactor(loop=loop)
            try:
                connection = loop.run_until_complete(
                    irc_client.server().connect("irc.chat.twitch.tv", 6667, twitch_username, password=twitch_token)
                )
            except irc.client.ServerConnectionError:
                print(sys.exc_info()[1])
                raise SystemExit(1) from None

            listener = TwitchChatListener(channel_name, message_queue)

            connection.add_global_handler("welcome", listener.on_connect)
            connection.add_global_handler("pubmsg", listener.on_pubmsg)

        if "youtube" in url_parts.netloc:
            query_params = parse_qs(url_parts.query)
            video_id = query_params.get("v", [None])[0]

            # SEE: https://developers.google.com/resources/api-libraries/documentation/youtube/v3/python/latest/youtube_v3.liveChatMessages.html
            try:
                youtube = build("youtube", "v3", developerKey=youtube_api_key)

                # Retrieve the live chat ID for the livestream
                live_chat_id = youtube.videos().list(
                    part="liveStreamingDetails",
                    id=video_id
                ).execute()["items"][0]["liveStreamingDetails"]["activeLiveChatId"]

                _ = loop.create_task(list_youtube_stream_chats_task(youtube, live_chat_id, message_queue))
            except HttpError as e:
                print(f"An HTTP error occurred: {e}")
                sys.exit(1)

    _ = loop.create_task(chat_listener_task(message_queue))

    loop.run_forever()

config.json

{
    "twitch": {
        "username": "username",
        "token": "oauth:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
    },
    "youtube": {
        "api_key": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
    }
}

requirements.in

google-api-python-client
irc
keyboard

requirements.txt

#
# This file is autogenerated by pip-compile with Python 3.11
# by the following command:
#
#    pip-compile
#
autocommand==2.2.2
    # via jaraco-text
backports-tarfile==1.1.0
    # via jaraco-context
cachetools==5.3.3
    # via google-auth
certifi==2024.2.2
    # via requests
charset-normalizer==3.3.2
    # via requests
google-api-core==2.18.0
    # via google-api-python-client
google-api-python-client==2.126.0
    # via -r requirements.in
google-auth==2.29.0
    # via
    #   google-api-core
    #   google-api-python-client
    #   google-auth-httplib2
google-auth-httplib2==0.2.0
    # via google-api-python-client
googleapis-common-protos==1.63.0
    # via google-api-core
httplib2==0.22.0
    # via
    #   google-api-python-client
    #   google-auth-httplib2
idna==3.7
    # via requests
inflect==7.2.0
    # via jaraco-text
irc==20.4.0
    # via -r requirements.in
jaraco-collections==5.0.1
    # via irc
jaraco-context==5.3.0
    # via jaraco-text
jaraco-functools==4.0.0
    # via
    #   irc
    #   jaraco-text
    #   tempora
jaraco-logging==3.3.0
    # via irc
jaraco-stream==3.0.3
    # via irc
jaraco-text==3.12.0
    # via
    #   irc
    #   jaraco-collections
keyboard==0.13.5
    # via -r requirements.in
more-itertools==10.2.0
    # via
    #   inflect
    #   irc
    #   jaraco-functools
    #   jaraco-text
proto-plus==1.23.0
    # via google-api-core
protobuf==4.25.3
    # via
    #   google-api-core
    #   googleapis-common-protos
    #   proto-plus
pyasn1==0.6.0
    # via
    #   pyasn1-modules
    #   rsa
pyasn1-modules==0.4.0
    # via google-auth
pyparsing==3.1.2
    # via httplib2
pytz==2024.1
    # via
    #   irc
    #   tempora
requests==2.31.0
    # via google-api-core
rsa==4.9
    # via google-auth
tempora==5.5.1
    # via
    #   irc
    #   jaraco-logging
typeguard==4.2.1
    # via inflect
typing-extensions==4.11.0
    # via
    #   inflect
    #   typeguard
uritemplate==4.1.1
    # via google-api-python-client
urllib3==2.2.1
    # via requests