Move all files into dakiny_bot package.
This commit is contained in:
@@ -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 database_setup.sql_data_classes import (
|
||||
Points,
|
||||
Presences,
|
||||
Votes,
|
||||
engine,
|
||||
)
|
||||
from 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)
|
||||
Reference in New Issue
Block a user