Test for Eventsub

haugebot_twitch/eventsub_cog.py

+import os
+from functools import partial
+
+from twitchio.ext import eventsub, commands
+from dotenv import load_dotenv
+
+load_dotenv()
+
+client_id = os.getenv("CLIENT_ID")
+client_secret = os.getenv("CLIENT_SECRET")
+client = commands.Bot.from_client_credentials(client_id, client_secret)
+
+class EventSubCog(commands.Cog):
+    def __init__(self, bot):
+        global client
+
+        self.bot = bot
+        self.eventsub_client = eventsub.EventSubClient(client, "trololololo",
+                                                       "https://bottest.copycat-games.de/haugebot/callback")
+        self.bot.loop.create_task(self.eventsub_client.listen(port=23456))
+
+    async def subscribe(self):
+        for subscription in await self.eventsub_client.get_subscriptions():
+            await self.eventsub_client.delete_subscription(subscription.id)
+        await self.eventsub_client.subscribe_channel_stream_start(87637599)
+        await self.eventsub_client.subscribe_channel_stream_end(87637599)
+        lol = await self.eventsub_client.get_subscriptions()
+        print(lol)
+
+    @client.event()
+    async def event_eventsub_notification_stream_start(payload: eventsub.NotificationEvent):
+        print(payload)
+
+    @client.event()
+    async def event_eventsub_notification_stream_end(payload: eventsub.NotificationEvent):
+        print(payload)

haugebot_twitch/haugebot.py

 import os
 from abc import ABC
 
+import twitchio
 from dotenv import load_dotenv
 from twitchio import Channel, Message
 from twitchio.ext.commands import Context, Bot
@@ -8,6 +9,10 @@ from twitchio.ext.commands import Context, Bot
 from vote_cog import VoteCog
 from wusstest_du_schon import WusstestDuSchon
 from wordcloud import Wordcloud
+from eventsub_cog import EventSubCog
+
+import logging
+logging.basicConfig(filename="haugebot.log", filemode="w", level=logging.DEBUG, )
 
 
 class HaugeBot(Bot, ABC):
@@ -24,6 +29,7 @@ class HaugeBot(Bot, ABC):
         self.add_cog(VoteCog(self))
         self.add_cog(WusstestDuSchon(self))
         self.add_cog(Wordcloud(self))
+        self.add_cog(EventSubCog(self))
 
     @staticmethod
     async def send_me(ctx, content):
@@ -41,6 +47,8 @@ class HaugeBot(Bot, ABC):
             wusstest_du_schon.loop.start()
         if vote_cog := self.cogs.get("VoteCog"):
             vote_cog.manage_vote.start()
+        if eventsub_cog := self.cogs.get("EventSubCog"):
+            await eventsub_cog.subscribe()
 
     @staticmethod
     def get_percentage(part, total):