""" Commands for viewing information in the database. """ import nextcord import pandas as pd from nextcord.ext import commands from numpy import indices from sqlalchemy import select from sqlalchemy.orm import Session from database_setup.sql_data_classes import ( ParticipationPoints, Presences, Votes, engine, ) 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=[1000247361639030804]) 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 == quarter_name) .statement, con=engine, ) else: df = pd.read_sql_query( sql=session.query(ParticipationPoints).statement, con=engine, ) cycles = df.groupby("Cycle") content = "" for cycle in cycles.groups: content += f"```{cycle}```\n" group = cycles.get_group(cycle) for row in group.itertuples(): userid = row.UserId points = row.Points discord_member = row.DiscordMember content += ( f"<@{userid}>: {points}\n" if discord_member else f"{userid}: {points}\n" ) if len(content) > 2000: await interaction.followup.send( "Message length is too long, please choose a quarter." ) 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." ), ): """ 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) 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])