nise/nise-discord/bot.py

import asyncio
import logging
import os
import sys

import discord
from discord.ext import commands


def setup_logging() -> None:
    logging.getLogger("discord").setLevel(logging.INFO)
    logging.getLogger("discord.http").setLevel(logging.WARNING)

    log = logging.getLogger()
    log.setLevel(logging.INFO)

    log_format = logging.Formatter("%(asctime)s %(levelname)s - %(message)s", '%Y-%m-%d %H:%M:%S')

    stdout_handler = logging.StreamHandler(sys.stdout)
    stdout_handler.setFormatter(log_format)
    log.addHandler(stdout_handler)

    log.info('Done setting up logging...')


class NiseBot(commands.Bot):
    def __init__(self, bot_loop):
        intents = discord.Intents.default()
        intents.members = True
        super().__init__(
            command_prefix='!',
            help_command=None,
            intents=intents
        )

        self.loop = bot_loop
        self.log = logging.getLogger()

        self.reaction_channel_id = int(os.getenv("REACTION_CHANNEL_ID"))
        self.reaction_emoji_id = int(os.getenv("REACTION_EMOJI_ID"))

    async def on_ready(self):
        self.log.info('Bot loaded!')

    def run(self):
        super().run(os.getenv("DISCORD_TOKEN"), reconnect=True)

    async def on_raw_reaction_add(self, payload):
        if payload.channel_id != self.reaction_channel_id or payload.emoji.id != self.reaction_emoji_id:
            return

        role = discord.utils.get(payload.member.guild.roles, name="ping-when-update")
        await payload.member.add_roles(role)

    async def on_raw_reaction_remove(self, payload):
        if payload.channel_id != self.reaction_channel_id or payload.emoji.id != self.reaction_emoji_id:
            return

        guild_id = payload.guild_id
        guild = discord.utils.find(lambda g: g.id == guild_id, self.guilds)

        user_id = payload.user_id
        member = guild.get_member(user_id)

        role = discord.utils.get(guild.roles, name="ping-when-update")
        await member.remove_roles(role)

    async def on_command_error(self, ctx, error):
        if isinstance(error, commands.CommandOnCooldown):
            await ctx.send('This command is currently on cooldown. Please try again in a second!')
        else:
            self.log.error(error)


if __name__ == "__main__":
    setup_logging()
    loop = asyncio.get_event_loop()
    bot = NiseBot(bot_loop=loop)
    bot.run()