Source code for src.discord_bot
"""
This module contains the discord bot implementation with definitions for the bot and its commands.
The bot is implemented using the discord.py library and provides a simple command to test the bot.
"""
import discord
from discord.ext import commands, tasks
from .game_setup import setup_game, evaluate_game
from .file_utils import import_tasks, export_tasks
from .game_1 import practice_game1, game1
from .game import show_league_table
from .reaction_tracker import schedule_reaction_tracker_add, schedule_reaction_tracker_remove
[docs]
class DiscordBot:
"""
DiscordBot class to create a discord bot with the given configuration. This is
necessary because of a own implementation with user configuration and
pydantic validation.
"""
def __init__(self, config):
self.config = config
intents = discord.Intents.default()
intents.members = True
intents.message_content = True
intents.reactions = True
self.bot = commands.Bot(command_prefix="!", intents=intents)
@self.bot.event
async def on_ready():
await self.on_ready()
@self.bot.event
async def on_raw_reaction_add(payload):
if payload.user_id == self.bot.user.id:
return
await schedule_reaction_tracker_add(self.bot, self.config, payload)
@self.bot.event
async def on_raw_reaction_remove(payload):
# Not possible to check if the bot is the user who removed the reaction
# if payload.user_id == self.bot.user.id:
# return
await schedule_reaction_tracker_remove(self.config, payload)
self.register_commands()
[docs]
async def start(self):
"""
Function to start the bot with the given token from the configuration.
This function is called in the main function to start the bot.
"""
await self.bot.start(self.config.dc.token)
[docs]
async def on_ready(self):
"""
Event function to print a message when the bot is online.
"""
self.config.watcher.logger.info(f"{self.bot.user} ist online")
synced = await self.bot.tree.sync()
self.config.watcher.logger.info(f"Slash Commands synchronisiert: {len(synced)}")
await self.bot.change_presence(
status=discord.Status.online, activity=discord.Game("Don't Starve Together")
)
self.config.watcher.logger.info("start reaction tracker")
self.reaction_tracker.start()
[docs]
def register_commands(self):
"""
Function to register the commands for the bot. This function is called in the
constructor to register the commands.
"""
async def wrapped_game1_command(interaction: discord.Interaction):
await game1(interaction, self.config)
async def wrapped_evaluate_game(interaction: discord.Interaction):
await evaluate_game(interaction, self.config)
async def wrapped_practice_game1_command(interaction: discord.Interaction):
await practice_game1(interaction, self.config)
async def wrapped_setup_game(interaction: discord.Interaction):
await setup_game(interaction, self.config)
async def wrapped_show_league_table(interaction: discord.Interaction):
await show_league_table(interaction, self.config)
async def wrapped_import_tasks(interaction: discord.Interaction):
await import_tasks(interaction, self.config)
async def wrapped_export_tasks(interaction: discord.Interaction):
await export_tasks(interaction, self.config)
self.bot.tree.command(
name="fast_and_hungry_task_hunt",
description=(
"Complete all tasks and survive. The game ends as "
"soon as one player has completed all tasks"
),
)(wrapped_game1_command)
self.bot.tree.command(
name="evaluate_game",
description=(
"Evaluate the game, check all reaktions and calculate the winner."
),
)(wrapped_evaluate_game)
self.bot.tree.command(
name="prac_fast_and_hungry_task_hunt",
description=(
"Practice the game 'Fast and hungry, "
"task hunt' with a user selection menu."
),
)(wrapped_practice_game1_command)
self.bot.tree.command(
name="setup_game",
description="Switch game state to specific status like running, paused, finished, etc.",
)(wrapped_setup_game)
self.bot.tree.command(
name="show_league_table",
description="Show the current league table with all players and their scores.",
)(wrapped_show_league_table)
self.bot.tree.command(
name="import_tasks",
description="Import and update current tasks from an Excel spreadsheet to database.",
)(wrapped_import_tasks)
self.bot.tree.command(
name="export_tasks",
description="Export current tasks from database to an Excel spreadsheet.",
)(wrapped_export_tasks)
@tasks.loop(seconds=10)
async def reaction_tracker(self):
"""
Experimental reaction tracker task that checks for reactions
"""
# await schedule_reaction_tracker(self.bot, self.config)
[docs]
@reaction_tracker.before_loop
async def init_reaction_tracker(self):
"""
Function to initialize the reaction tracker before it starts.
"""
await self.bot.wait_until_ready()