243 lines
8.3 KiB
Python
243 lines
8.3 KiB
Python
"""
|
|
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 EventPoints, Presences, Votes, engine
|
|
from utils import 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()
|
|
async def view(self, interaction: nextcord.Interaction):
|
|
"""
|
|
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,
|
|
),
|
|
):
|
|
# just in case
|
|
await interaction.response.defer(ephemeral=True)
|
|
|
|
with Session(engine) as session:
|
|
if quarter_name:
|
|
df = pd.read_sql_query(
|
|
sql=session.query(EventPoints)
|
|
.where(EventPoints.Cycle.startswith(quarter_name))
|
|
.statement,
|
|
con=engine,
|
|
)
|
|
else:
|
|
df = pd.read_sql_query(
|
|
sql=session.query(EventPoints).statement,
|
|
con=engine,
|
|
)
|
|
|
|
if df.empty:
|
|
await interaction.followup.send(
|
|
f'Quarter "{quarter_name}" is invalid.', ephemeral=True
|
|
)
|
|
return
|
|
|
|
cycles = df.groupby("Cycle")
|
|
content = ""
|
|
for cycle in cycles.groups:
|
|
content += f"**{cycle}**\n"
|
|
unsorted_all = cycles.get_group(cycle).groupby("DiscordMember")
|
|
members = unsorted_all.get_group(True)
|
|
non_members = unsorted_all.get_group(False)
|
|
sorted_members = members.sort_values(
|
|
by="UserId",
|
|
key=lambda user_ids: (
|
|
[
|
|
interaction.guild.get_member(int(id)).display_name
|
|
for id in user_ids
|
|
]
|
|
),
|
|
)
|
|
sorted_non_members = non_members.sort_values(by="UserId")
|
|
sorted_all = pd.concat(
|
|
[sorted_members, sorted_non_members], ignore_index=True, sort=False
|
|
)
|
|
for row in sorted_all.itertuples():
|
|
userid = row.UserId
|
|
points = row.Points
|
|
discord_member = row.DiscordMember
|
|
content += (
|
|
f"<@{userid}>: {points}\n"
|
|
if discord_member
|
|
else f"{userid}: {points}\n"
|
|
)
|
|
content += "\n"
|
|
|
|
if len(content) > 2000:
|
|
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,
|
|
),
|
|
):
|
|
"""
|
|
Gathers all presences from the database and responds with a message showing them.
|
|
|
|
Args:
|
|
interaction (nextcord.Integration): User's interaction with the command.
|
|
"""
|
|
|
|
# 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,
|
|
),
|
|
):
|
|
"""
|
|
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.
|
|
"""
|
|
|
|
stmt = (
|
|
select(Votes)
|
|
.where(Votes.ElectionCycle == 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
|
|
):
|
|
"""
|
|
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.ElectionCycle).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")
|
|
async def event_points_autocomplete(
|
|
self, interaction: nextcord.Interaction, quarter_name: str
|
|
):
|
|
"""
|
|
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(EventPoints.Cycle).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])
|