Discord Bot That Announces Your Twitch Streams and YouTube Videos

discord.py
aiohttp
python-dotenv

DISCORD_TOKEN=paste-the-bot-token-here
ALERTS_CHANNEL=123456789012345678
MENTION=<@&123456789012345678>
TWITCH_CLIENT_ID=your-client-id
TWITCH_CLIENT_SECRET=your-client-secret
TWITCH_USERS=your_channel,your_friends_channel
YOUTUBE_CHANNELS=UCxxxxxxxxxxxxxxxxxxxxxx

import json
import os
import xml.etree.ElementTree as ET

import aiohttp
import discord
from discord.ext import tasks
from dotenv import load_dotenv

load_dotenv()  # reads the .env

ALERTS_CHANNEL = int(os.environ["ALERTS_CHANNEL"])
TWITCH_ID = os.environ.get("TWITCH_CLIENT_ID", "")
TWITCH_SECRET = os.environ.get("TWITCH_CLIENT_SECRET", "")
TWITCH = [u.strip().lower() for u in os.environ.get("TWITCH_USERS", "").split(",") if u.strip()]
YOUTUBE = [c.strip() for c in os.environ.get("YOUTUBE_CHANNELS", "").split(",") if c.strip()]
MENTION = os.environ.get("MENTION", "")  # <@&ROLE_ID>, @everyone or nothing
FILE = "announced.json"  # what was already announced, so restarts don't repeat it
NS = {"a": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015"}

client = discord.Client(intents=discord.Intents.default())
twitch_token = None

try:
    with open(FILE) as f:
        announced = json.load(f)
except FileNotFoundError:
    announced = {"twitch": {}, "youtube": {}}


async def get_token(session):
    global twitch_token
    data = {"client_id": TWITCH_ID, "client_secret": TWITCH_SECRET, "grant_type": "client_credentials"}
    async with session.post("https://id.twitch.tv/oauth2/token", params=data) as r:
        r.raise_for_status()
        twitch_token = (await r.json())["access_token"]


async def live_streams(session):
    """The TWITCH_USERS channels that are live right now."""
    if not TWITCH:
        return []
    if twitch_token is None:
        await get_token(session)
    for _ in range(2):
        headers = {"Client-Id": TWITCH_ID, "Authorization": f"Bearer {twitch_token}"}
        query = [("user_login", u) for u in TWITCH]
        async with session.get("https://api.twitch.tv/helix/streams", params=query, headers=headers) as r:
            if r.status == 401:  # the token expires: get another and retry
                await get_token(session)
                continue
            r.raise_for_status()
            return (await r.json())["data"]
    return []


async def videos(session, channel):
    """The latest videos of a YouTube channel, newest first."""
    url = "https://www.youtube.com/feeds/videos.xml"
    async with session.get(url, params={"channel_id": channel}) as r:
        r.raise_for_status()
        root = ET.fromstring(await r.text())
    return [
        {
            "id": e.find("yt:videoId", NS).text,
            "title": e.find("a:title", NS).text,
            "author": e.find("a:author/a:name", NS).text,
            "link": e.find("a:link", NS).get("href"),
        }
        for e in root.findall("a:entry", NS)
    ]


@tasks.loop(minutes=2)
async def watch():
    channel = client.get_channel(ALERTS_CHANNEL) or await client.fetch_channel(ALERTS_CHANNEL)
    async with aiohttp.ClientSession() as session:
        try:
            for s in await live_streams(session):
                if announced["twitch"].get(s["user_login"]) == s["id"]:
                    continue  # this stream was already announced
                announced["twitch"][s["user_login"]] = s["id"]
                card = discord.Embed(
                    title=s["title"] or "Live",
                    url=f"https://www.twitch.tv/{s['user_login']}",
                    description=f"Playing {s['game_name']}" if s["game_name"] else None,
                    color=0x9146FF,
                )
                card.set_image(url=s["thumbnail_url"].replace("{width}", "1280").replace("{height}", "720"))
                await channel.send(f"{MENTION} 🔴 **{s['user_name']}** is live".strip(), embed=card)
        except Exception as error:  # Twitch isn't answering: try again next round
            print("Twitch:", error)

        for c in YOUTUBE:
            try:
                items = await videos(session, c)
            except Exception as error:  # the feed sometimes fails: next round
                print("YouTube:", c, error)
                continue
            seen = announced["youtube"].get(c)
            announced["youtube"][c] = [v["id"] for v in items]
            if seen is None:
                continue  # the first time it only notes the existing ones
            for v in reversed(items):
                if v["id"] not in seen:
                    await channel.send(f"{MENTION} 📺 **{v['author']}** uploaded “{v['title']}”\n{v['link']}".strip())

    with open(FILE, "w") as f:
        json.dump(announced, f)


@client.event
async def on_ready():
    print(f"Logged in as {client.user}")
    if not watch.is_running():
        watch.start()


client.run(os.environ["DISCORD_TOKEN"])