@@ -0,0 +1,91 @@
|
||||
"""File contains logic for creating a webhook listener, pulling from git, and updating all messages."""
|
||||
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import uvicorn
|
||||
from fastapi import FastAPI
|
||||
from git import Repo
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.classes.gitea_webhook import GiteaWebhook
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.utils import REPO_DIR, edit_clan_council_info, update_clan_messages
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import nextcord
|
||||
|
||||
app = FastAPI()
|
||||
REPO_URL = "https://gitea.cheezers.org/imcompiling/dakiny-messages.git"
|
||||
|
||||
|
||||
class ClanMessages(commands.Cog, name="Clan Messages"):
|
||||
"""
|
||||
Cog to be loaded into the bot.
|
||||
|
||||
Used to be able to have a fastapi post listener and access to the bot at
|
||||
the same time.
|
||||
"""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
self.repo_setup()
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self) -> None:
|
||||
"""Triggers when the bot is in ready state. Adds the webhook listener to FastAPI and serves."""
|
||||
app.add_api_route("/messages/", self.update_messages, methods=["POST"])
|
||||
# Bind to all interfaces, don't know the docker interface ahead of time.
|
||||
config = uvicorn.Config(app=app, host="0.0.0.0", port=5000, log_level="info") # noqa: S104
|
||||
server = uvicorn.Server(config)
|
||||
await server.serve()
|
||||
|
||||
async def update_messages(self, hook_info: GiteaWebhook) -> dict:
|
||||
"""
|
||||
Triggered by webhook.
|
||||
|
||||
Pulls from git and then calls method to update all the discord messages as described by the
|
||||
files in the git repo.
|
||||
"""
|
||||
# Get all modified/added files
|
||||
modified_files: list[str] = [
|
||||
file for files in [commit.modified + commit.added for commit in hook_info.commits] for file in files
|
||||
]
|
||||
|
||||
template_updates = {file.split("/")[1] for file in modified_files if "message_templates" in file}
|
||||
channel_updates = {file.split("/")[1] for file in modified_files if file.split("/")[1] not in template_updates}
|
||||
|
||||
self.origin.pull()
|
||||
guild: nextcord.Guild = self.bot.get_guild(settings.params.guild_id)
|
||||
|
||||
await update_clan_messages(guild, channel_updates)
|
||||
|
||||
# if the council members template has changed, update the message
|
||||
if "council_members.txt" in template_updates:
|
||||
await edit_clan_council_info(self.bot.get_guild(settings.params.guild_id))
|
||||
|
||||
return {"message": "Successfully updated all discord messages."}
|
||||
|
||||
def repo_setup(self) -> None:
|
||||
"""Checks if the repo exists and creates repo object, clones if it does not."""
|
||||
# check if repo has been cloned. open repo object
|
||||
if Path(REPO_DIR).is_dir():
|
||||
self.repo = Repo(REPO_DIR)
|
||||
else:
|
||||
self.repo = Repo.clone_from(REPO_URL, REPO_DIR)
|
||||
|
||||
self.origin = self.repo.remote("origin")
|
||||
|
||||
if settings.params.git_branch == "test":
|
||||
self.repo.create_head("test", self.origin.refs.test).set_tracking_branch(self.origin.refs.test).checkout()
|
||||
else:
|
||||
self.repo.create_head("main", self.origin.refs.main).set_tracking_branch(self.origin.refs.main).checkout()
|
||||
|
||||
self.origin.pull()
|
||||
|
||||
|
||||
# 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:
|
||||
"""Adds function to bot."""
|
||||
bot.add_cog(ClanMessages(bot))
|
||||
@@ -0,0 +1,28 @@
|
||||
"""Cog loader. Adds all command cogs to the bot."""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.cogs.commands.record import RecordCog
|
||||
from dakiny_bot.cogs.commands.refresh import RefreshCog
|
||||
from dakiny_bot.cogs.commands.remove import RemoveCog
|
||||
from dakiny_bot.cogs.commands.request import RequestCog
|
||||
from dakiny_bot.cogs.commands.set import SetCog
|
||||
from dakiny_bot.cogs.commands.settings import SettingsCog
|
||||
from dakiny_bot.cogs.commands.test import TestCog
|
||||
from dakiny_bot.cogs.commands.time import TimeCog
|
||||
from dakiny_bot.cogs.commands.view import ViewCog
|
||||
|
||||
|
||||
# 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:
|
||||
"""Adds all command classes to bot."""
|
||||
bot.add_cog(RecordCog(bot))
|
||||
bot.add_cog(RefreshCog(bot))
|
||||
bot.add_cog(RemoveCog())
|
||||
bot.add_cog(RequestCog(bot))
|
||||
bot.add_cog(SetCog())
|
||||
bot.add_cog(SettingsCog())
|
||||
bot.add_cog(TestCog(bot))
|
||||
bot.add_cog(TimeCog())
|
||||
bot.add_cog(ViewCog())
|
||||
@@ -0,0 +1,358 @@
|
||||
"""Contained commands and views for record commands."""
|
||||
|
||||
import nextcord
|
||||
from discord import TextInputStyle
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.utils import GUILDS, MAX_DISCORD_MESSAGE_LENGTH, PointType, record_points
|
||||
|
||||
|
||||
class RecordCog(commands.Cog, name="Record Commands"):
|
||||
"""Class holds record commands."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""Default constructor for RecordCog class. Does nothing."""
|
||||
super().__init__()
|
||||
self.bot: commands.Bot = bot
|
||||
|
||||
@nextcord.slash_command(guild_ids=GUILDS)
|
||||
async def record(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for record. Never called."""
|
||||
|
||||
@record.subcommand(description="Record an event and its participants.")
|
||||
async def event(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
name: str = nextcord.SlashOption(required=True, description="Name of the event that was hosted."),
|
||||
host: nextcord.Member = nextcord.SlashOption(
|
||||
required=False,
|
||||
description="An alternative host if you are recording event for someone else.",
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Create and send recording event.
|
||||
|
||||
This command creates the base event recording embed and sends a message with it and the EventView
|
||||
for gathering more information on the event.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): interaction with this command.
|
||||
name (str): Name of the event.
|
||||
host (nextcord.Member, optional): Alternative host to user using command.
|
||||
"""
|
||||
# check role
|
||||
if interaction.guild.get_role(settings.role_ids.event_host) not in interaction.user.roles:
|
||||
await interaction.response.send_message(
|
||||
f"You must have the <@&{settings.role_ids.event_host}> role to record an event.",
|
||||
ephemeral=True,
|
||||
)
|
||||
return
|
||||
|
||||
if host is None:
|
||||
host = interaction.user
|
||||
|
||||
embed: nextcord.Embed = nextcord.Embed(title=name, description="")
|
||||
embed.add_field(name="Participants", value="")
|
||||
embed.add_field(name="Non-Discord Participants", value="")
|
||||
embed.add_field(name="Hosted by", value=f"<@{host.id}>", inline=False)
|
||||
|
||||
await interaction.response.send_message(
|
||||
"Fill out event details and select all members who participated!",
|
||||
view=EventView(name, host.id),
|
||||
ephemeral=True,
|
||||
embed=embed,
|
||||
)
|
||||
|
||||
@record.subcommand(description="Record a miscellaneous point for a clan member.")
|
||||
async def point(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
activity: str = nextcord.SlashOption(
|
||||
description="The type of activity to receive points for.",
|
||||
required=True,
|
||||
choices=[
|
||||
"Clan chat activity",
|
||||
"Teaching or sharing in-game tips",
|
||||
"Recruiting efforts",
|
||||
"Successful recruitment",
|
||||
],
|
||||
),
|
||||
justification: str = nextcord.SlashOption(
|
||||
required=True,
|
||||
description="Reason to give the recipient a point.",
|
||||
max_length=int(MAX_DISCORD_MESSAGE_LENGTH / 2),
|
||||
),
|
||||
recipient: nextcord.Member = nextcord.SlashOption(
|
||||
required=False,
|
||||
description="Discord user to give point to.",
|
||||
),
|
||||
non_discord_recipient: str = nextcord.SlashOption(
|
||||
required=False, description="Non Discord user to give point to.", max_length=32
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Gives a point to the provided recipient. Sends recording message to event participation channel.
|
||||
|
||||
Only one recipient should be given point/provided to the command. Fails if not exactly one recipient
|
||||
is given.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): interaction with this command.
|
||||
activity (str): Choice list of activities to give point for.
|
||||
justification (str): Reason to give point.
|
||||
recipient (nextcord.Member): Discord member to give point to.
|
||||
non_discord_recipient (str): Non Discord member to give point to.
|
||||
"""
|
||||
# role check
|
||||
if interaction.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
|
||||
await interaction.response.send_message(
|
||||
f"You must have the <@&{settings.role_ids.clan_member}> role to record points.",
|
||||
ephemeral=True,
|
||||
)
|
||||
return
|
||||
|
||||
# pylint or whatever seems to think that there are lines of unreachable code. That is false.
|
||||
if recipient and non_discord_recipient:
|
||||
await interaction.response.send_message("Please only provide 1 recipient.", ephemeral=True)
|
||||
return
|
||||
elif recipient is not None: # noqa: RET505
|
||||
record_points(
|
||||
[str(recipient.id)],
|
||||
settings.point_settings.clan_points,
|
||||
PointType.CLAN,
|
||||
is_member=True,
|
||||
)
|
||||
recipient_str = f"<@{recipient.id}>"
|
||||
elif non_discord_recipient is not None:
|
||||
record_points(
|
||||
[non_discord_recipient],
|
||||
settings.point_settings.clan_points,
|
||||
PointType.CLAN,
|
||||
is_member=False,
|
||||
)
|
||||
recipient_str = f'"{non_discord_recipient}"'
|
||||
else:
|
||||
await interaction.response.send_message("You must provide a recipient.", ephemeral=True)
|
||||
return
|
||||
|
||||
event_channel: nextcord.TextChannel = interaction.guild.get_channel(settings.channel_ids.event_post)
|
||||
|
||||
msg = f"<@{interaction.user.id}> has given {recipient_str} points for the {activity} activity. Justification: {justification}" # noqa: E501
|
||||
if not msg.endswith("."):
|
||||
msg += "."
|
||||
await event_channel.send(msg)
|
||||
await interaction.response.send_message("Point has been added.", ephemeral=True)
|
||||
|
||||
|
||||
class EventView(nextcord.ui.View):
|
||||
"""
|
||||
Class handles gathering information on an event.
|
||||
|
||||
Has buttons to edit the information, cancel, and submit.
|
||||
Also contains a UserSelect object to pick discord members.
|
||||
"""
|
||||
|
||||
def __init__(self, name: str, host_id: int):
|
||||
super().__init__(timeout=60 * 60 * 2) # timeout 2 hours
|
||||
|
||||
self.name: str = name
|
||||
self.host_id: int = host_id
|
||||
|
||||
self.event_participants = ParticipantSelect()
|
||||
|
||||
self.add_item(self.event_participants)
|
||||
|
||||
@nextcord.ui.button(label="Cancel", style=nextcord.ButtonStyle.red, row=1)
|
||||
async def cancel(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""Cancels event submission. Removes embed and view."""
|
||||
await interaction.response.edit_message("Canceled", ephemeral=True, embed=None, view=None)
|
||||
self.stop()
|
||||
|
||||
@nextcord.ui.button(label="Edit", style=nextcord.ButtonStyle.blurple, row=1)
|
||||
async def edit(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""Calls modal to update the text input information of the event."""
|
||||
await interaction.response.send_modal(ParticipantModal(interaction.message.embeds[0]))
|
||||
|
||||
@nextcord.ui.button(label="Confirm", style=nextcord.ButtonStyle.green, row=1)
|
||||
async def confirm(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Submits the event embed to the event channel.
|
||||
|
||||
Removes view and embed from recording message.
|
||||
Sends point information to database.
|
||||
"""
|
||||
if settings.point_settings.verify_events:
|
||||
verification_channel: nextcord.TextChannel = interaction.guild.get_channel(settings.channel_ids.admin)
|
||||
verification_view = EventVerification(
|
||||
self.event_participants.values.members, interaction.message.embeds[0].fields[1].value, self.host_id
|
||||
)
|
||||
await verification_channel.send(
|
||||
content="Event record request.", embed=interaction.message.embeds[0], view=verification_view
|
||||
)
|
||||
submission_response = "Event has been submitted to admins for approval."
|
||||
else:
|
||||
process_event(
|
||||
self.event_participants.values.members,
|
||||
interaction.message.embeds[0].fields[1].value,
|
||||
self.host_id,
|
||||
)
|
||||
|
||||
event_channel: nextcord.TextChannel = interaction.guild.get_channel(settings.channel_ids.event_post)
|
||||
await event_channel.send(embed=interaction.message.embeds[0])
|
||||
submission_response = "Event has been submitted."
|
||||
|
||||
await interaction.response.edit_message(content=submission_response, view=None, embed=None)
|
||||
self.stop()
|
||||
|
||||
|
||||
class EventVerification(nextcord.ui.View):
|
||||
"""Class contains code to approve/deny event submissions."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
discord_users: list[nextcord.Member],
|
||||
non_discord_users: str,
|
||||
host_id: int,
|
||||
):
|
||||
super().__init__(timeout=None, auto_defer=True)
|
||||
self.discord_users: list[nextcord.Member] = discord_users
|
||||
self.non_discord_users: str = non_discord_users
|
||||
self.host_id: int = host_id
|
||||
|
||||
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green)
|
||||
async def approve(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Approve button for an event submission.
|
||||
|
||||
Calls the process event function. Records who approved the event.
|
||||
Removes the view from the interaction message.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
process_event(self.discord_users, self.non_discord_users, self.host_id)
|
||||
event_channel: nextcord.TextChannel = interaction.guild.get_channel(settings.channel_ids.event_post)
|
||||
await event_channel.send(embed=interaction.message.embeds[0])
|
||||
|
||||
content = f"\nApproved by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
|
||||
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red)
|
||||
async def deny(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Deny button for an event submission. Just removes the view and logs who denied the request.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
content = f"\nDenied by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
|
||||
|
||||
class ParticipantModal(nextcord.ui.Modal):
|
||||
"""
|
||||
This modal contains all the text edits for an event.
|
||||
|
||||
You can change the name, description, and add the usernames of participants who are not in the discord.
|
||||
"""
|
||||
|
||||
# consider turning this embed into a dict
|
||||
def __init__(self, embed: nextcord.Embed):
|
||||
super().__init__(
|
||||
"Edit the event.",
|
||||
timeout=5 * 60,
|
||||
)
|
||||
self.embed: nextcord.Embed = embed
|
||||
self.name: nextcord.ui.TextInput = nextcord.ui.TextInput(
|
||||
required=True,
|
||||
label="Event Name",
|
||||
default_value=self.embed.title,
|
||||
)
|
||||
self.description: nextcord.ui.TextInput = nextcord.ui.TextInput(
|
||||
required=False,
|
||||
label="Event Description",
|
||||
default_value=self.embed.description,
|
||||
style=TextInputStyle.paragraph,
|
||||
placeholder="Add a description if desired.",
|
||||
max_length=1000,
|
||||
)
|
||||
self.participants: nextcord.ui.TextInput = nextcord.ui.TextInput(
|
||||
required=False,
|
||||
label="Non-Discord Usernames",
|
||||
style=TextInputStyle.paragraph,
|
||||
placeholder="Please separate usernames with a new line.",
|
||||
default_value=self.embed.fields[1].value,
|
||||
)
|
||||
self.add_item(self.name)
|
||||
self.add_item(self.description)
|
||||
self.add_item(self.participants)
|
||||
|
||||
async def callback(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Edits the interaction message to contain the Non-Discord participants.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction of the callback.
|
||||
"""
|
||||
self.embed.title = self.name.value
|
||||
self.embed.description = self.description.value
|
||||
self.embed.set_field_at(index=1, name="Non-Discord Participants", value=self.participants.value)
|
||||
|
||||
await interaction.response.edit_message(embed=self.embed)
|
||||
|
||||
|
||||
class ParticipantSelect(nextcord.ui.UserSelect):
|
||||
"""A selection drop down for discord members. Updates the embed with all users selected."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
super().__init__(
|
||||
min_values=1,
|
||||
max_values=25,
|
||||
placeholder="Select all participants",
|
||||
row=0,
|
||||
)
|
||||
|
||||
async def callback(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Edits the interaction message to contain the Discord participants.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction of the callback.
|
||||
"""
|
||||
embed: nextcord.Embed = interaction.message.embeds[0]
|
||||
participants: str = "\n".join([f"<@{participant.id}>" for participant in self.values.members])
|
||||
embed.set_field_at(0, name="Participants", value=participants)
|
||||
await interaction.response.edit_message(embed=embed)
|
||||
|
||||
|
||||
def process_event(discord_users: list[nextcord.Member], non_discord_users: str, host_id: int) -> None:
|
||||
"""
|
||||
Processes all points to give to participants and host for an event.
|
||||
|
||||
Args:
|
||||
discord_users (list[nextcord.Member]): List of discord members that participated in event.
|
||||
non_discord_users (str): Newline separated list of non discord members that participated in event.
|
||||
host_id (int): Discord ID of host.
|
||||
"""
|
||||
if discord_users:
|
||||
discord_users_list: list[str] = [str(user.id) for user in discord_users if user.id is not host_id]
|
||||
record_points(
|
||||
discord_users_list,
|
||||
settings.point_settings.participate_points,
|
||||
PointType.EVENT,
|
||||
is_member=True,
|
||||
)
|
||||
|
||||
if non_discord_users:
|
||||
non_discord_users_list: list[str] = non_discord_users.split("\n")
|
||||
record_points(
|
||||
non_discord_users_list,
|
||||
settings.point_settings.participate_points,
|
||||
PointType.EVENT,
|
||||
is_member=False,
|
||||
)
|
||||
|
||||
record_points([host_id], settings.point_settings.host_points, PointType.EVENT, is_member=True)
|
||||
@@ -0,0 +1,65 @@
|
||||
"""Contained commands and views for refresh commands."""
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.utils import MESSAGES_DIR, edit_clan_council_info, update_clan_messages
|
||||
|
||||
|
||||
class RefreshCog(commands.Cog, name="Refresh Commands"):
|
||||
"""Class holds refresh commands."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""Default constructor for RefreshCog class. Does nothing."""
|
||||
super().__init__()
|
||||
self.bot: commands.Bot = bot
|
||||
|
||||
@nextcord.slash_command(default_member_permissions=nextcord.Permissions(administrator=True))
|
||||
async def refresh(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for refresh. Never called."""
|
||||
|
||||
@refresh.subcommand(description="Force refresh all of the bot managed messages.")
|
||||
async def messages(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
description="Channel to refresh messages in.",
|
||||
required=False,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Updates all messages.
|
||||
|
||||
Calls update_clan_messages() to update all messages.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the command.
|
||||
channel (nextcord.TextChannel): channel to refresh messages in.
|
||||
"""
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
if channel is None:
|
||||
channel_updates = [
|
||||
dir_name for dir_name in os.listdir(MESSAGES_DIR) if Path(f"{MESSAGES_DIR}/{dir_name}").is_dir()
|
||||
]
|
||||
else:
|
||||
channel_updates = [channel.name]
|
||||
|
||||
await update_clan_messages(interaction.guild, channel_updates)
|
||||
await interaction.followup.send("Messages have been refreshed.", ephemeral=True)
|
||||
|
||||
@refresh.subcommand(description="Updates the current council info message.")
|
||||
async def council_info(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Edits the clan council info message.
|
||||
|
||||
Calls edit_clan_council_info().
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the command.
|
||||
"""
|
||||
await edit_clan_council_info(interaction.guild)
|
||||
await interaction.response.send_message("Clan Council Info message updated.", ephemeral=True)
|
||||
@@ -0,0 +1,67 @@
|
||||
"""Commands for removing information in the database."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.database.sql_data_classes import Presences, engine
|
||||
|
||||
|
||||
class RemoveCog(commands.Cog, name="Remove Commands"):
|
||||
"""Holds the view commands."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Default constructor for RemoveCog class. Does nothing."""
|
||||
|
||||
@nextcord.slash_command(
|
||||
default_member_permissions=nextcord.Permissions(administrator=True),
|
||||
)
|
||||
async def remove(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for set. Never called."""
|
||||
|
||||
@remove.subcommand(description="Remove a presence from the database.")
|
||||
async def presence(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
presence_str: str = nextcord.SlashOption(description="The presence to remove", required=True),
|
||||
) -> None:
|
||||
"""
|
||||
Deletes the given presence id from the database.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction that invoked the command.
|
||||
presence_str (str): Is SlashOption. The presence to remove. Is the ID of the presence.
|
||||
"""
|
||||
presence_id = int(presence_str)
|
||||
with Session(engine) as session:
|
||||
stmt = delete(Presences).where(Presences.Id == presence_id)
|
||||
session.execute(stmt)
|
||||
session.commit()
|
||||
|
||||
await interaction.response.send_message("Presence deleted.", ephemeral=True)
|
||||
|
||||
@presence.on_autocomplete("presence_str")
|
||||
async def presence_autocomplete(self, interaction: nextcord.Interaction, presence_str: str) -> None:
|
||||
"""
|
||||
Collects all presences that start with the presence string.
|
||||
|
||||
Sends autocomplete dict {Presence Name, Presence Id}
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with command.
|
||||
presence_str (str): String to run autocomplete on.
|
||||
"""
|
||||
with Session(engine) as session:
|
||||
stmt = select(Presences)
|
||||
presences_dict: dict[str, str] = {pres.Presence: str(pres.Id) for pres in session.scalars(stmt)}
|
||||
|
||||
# if empty compare string
|
||||
if not presence_str:
|
||||
await interaction.response.send_autocomplete(dict(list(presences_dict.items())[:25]))
|
||||
return
|
||||
|
||||
autofilled_presences: dict[str, str] = {
|
||||
pres: pres_id for pres, pres_id in presences_dict.items() if presence_str.lower() in pres.lower()
|
||||
}
|
||||
await interaction.response.send_autocomplete(autofilled_presences[:25])
|
||||
@@ -0,0 +1,168 @@
|
||||
"""Contains commands and views for request commands."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.database.sql_data_classes import Presences, engine
|
||||
|
||||
|
||||
class RequestCog(commands.Cog, name="Request Commands"):
|
||||
"""Class holds request commands."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""Default constructor for RequestCog class. Does nothing."""
|
||||
self.bot: commands.Bot = bot
|
||||
|
||||
@nextcord.slash_command()
|
||||
async def request(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for request. Never called."""
|
||||
|
||||
@request.subcommand(description="Request to add a presence to the bot's presence list.")
|
||||
async def add_presence(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
presence: str = nextcord.SlashOption(
|
||||
required=True,
|
||||
min_length=1,
|
||||
max_length=100,
|
||||
description="The presence you want to add.",
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sends a message to admins to request to add a presence.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction that invoked the command.
|
||||
presence (str): The presence to add.
|
||||
"""
|
||||
content = f'<@{interaction.user.id}> would like to add presence "{presence}."'
|
||||
view = PresenceChange(presence)
|
||||
await self.bot.get_channel(settings.channel_ids.admin).send(content=content, view=view)
|
||||
await interaction.response.send_message("Request sent to admins.", ephemeral=True)
|
||||
|
||||
@request.subcommand(description="Request to change your Dakiny nickname")
|
||||
async def change_nickname(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
new_nickname: str = nextcord.SlashOption(
|
||||
required=True,
|
||||
min_length=2,
|
||||
max_length=32,
|
||||
description="Your requested new nickname.",
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sends a message to admins to request a nickname change.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction that invoked the command.
|
||||
new_nickname(str): The username to change to.
|
||||
"""
|
||||
content = f'<@{interaction.user.id}> would like to change their username to "{new_nickname}."'
|
||||
view = UsernameRequest(interaction.user.id, new_nickname)
|
||||
await self.bot.get_channel(settings.channel_ids.admin).send(content=content, view=view)
|
||||
await interaction.response.send_message("Request sent to admins.", ephemeral=True)
|
||||
|
||||
|
||||
class UsernameRequest(nextcord.ui.View):
|
||||
"""View provides approve and deny buttons and action for a username change request."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
user_id: int,
|
||||
new_nickname: str,
|
||||
):
|
||||
super().__init__(timeout=None, auto_defer=True)
|
||||
self.user_id = user_id
|
||||
self.new_nickname = new_nickname
|
||||
|
||||
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green)
|
||||
async def approve(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Approve button for a nickname change request.
|
||||
|
||||
Changes the target's nickname to self.new_nickname.
|
||||
Catches an error if the target as a higher role than the bot.
|
||||
Removes the view from the interaction message.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
user = interaction.guild.get_member(self.user_id)
|
||||
if user is not None:
|
||||
try:
|
||||
await user.edit(nick=self.new_nickname)
|
||||
content = interaction.message.content + f"\nApproved by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
except nextcord.errors.Forbidden:
|
||||
await interaction.response.send_message(
|
||||
content="Please change manually.",
|
||||
ephemeral=True,
|
||||
)
|
||||
content = (
|
||||
interaction.message.content
|
||||
+ "\n\nFailed. User has too high permissions for bot to change nickname."
|
||||
)
|
||||
await interaction.followup.edit_message(interaction.message.id, content=content, view=None)
|
||||
else:
|
||||
await interaction.response.edit_message(
|
||||
content="An error occurred while changing the user's nickname.",
|
||||
view=None,
|
||||
)
|
||||
|
||||
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red)
|
||||
async def deny(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Deny button for a nickname request. Just removes the view and logs who denied the request.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
content = interaction.message.content + f"\nDenied by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
|
||||
|
||||
class PresenceChange(nextcord.ui.View):
|
||||
"""View provides approve and deny buttons and action for a username change request."""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
presence: str,
|
||||
):
|
||||
super().__init__(timeout=None, auto_defer=True)
|
||||
self.presence = presence
|
||||
|
||||
@nextcord.ui.button(label="Approve", style=nextcord.ButtonStyle.green)
|
||||
async def approve(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Approve button for a presence add request.
|
||||
|
||||
Adds the presence to the database.
|
||||
Removes the view from the interaction message.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
with Session(engine) as session:
|
||||
session.add(Presences(Presence=self.presence))
|
||||
session.commit()
|
||||
|
||||
content = interaction.message.content + f"\nApproved by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
|
||||
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red)
|
||||
async def deny(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None: # noqa: ARG002
|
||||
"""
|
||||
Deny button for a presence add request. Just removes the view and logs who denied the request.
|
||||
|
||||
Args:
|
||||
button (nextcord.ui.Button): Unused.
|
||||
interaction (nextcord.Interaction): The interaction of the user clicking the button.
|
||||
"""
|
||||
content = interaction.message.content + f"\nDenied by <@{interaction.user.id}>."
|
||||
await interaction.response.edit_message(content=content, view=None)
|
||||
@@ -0,0 +1,76 @@
|
||||
"""Contains commands and views for set commands."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import delete, select
|
||||
from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.app import birthday_list
|
||||
from dakiny_bot.database.sql_data_classes import UserBirthday, engine
|
||||
|
||||
|
||||
class SetCog(commands.Cog, name="Set Commands"):
|
||||
"""Class holds set commands."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Default constructor for SetCog class. Does nothing."""
|
||||
|
||||
@nextcord.slash_command()
|
||||
async def set(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for set. Never called."""
|
||||
|
||||
@set.subcommand(description="Sets your birth month and day for the Birthday Fun. Use 0/0 to remove yourself.")
|
||||
async def birthday(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
month: int = nextcord.SlashOption(required=True, min_value=0, max_value=12, description="Your birth month."),
|
||||
day: int = nextcord.SlashOption(required=True, min_value=0, max_value=31, description="Your birth day."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets a user's birthday in the database. Removes on input 0/0. 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.
|
||||
"""
|
||||
# Must be deferred, apparently datetime takes a little bit on invalid dates.
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
if month == 0 and day == 0:
|
||||
stmt = delete(UserBirthday).where(UserBirthday.UserId == interaction.user.id)
|
||||
with Session(engine) as session:
|
||||
session.execute(stmt)
|
||||
session.commit()
|
||||
await interaction.followup.send("Removed birthday.", ephemeral=True)
|
||||
return
|
||||
|
||||
# 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(UserId=interaction.user.id, Birthday=f"{month}/{day}")
|
||||
with Session(engine) as session:
|
||||
try:
|
||||
session.add(new_birthday)
|
||||
session.commit()
|
||||
except IntegrityError:
|
||||
# User already has birthday, change it instead.
|
||||
session.rollback()
|
||||
stmt = select(UserBirthday).where(UserBirthday.UserId == interaction.user.id)
|
||||
new_birthday = session.scalar(stmt)
|
||||
new_birthday.Birthday = f"{month}/{day}"
|
||||
session.commit()
|
||||
|
||||
await interaction.followup.send(f"Birthday set to {month}/{day}.", ephemeral=True)
|
||||
@@ -0,0 +1,379 @@
|
||||
"""Contains commands and views for settings commands."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.utils import GUILDS
|
||||
|
||||
|
||||
class SettingsCog(commands.Cog, name="Settings Commands"):
|
||||
"""Class holds settings commands."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Default constructor for SettingsCog class. Does nothing."""
|
||||
|
||||
@nextcord.slash_command(guild_ids=GUILDS, default_member_permissions=nextcord.Permissions(administrator=True))
|
||||
async def settings(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for settings. Never called."""
|
||||
|
||||
@settings.subcommand()
|
||||
async def channel_ids(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Subcommand parent. Never called."""
|
||||
|
||||
@channel_ids.subcommand(description="Sets channel that bot sends verification requests to admins.")
|
||||
async def admin(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.admin value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.admin = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Admin verification channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@channel_ids.subcommand(description="Sets announcement channel.")
|
||||
async def announcement(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.announcement value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.announcement = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Announcement channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@channel_ids.subcommand(description="Sets clan info channel. Channel used for tracking current council.")
|
||||
async def clan_info(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.clan_info value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.clan_info = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Clan information channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@channel_ids.subcommand(
|
||||
description="Sets event post channel. Channel is where event tracking information is posted."
|
||||
)
|
||||
async def event_post(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.event_post value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.event_post = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event posting channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@channel_ids.subcommand(description="Sets general channel.")
|
||||
async def general(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.general value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.general = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"General channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@channel_ids.subcommand(description="Sets voting channel. Channel used as parent for voting threads.")
|
||||
async def voting(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
channel: nextcord.TextChannel = nextcord.SlashOption(
|
||||
required=False, description="Channel to set to. If left blank, uses current channel."
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings channel_ids.voting value to given channel id or current channel id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
channel (nextcord.TextChannel, optional): Channel to set in settings.
|
||||
"""
|
||||
channel_id = interaction.channel_id if channel is None else channel.id
|
||||
settings.channel_ids.voting = channel_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Voting channel set to <#{channel_id}>.", ephemeral=True)
|
||||
|
||||
@settings.subcommand()
|
||||
async def point_settings(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Subcommand parent. Never called."""
|
||||
|
||||
@point_settings.subcommand(description="Sets Points value.")
|
||||
async def clan_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
points: int = nextcord.SlashOption(
|
||||
required=True, description="Number of Clan Points the recipient earns.", min_value=0
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.clan_points to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
points (int): Number of points to assign to PointSettings.points.
|
||||
"""
|
||||
settings.point_settings.clan_points = points
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Points set to {points}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets Event Weekly Limit value for Clan Points.")
|
||||
async def clan_points_weekly_limit(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
points: int = nextcord.SlashOption(
|
||||
required=True, description="Number of Clan Points a clan member can earn in a week.", min_value=0
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.clan_points_weekly_limit to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
points (int): Number of points to set.
|
||||
"""
|
||||
settings.point_settings.clan_points_weekly_limit = points
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Clan Points Weekly Limit set to {points}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets Event Weekly Limit value.")
|
||||
async def events_weekly_limit(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
points: int = nextcord.SlashOption(
|
||||
required=True, description="Number of Event Points a clan member can earn in a week.", min_value=0
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.events_weekly_limit to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
points (int): Number of points to set.
|
||||
"""
|
||||
settings.point_settings.events_weekly_limit = points
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event Weekly Limit set to {points}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets Event Host Points value.")
|
||||
async def host_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
points: int = nextcord.SlashOption(
|
||||
required=True, description="Number of Event Points a host earns when hosting event.", min_value=0
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.host_points to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
points (int): Number of points to set.
|
||||
"""
|
||||
settings.point_settings.host_points = points
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event Host Points set to {points}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets Event Participation Points value.")
|
||||
async def participate_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
points: int = nextcord.SlashOption(
|
||||
required=True,
|
||||
description="Number of Event Points a participant earns when attending an event.",
|
||||
min_value=0,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.participate_points to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
points (int): Number of points to set.
|
||||
"""
|
||||
settings.point_settings.participate_points = points
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event Participation Points set to {points}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets whether to verify a clan point entry.")
|
||||
async def verify_clan_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
do_verification: bool = nextcord.SlashOption( # noqa: FBT001
|
||||
required=True,
|
||||
description="Whether an event should be verified or not.",
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.verify_clan_points to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
do_verification (bool): If the bot should verify events.
|
||||
"""
|
||||
settings.point_settings.verify_clan_points = do_verification
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event verification set to {do_verification}.", ephemeral=True)
|
||||
|
||||
@point_settings.subcommand(description="Sets whether an event.")
|
||||
async def verify_events(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
do_verification: bool = nextcord.SlashOption( # noqa: FBT001
|
||||
required=True,
|
||||
description="Whether an event should be verified or not.",
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings point_settings.verify_events to the given number of points.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
do_verification (bool): If the bot should verify events.
|
||||
"""
|
||||
settings.point_settings.verify_events = do_verification
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event verification set to {do_verification}.", ephemeral=True)
|
||||
|
||||
@settings.subcommand()
|
||||
async def role_ids(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Subcommand parent. Never called."""
|
||||
|
||||
@role_ids.subcommand(description="Sets Clan Council role id.")
|
||||
async def clan_council(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
role: nextcord.Role = nextcord.SlashOption(required=True, description="Role to set."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings role_ids.clan_council to the given role id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
role (nextcord.TextChannel): Role to set in settings.
|
||||
"""
|
||||
role_id = role.id
|
||||
settings.role_ids.clan_council = role_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Clan Council role set to <@&{role_id}>.", ephemeral=True)
|
||||
|
||||
@role_ids.subcommand(description="Sets Clan Council Elect role id.")
|
||||
async def clan_council_elect(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
role: nextcord.Role = nextcord.SlashOption(required=True, description="Role to set."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings role_ids.clan_council_elect to the given role id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
role (nextcord.TextChannel): Role to set in settings.
|
||||
"""
|
||||
role_id = role.id
|
||||
settings.role_ids.clan_council_elect = role_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Clan Council Elect role set to <@&{role_id}>.", ephemeral=True)
|
||||
|
||||
@role_ids.subcommand(description="Sets Clan Member role id.")
|
||||
async def clan_member(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
role: nextcord.Role = nextcord.SlashOption(required=True, description="Role to set."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings role_ids.clan_member to the given role id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
role (nextcord.TextChannel): Role to set in settings.
|
||||
"""
|
||||
role_id = role.id
|
||||
settings.role_ids.clan_member = role_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Clan Member role set to <@&{role_id}>.", ephemeral=True)
|
||||
|
||||
@role_ids.subcommand(description="Sets Event Host role id.")
|
||||
async def event_host(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
role: nextcord.Role = nextcord.SlashOption(required=True, description="Role to set."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings role_ids.event_host to the given role id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
role (nextcord.TextChannel): Role to set in settings.
|
||||
"""
|
||||
role_id = role.id
|
||||
settings.role_ids.event_host = role_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"Event Host role set to <@&{role_id}>.", ephemeral=True)
|
||||
|
||||
@role_ids.subcommand(description="Sets High Council role id.")
|
||||
async def high_council(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
role: nextcord.Role = nextcord.SlashOption(required=True, description="Role to set."),
|
||||
) -> None:
|
||||
"""
|
||||
Sets DakinySettings role_ids.high_council to the given role id.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of command.
|
||||
role (nextcord.TextChannel): Role to set in settings.
|
||||
"""
|
||||
role_id = role.id
|
||||
settings.role_ids.high_council = role_id
|
||||
settings.save()
|
||||
await interaction.response.send_message(f"High Council role set to <@&{role_id}>.", ephemeral=True)
|
||||
@@ -0,0 +1,155 @@
|
||||
"""Commands for testing."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import application_checks, commands
|
||||
|
||||
from dakiny_bot.classes.election import Election
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.utils import GUILDS, get_active_election_cycle, get_quarter_name
|
||||
|
||||
ME = 206602713511886849
|
||||
|
||||
|
||||
def is_me(): # noqa: ANN201
|
||||
"""Predicate that makes sure only I can run the test commands."""
|
||||
|
||||
def predicate(interaction: nextcord.Interaction) -> bool:
|
||||
return interaction.user.id == ME
|
||||
|
||||
return application_checks.check(predicate)
|
||||
|
||||
|
||||
class TestCog(commands.Cog, name="Test commands."):
|
||||
"""Class holds commands for testing."""
|
||||
|
||||
def __init__(self, bot: commands.Bot) -> None:
|
||||
"""Default constructor for HelpCog class. Does nothing."""
|
||||
self.bot: commands.Bot = bot
|
||||
|
||||
@nextcord.slash_command(guild_ids=GUILDS)
|
||||
async def test(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for help. Never called."""
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def election_start(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Force starts election.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.election_start_action()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def voting_start(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Force starts voting.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.voting_start_action()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def voting_end(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Force ends voting.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.voting_end_action()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def highcouncil_vote(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Force starts highcouncil voting.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.highcouncil_election_start_action()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def election_reminder(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Forces election reminder.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.election_reminder()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def daily(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Forces daily function.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.check_date()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def term_start(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Forces term start.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
cycle = Election(
|
||||
get_active_election_cycle(),
|
||||
self.bot.get_guild(settings.params.guild_id),
|
||||
)
|
||||
await cycle.term_start_action()
|
||||
await interaction.send("Test command sent.", ephemeral=True)
|
||||
|
||||
@test.subcommand()
|
||||
@is_me()
|
||||
async def get_quarter_name(self, interaction: nextcord.Interaction) -> None:
|
||||
"""
|
||||
Tests getting the quarter name.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
"""
|
||||
name = get_quarter_name()
|
||||
await interaction.send(name)
|
||||
@@ -0,0 +1,109 @@
|
||||
"""Contains commands for time commands."""
|
||||
|
||||
from datetime import MAXYEAR, MINYEAR, UTC, datetime
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.utils import GUILDS
|
||||
|
||||
|
||||
class TimeCog(commands.Cog, name="Time Commands"):
|
||||
"""Class holds time commands."""
|
||||
|
||||
@nextcord.slash_command(guild_ids=GUILDS)
|
||||
async def time(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command. Never called."""
|
||||
|
||||
@time.subcommand(
|
||||
description="Get a discord timestamp of the given game time. If date information is left blank, uses current day." # noqa: E501
|
||||
)
|
||||
async def at( # noqa: PLR0913
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
hour: int = nextcord.SlashOption(
|
||||
description="The hour you want a timestamp of. Given in gametime/UTC.",
|
||||
min_value=0,
|
||||
max_value=23,
|
||||
required=True,
|
||||
),
|
||||
minute: int = nextcord.SlashOption(
|
||||
description="The minute you want a timestamp of. Given in gametime/UTC.",
|
||||
min_value=0,
|
||||
max_value=59,
|
||||
required=True,
|
||||
),
|
||||
year: int = nextcord.SlashOption(
|
||||
description="Year to make timestamp of.", required=False, min_value=MINYEAR, max_value=MAXYEAR
|
||||
),
|
||||
month: int = nextcord.SlashOption(
|
||||
description="Month to make timestamp of.", required=False, min_value=1, max_value=12
|
||||
),
|
||||
day: int = nextcord.SlashOption(
|
||||
description="Day to make timestamp of.", required=False, min_value=1, max_value=31
|
||||
),
|
||||
ephemeral: bool = nextcord.SlashOption( # noqa: FBT001
|
||||
description="Whether the replying message will be public or not.", required=False, default=True
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Replies with a discord timestamp of given UTC time.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the command.
|
||||
hour (int): The hour in UTC to make the timestamp of..
|
||||
minute (int): The minute in UTC to make the timestamp of..
|
||||
year (int): The year to make the timestamp of.
|
||||
month (int): The month to make the timestamp of.
|
||||
day (int): The day to make the timestamp of.
|
||||
ephemeral (bool): Whether the response is ephemeral or not.
|
||||
"""
|
||||
gametime_today: datetime = datetime.now(tz=UTC).replace(hour=hour, minute=minute)
|
||||
|
||||
if year is not None:
|
||||
gametime_today = gametime_today.replace(year=year)
|
||||
|
||||
if month is not None and day is not None:
|
||||
try:
|
||||
gametime_today = gametime_today.replace(month=month, day=day)
|
||||
except ValueError:
|
||||
await interaction.response.send_message(f"Day {day} is out of range for month {month}.", ephemeral=True)
|
||||
return
|
||||
else:
|
||||
if month is not None:
|
||||
try:
|
||||
gametime_today = gametime_today.replace(month=month)
|
||||
except ValueError:
|
||||
await interaction.response.send_message(
|
||||
f"Today ({gametime_today.day}) is out of range for month {month}.", ephemeral=True
|
||||
)
|
||||
return
|
||||
|
||||
if day is not None:
|
||||
try:
|
||||
gametime_today = gametime_today.replace(day=day)
|
||||
except ValueError:
|
||||
await interaction.response.send_message(
|
||||
f"Day {day} is out of range for this month {gametime_today.month}.", ephemeral=True
|
||||
)
|
||||
return
|
||||
|
||||
await interaction.response.send_message(content=f"<t:{int(gametime_today.timestamp())}:t>", ephemeral=ephemeral)
|
||||
|
||||
@time.subcommand(description="Get the current time in game.")
|
||||
async def gametime(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
ephemeral: bool = nextcord.SlashOption( # noqa: FBT001
|
||||
description="Whether the replying message will be public or not.", required=False, default=True
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Replies with the current time in game.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the command.
|
||||
ephemeral: (bool): Whether the response is ephemeral or not.
|
||||
"""
|
||||
gametime = datetime.now(tz=UTC)
|
||||
await interaction.response.send_message(content=f"{gametime.hour}:{gametime.minute}", ephemeral=ephemeral)
|
||||
@@ -0,0 +1,297 @@
|
||||
"""Commands for viewing information in the database."""
|
||||
|
||||
import nextcord
|
||||
import pandas as pd
|
||||
from nextcord.ext import commands
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.database.sql_data_classes import (
|
||||
Points,
|
||||
Presences,
|
||||
Votes,
|
||||
engine,
|
||||
)
|
||||
from dakiny_bot.utils import GUILDS, MAX_DISCORD_MESSAGE_LENGTH, PointType, VoteType
|
||||
|
||||
|
||||
class ViewCog(commands.Cog, name="View Commands"):
|
||||
"""Holds the view commands."""
|
||||
|
||||
def __init__(self) -> None:
|
||||
"""Default constructor for ViewCog class. Does nothing."""
|
||||
|
||||
@nextcord.slash_command(guild_ids=GUILDS)
|
||||
async def view(self, interaction: nextcord.Interaction) -> None:
|
||||
"""Parent slash command for set. Never called."""
|
||||
|
||||
@view.subcommand(description="Show event points.")
|
||||
async def event_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
quarter_name: str = nextcord.SlashOption(
|
||||
description="The quarter to view",
|
||||
required=False,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
View the event points.
|
||||
|
||||
Builds a message with the event points for the quarter that matches with quarter_name.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
quarter_name (str, optional): Same of quarter to autofill with. Is SlashOption.
|
||||
"""
|
||||
# just in case
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
with Session(engine) as session:
|
||||
if quarter_name:
|
||||
ppdf = pd.read_sql_query(
|
||||
sql=session.query(Points).where(Points.QuarterName.startswith(quarter_name)).statement,
|
||||
con=engine,
|
||||
)
|
||||
else:
|
||||
ppdf = pd.read_sql_query(
|
||||
sql=session.query(Points).statement,
|
||||
con=engine,
|
||||
)
|
||||
|
||||
if ppdf.empty:
|
||||
await interaction.followup.send(f'Quarter "{quarter_name}" is invalid.', ephemeral=True)
|
||||
return
|
||||
|
||||
quarters = ppdf.groupby("QuarterName")
|
||||
content = create_points_string(interaction, quarters, PointType.EVENT)
|
||||
|
||||
if len(content) > MAX_DISCORD_MESSAGE_LENGTH:
|
||||
await interaction.followup.send("Message length is too long, please select search to limit results.")
|
||||
return
|
||||
|
||||
await interaction.followup.send(content=content, ephemeral=True)
|
||||
|
||||
@view.subcommand(description="Show event points.")
|
||||
async def clan_points(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
quarter_name: str = nextcord.SlashOption(
|
||||
description="The quarter to view",
|
||||
required=False,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
View the clan points.
|
||||
|
||||
Builds a message with the clan points for the quarter that matches with quarter_name.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with the message.
|
||||
quarter_name (str, optional): Same of quarter to autofill with. Is SlashOption.
|
||||
"""
|
||||
# just in case
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
with Session(engine) as session:
|
||||
if quarter_name:
|
||||
ppdf = pd.read_sql_query(
|
||||
sql=session.query(Points).where(Points.QuarterName.startswith(quarter_name)).statement,
|
||||
con=engine,
|
||||
)
|
||||
else:
|
||||
ppdf = pd.read_sql_query(
|
||||
sql=session.query(Points).statement,
|
||||
con=engine,
|
||||
)
|
||||
|
||||
if ppdf.empty:
|
||||
await interaction.followup.send(f'Quarter "{quarter_name}" is invalid.', ephemeral=True)
|
||||
return
|
||||
|
||||
quarters = ppdf.groupby("QuarterName")
|
||||
content = create_points_string(interaction, quarters, PointType.CLAN)
|
||||
|
||||
if len(content) > MAX_DISCORD_MESSAGE_LENGTH:
|
||||
await interaction.followup.send("Message length is too long, please select search to limit results.")
|
||||
return
|
||||
|
||||
await interaction.followup.send(content=content, ephemeral=True)
|
||||
|
||||
@view.subcommand(description="Show all possible presence choices.")
|
||||
async def presences(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
search: str = nextcord.SlashOption(
|
||||
description="A word or string to search for in the presences.",
|
||||
required=False,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
Gathers all presences from the database and responds with a message showing them.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Integration): User's interaction with the command.
|
||||
search (str): string to search for in presences. Is SlashOption.
|
||||
"""
|
||||
# might take a while depending on number of presences
|
||||
await interaction.response.defer(ephemeral=True)
|
||||
|
||||
with Session(engine) as session:
|
||||
stmt = select(Presences.Presence)
|
||||
presences = session.scalars(stmt)
|
||||
|
||||
if search:
|
||||
message = "\n".join([presence for presence in presences if search.lower() in presence.lower()])
|
||||
else:
|
||||
message = "\n".join(list(presences))
|
||||
|
||||
await interaction.followup.send(content=message, ephemeral=True)
|
||||
|
||||
@view.subcommand(description="Shows all voters and who they voted for in a given election.")
|
||||
async def votes(
|
||||
self,
|
||||
interaction: nextcord.Interaction,
|
||||
election_name: str = nextcord.SlashOption(
|
||||
required=True, description="The election you want to view votes for."
|
||||
),
|
||||
election_type: str = nextcord.SlashOption(
|
||||
required=True,
|
||||
description="The election type. Council, Discord Manager, or Clan Leader.",
|
||||
choices=VoteType,
|
||||
),
|
||||
) -> None:
|
||||
"""
|
||||
View all voters and who they voted for.
|
||||
|
||||
Takes in an election name and sends a response message to the user with a list of people who voted in that
|
||||
election and who they voted for. Notes those tho abstained. Tells a user if the election is invalid. Uses
|
||||
autocompletion to assist with getting elections.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): User's interaction with the command.
|
||||
election_name (str): Is SlashOption. The name of the election to view votes for.
|
||||
election_type(str): Is SlashOption. The type of election to search for. Choice from StrEnum VoteType.
|
||||
"""
|
||||
stmt = select(Votes).where(Votes.ElectionName == election_name).where(Votes.Type == election_type)
|
||||
|
||||
with Session(engine) as session:
|
||||
# Check if election exists.
|
||||
if session.scalar(stmt) is None:
|
||||
await interaction.response.send_message(f"Invalid election: {election_name}.", ephemeral=True)
|
||||
return
|
||||
|
||||
votes = session.scalars(stmt)
|
||||
content = ""
|
||||
for vote in votes:
|
||||
content += (
|
||||
f"<@{vote.VoterId}> voted for <@{vote.VotedForId}>.\n"
|
||||
if vote.VotedForId is not None
|
||||
else f"<@{vote.VoterId}> abstained from voted.\n"
|
||||
)
|
||||
|
||||
await interaction.response.send_message(content=content, ephemeral=True)
|
||||
|
||||
@votes.on_autocomplete("election_name")
|
||||
async def election_autocomplete(self, interaction: nextcord.Interaction, election_name: str) -> None:
|
||||
"""
|
||||
Collects all elections that start with the election string.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with command.
|
||||
election_name (str): String to run autocomplete on.
|
||||
"""
|
||||
with Session(engine) as session:
|
||||
stmt = select(Votes.ElectionName).distinct()
|
||||
elections = session.scalars(stmt)
|
||||
|
||||
# if empty compare string
|
||||
if not election_name:
|
||||
await interaction.response.send_autocomplete(list(elections)[:25])
|
||||
return
|
||||
|
||||
autofilled_elections: list[str] = [
|
||||
election for election in elections if election.lower().startswith(election_name.lower())
|
||||
]
|
||||
await interaction.response.send_autocomplete(autofilled_elections[:25])
|
||||
|
||||
@event_points.on_autocomplete("quarter_name")
|
||||
@clan_points.on_autocomplete("quarter_name")
|
||||
async def event_points_autocomplete(self, interaction: nextcord.Interaction, quarter_name: str) -> None:
|
||||
"""
|
||||
Collects all elections that start with the election string.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Interaction with command.
|
||||
quarter_name (str): String to run autocomplete on.
|
||||
"""
|
||||
with Session(engine) as session:
|
||||
stmt = select(Points.QuarterName).distinct()
|
||||
quarters = session.scalars(stmt)
|
||||
|
||||
# if empty compare string
|
||||
if not quarter_name:
|
||||
await interaction.response.send_autocomplete(list(quarters)[:25])
|
||||
return
|
||||
|
||||
autofilled_quarters: list[str] = [
|
||||
quarter for quarter in quarters if quarter.lower().startswith(quarter_name.lower())
|
||||
]
|
||||
await interaction.response.send_autocomplete(autofilled_quarters[:25])
|
||||
|
||||
|
||||
def create_points_string(interaction: nextcord.Interaction, quarters: any, point_type: PointType) -> str:
|
||||
"""
|
||||
Creates the string to show how many clan/event points users have.
|
||||
|
||||
Args:
|
||||
interaction (nextcord.Interaction): Discord interaction of the command.
|
||||
quarters (any): Pandas objects grouped by quarters.
|
||||
point_type (PointType): The type of points to create the message for.
|
||||
|
||||
Returns:
|
||||
str: The string with the message content.
|
||||
"""
|
||||
content: str = ""
|
||||
for quarter in quarters.groups:
|
||||
content += f"**{quarter}**\n"
|
||||
unsorted_all = quarters.get_group(quarter).groupby("DiscordMember")
|
||||
members = unsorted_all.get_group(name=True)
|
||||
|
||||
# there can be no non-discord members sometimes.
|
||||
try:
|
||||
sorted_non_members = unsorted_all.get_group(name=False).sort_values(by="UserId")
|
||||
except KeyError:
|
||||
sorted_non_members = None
|
||||
sorted_members = members.sort_values(
|
||||
by="UserId",
|
||||
key=lambda user_ids: ([member_lambda_string(interaction.guild, user_id) for user_id in user_ids]),
|
||||
)
|
||||
sorted_all = pd.concat([sorted_members, sorted_non_members], ignore_index=True, sort=False)
|
||||
for row in sorted_all.itertuples():
|
||||
points = row.EventPoints if point_type == PointType.EVENT else row.ClanPoints
|
||||
discord_member = row.DiscordMember
|
||||
if points > 0:
|
||||
content += f"<@{row.UserId}>: {points}\n" if discord_member else f"{row.UserId}: {points}\n"
|
||||
content += "\n"
|
||||
return content
|
||||
|
||||
|
||||
def member_lambda_string(guild: nextcord.Guild, user_id: int) -> str:
|
||||
"""
|
||||
Finds a user in a guild and returns the display name.
|
||||
|
||||
Returns the user id in string form if user is no longer in guild.
|
||||
I don't know if I like this name very much.
|
||||
|
||||
Args:
|
||||
guild (nextcord.Guild): Guild object to find the user in.
|
||||
user_id (int): User id to get display name of.
|
||||
|
||||
Returns:
|
||||
str: String to use for sorting.
|
||||
"""
|
||||
try:
|
||||
return guild.get_member(int(user_id)).display_name.lower()
|
||||
except AttributeError:
|
||||
# The user is no longer a part of the guild
|
||||
return str(user_id)
|
||||
@@ -0,0 +1,12 @@
|
||||
"""Initializing code for all loops."""
|
||||
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.cogs.loops.bot_presence import BotPresence
|
||||
from dakiny_bot.cogs.loops.daily_loop import DailyLoop
|
||||
|
||||
|
||||
def setup(bot: commands.Bot) -> None:
|
||||
"""Adds functions to bot."""
|
||||
bot.add_cog(BotPresence(bot))
|
||||
bot.add_cog(DailyLoop(bot))
|
||||
@@ -0,0 +1,38 @@
|
||||
"""Functions for controlling the bot's presence. Also contains the loop."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands, tasks
|
||||
from sqlalchemy import func, select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.database.sql_data_classes import Presences, engine
|
||||
|
||||
|
||||
class BotPresence(commands.Cog, name="Bot Presence"):
|
||||
"""Class for bot presence functions."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_guild_join(self, guild: nextcord.Guild) -> None: # noqa: ARG002
|
||||
"""Listener for Guild Join. Updated presence on guild join."""
|
||||
await self.change_presence()
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self) -> None:
|
||||
"""Starts the loop when bot is ready."""
|
||||
await self.bot.wait_until_ready()
|
||||
self.periodic_change_presence.start()
|
||||
|
||||
@tasks.loop(minutes=5)
|
||||
async def periodic_change_presence(self) -> None:
|
||||
"""Loops every 5 minutes and changes presence."""
|
||||
await self.change_presence()
|
||||
|
||||
async def change_presence(self) -> None:
|
||||
"""Randomly chooses a presence and updates bot's presence."""
|
||||
with Session(engine) as session:
|
||||
presence = session.scalar(select(Presences.Presence).order_by(func.random()))
|
||||
|
||||
await self.bot.change_presence(activity=nextcord.CustomActivity(presence))
|
||||
@@ -0,0 +1,128 @@
|
||||
"""Defines the daily loop that runs the Dakiny bot functions."""
|
||||
|
||||
import random
|
||||
from datetime import UTC, datetime, time
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from nextcord.ext import commands, tasks
|
||||
from pytz import timezone
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from dakiny_bot.app import birthday_list
|
||||
from dakiny_bot.classes.election import Election
|
||||
from dakiny_bot.DakinySettings import settings
|
||||
from dakiny_bot.database.sql_data_classes import UserBirthday, engine
|
||||
from dakiny_bot.utils import get_active_election_cycle
|
||||
|
||||
if TYPE_CHECKING:
|
||||
import nextcord
|
||||
|
||||
KANDY = 239547993106284544
|
||||
|
||||
IGNORE_CAKEDAY = [266378407682506754]
|
||||
|
||||
|
||||
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()
|
||||
|
||||
# pylint doesn't recognize that daily_loop is subclassed from tasks.loop.
|
||||
self.daily_loop.start() # pylint: disable=no-member
|
||||
|
||||
@tasks.loop(time=time(hour=0, tzinfo=timezone("UTC")))
|
||||
async def daily_loop(self) -> None:
|
||||
"""
|
||||
Runs daily.
|
||||
|
||||
Checks if the current election is None. If so, checks if there is an active election. If there is
|
||||
an active election, runs the check_date() function for the election to perform actions. Finally checks if the
|
||||
election is over to delete the current election object.
|
||||
"""
|
||||
if self.current_election is None:
|
||||
self.current_election = self.get_election()
|
||||
|
||||
if self.current_election is not None:
|
||||
await self.current_election.check_date()
|
||||
if self.current_election.election_over:
|
||||
self.current_election = None
|
||||
|
||||
self.update_birthdays()
|
||||
|
||||
if KANDY in birthday_list:
|
||||
# inform kandy that it is her "birthday"
|
||||
kandy: nextcord.User = await self.bot.fetch_user(KANDY)
|
||||
await kandy.send('Happy "Birthday"!')
|
||||
|
||||
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()
|
||||
|
||||
# Bot needs to know guilds to get the current cycle.
|
||||
self.current_election = self.get_election()
|
||||
|
||||
def get_election(self) -> Election:
|
||||
"""
|
||||
Creates an object for the current election cycle if one is active.
|
||||
|
||||
An election is considered active between the election start and the term start.
|
||||
|
||||
Returns:
|
||||
Election: Current election cycle, None if there is no election happening.
|
||||
"""
|
||||
cycle = get_active_election_cycle()
|
||||
|
||||
# prevent exception from using cycle as an Int in an array
|
||||
if cycle is None:
|
||||
return None
|
||||
|
||||
election = Election(cycle, self.bot.get_guild(settings.params.guild_id))
|
||||
|
||||
return election if election.active else None
|
||||
|
||||
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()
|
||||
|
||||
# kandy's 1% chance
|
||||
if random.randint(0, 99) == 0: # noqa: S311 isn't cryptography related.
|
||||
birthday_list.append(KANDY)
|
||||
|
||||
today = datetime.now(tz=UTC).date()
|
||||
birthday = f"{today.month}/{today.day}"
|
||||
|
||||
stmt = select(UserBirthday.UserId).where(UserBirthday.Birthday == birthday)
|
||||
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()
|
||||
guild: nextcord.Guild = self.bot.get_guild(settings.params.guild_id)
|
||||
announce_channel: nextcord.TextChannel = guild.get_channel(settings.channel_ids.general)
|
||||
|
||||
cakedays: list[nextcord.Member] = [
|
||||
human
|
||||
for human in guild.humans
|
||||
if human.joined_at.month == today.month
|
||||
and human.joined_at.day == today.day
|
||||
and human.id not in IGNORE_CAKEDAY
|
||||
]
|
||||
|
||||
for member in cakedays:
|
||||
years = today.year - member.joined_at.year
|
||||
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,40 @@
|
||||
"""Defines cog for on message listener."""
|
||||
|
||||
import nextcord
|
||||
from nextcord.ext import commands
|
||||
|
||||
from dakiny_bot.app import birthday_list
|
||||
|
||||
|
||||
class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||
"""Sets up message listener."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_message(self, message: nextcord.Message) -> None:
|
||||
"""Listener 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:
|
||||
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:
|
||||
"""Adds function to bot."""
|
||||
bot.add_cog(SubmissionListener(bot))
|
||||
|
||||
|
||||
async def birthday_action(message: nextcord.Message) -> None:
|
||||
"""
|
||||
Checks a message's auther 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("🎂")
|
||||
Reference in New Issue
Block a user