""" 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 ( ParticipationPoints, 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(ParticipationPoints) .where(ParticipationPoints.Cycle.startswith(quarter_name)) .statement, con=engine, ) else: df = pd.read_sql_query( sql=session.query(ParticipationPoints).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(ParticipationPoints.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])