feat: add twitch alerts

.env.local

@@ -1,4 +1,6 @@
 BOT_TOKEN=/secrets/bot-token
+TWITCH_CLIENT_ID=/secrets/twitch-client-id
+TWITCH_CLIENT_SECRET=/secrets/twitch-client-secret
 CONFIG_FILE=/config/config.yaml.local
 DB_DATABASE=SCUFFBOT
 DB_HOST=db

README.md

@@ -14,5 +14,7 @@ If you would like to run locally:
    | `bot-token` | Token for Discord bot |
    | `db-password` | Default password for database |
    | `db-root-password` | Root password for the database |
+   | `twitch-client-id` | Twitch app client id |
+   | `twitch-client-secret` | Twitch app client secret |
 4. Copy [/db/.env.template](./db/.env.template) to `db/.env`, and adjust values as necessary.
 5. Run the bot with `docker compose up -d --build`.

config.yaml

@@ -58,3 +58,7 @@ LEGACY_ROLES:
 
 SIX_MAN:
   CATEGORY: 1330751291223183491
+
+TWITCH_ALERTS:
+  TRIGGER: "scuffcord"
+  CHANNEL: 1497144541549957230

infra/values.yaml.gotmpl

@@ -14,11 +14,15 @@ deployments:
       DB_HOST: "service-scuffbot-db"
       DB_PASSWORD_FILE: "/secrets/db-password"
       DB_USER: "SCUFFBOT"
+      TWITCH_CLIENT_ID: "/secrets/twitch-client-id"
+      TWITCH_CLIENT_SECRET: "/secrets/twitch-client-secret"
     secretProviderClass:
       projectId: "e1420e9b-7c7d-4f54-b916-74d17f594a83"
       secrets:
         - secretKey: "BOT_TOKEN"
         - secretKey: "DB_PASSWORD"
+        - secretKey: "TWITCH_CLIENT_ID"
+        - secretKey: "TWITCH_CLIENT_SECRET"
 
 statefulSets:
   - name: "scuffbot-db"

src/lib/bot/__init__.py

@@ -25,6 +25,7 @@ def __init__(self, is_dev):
 
     async def setup_hook(self):
         self.setup_logger()
+        self.appinfo = await super().application_info()
         await self.load_cog_manager()
 
     def setup_logger(self):
@@ -45,7 +46,7 @@ def create_embed(self, title, description, colour):
         embed = discord.Embed(title=None, description=description,
                               colour=colour if colour else 0xDC3145, timestamp=discord.utils.utcnow())
         embed.set_author(name=title if title else None,
-                         icon_url=self.avatar_url)
+                         icon_url=self.appinfo.icon.url)
         return embed
 
     #  Doesn't work, need to look into
@@ -64,8 +65,6 @@ def is_developer(interaction: discord.Interaction):
         return interaction.user.id in config["DEVELOPERS"]
 
     async def on_ready(self):
-        self.appinfo = await super().application_info()
-        self.avatar_url = self.appinfo.icon.url if self.appinfo.icon is not None else None
         self.logger.info(
             f"Connected on {self.user.name} | d.py v{str(discord.__version__)}"
         )

src/lib/cogs/TwitchAlerts.py

@@ -0,0 +1,63 @@
+from dotenv import find_dotenv, load_dotenv
+from discord.ext import commands, tasks
+from discord.ui import Button, View
+from src.lib.bot import config
+import logging
+import requests
+import discord
+import os
+
+env_file = find_dotenv(".env.local")
+load_dotenv(env_file)
+
+
+class TwitchAlerts(commands.Cog):
+
+    with open(os.environ["TWITCH_CLIENT_ID"], "r", encoding="utf-8") as f:
+        client_id = f.readline().strip()
+
+    with open(os.environ["TWITCH_CLIENT_SECRET"], "r", encoding="utf-8") as f:
+        client_secret = f.readline().strip()
+
+    def __init__(self, bot):
+        self.bot = bot
+        self.logger = logging.getLogger(__name__)
+        self.bearer_token = None
+        self.livestreams = set()
+
+    async def cog_load(self):
+        self.logger.info(f"[COG] Loaded {self.__class__.__name__}")
+        self.check_live.start()
+
+    async def cog_unload(self):
+        self.check_live.stop()
+
+    @tasks.loop(minutes=5)
+    async def check_live(self):
+        await self.bot.wait_until_ready()
+        if self.bearer_token == None or (requests.get("https://id.twitch.tv/oauth2/validate", headers={"Authorization": f"OAuth {self.bearer_token}"})).status_code != 200:
+            self.bearer_token = (requests.post("https://id.twitch.tv/oauth2/token", params={"client_id": self.client_id, "client_secret": self.client_secret, "grant_type": "client_credentials"})).json()["access_token"]
+        if len(streams := (requests.get("https://api.twitch.tv/helix/streams", params={"user_login": config["TWITCH_ALERTS"]["TRIGGER"]}, headers={"Authorization": f"Bearer {self.bearer_token}", "Client-Id": self.client_id})).json()["data"]) == 0:
+            if (user_login := config["TWITCH_ALERTS"]["TRIGGER"]) in self.livestreams:
+                self.livestreams.remove(user_login)
+            return
+        stream = streams[0]
+        if ((user_login := stream["user_login"]) in self.livestreams):
+            return
+        self.livestreams.add(user_login)
+        embed = discord.Embed(title=stream["title"], description=f"**{stream['user_name']}** is now live streaming {stream['game_name']} on Twitch!", colour=0xDC3145, timestamp=discord.utils.utcnow())
+        embed.set_author(name="ScuffBot Twitch Alerts", icon_url=self.bot.user.display_avatar.url)
+        embed.set_image(url=stream["thumbnail_url"].replace("{width}", "1280").replace("{height}", "720"))
+        view = View()
+        view.add_item(
+            Button(
+                label="Watch on Twitch",
+                style=discord.ButtonStyle.link,
+                url=f"https://twitch.tv/{stream['user_login']}",
+            )
+        )
+        channel = await self.bot.fetch_channel(int(config["TWITCH_ALERTS"]["CHANNEL"]))
+        await channel.send("@everyone", embed=embed, view=view)
+
+async def setup(bot):
+    await bot.add_cog(TwitchAlerts(bot))