Bot de Discord que avisa de tus directos de Twitch y tus vídeos de YouTube

discord.py
aiohttp
python-dotenv

DISCORD_TOKEN=pega-aquí-el-token-del-bot
CANAL_AVISOS=123456789012345678
MENCION=<@&123456789012345678>
TWITCH_CLIENT_ID=tu-client-id
TWITCH_CLIENT_SECRET=tu-client-secret
TWITCH_USUARIOS=tu_canal,el_de_tu_amigo
YOUTUBE_CANALES=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()  # lee el .env

CANAL_AVISOS = int(os.environ["CANAL_AVISOS"])
TWITCH_ID = os.environ.get("TWITCH_CLIENT_ID", "")
TWITCH_SECRETO = os.environ.get("TWITCH_CLIENT_SECRET", "")
TWITCH = [u.strip().lower() for u in os.environ.get("TWITCH_USUARIOS", "").split(",") if u.strip()]
YOUTUBE = [c.strip() for c in os.environ.get("YOUTUBE_CANALES", "").split(",") if c.strip()]
MENCION = os.environ.get("MENCION", "")  # <@&ID_DEL_ROL>, @everyone o nada
FICHERO = "avisados.json"  # lo ya avisado, para no repetir al reiniciar
NS = {"a": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015"}

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

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


async def pedir_token(sesion):
    global token_twitch
    datos = {"client_id": TWITCH_ID, "client_secret": TWITCH_SECRETO, "grant_type": "client_credentials"}
    async with sesion.post("https://id.twitch.tv/oauth2/token", params=datos) as r:
        r.raise_for_status()
        token_twitch = (await r.json())["access_token"]


async def directos(sesion):
    """Los canales de TWITCH_USUARIOS que están en directo ahora mismo."""
    if not TWITCH:
        return []
    if token_twitch is None:
        await pedir_token(sesion)
    for _ in range(2):
        cabeceras = {"Client-Id": TWITCH_ID, "Authorization": f"Bearer {token_twitch}"}
        consulta = [("user_login", u) for u in TWITCH]
        async with sesion.get("https://api.twitch.tv/helix/streams", params=consulta, headers=cabeceras) as r:
            if r.status == 401:  # el token caduca: se pide otro y se repite
                await pedir_token(sesion)
                continue
            r.raise_for_status()
            return (await r.json())["data"]
    return []


async def videos(sesion, canal):
    """Los últimos vídeos de un canal de YouTube, del más nuevo al más viejo."""
    url = "https://www.youtube.com/feeds/videos.xml"
    async with sesion.get(url, params={"channel_id": canal}) as r:
        r.raise_for_status()
        raiz = ET.fromstring(await r.text())
    return [
        {
            "id": e.find("yt:videoId", NS).text,
            "titulo": e.find("a:title", NS).text,
            "autor": e.find("a:author/a:name", NS).text,
            "enlace": e.find("a:link", NS).get("href"),
        }
        for e in raiz.findall("a:entry", NS)
    ]


@tasks.loop(minutes=2)
async def vigilar():
    canal = client.get_channel(CANAL_AVISOS) or await client.fetch_channel(CANAL_AVISOS)
    async with aiohttp.ClientSession() as sesion:
        try:
            for d in await directos(sesion):
                if avisados["twitch"].get(d["user_login"]) == d["id"]:
                    continue  # de este directo ya se avisó
                avisados["twitch"][d["user_login"]] = d["id"]
                tarjeta = discord.Embed(
                    title=d["title"] or "En directo",
                    url=f"https://www.twitch.tv/{d['user_login']}",
                    description=f"Jugando a {d['game_name']}" if d["game_name"] else None,
                    color=0x9146FF,
                )
                tarjeta.set_image(url=d["thumbnail_url"].replace("{width}", "1280").replace("{height}", "720"))
                await canal.send(f"{MENCION} 🔴 **{d['user_name']}** está en directo".strip(), embed=tarjeta)
        except Exception as error:  # Twitch no responde: se prueba en la siguiente vuelta
            print("Twitch:", error)

        for c in YOUTUBE:
            try:
                lista = await videos(sesion, c)
            except Exception as error:  # el feed a veces falla: siguiente vuelta
                print("YouTube:", c, error)
                continue
            vistos = avisados["youtube"].get(c)
            avisados["youtube"][c] = [v["id"] for v in lista]
            if vistos is None:
                continue  # la primera vez solo apunta los que ya había
            for v in reversed(lista):
                if v["id"] not in vistos:
                    await canal.send(f"{MENCION} 📺 **{v['autor']}** ha subido «{v['titulo']}»\n{v['enlace']}".strip())

    with open(FICHERO, "w") as f:
        json.dump(avisados, f)


@client.event
async def on_ready():
    print(f"Conectado como {client.user}")
    if not vigilar.is_running():
        vigilar.start()


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