Archived
generated from imcompiling/discord-bot-template
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Empty init file for off_cooldown_bot module."""
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Cog loader. Adds all command cogs to the bot."""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from cogs.commands.birthday import BirthdayCog
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
# It is necessary to add these functions to the bot.
|
||||
def setup(bot: commands.Bot) -> None:
|
||||
"""Add all command classes to bot."""
|
||||
bot.add_cog(BirthdayCog())
|
||||
@@ -0,0 +1,94 @@
|
||||
"""Contains commands for birthday feature."""
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from pytz import UTC
|
||||
from sqlmodel import Session, col, select
|
||||
|
||||
from database.db import engine
|
||||
from database.models import UserBirthday
|
||||
from main import birthday_list
|
||||
|
||||
|
||||
class BirthdayCog(commands.Cog, name="Birthday Commands"):
|
||||
"""Class holds birthday commands."""
|
||||
|
||||
@nextcord.slash_command(guild_ids=[1000247361639030804, 822199658306076672])
|
||||
async def birthday(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for birthday. Never called."""
|
||||
|
||||
@birthday.subcommand(description="Set your birthday month and day for birthday fun!")
|
||||
async def set(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
month: int = nextcord.SlashOption(required=True, min_value=1, max_value=12, description="Your birth month."),
|
||||
day: int = nextcord.SlashOption(required=True, min_value=1, max_value=31, description="Your birth day."),
|
||||
) -> None:
|
||||
"""
|
||||
Set a user's birthday in the database. If birthday is today, adds to the birthday_list.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction that invoked the command.
|
||||
month (int, optional): The month a user was born. Is SlashOption.
|
||||
day (int, optional): The day a user was born. Is SlashOption.
|
||||
|
||||
"""
|
||||
# Ensure that a person used the command.
|
||||
if interaction.user is None:
|
||||
return
|
||||
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
# check if valid day
|
||||
today = datetime.now(tz=UTC)
|
||||
try:
|
||||
birthday = datetime(today.year, month, day, tzinfo=UTC)
|
||||
except ValueError:
|
||||
await interaction.followup.send(f"Invalid date: {month}/{day}.", ephemeral=True)
|
||||
return
|
||||
|
||||
# If the birthday is today, add to birthday_list
|
||||
if today.date() == birthday.date():
|
||||
birthday_list.append(interaction.user.id)
|
||||
|
||||
new_birthday = UserBirthday(user_id=interaction.user.id, month=month, day=day)
|
||||
with Session(engine) as session:
|
||||
stmt = select(UserBirthday).where(UserBirthday.user_id == interaction.user.id)
|
||||
if (new_birthday := session.scalar(stmt)) is None:
|
||||
new_birthday = UserBirthday(user_id=interaction.user.id, month=month, day=day)
|
||||
session.add(new_birthday)
|
||||
else:
|
||||
new_birthday.month = month
|
||||
new_birthday.day = day
|
||||
session.commit()
|
||||
|
||||
await interaction.followup.send(f"Birthday set to {month}/{day}.", ephemeral=True)
|
||||
|
||||
@birthday.subcommand(description="Remove yourself from the birthday list.")
|
||||
async def clear(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Remove the user from the birthday database.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction that invoked the command.
|
||||
|
||||
"""
|
||||
# Ensure that a person used the command.
|
||||
if interaction.user is None:
|
||||
return
|
||||
|
||||
if interaction.user.id in birthday_list:
|
||||
birthday_list.remove(interaction.user.id)
|
||||
|
||||
with Session(engine) as session:
|
||||
stmt = select(UserBirthday).where(col(UserBirthday.user_id) == interaction.user.id)
|
||||
if (user := session.scalar(stmt)) is None:
|
||||
await interaction.response.send_message("No birthday to clear.", ephemeral=True)
|
||||
return
|
||||
|
||||
session.delete(user)
|
||||
session.commit()
|
||||
|
||||
await interaction.response.send_message("Birthday cleared.", ephemeral=True)
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Initializing code for all loops."""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from cogs.loops.bot_presence import BotPresence
|
||||
from cogs.loops.daily_loop import DailyLoop
|
||||
|
||||
|
||||
def setup(bot: commands.Bot) -> None:
|
||||
"""Add functions to bot."""
|
||||
bot.add_cog(DailyLoop(bot))
|
||||
bot.add_cog(BotPresence(bot))
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Functions for controlling the bot's presence. Also contains the loop."""
|
||||
|
||||
import random
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands, tasks
|
||||
|
||||
|
||||
class BotPresence(commands.Cog, name="Bot Presence"):
|
||||
"""Class for bot presence functions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self) -> None:
|
||||
"""Start the loop when bot is ready."""
|
||||
await self.bot.wait_until_ready()
|
||||
self.periodic_change_presence.start()
|
||||
|
||||
@tasks.loop(minutes=30)
|
||||
async def periodic_change_presence(self) -> None:
|
||||
"""Loops every 30 minutes and changes presence."""
|
||||
await self.change_presence()
|
||||
|
||||
async def change_presence(self) -> None:
|
||||
"""Randomly chooses a presence and updates bot's presence."""
|
||||
dungeons = [
|
||||
"Cinderbrew Meadery",
|
||||
"Darkflame Cleft",
|
||||
"The Rookery",
|
||||
"Priory of the Sacred Flame",
|
||||
"Operation: Floodgate",
|
||||
"The MOTHERLODE!!",
|
||||
"Theater of Pain",
|
||||
"Operation: Mechagon - Workshop",
|
||||
]
|
||||
|
||||
dungeon = random.choice(dungeons) # noqa: S311
|
||||
plus = random.randint(2, 12) # noqa: S311
|
||||
|
||||
presence = f"Running +{plus} {dungeon}"
|
||||
|
||||
await self.bot.change_presence(activity=nextcord.CustomActivity(presence))
|
||||
@@ -0,0 +1,81 @@
|
||||
"""Defines the daily loop that runs the Off Cooldown bot functions."""
|
||||
|
||||
from datetime import UTC, datetime, time
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands, tasks
|
||||
from pytz import timezone
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
from sqlmodel import col
|
||||
|
||||
from database.db import engine
|
||||
from database.models import UserBirthday
|
||||
from logger import ocd_logger
|
||||
from main import birthday_list
|
||||
from settings import settings
|
||||
|
||||
|
||||
class DailyLoop(commands.Cog, name="Daily Loop"):
|
||||
"""Class for performing daily functions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
self.current_election = None
|
||||
self.update_birthdays()
|
||||
self.daily_loop.start()
|
||||
|
||||
@tasks.loop(time=time(hour=0, tzinfo=timezone("UTC")))
|
||||
async def daily_loop(self) -> None:
|
||||
"""Run daily repeated functions."""
|
||||
self.update_birthdays()
|
||||
|
||||
await self.cakedays()
|
||||
|
||||
@daily_loop.before_loop
|
||||
async def before_loop(self) -> None:
|
||||
"""Actions run before the loop starts. Want to wait for bot to be ready and get the current cycle."""
|
||||
await self.bot.wait_until_ready()
|
||||
|
||||
def update_birthdays(self) -> None:
|
||||
"""Goes through database to collect all birthdays that are today. Adds user's id to birthday list."""
|
||||
birthday_list.clear()
|
||||
|
||||
today = datetime.now(tz=UTC).date()
|
||||
|
||||
stmt = (
|
||||
select(col(UserBirthday.user_id))
|
||||
.where(col(UserBirthday.month) == today.month)
|
||||
.where(col(UserBirthday.day) == today.day)
|
||||
)
|
||||
with Session(engine) as session:
|
||||
to_add = session.scalars(stmt)
|
||||
birthday_list.extend(to_add.all())
|
||||
|
||||
async def cakedays(self) -> None:
|
||||
"""
|
||||
Loops through the guild's humans checking for joined_at dates that match today's date.
|
||||
|
||||
Sends a message saying happy cakeday.
|
||||
"""
|
||||
today = datetime.now(tz=UTC).date()
|
||||
if (guild := self.bot.get_guild(settings.params.guild_id)) is None:
|
||||
ocd_logger.warning("Guild not found in bot.")
|
||||
return
|
||||
if (announce_channel := guild.get_channel(settings.channel_ids.general)) is None or not isinstance(
|
||||
announce_channel, nextcord.TextChannel
|
||||
):
|
||||
ocd_logger.warning("Set general channel isn't found or isn't a text channel.")
|
||||
return
|
||||
|
||||
cakedays: list[nextcord.Member] = [
|
||||
human
|
||||
for human in guild.humans
|
||||
if human.joined_at is not None and human.joined_at.month == today.month and human.joined_at.day == today.day
|
||||
]
|
||||
|
||||
for member in cakedays:
|
||||
years = today.year - member.joined_at.year # type: ignore reportOptionalMemberAccess
|
||||
year_str = "years" if years > 1 else "year"
|
||||
msg = f"Happy Cake day! <@{member.id}> joined the discord {years} {year_str} ago!"
|
||||
await announce_channel.send(msg)
|
||||
@@ -0,0 +1,41 @@
|
||||
"""Defines cog for on message listener."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from main import birthday_list
|
||||
|
||||
|
||||
class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
"""Sets up message listener."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message: nextcord.Message) -> None:
|
||||
"""Listen for message creation. Adds score and attachments to submission."""
|
||||
# don't want to process any messages from bot.
|
||||
if message.author.id == self.bot.user.id: # type: ignore reportOptionalMemberAccess
|
||||
return
|
||||
|
||||
await birthday_action(message)
|
||||
|
||||
|
||||
# This function will be called when this extension is loaded.
|
||||
# It is necessary to add these functions to the bot.
|
||||
def setup(bot: commands.Bot) -> None:
|
||||
"""Add function to bot."""
|
||||
bot.add_cog(SubmissionListener(bot))
|
||||
|
||||
|
||||
async def birthday_action(message: nextcord.Message) -> None:
|
||||
"""
|
||||
Check a message's author and mentions to see if they are in the birthday list.
|
||||
|
||||
Args:
|
||||
message (nextcord.Message): The message to check for a birthday person.
|
||||
|
||||
"""
|
||||
if message.author.id in birthday_list or bool(set(message.raw_mentions) & set(birthday_list)):
|
||||
await message.add_reaction("🎂")
|
||||
@@ -0,0 +1 @@
|
||||
"""Package for database interaction."""
|
||||
@@ -0,0 +1,11 @@
|
||||
"""Simple file to set up the sqlmodel connection to database."""
|
||||
|
||||
from sqlmodel import SQLModel, create_engine
|
||||
|
||||
from . import models # noqa: F401 - Need this import for creating the models.
|
||||
|
||||
url = "sqlite:///database.db"
|
||||
engine = create_engine(url)
|
||||
|
||||
# Don't need to create all when connected to prod.
|
||||
SQLModel.metadata.create_all(engine)
|
||||
@@ -0,0 +1,20 @@
|
||||
"""File holds all model definitions of SQL tables."""
|
||||
|
||||
from sqlalchemy.dialects.mysql import BIGINT, SMALLINT
|
||||
from sqlmodel import Column, Field, SQLModel
|
||||
|
||||
|
||||
class UserBirthday(SQLModel, table=True):
|
||||
"""
|
||||
Represents a user's birthday.
|
||||
|
||||
Attributes:
|
||||
user_id: The user's discord id. Primary key.
|
||||
month: The month the user was born.
|
||||
day: The day the user was born.
|
||||
|
||||
"""
|
||||
|
||||
user_id: int = Field(sa_column=Column(BIGINT(unsigned=True), primary_key=True, nullable=False))
|
||||
month: int = Field(sa_column=Column(SMALLINT(unsigned=True), nullable=False))
|
||||
day: int = Field(sa_column=Column(SMALLINT(unsigned=True), nullable=False))
|
||||
@@ -0,0 +1,10 @@
|
||||
"""Creates logger for Agon bot."""
|
||||
|
||||
import logging
|
||||
import sys
|
||||
|
||||
ocd_logger = logging.getLogger("default")
|
||||
ocd_logger.setLevel(logging.INFO)
|
||||
handler = logging.StreamHandler(sys.stdout)
|
||||
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
|
||||
ocd_logger.addHandler(handler)
|
||||
@@ -0,0 +1,57 @@
|
||||
"""Main file for bot."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import pathlib
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from .logger import ocd_logger
|
||||
from .settings import settings
|
||||
|
||||
nextcord_logger = logging.getLogger("nextcord")
|
||||
nextcord_logger.setLevel(logging.DEBUG)
|
||||
nextcord_handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8", mode="w")
|
||||
nextcord_handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
|
||||
nextcord_logger.addHandler(nextcord_handler)
|
||||
|
||||
birthday_list: list[int] = []
|
||||
|
||||
|
||||
def main() -> None:
|
||||
"""Set config, intents, then connects the bot to discord."""
|
||||
# Allows privileged intents for monitoring members joining, roles editing, and role assignments
|
||||
# These need to be enabled in the developer portal as well
|
||||
intents = nextcord.Intents.default()
|
||||
|
||||
# Required in order to read messages (eg. prefix commands)
|
||||
intents.message_content = True
|
||||
|
||||
# To enable the guilds privileged intent:
|
||||
# intents.guilds = True
|
||||
|
||||
# To enable the members privileged intent:
|
||||
intents.members = True
|
||||
|
||||
bot = commands.Bot(intents=intents)
|
||||
|
||||
# Get the modules of all cogs whose directory structure is ./cogs/<module_name>
|
||||
for folder in pathlib.Path("cogs").iterdir():
|
||||
bot.load_extension(f"cogs.{folder.name}")
|
||||
|
||||
@bot.listen()
|
||||
async def on_ready() -> None:
|
||||
"""When Discord is connected."""
|
||||
if bot.user is None:
|
||||
ocd_logger.critical("Discord connection failed.")
|
||||
return
|
||||
ocd_logger.info(f"{bot.user.name} has connected to Discord!")
|
||||
|
||||
# Run Discord bot
|
||||
bot.run(settings.params.discord_token)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,41 @@
|
||||
"""File contains classes defining off-cooldown-bot settings structure."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from dataclass_wizard import YAMLWizard
|
||||
|
||||
CONFIG_LOCATION = "conf/config.yaml"
|
||||
|
||||
|
||||
@dataclass
|
||||
class ChannelIds:
|
||||
"""Class holds all channel ids needed by bot."""
|
||||
|
||||
general: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Params:
|
||||
"""Class defines start parameters that shouldn't be changed."""
|
||||
|
||||
discord_token: str
|
||||
guild_id: int
|
||||
|
||||
|
||||
@dataclass
|
||||
class Settings(YAMLWizard):
|
||||
"""Class defines all configs required for off-cooldown-bot."""
|
||||
|
||||
channel_ids: ChannelIds
|
||||
params: Params
|
||||
|
||||
def save(self) -> None:
|
||||
"""Save settings to the yaml file."""
|
||||
self.to_yaml_file(CONFIG_LOCATION)
|
||||
|
||||
|
||||
settings = Settings.from_yaml_file(CONFIG_LOCATION)
|
||||
|
||||
# I'm not expecting a list to be returned.
|
||||
if isinstance(settings, list):
|
||||
settings = settings[0]
|
||||
Reference in New Issue
Block a user