IT Projekte / Home Lab / Smart Home / Games

Allgemein, Streaming

Twitch Chatbot TwitchIO Commands mit PyCharm 2023

Erstelle Deinen eigenen Chatbot, lege Commands an und sortiere sie in Cogs. Wie lange ist der Streamer schon online, auch das zeige ich Dir mit dem uptime Command.

Du benötigst Folgendes:

  • PyCharm installiert
  • Twitch Anwendung

Twitch Anwendung anlegen 2023

1. TwitchIO Packet installieren

Installiere TwitchIO mittels der Python Packages in PyCharm. Achte darauf, dass Dein Python Interpreter schon die Version 3.10.x verwendet. Da es hierfür von TwitchIO schon vereinfachte Befehle gibt.

TwitchIO - TwitchIO installieren
TwitchIO – TwitchIO installieren

2. Basis aufbauen

Lege in Deinem neuen Projekt eine neue Main.py Datei an.

In diese Datei schreibst Du zu Beginn erst einmal folgenden Code:

from twitchio.ext import commands


class Bot(commands.Bot):

    def __init__(self):
        # Initialise our Bot with our access token, prefix and a list of channels to join on boot...
        # prefix can be a callable, which returns a list of strings or a string...
        # initial_channels can also be a callable which returns a list of strings...
        super().__init__(token='DEIN_ACCESS_TOKEN', prefix='!', initial_channels=['DER_KANAL_CHANNEL'])

    async def event_ready(self):
        # Notify us when everything is ready!
        # We are logged in and ready to chat and use commands...
        print(f'Logged in as | {self.nick}')
        print(f'User id is | {self.user_id}')

    async def event_message(self, message):
        # Messages with echo set to True are messages sent by the bot...
        # For now we just want to ignore them...
        if message.echo:
            return

        # Print the contents of our message to console...
        print(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)

    @commands.command()
    async def hello(self, ctx: commands.Context):
        # Here we have a command hello, we can invoke our command with our prefix and command name
        # e.g ?hello
        # We can also give our commands aliases (different names) to invoke with.

        # Send a hello back!
        # Sending a reply back to the channel is easy... Below is an example.
        await ctx.send(f'Hello {ctx.author.name}!')


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

Aber was passiert hier im Einzelnen?

from twitchio.ext import commands

Um für dieses Beispiel die Commands nutzen zu können, müssen diese zunächst aus dem TwitchIO Paket importiert werden.

Danach legst Du eine Klasse an, die Klasse bekommt im Anschluss eine init Methode, in der Dein Access Token, Prefix, welcher vor Deinem Command steht und der Twitch Kanal angegeben werden. Füge nun hier Deinen Access Token ein. Dieser sollte von dem Twitch Account sein, welcher als Bot genutzt werden soll.

class Bot(commands.Bot):

    def __init__(self):
        # Initialise our Bot with our access token, prefix and a list of channels to join on boot...
        # prefix can be a callable, which returns a list of strings or a string...
        # initial_channels can also be a callable which returns a list of strings...
        super().__init__(token='DEIN_ACCESS_TOKEN', prefix='!', initial_channels=['DER_KANAL_CHANNEL'])

Die Methode event_ready gibt Dir im Log Bescheid, wenn Dein Bot eingeloggt und bereit zum Interagieren ist.

async def event_ready(self):
        # Notify us when everything is ready!
        # We are logged in and ready to chat and use commands...
        print(f'Logged in as | {self.nick}')
        print(f'User id is | {self.user_id}')

Dank der event_message-Methode kannst Du alle Nachrichten, welche in Deinen Kanälen geschrieben werden, lesen und damit bestimmt noch viel mehr anstellen, weil das Message-Objekt mehr als nur den Content bereitstellen kann. Hier möchten wir aber erst einmal, dass unser Bot nicht auf seine eigenen Nachrichten reagieren soll. Somit setzten wir ein return, wenn das echo true ist, weil dann kommt die Nachricht von Deinem Bot selbst.
Da Du aber die event_message-Methode überschrieben hast, kannst Du mit Commands nicht mehr interagieren. Dies musst Du wieder hinzufügen am Ende.

async def event_message(self, message):
        # Messages with echo set to True are messages sent by the bot...
        # For now we just want to ignore them...
        if message.echo:
            return

        # Print the contents of our message to console...
        print(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)

2.1 Command hello

Jetzt aber kommen wir zum wichtigen Teil, wie lege ich einen Command in meinem Bot an. Lege eine Methode an, welche vom Typ Command ist. Der Context eines Command gibt Dir zum Beispiel das Objekt Autor und dieses bringt Dir den Namen des Autors mit, wo durch Du persönlich den Command Auslöser ansprechen kannst und die Nachricht auch im Chat für den Nutzer hervorgehoben wird.

@commands.command()
    async def hello(self, ctx: commands.Context):
        # Here we have a command hello, we can invoke our command with our prefix and command name
        # e.g ?hello
        # We can also give our commands aliases (different names) to invoke with.

        # Send a hello back!
        # Sending a reply back to the channel is easy... Below is an example.
        await ctx.send(f'Hello {ctx.author.name}!')

Führe zum Schluss die Methode run von Deiner Klasse Bot aus, um Deinen Bot zu starten.

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

Führe die Datei aus und gehe in den Channel von dem Twitch Account, den Du in der init Methode angegeben hast. Gib hier einmal !hello ein und Du wirst im Chat eine Antwort Hello (User) erhalten und auch in der Konsole kannst Du sehe, dass eine Nachricht im Chat geschrieben wurde, wenn diese nicht von Deinem Bot selbst ist.

3. Cogs verwenden und Uptime Command

Lege einen neuen Ordner an und nenne diesen cogs. Cogs können dazu verwendet werden, um Deinen Python Code besser lesen zu können und diesen in eine Art Klasse aufzuteilen.
Lege ein neues Python File an und nenne dieses zum Beispiel API.

Diese Datei soll dafür dienen, mittels API Request in diesem Falle einmal die Uptime des Streams anzuzeigen. Wie lange der Stream schon läuft.

Diese Datei wird zu Beginn mit folgenden Zeilen Code befüllt:

import datetime

import requests
from twitchio.ext import commands

USER_LOGIN = 'USER_LOGIN'
CLIENT_ID = 'CLIENT_ID'
TOKEN = 'TOKEN'

class API(commands.Cog):

    def __init__(self, bot: commands.Bot):
        self.bot = bot

    @commands.command()
    async def uptime(self, ctx: commands.Context):
        # send API request
        url = f'https://api.twitch.tv/helix/streams?user_login={USER_LOGIN}'
        headers = {
            'Authorization': f'Bearer {TOKEN}',
            'Client-ID': CLIENT_ID
        }
        stream_request = requests.get(url, headers=headers)
        stream_data = stream_request.json()
        #print(stream_data)

        if "data" in stream_data.keys():
            if stream_data["data"]:
                datetime_now = datetime.datetime.now()
                stream_started_at = datetime.datetime.strptime(stream_data["data"][0]["started_at"],
                                                               "%Y-%m-%dT%H:%M:%SZ")
                raw_uptime = datetime_now - stream_started_at
                total_seconds = raw_uptime.total_seconds()
                days = int(total_seconds // 86400)
                hours = int((total_seconds % 86400) // 3600)
                minutes = int(((total_seconds % 86400) % 3600) // 60)
                seconds = int(total_seconds % 60)
                uptime = f"{days} Tagen, {hours} Stunden, {minutes}" \
                         f" Minuten, {seconds} Sekunden"
                #print(uptime)
                await ctx.send(f"/me Der Stream ist schon seit {uptime} online | @{ctx.author.name}")
            else:
                await ctx.send(f"/me Der Stream ist derzeit offline | @{ctx.author.name}")


def prepare(bot: commands.Bot):
    # Load our cog with this module...
    bot.add_cog(API(bot))

Lege zunächst die drei Variablen USER_LOGIN, CLIENT_ID und Access Token an.

USER_LOGIN = 'USER_LOGIN'
CLIENT_ID = 'CLIENT_ID'
TOKEN = 'TOKEN'

Darauf hin legst Du eine Klasse API an, welche vom Typ command.Cog erben soll. Dabei initialisierst Du in der __init__ Methode den Bot aus dem Main.py File.

class API(commands.Cog):

    def __init__(self, bot: commands.Bot):
        self.bot = bot

Jetzt kommt die Uptime Methode, welche con der commands.command Methode erbt. Zuerst musst Du den API Request senden, um in einem JSON die Informationen zu erhalten. Dabei legst Du zunächst die URL an und im Header deklarierst Du die Autorisierung und Client ID. Wird der Request ausgeführt, dann werden dieser in einem JSON in der Variable stream_data gespeichert.

@commands.command()
    async def uptime(self, ctx: commands.Context):
        # send API request
        url = f'https://api.twitch.tv/helix/streams?user_login={USER_LOGIN}'
        headers = {
            'Authorization': f'Bearer {TOKEN}',
            'Client-ID': CLIENT_ID
        }
        stream_request = requests.get(url, headers=headers)
        stream_data = stream_request.json()
        #print(stream_data)

Im zweiten Teil des Uptime Commands wird zunächst geprüft, ob der Key data im request verfügbar ist. Wenn nicht, dann ist der Stream offline. Wenn ja, dann erhält man viel mehr Informationen als nur bloß den Streamstart.
Mit datetime_now wird die aktuelle Uhrzeit gespeichert, um diese am Ende von der Startzeit abzuziehen. Dann wird die started_at Variable aus dem JSON in ebenfalls ein datetime Format konvertiert, um im Anschluss diese von der aktuellen Zeit abzuziehen. Ist dies geschehen, werden die reinen Sekunden aus dem Ergebnis umgerechnet und im Anschluss daran zur Anzeige gebracht.

        if "data" in stream_data.keys():
            if stream_data["data"]:
                datetime_now = datetime.datetime.now()
                stream_started_at = datetime.datetime.strptime(stream_data["data"][0]["started_at"],
                                                               "%Y-%m-%dT%H:%M:%SZ")
                raw_uptime = datetime_now - stream_started_at
                total_seconds = raw_uptime.total_seconds()
                days = int(total_seconds // 86400)
                hours = int((total_seconds % 86400) // 3600)
                minutes = int(((total_seconds % 86400) % 3600) // 60)
                seconds = int(total_seconds % 60)
                uptime = f"{days} Tagen, {hours} Stunden, {minutes}" \
                         f" Minuten, {seconds} Sekunden"
                #print(uptime)
                await ctx.send(f"/me Der Stream ist schon seit {uptime} online | @{ctx.author.name}")
            else:
                await ctx.send(f"/me Der Stream ist derzeit offline | @{ctx.author.name}")

Zum Ende der Datei muss auch wiederum diese Klasse dem Bot als Modul zugewiesen werden.

def prepare(bot: commands.Bot):
    # Load our cog with this module...
    bot.add_cog(API(bot))

Quellen:

Titelbild von TwitchIO und Twitch