Compare commits

..
3 Commits
Author SHA1 Message Date
imcompiling 1c8e8c4d5a Merge branch 'main' into mock-election
build_develop / build (push) Successful in 1m29s
2023-11-07 14:44:40 -05:00
imcompiling 0b5423a62b Merge branch 'main' into mock-election
build_develop / build (push) Successful in 1m22s
2023-11-06 19:47:39 -05:00
imcompiling dbbb7c1c62 Change votes to 2 for mock test
build_develop / build (push) Successful in 1m26s
2023-11-04 09:20:19 -04:00
28 changed files with 203 additions and 2158 deletions
+2 -3
View File
@@ -130,7 +130,7 @@ dmypy.json
# logging
nextcord.log
# bot files
# data file for bot
highscores_data.json
highscores_config.json
deleted_backups/
@@ -139,5 +139,4 @@ TODO.txt
sql_tester.py
conf
outputs
database
dakiny-messages/
database
+7 -4
View File
@@ -1,7 +1,7 @@
FROM python:3.10.6-slim
# install dependencies
RUN apt-get update && apt-get -y upgrade && apt-get -y install git
# install dependencies for mysqlclient
# RUN apt-get update && apt-get -y upgrade && apt-get -y install python3-dev default-libmysqlclient-dev build-essential
RUN pip install --upgrade pip
@@ -10,7 +10,7 @@ WORKDIR /home
COPY requirements.txt requirements.txt
COPY config.py config.py
COPY main.py main.py
COPY dakiny_bot.py dakiny_bot.py
COPY utils.py utils.py
COPY bot_messages bot_messages/
@@ -20,4 +20,7 @@ COPY database_setup database_setup/
RUN pip install -r requirements.txt
CMD ["python3", "-u", "./main.py"]
# remove build deps that aren't needed to run
# RUN apt-get -y remove python3-dev build-essential && apt-get -y autoremove
CMD ["python3", "-u", "./dakiny_bot.py"]
+1 -5
View File
@@ -1,9 +1,5 @@
# Dakiny Bot
Dakiny clan's own discord bot!
### Assumptions
- For Elections: Bot assumes that it always did the step before it. If the step before an action didn't happen, behavior is undefined.
Bot assumes that it always did the step before it. If the step before an action didn't happen, behavior is undefined.
Object creation in the middle of a cycle is only there for restarting the bot. All relevant information should already
exist in the database.
+2 -1
View File
@@ -7,6 +7,7 @@ From {highcouncil_election_announce} to {highcouncil_election_start}, you will b
Clan Leader
Discord Server Manager
Each member of the CCE can vote once in each category (Discord Server Manager and Clan Leader). Alternatively, you may choose to intentionally abstain.
Each member of the CCE can cast one for each category by using the emotes: <:Rank_Owner:479008325137072155> for Clan Leader and <:Rank_Deputy:479008295319896064> for Discord Server Manager.
If needed, the icon 🤷 can be used to abstain to vote in any or both categories and it should be applied to all candidates of that category.
Votes can be cast until {highcouncil_election_end}, so we can announce and have everyone set up to start the term by {term_start}.
+58 -218
View File
@@ -6,28 +6,14 @@ import collections
from datetime import datetime, timedelta
import nextcord
from sqlalchemy import select, true
from sqlalchemy import select
from sqlalchemy.orm import Session
from classes.nominate_highcouncil_views import NominateHighCouncilView
from classes.nominate_views import NominateView
from classes.voting_highcouncil_views import VoteHighcouncilView
from classes.voting_views import VoteView
from config import DakinyConfig
from database_setup.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from utils import (
build_highcouncil_votes_str,
build_votes_str,
get_cl_election_name,
get_datetime,
get_dm_election_name,
get_election_cycle_name,
)
from database_setup.sql_data_classes import CouncilMembers, ElectionInformation, Votes, engine
from utils import build_votes_str, get_datetime, get_election_cycle_name
class ElectionCycle:
@@ -54,15 +40,9 @@ class ElectionCycle:
self.term_end: datetime = get_datetime(self.election_cycle.term_end)
self.voting_start: datetime = get_datetime(self.election_cycle.voting_start)
self.voting_end: datetime = get_datetime(self.election_cycle.voting_end)
self.highcouncil_election_announce: datetime = get_datetime(
self.election_cycle.highcouncil_election_announce
)
self.highcouncil_election_start: datetime = get_datetime(
self.election_cycle.highcouncil_election_start
)
self.highcouncil_election_end: datetime = get_datetime(
self.election_cycle.highcouncil_election_end
)
self.highcouncil_election_announce: datetime = get_datetime(self.election_cycle.highcouncil_election_announce)
self.highcouncil_election_start: datetime = get_datetime(self.election_cycle.highcouncil_election_start)
self.highcouncil_election_end: datetime = get_datetime(self.election_cycle.highcouncil_election_end)
self.guild: nextcord.Guild = guild
self.election_over: bool = False
@@ -75,15 +55,11 @@ class ElectionCycle:
long.
"""
with open(
"bot_messages/election_announcement.txt", mode="r", encoding="utf8"
) as file:
with open("bot_messages/election_announcement.txt", mode="r", encoding="utf8") as file:
message = file.read()
next_election = get_datetime(
self.config.election_cycles[
(self.cycle + 1) % len(self.config.election_cycles)
].election_start
self.config.election_cycles[(self.cycle + 1) % len(self.config.election_cycles)].election_start
)
message = message.format(
@@ -108,22 +84,14 @@ class ElectionCycle:
str: Message to send.
"""
with open(
"bot_messages/highcouncil_election.txt", mode="r", encoding="utf8"
) as file:
with open("bot_messages/highcouncil_election.txt", mode="r", encoding="utf8") as file:
message = file.read()
return message.format(
clan_council_elect_role_id=self.config.params.clan_council_elect_role_id,
highcouncil_election_announce=self.highcouncil_election_announce.strftime(
"%B %d"
),
highcouncil_election_start=self.highcouncil_election_start.strftime(
"%B %d"
),
highcouncil_election_end=self.highcouncil_election_end.strftime(
"%B %d"
),
highcouncil_election_announce=self.highcouncil_election_announce.strftime("%B %d"),
highcouncil_election_start=self.highcouncil_election_start.strftime("%B %d"),
highcouncil_election_end=self.highcouncil_election_end.strftime("%B %d"),
term_start=self.term_start.strftime("%B %d"),
)
@@ -135,9 +103,7 @@ class ElectionCycle:
str: Message to send.
"""
with open(
"bot_messages/highcouncil_vote.txt", mode="r", encoding="utf8"
) as file:
with open("bot_messages/highcouncil_vote.txt", mode="r", encoding="utf8") as file:
message = file.read()
return message.format(
@@ -155,9 +121,7 @@ class ElectionCycle:
with open("bot_messages/vote_start.txt", mode="r", encoding="utf8") as file:
message = file.read()
vote_end: datetime = get_datetime(
self.config.election_cycles[self.cycle].voting_end
)
vote_end: datetime = get_datetime(self.config.election_cycles[self.cycle].voting_end)
return message.format(
clan_member_id=self.config.params.clan_member_role_id,
@@ -176,11 +140,7 @@ class ElectionCycle:
await self.election_start_action()
elif today.date() == self.voting_start.date():
await self.voting_start_action()
elif (
(self.voting_end - timedelta(3)).date()
<= today.date()
< self.voting_end.date()
):
elif (self.voting_end - timedelta(3)).date() <= today.date() < self.voting_end.date():
await self.election_reminder()
elif today.date() == self.voting_end.date():
await self.voting_end_action()
@@ -194,16 +154,17 @@ class ElectionCycle:
Performs required actions for when the election cycle starts.
"""
# create new channel for election
parent: nextcord.TextChannel = self.guild.get_channel(
self.config.params.voting_parent_id
)
category: nextcord.CategoryChannel = self.guild.get_channel(self.config.params.active_category_id)
quarter_name: str = get_election_cycle_name(self.config)
thread_msg: nextcord.Message = await parent.send(content=quarter_name)
thread: nextcord.Thread = await parent.create_thread(
name=quarter_name, message=thread_msg
)
# create new entry into database.
election_channel: nextcord.TextChannel = await category.create_text_channel(quarter_name)
clan_member_role: nextcord.Role = self.guild.get_role(self.config.params.clan_member_role_id)
await election_channel.set_permissions(clan_member_role, add_reactions=True)
with Session(engine) as session:
new_cycle = ElectionInformation(ElectionCycle=quarter_name, ElectionChannel=election_channel.id)
session.add(new_cycle)
session.commit()
# post message announcing election
# election start message is returned as an array of string because it is too long. send each string
@@ -212,21 +173,9 @@ class ElectionCycle:
for index, message in enumerate(messages):
# if last item, add the nominate button
if index + 1 == len(messages):
nominate_msg: nextcord.Message = await thread.send(
message, view=NominateView(self.config, self.guild)
)
await election_channel.send(message, view=NominateView(self.config))
else:
await thread.send(message)
# create new entry into database.
with Session(engine) as session:
new_cycle = ElectionInformation(
ElectionCycle=quarter_name,
ThreadId=thread.id,
CouncilNominationMsgId=nominate_msg.id,
)
session.add(new_cycle)
session.commit()
await election_channel.send(message)
async def voting_start_action(self) -> None:
"""
@@ -234,29 +183,20 @@ class ElectionCycle:
Removes write permission from the general clan member role.
"""
message = self.build_vote_start()
election_thread: nextcord.Thread = self.guild.get_thread(
self.get_election_info().ThreadId
)
await election_thread.send(message)
vote_msg = await election_thread.send(
build_votes_str(self.config, self.guild),
view=VoteView(self.config, self.guild),
)
clan_member_role: nextcord.Role = self.guild.get_role(self.config.params.clan_member_role_id)
election_channel: nextcord.TextChannel = self.guild.get_channel(self.get_election_info().ElectionChannel)
await election_channel.send(message)
vote_msg = await election_channel.send(build_votes_str(self.config), view=VoteView(self.config, self.guild))
with Session(engine) as session:
stmt = select(ElectionInformation).where(
ElectionInformation.ElectionCycle
== get_election_cycle_name(self.config)
ElectionInformation.ElectionCycle == get_election_cycle_name(self.config)
)
info = session.scalar(stmt)
info.CouncilVoteMsgId = vote_msg.id
session.commit()
thread: nextcord.Thread = self.guild.get_thread(info.ThreadId)
nominate_msg: nextcord.Message = await thread.fetch_message(
info.CouncilNominationMsgId
)
await nominate_msg.edit(view=None)
await election_channel.set_permissions(clan_member_role, send_messages=False)
async def voting_end_action(self) -> None:
"""
@@ -264,40 +204,27 @@ class ElectionCycle:
"""
message = self.build_highcouncil_election_announce()
council_elect_role: nextcord.Role = self.guild.get_role(
self.config.params.clan_council_elect_role_id
)
election_thread: nextcord.Thread = self.guild.get_thread(
self.get_election_info().ThreadId
)
nominate_msg: nextcord.Message = await election_thread.send(
message, view=NominateHighCouncilView(self.config, self.guild)
)
council_elect_role: nextcord.Role = self.guild.get_role(self.config.params.clan_council_elect_role_id)
election_channel: nextcord.TextChannel = self.guild.get_channel(self.get_election_info().ElectionChannel)
clan_member_role: nextcord.Role = self.guild.get_role(self.config.params.clan_member_role_id)
# remove vote button from vote message
election_name = get_election_cycle_name(self.config)
with Session(engine) as session:
# remove vote button from vote message
stmt = select(ElectionInformation).where(
stmt = select(ElectionInformation.CouncilVoteMsgId).where(
ElectionInformation.ElectionCycle == election_name
)
info = session.scalar(stmt)
vote_msg: nextcord.Message = await election_thread.fetch_message(
info.CouncilVoteMsgId
)
await vote_msg.edit(view=None)
vote_msg_id = session.scalar(stmt)
vote_msg: nextcord.Message = await election_channel.fetch_message(vote_msg_id)
await vote_msg.edit(view=None)
# add the highcouncil nomination message to election information
info.HighCouncilNominationMsgId = nominate_msg.id
session.commit()
# add the highcouncil role to everyone with 3 or more votes. set elected to true in database
# add the highcouncil role to everyone with 3 or more votes. set elected to true in database
with Session(engine) as session:
stmt = select(Votes.VotedForId).where(Votes.ElectionCycle == election_name)
vote_count = collections.Counter(session.scalars(stmt))
for user_id, votes in vote_count.items():
if votes >= 3:
if votes >= 2:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == election_name)
@@ -307,12 +234,13 @@ class ElectionCycle:
council_member.Elected = True
session.commit()
member: nextcord.Member = self.guild.get_member(user_id)
user: nextcord.Member = self.guild.get_member(user_id)
await member.add_roles(
council_elect_role,
reason=f"{member.display_name} has won a nomination and become a Council Elect.",
)
await user.add_roles(council_elect_role)
await election_channel.send(message)
await election_channel.set_permissions(council_elect_role, send_messages=True, add_reactions=True)
await election_channel.set_permissions(clan_member_role, add_reactions=False)
async def highcouncil_election_start_action(self) -> None:
"""
@@ -320,104 +248,19 @@ class ElectionCycle:
"""
message = self.build_highcouncil_vote_start()
election_thread: nextcord.Thread = self.guild.get_thread(
self.get_election_info().ThreadId
)
await election_thread.send(message)
vote_msg = await election_thread.send(
build_highcouncil_votes_str(self.config, self.guild),
view=VoteHighcouncilView(self.config, self.guild),
)
with Session(engine) as session:
stmt = select(ElectionInformation).where(
ElectionInformation.ElectionCycle
== get_election_cycle_name(self.config)
)
info = session.scalar(stmt)
info.HighCouncilVoteMsgId = vote_msg.id
session.commit()
# remove highcouncil nomination view
nomination_msg: nextcord.Message = await election_thread.fetch_message(
info.HighCouncilNominationMsgId
)
await nomination_msg.edit(view=None)
election_channel: nextcord.TextChannel = self.guild.get_channel(self.get_election_info().ElectionChannel)
await election_channel.send(message)
async def term_start_action(self) -> None:
"""
Marks the election as over.
Moves the election channel into the archive category.
"""
self.election_over = True
election_channel: nextcord.TextChannel = self.guild.get_channel(self.get_election_info().ElectionChannel)
archive: nextcord.CategoryChannel = self.guild.get_channel(self.config.params.archive_channel_group_id)
elect_role: nextcord.Role = self.guild.get_role(
self.config.params.clan_council_elect_role_id
)
council_role: nextcord.Role = self.guild.get_role(
self.config.params.clan_council_role_id
)
with Session(engine) as session:
# role management
stmt = (
select(CouncilMembers.MemberId)
.where(
CouncilMembers.ElectionCycle == get_election_cycle_name(self.config)
)
.where(CouncilMembers.Elected == true())
)
for member_id in session.scalars(stmt):
member = self.guild.get_member(member_id)
await member.remove_roles(
elect_role,
reason=f"{member.display_name} no longer needs the elect role.",
)
await member.add_roles(
council_role,
reason=f"{member.display_name} has been elected to council.",
)
# announce high council members
# ties are undefined behavior.
dm_election = get_dm_election_name(self.config)
cl_election = get_cl_election_name(self.config)
stmt = select(Votes.VotedForId).where(Votes.ElectionCycle == dm_election)
dm_votes = collections.Counter(session.scalars(stmt))
stmt = select(Votes.VotedForId).where(Votes.ElectionCycle == cl_election)
cl_votes = collections.Counter(session.scalars(stmt))
try:
dm_winner = max(dm_votes, key=dm_votes.get)
except ValueError:
dm_winner = "No winner"
try:
cl_winner = max(cl_votes, key=cl_votes.get)
except ValueError:
cl_winner = "No winner"
election_thread: nextcord.Thread = self.guild.get_thread(
self.get_election_info().ThreadId
)
# clear view from vote message
stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
ElectionInformation.ElectionCycle == get_election_cycle_name(self.config)
)
vote_msg_id = session.scalar(stmt)
vote_msg: nextcord.Message = await election_thread.fetch_message(vote_msg_id)
await vote_msg.edit(view=None)
highcouncil_msg: str = f"Congratulations for our high council members!\n Clan Leader: <@{cl_winner}>\n"
highcouncil_msg += f"Discord Manager: <@{dm_winner}>"
await election_thread.send(highcouncil_msg)
await election_thread.send(
"This election cycle is now over. Thank you everyone for participating!"
)
await election_thread.edit(archived=True, locked=True)
await election_channel.move(beginning=True, category=archive)
def get_election_info(self) -> ElectionInformation:
"""
@@ -429,8 +272,7 @@ class ElectionCycle:
with Session(engine) as session:
stmt = select(ElectionInformation).where(
ElectionInformation.ElectionCycle
== get_election_cycle_name(self.config)
ElectionInformation.ElectionCycle == get_election_cycle_name(self.config)
)
return session.scalar(stmt)
@@ -439,11 +281,9 @@ class ElectionCycle:
Sends a message to general chat reminder clan about election.
"""
general: nextcord.TextChannel = self.guild.get_channel(
self.config.params.general_channel_id
)
general: nextcord.TextChannel = self.guild.get_channel(self.config.params.general_channel_id)
message = f"Reminder to <@&{self.config.params.clan_member_role_id}>! Clan elections are happening! Head over "
message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!"
message += f"to <#{self.get_election_info().ElectionChannel}> to cast your vote!"
await general.send(content=message)
-266
View File
@@ -1,266 +0,0 @@
"""
Contains various views to be used in the high council nomination process.
"""
import nextcord
from nextcord.interactions import Interaction
from sqlalchemy import select
from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import CouncilMembers, engine
from utils import (
NOMINEE_SPLITTER,
decorate_cl_nomination,
decorate_dm_nomination,
get_election_cycle_name,
)
class NominateDMModal(nextcord.ui.Modal):
"""
Modal for nominating oneself. User writes a speech and then the callback sends the speech into the correct channel.
"""
def __init__(
self,
config: DakinyConfig,
usr_id: int,
usr_nominated: bool,
speech_content: str,
):
super().__init__(
"Discord Manager Nomination",
timeout=5 * 60, # 5 minutes
)
self.usr_id: int = usr_id
self.config: DakinyConfig = config
self.usr_nominated = usr_nominated
self.speech = nextcord.ui.TextInput(
label="Nomination Speech",
style=nextcord.TextInputStyle.paragraph,
placeholder="Please write your speech here.",
required=True,
max_length=1500,
default_value=speech_content,
)
self.add_item(self.speech)
async def callback(self, interaction: Interaction) -> None:
"""
Sends the nomination speech into the voting channel and enters the user into the voting database.
Args:
interaction (Interaction): Interaction with user.
"""
speech = decorate_dm_nomination(interaction.user.id, self.speech.value)
with Session(engine) as session:
# user already in table
cycle = get_election_cycle_name(self.config)
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == cycle)
.where(CouncilMembers.MemberId == self.usr_id)
)
council_elect = session.scalar(stmt)
if not self.usr_nominated:
msg = await interaction.channel.send(speech)
council_elect.DiscordManagerSpeechId = msg.id
session.commit()
else:
msg: nextcord.Message = await interaction.channel.fetch_message(
council_elect.DiscordManagerSpeechId
)
await msg.edit(
content=decorate_dm_nomination(
interaction.user.id, self.speech.value
)
)
class NominateCLModal(nextcord.ui.Modal):
"""
Modal for nominating oneself. User writes a speech and then the callback sends the speech into the correct channel.
"""
def __init__(
self,
config: DakinyConfig,
usr_id: int,
usr_nominated: bool,
speech_content: str,
):
super().__init__(
"Clan Leader Nomination",
timeout=5 * 60, # 5 minutes
)
self.usr_id: int = usr_id
self.config: DakinyConfig = config
self.usr_nominated = usr_nominated
self.speech = nextcord.ui.TextInput(
label="Nomination Speech",
style=nextcord.TextInputStyle.paragraph,
placeholder="Please write your speech here.",
required=True,
max_length=1500,
default_value=speech_content,
)
self.add_item(self.speech)
async def callback(self, interaction: Interaction) -> None:
"""
Sends the nomination speech into the voting channel and enters the user into the voting database.
Args:
interaction (Interaction): Interaction with user.
"""
speech = decorate_cl_nomination(interaction.user.id, self.speech.value)
with Session(engine) as session:
# user already in table
cycle = get_election_cycle_name(self.config)
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == cycle)
.where(CouncilMembers.MemberId == self.usr_id)
)
council_elect = session.scalar(stmt)
if not self.usr_nominated:
msg = await interaction.channel.send(speech)
council_elect.ClanLeaderSpeechId = msg.id
session.commit()
else:
msg: nextcord.Message = await interaction.channel.fetch_message(
council_elect.ClanLeaderSpeechId
)
await msg.edit(
content=decorate_cl_nomination(
interaction.user.id, self.speech.value
)
)
class NominateHighCouncilView(nextcord.ui.View):
"""
The buttons to press to start your High Council nomination.
"""
def __init__(self, config: DakinyConfig, guild: nextcord.Guild):
super().__init__(timeout=None)
self.config: DakinyConfig = config
self.guild: nextcord.Guild = guild
@nextcord.ui.button(
label="Nominate for Discord Manager",
style=nextcord.ButtonStyle.primary,
custom_id="nominate-dm",
)
async def nominate_dm(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
"""
Gathers previous information about the user if they have already nominated themselves, then launches
the nomination modal to get their speech.
Args:
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
# enforce council elect only
if (
self.guild.get_role(self.config.params.clan_council_elect_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_council_elect_role_id}> role to vote.",
ephemeral=True,
)
return
cycle = get_election_cycle_name(self.config)
usr_nominated = False
speech_content = ""
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == cycle)
.where(CouncilMembers.MemberId == interaction.user.id)
)
obj = session.scalar(stmt)
# user needs to exist to get here anyways...
if obj.DiscordManagerSpeechId is not None:
usr_nominated = True
msg = await interaction.channel.fetch_message(
obj.DiscordManagerSpeechId
)
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(
NominateDMModal(
self.config, interaction.user.id, usr_nominated, speech_content
)
)
@nextcord.ui.button(
label="Nominate for Clan Leader",
style=nextcord.ButtonStyle.primary,
custom_id="nominate-cl",
)
async def nominate_cl(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
"""
Gathers previous information about the user if they have already nominated themselves, then launches
the nomination modal to get their speech.
Args:
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
# enforce council elect only
if (
self.guild.get_role(self.config.params.clan_council_elect_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_council_elect_role_id}> role to vote.",
ephemeral=True,
)
return
cycle = get_election_cycle_name(self.config)
usr_nominated = False
speech_content = ""
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == cycle)
.where(CouncilMembers.MemberId == interaction.user.id)
)
obj = session.scalar(stmt)
# user needs to exist to get here anyways...
if obj.ClanLeaderSpeechId is not None:
usr_nominated = True
msg = await interaction.channel.fetch_message(obj.ClanLeaderSpeechId)
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(
NominateCLModal(
self.config, interaction.user.id, usr_nominated, speech_content
)
)
+28 -38
View File
@@ -9,7 +9,9 @@ from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import CouncilMembers, engine
from utils import NOMINEE_SPLITTER, decorate_nomination, get_election_cycle_name
from utils import get_election_cycle_name
NOMINEE_SPLITTER = "========================================\n\n"
class NominateModal(nextcord.ui.Modal):
@@ -17,13 +19,7 @@ class NominateModal(nextcord.ui.Modal):
Modal for nominating oneself. User writes a speech and then the callback sends the speech into the correct channel.
"""
def __init__(
self,
config: DakinyConfig,
usr_id: int,
usr_nominated: bool,
speech_content: str,
):
def __init__(self, config: DakinyConfig, usr_id: int, usr_nominated: bool, speech_content: str):
super().__init__(
"Nomination",
timeout=5 * 60, # 5 minutes
@@ -74,12 +70,8 @@ class NominateModal(nextcord.ui.Modal):
)
obj = session.scalar(stmt)
msg: nextcord.Message = await interaction.channel.fetch_message(
obj.SpeechMsgId
)
await msg.edit(
content=decorate_nomination(interaction.user.id, self.speech.value)
)
msg: nextcord.Message = await interaction.channel.fetch_message(obj.SpeechMsgId)
await msg.edit(content=decorate_nomination(interaction.user.id, self.speech.value))
class NominateView(nextcord.ui.View):
@@ -87,19 +79,12 @@ class NominateView(nextcord.ui.View):
The button to press to start your nomination.
"""
def __init__(self, config: DakinyConfig, guild: nextcord.Guild):
def __init__(self, config):
super().__init__(timeout=None)
self.config: DakinyConfig = config
self.guild: nextcord.Guild = guild
@nextcord.ui.button(
label="Nominate Yourself",
style=nextcord.ButtonStyle.primary,
custom_id="nominate",
)
async def nominate(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
@nextcord.ui.button(label="Nominate Yourself", style=nextcord.ButtonStyle.primary, custom_id="nominate")
async def nominate(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None:
"""
Gathers previous information about the user if they have already nominated themselves, then launches
the nomination modal to get their speech.
@@ -109,17 +94,6 @@ class NominateView(nextcord.ui.View):
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
# enforce members only
if (
self.guild.get_role(self.config.params.clan_member_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_member_role_id}> role to vote.",
ephemeral=True,
)
return
cycle = get_election_cycle_name(self.config)
usr_nominated = False
speech_content = ""
@@ -136,7 +110,23 @@ class NominateView(nextcord.ui.View):
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(
NominateModal(
self.config, interaction.user.id, usr_nominated, speech_content
)
NominateModal(self.config, interaction.user.id, usr_nominated, speech_content)
)
def decorate_nomination(user_id: int, speech: str) -> str:
"""
Decorates a nomination speech so that it sits within multiple speeches better.
Args:
user_id (int): ID of user sending the speech
speech (str): Speech from the user
Returns:
str: Decorated speech
"""
msg = f"**Clan Council Nominee: <@{user_id}>**\n"
msg += NOMINEE_SPLITTER
msg += speech
return msg
-259
View File
@@ -1,259 +0,0 @@
"""
Contains various views to be used while voting for high council. Controls the logic when performing the vote.
"""
from typing import List
import nextcord
from nextcord import SelectOption
from nextcord.interactions import Interaction
from sqlalchemy import exc, select, update
from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from utils import (
build_highcouncil_votes_str,
get_cl_election_name,
get_dm_election_name,
get_election_cycle_name,
)
class VoteSelect(nextcord.ui.StringSelect):
def __init__(
self, options: List["SelectOption"], config: DakinyConfig, election_type: str
) -> None:
super().__init__(
placeholder="Choose 1",
min_values=1,
max_values=1,
options=options,
)
self.config = config
self.election_type = election_type
async def callback(self, interaction: Interaction):
# decide on election type
match self.election_type:
case "cl":
election_name: str = get_cl_election_name(self.config)
case "dm":
election_name: str = get_dm_election_name(self.config)
case _:
# should never happen
raise ValueError('election type was not "cl" or "dm"')
cycle: str = get_election_cycle_name(self.config)
with Session(engine) as session:
stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
ElectionInformation.ElectionCycle == cycle
)
vote_msg_id = session.scalar(stmt)
if self.values[0] == "Abstain":
voted_for = None
else:
voted_for = int(self.values[0])
new_vote = Votes(
ElectionCycle=election_name,
VoterId=interaction.user.id,
VotedForId=voted_for,
)
try:
session.add(new_vote)
session.commit()
except exc.IntegrityError:
session.rollback()
stmt = (
update(Votes)
.where(Votes.ElectionCycle == election_name)
.where(Votes.VoterId == interaction.user.id)
.values(VotedForId=voted_for)
)
session.execute(stmt)
session.commit()
vote_message_str = build_highcouncil_votes_str(self.config, interaction.guild)
vote_msg: nextcord.Message = await interaction.channel.fetch_message(
vote_msg_id
)
await vote_msg.edit(content=vote_message_str)
if voted_for is not None:
msg = f"You have voted for <@{voted_for}>"
else:
msg = "You have abstained from voting."
await interaction.edit(content=msg, view=None)
class VoteHighcouncilView(nextcord.ui.View):
"""
The button to press to start voting for HighCouncil.
"""
def __init__(self, config: DakinyConfig, guild: nextcord.Guild):
super().__init__(timeout=None)
self.config = config
self.guild = guild
@nextcord.ui.button(
label="Vote for Clan Leader",
style=nextcord.ButtonStyle.primary,
custom_id="vote-cl",
)
async def vote_cl(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
"""
Enforces only council elect vote. Gets election cycle, creates voting
options for clan leader (checking if user has voted already),
then sends the voting view.
Args:
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
# enforce council elect only
if (
self.guild.get_role(self.config.params.clan_council_elect_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_council_elect_role_id}> role to vote.",
ephemeral=True,
)
return
cycle: str = get_election_cycle_name(self.config)
election: str = get_cl_election_name(self.config)
with Session(engine) as session:
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionCycle == election)
.where(Votes.VoterId == interaction.user.id)
)
voted_for = session.scalar(stmt)
stmt = (
select(CouncilMembers.MemberId)
.where(CouncilMembers.ElectionCycle == cycle)
.filter(CouncilMembers.ClanLeaderSpeechId.is_not(None))
)
options = [
SelectOption(
label=f"{self.guild.get_member(nominee).display_name} ☑️"
if voted_for == nominee
else f"{self.guild.get_member(nominee).display_name}",
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name,
)
]
options.append(
SelectOption(
label="Abstain ☑️" if voted_for is None else "Abstain",
value="Abstain",
)
)
view = nextcord.ui.View()
view.add_item(
VoteSelect(options=options, config=self.config, election_type="cl")
)
await interaction.response.send_message(
content="Please vote for a Clan Leader!",
view=view,
ephemeral=True,
)
@nextcord.ui.button(
label="Vote for Discord Manager",
style=nextcord.ButtonStyle.primary,
custom_id="vote-dm",
)
async def vote_dm(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
"""
Enforces only council elect vote. Gets election cycle, creates voting
options for discord manager (checking if user has voted already),
then sends the voting view.
Args:
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
# enforce council elect only
if (
self.guild.get_role(self.config.params.clan_council_elect_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_council_elect_role_id}> role to vote.",
ephemeral=True,
)
return
cycle: str = get_election_cycle_name(self.config)
election: str = get_dm_election_name(self.config)
with Session(engine) as session:
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionCycle == election)
.where(Votes.VoterId == interaction.user.id)
)
voted_for = session.scalar(stmt)
stmt = (
select(CouncilMembers.MemberId)
.where(CouncilMembers.ElectionCycle == cycle)
.filter(CouncilMembers.DiscordManagerSpeechId.is_not(None))
)
options = [
SelectOption(
label=f"{self.guild.get_member(nominee).display_name} ☑️"
if voted_for == nominee
else f"{self.guild.get_member(nominee).display_name}",
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name,
)
]
options.append(
SelectOption(
label="Abstain ☑️" if voted_for is None else "Abstain",
value="Abstain",
)
)
view = nextcord.ui.View()
view.add_item(
VoteSelect(options=options, config=self.config, election_type="dm")
)
await interaction.response.send_message(
content="Please vote for a Discord Manager!",
view=view,
ephemeral=True,
)
+18 -48
View File
@@ -7,16 +7,11 @@ from typing import List
import nextcord
from nextcord import SelectOption
from nextcord.interactions import Interaction
from sqlalchemy import exc, select, update
from sqlalchemy import exc, select
from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from database_setup.sql_data_classes import CouncilMembers, ElectionInformation, Votes, engine
from utils import build_votes_str, get_election_cycle_name
@@ -43,11 +38,7 @@ class VoteSelect(nextcord.ui.StringSelect):
else:
voted_for = int(self.values[0])
new_vote = Votes(
ElectionCycle=election_name,
VoterId=interaction.user.id,
VotedForId=voted_for,
)
new_vote = Votes(ElectionCycle=election_name, VoterId=interaction.user.id, VotedForId=voted_for)
try:
session.add(new_vote)
@@ -56,18 +47,16 @@ class VoteSelect(nextcord.ui.StringSelect):
except exc.IntegrityError:
session.rollback()
stmt = (
update(Votes)
select(Votes)
.where(Votes.ElectionCycle == election_name)
.where(Votes.VoterId == interaction.user.id)
.values(VotedForId=voted_for)
)
session.execute(stmt)
vote = session.scalar(stmt)
vote.VotedForId = voted_for
session.commit()
vote_message_str = build_votes_str(self.config, interaction.guild)
vote_msg: nextcord.Message = await interaction.channel.fetch_message(
vote_msg_id
)
vote_message_str = build_votes_str(self.config)
vote_msg: nextcord.Message = await interaction.channel.fetch_message(vote_msg_id)
await vote_msg.edit(content=vote_message_str)
if voted_for is not None:
@@ -88,15 +77,10 @@ class VoteView(nextcord.ui.View):
self.config = config
self.guild = guild
@nextcord.ui.button(
label="Vote", style=nextcord.ButtonStyle.primary, custom_id="vote"
)
async def vote(
self, button: nextcord.ui.Button, interaction: nextcord.Interaction
) -> None:
@nextcord.ui.button(label="Vote", style=nextcord.ButtonStyle.primary, custom_id="vote")
async def vote(self, button: nextcord.ui.Button, interaction: nextcord.Interaction) -> None:
"""
Enforces only members vote. Gets election cycle, creates voting options (checking if user has voted already),
then sends the voting view.
TODO
Args:
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
@@ -104,13 +88,9 @@ class VoteView(nextcord.ui.View):
"""
# enforce members only
if (
self.guild.get_role(self.config.params.clan_member_role_id)
not in interaction.user.roles
):
if self.guild.get_role(self.config.params.clan_member_role_id) not in interaction.user.roles:
await interaction.response.send_message(
f"You must have the <@&{self.config.params.clan_member_role_id}> role to vote.",
ephemeral=True,
f"You must have the <@&{self.config.params.clan_member_role_id}> role to vote.", ephemeral=True
)
return
@@ -124,28 +104,18 @@ class VoteView(nextcord.ui.View):
)
voted_for = session.scalar(stmt)
stmt = select(CouncilMembers.MemberId).where(
CouncilMembers.ElectionCycle == election
)
stmt = select(CouncilMembers.MemberId).where(CouncilMembers.ElectionCycle == election)
options = [
SelectOption(
label=f"{self.guild.get_member(nominee).display_name} ☑️"
label=f"{self.guild.get_member(nominee).nick} ☑️"
if voted_for == nominee
else f"{self.guild.get_member(nominee).display_name}",
else f"{self.guild.get_member(nominee).nick}",
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name,
)
for nominee in session.scalars(stmt)
]
options.append(
SelectOption(
label="Abstain ☑️" if voted_for is None else "Abstain",
value="Abstain",
)
)
options.append(SelectOption(label="Abstain ☑️" if voted_for is None else "Abstain", value="Abstain"))
view = nextcord.ui.View()
view.add_item(VoteSelect(options=options, config=self.config))
-82
View File
@@ -1,82 +0,0 @@
"""
File contains logic for creating a webhook listener, pulling from git, and updating all messages.
"""
import os
import nextcord
import uvicorn
from fastapi import FastAPI
from git import Repo
from nextcord.ext import commands
from config import DakinyConfig
from utils import REPO_DIR, update_clan_messages
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.config: DakinyConfig = self.bot.config
self.repo_setup()
@commands.Cog.listener()
async def on_ready(self):
"""
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"])
config = uvicorn.Config(app=app, host="0.0.0.0", port=5000, log_level="info")
server = uvicorn.Server(config)
await server.serve()
async def update_messages(self):
"""
Pulls from git and then calls method to update all the discord messages as described by the
files in the git repo.
"""
self.origin.pull()
guild: nextcord.Guild = self.bot.get_guild(self.config.params.guild_id)
await update_clan_messages(guild)
return {"message": "Successfully updated all discord messages."}
def repo_setup(self):
"""
Checks if the repo exists and creates repo object, clones if it does not.
"""
# check if repo has been cloned. open repo object
if os.path.isdir(REPO_DIR):
self.repo = Repo(REPO_DIR)
else:
self.repo = Repo.clone_from(REPO_URL, REPO_DIR)
self.origin = self.repo.remote("origin")
if self.config.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):
"""Adds function to bot"""
bot.add_cog(ClanMessages(bot))
+1 -7
View File
@@ -4,9 +4,6 @@ Cog loader. Adds all command cogs to the bot.
from nextcord.ext import commands
from cogs.commands.record import RecordCog
from cogs.commands.refresh import RefreshCog
from cogs.commands.remove import RemoveCog
from cogs.commands.request import RequestCog
from cogs.commands.set import setCog
from cogs.commands.test import TestCog
@@ -20,10 +17,7 @@ def setup(bot: commands.Bot):
Adds all command classes to bot.
"""
bot.add_cog(RecordCog(bot))
bot.add_cog(RefreshCog())
bot.add_cog(RemoveCog())
bot.add_cog(RequestCog(bot))
bot.add_cog(TestCog(bot))
bot.add_cog(ViewCog())
bot.add_cog(setCog())
bot.add_cog(ViewCog())
-218
View File
@@ -1,218 +0,0 @@
"""
Contained commands and views for record commands.
"""
import nextcord
from discord import TextInputStyle
from nextcord.ext import commands
from config import DakinyConfig
from utils import record_event_host, record_event_participation
class RecordCog(commands.Cog, name="Record Commands"):
"""
Class holds record commands.
"""
def __init__(self, bot) -> None:
"""
Default constructor for RecordCog class. Does nothing.
"""
super().__init__()
self.bot: commands.Bot = bot
@nextcord.slash_command()
async def record(self, interaction: nextcord.Interaction):
"""
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.",
),
):
"""
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(self.bot.config.params.event_host_role_id)
not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{self.bot.config.params.event_host_role_id}> 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, self.bot.config, host.id),
ephemeral=True,
embed=embed,
)
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, config: DakinyConfig, host_id: int):
super().__init__()
self.name: str = name
self.config: DakinyConfig = config
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
):
"""
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):
"""
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
):
"""
Submits the event embed to the event channel. Removes view and embed from recording message.
Sends point information to database.
"""
discord_users = self.event_participants.values.members
if discord_users:
discord_users_list: list[str] = [
str(user.id) for user in discord_users if user.id is not self.host_id
]
record_event_participation(self.config, discord_users_list, True)
non_discord_users = interaction.message.embeds[0].fields[1].value
if non_discord_users:
non_discord_users_list: list[str] = non_discord_users.split("\n")
record_event_participation(self.config, non_discord_users_list, False)
record_event_host(self.config, self.host_id)
event_channel: nextcord.TextChannel = interaction.guild.get_channel(
self.config.params.event_post_channel_id
)
await event_channel.send(embed=interaction.message.embeds[0])
await interaction.response.edit_message(
content="Event has been submitted.", view=None, embed=None
)
self.stop()
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.
"""
# TODO turn this embed into a dict
def __init__(self, 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):
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):
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)
-34
View File
@@ -1,34 +0,0 @@
"""
Contained commands and views for refresh commands.
"""
import nextcord
from nextcord.ext import commands
from utils import update_clan_messages
class RefreshCog(commands.Cog, name="Refresh Commands"):
"""
Class holds refresh commands.
"""
@nextcord.slash_command(
default_member_permissions=nextcord.Permissions(administrator=True)
)
async def refresh(self, interaction: nextcord.Interaction):
"""
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):
"""
Calls update_clan_messages() to update all messages.
Args:
interaction (nextcord.Interaction): Interaction with the command.
"""
await interaction.response.defer(ephemeral=True)
await update_clan_messages(interaction.guild)
await interaction.followup.send("Messages have been refreshed.", ephemeral=True)
-86
View File
@@ -1,86 +0,0 @@
"""
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 database_setup.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):
"""
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
),
):
"""
Deletes the given presence id from the database.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
presence (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
):
"""
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])
+10 -109
View File
@@ -4,10 +4,8 @@ Contains commands and views for request commands.
import nextcord
from nextcord.ext import commands
from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import Presences, engine
class RequestCog(commands.Cog, name="Request Commands"):
@@ -24,49 +22,15 @@ class RequestCog(commands.Cog, name="Request Commands"):
@nextcord.slash_command()
async def request(self, interaction: nextcord.Interaction):
"""
Parent slash command for request. Never called.
Parent slash command for help. 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.",
),
):
"""
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.
"""
config: DakinyConfig = self.bot.config
content = f'<@{interaction.user.id}> would like to add presence "{presence}."'
view = PresenceChange(presence)
await self.bot.get_channel(config.params.admin_channel_id).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.",
required=True, min_length=2, max_length=32, description="Your requested new nickname."
),
):
"""
@@ -80,12 +44,8 @@ class RequestCog(commands.Cog, name="Request Commands"):
config: DakinyConfig = self.bot.config
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(config.params.admin_channel_id).send(
content=content, view=view
)
await interaction.response.send_message(
"Request sent to admins.", ephemeral=True
)
await self.bot.get_channel(config.params.admin_channel_id).send(content=content, view=view)
await interaction.response.send_message("Request sent to admins.", ephemeral=True)
class UsernameRequest(nextcord.ui.View):
@@ -93,19 +53,13 @@ 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)
def __init__(self, user_id: int, new_nickname: str, *, timeout: float | None = 180, auto_defer: bool = True):
super().__init__(timeout=timeout, auto_defer=auto_defer)
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
):
async def approve(self, button: nextcord.ui.Button, interaction: nextcord.Interaction):
"""
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.
@@ -119,10 +73,7 @@ class UsernameRequest(nextcord.ui.View):
if user is not None:
try:
await user.edit(nick=self.new_nickname)
content = (
interaction.message.content
+ f"\nApproved by <@{interaction.user.id}>."
)
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(
@@ -133,13 +84,10 @@ class UsernameRequest(nextcord.ui.View):
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
)
await interaction.followup.edit_message(interaction.message.id, content=content, view=None)
else:
await interaction.response.edit_message(
content="An error occured while changing the user's nickname.",
view=None,
content="An error occured while changing the user's nickname.", view=None
)
@nextcord.ui.button(label="Deny", style=nextcord.ButtonStyle.red)
@@ -153,50 +101,3 @@ class UsernameRequest(nextcord.ui.View):
"""
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
):
"""
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):
"""
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)
+9 -25
View File
@@ -10,8 +10,8 @@ from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session
from dakiny_bot import birthday_list
from database_setup.sql_data_classes import UserBirthday, engine
from main import birthday_list
class setCog(commands.Cog, name="Set Commands"):
@@ -30,18 +30,12 @@ class setCog(commands.Cog, name="Set Commands"):
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."
)
@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."
),
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."),
):
"""
Sets a user's birthday in the database. Removes on input 0/0. If birthday is today, adds to the birthday_list.
@@ -56,9 +50,7 @@ class setCog(commands.Cog, name="Set Commands"):
await interaction.response.defer(ephemeral=True)
if month == 0 and day == 0:
stmt = delete(UserBirthday).where(
UserBirthday.UserId == interaction.user.id
)
stmt = delete(UserBirthday).where(UserBirthday.UserId == interaction.user.id)
with Session(engine) as session:
session.execute(stmt)
session.commit()
@@ -70,18 +62,14 @@ class setCog(commands.Cog, name="Set Commands"):
try:
birthday = datetime(today.year, month, day)
except ValueError:
await interaction.followup.send(
f"Invalid date: {month}/{day}.", ephemeral=True
)
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}"
)
new_birthday = UserBirthday(UserId=interaction.user.id, Birthday=f"{month}/{day}")
with Session(engine) as session:
try:
session.add(new_birthday)
@@ -89,13 +77,9 @@ class setCog(commands.Cog, name="Set Commands"):
except IntegrityError:
# User already has birthday, change it instead.
session.rollback()
stmt = select(UserBirthday).where(
UserBirthday.UserId == interaction.user.id
)
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
)
await interaction.followup.send(f"Birthday set to {month}/{day}.", ephemeral=True)
+31 -71
View File
@@ -7,18 +7,7 @@ from nextcord.ext import application_checks, commands
from classes.election_cycle import ElectionCycle
from config import DakinyConfig
from utils import get_current_election_cycle
def is_me():
"""
predicate that makes sure only I can run the test commands.
"""
def predicate(interaction: nextcord.Interaction):
return interaction.user.id == 206602713511886849
return application_checks.check(predicate)
from utils import get_current_cycle
class TestCog(commands.Cog, name="Test commands."):
@@ -32,14 +21,37 @@ class TestCog(commands.Cog, name="Test commands."):
"""
self.bot = bot
@nextcord.slash_command(guild_ids=[1000247361639030804])
def is_me():
def predicate(interaction: nextcord.Interaction):
return interaction.user.id == 206602713511886849
return application_checks.check(predicate)
@nextcord.slash_command()
@is_me()
async def test(self, interaction: nextcord.Interaction):
"""
Parent slash command for help. Never called.
"""
@test.subcommand()
@is_me()
async def msg_builder(self, interaction: nextcord.Interaction):
"""
Tests the message builder.
Args:
interaction (nextcord.Interaction): Interaction that invoked the command.
"""
cycle = ElectionCycle(0, self.bot)
message = cycle.build_election_announcement()
# doesn't tell discord that the message was responded to. Is a test command so I don't really care atm.
for m in message:
await interaction.channel.send(content=m)
@test.subcommand()
async def election_start(self, interaction: nextcord.Interaction):
"""
idc
@@ -49,15 +61,10 @@ class TestCog(commands.Cog, name="Test commands."):
"""
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
cycle = ElectionCycle(get_current_cycle(config), config, self.bot.get_guild(config.params.guild_id))
await cycle.election_start_action()
@test.subcommand()
@is_me()
async def voting_start(self, interaction: nextcord.Interaction):
"""
tests the voting start action
@@ -66,15 +73,10 @@ class TestCog(commands.Cog, name="Test commands."):
interaction (nextcord.Interaction): _description_
"""
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
cycle = ElectionCycle(get_current_cycle(config), config, self.bot.get_guild(config.params.guild_id))
await cycle.voting_start_action()
@test.subcommand()
@is_me()
async def voting_end(self, interaction: nextcord.Interaction):
"""
tests the voting start action
@@ -83,59 +85,17 @@ class TestCog(commands.Cog, name="Test commands."):
interaction (nextcord.Interaction): _description_
"""
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
cycle = ElectionCycle(get_current_cycle(config), config, self.bot.get_guild(config.params.guild_id))
await cycle.voting_end_action()
@test.subcommand()
@is_me()
async def highcouncil_vote(self, interaction: nextcord.Interaction):
"""
tests the voting start action
Args:
interaction (nextcord.Interaction): _description_
"""
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
await cycle.highcouncil_election_start_action()
@test.subcommand()
@is_me()
async def election_reminder(self, interaction: nextcord.Interaction):
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
cycle = ElectionCycle(get_current_cycle(config), config, self.bot.get_guild(config.params.guild_id))
await cycle.election_reminder()
@test.subcommand()
@is_me()
async def daily(self, interaction: nextcord.Interaction):
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
cycle = ElectionCycle(get_current_cycle(config), config, self.bot.get_guild(config.params.guild_id))
await cycle.check_date()
@test.subcommand()
@is_me()
async def term_start(self, interaction: nextcord.Interaction):
config: DakinyConfig = self.bot.config
cycle = ElectionCycle(
get_current_election_cycle(config),
config,
self.bot.get_guild(config.params.guild_id),
)
await cycle.term_start_action()
+3 -144
View File
@@ -3,17 +3,11 @@ 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 database_setup.sql_data_classes import ElectionInformation, Votes, engine
class ViewCog(commands.Cog, name="View Commands"):
@@ -32,113 +26,6 @@ class ViewCog(commands.Cog, name="View Commands"):
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."
)
@@ -193,12 +80,12 @@ class ViewCog(commands.Cog, name="View Commands"):
"""
with Session(engine) as session:
stmt = select(Votes.ElectionCycle).distinct()
stmt = select(ElectionInformation.ElectionCycle)
elections = session.scalars(stmt)
# if empty compare string
if not election_name:
await interaction.response.send_autocomplete(list(elections)[:25])
await interaction.response.send_autocomplete(elections)
return
autofilled_elections: "list[str]" = [
@@ -207,31 +94,3 @@ class ViewCog(commands.Cog, name="View Commands"):
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])
+1 -3
View File
@@ -4,8 +4,7 @@ Initializing code for all loops.
from nextcord.ext import commands
from cogs.loops.bot_presence import BotPresence
from cogs.loops.daily_loop import DailyLoop
from .daily_loop import DailyLoop
def setup(bot: commands.Bot):
@@ -13,5 +12,4 @@ def setup(bot: commands.Bot):
Adds functions to bot.
"""
bot.add_cog(BotPresence(bot))
bot.add_cog(DailyLoop(bot))
-48
View File
@@ -1,48 +0,0 @@
"""
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 database_setup.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
): # pylint: disable=unused-argument
"""Listener for Guild Join. Updated presence on guild join."""
await self.change_presence()
@commands.Cog.listener()
async def on_ready(self):
"""Starts the loop when bot is ready."""
await self.bot.wait_until_ready()
# pylint doesn't recognize that periodic_change_presense is subclassed from tasks.loop.
self.periodic_change_presence.start() # pylint: disable=no-member
@tasks.loop(minutes=5)
async def periodic_change_presence(self):
"""Loops every 5 minutes and changes presence."""
await self.change_presence()
async def change_presence(self):
"""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(name="Custom Status", state=presence)
)
+5 -13
View File
@@ -13,9 +13,9 @@ from sqlalchemy.orm import Session
from classes.election_cycle import ElectionCycle
from config import DakinyConfig
from dakiny_bot import birthday_list
from database_setup.sql_data_classes import UserBirthday, engine
from main import birthday_list
from utils import get_current_election_cycle
from utils import get_current_cycle
class DailyLoop(commands.Cog, name="Daily Loop"):
@@ -77,11 +77,7 @@ class DailyLoop(commands.Cog, name="Daily Loop"):
"""
config: DakinyConfig = self.bot.config
cycle = get_current_election_cycle(config)
# prevent exception from using cycle as an Int in an array
if cycle is None:
return None
cycle = get_current_cycle(config)
return ElectionCycle(cycle, config, self.bot.get_guild(config.params.guild_id))
@@ -111,14 +107,10 @@ class DailyLoop(commands.Cog, name="Daily Loop"):
today = datetime.today()
guild: nextcord.Guild = self.bot.get_guild(self.bot.config.params.guild_id)
announce_channel: nextcord.TextChannel = guild.get_channel(
self.bot.config.params.general_channel_id
)
announce_channel: nextcord.TextChannel = guild.get_channel(self.bot.config.params.general_channel_id)
cakedays: list[nextcord.Member] = [
human
for human in guild.humans
if human.joined_at.month == today.month and human.joined_at.day == today.day
human for human in guild.humans if human.joined_at.month == today.month and human.joined_at.day == today.day
]
for member in cakedays:
+2 -4
View File
@@ -5,7 +5,7 @@ Defines cog for on message listener.
import nextcord
from nextcord.ext import commands
from main import birthday_list
from dakiny_bot import birthday_list
class SubmissionListener(commands.Cog, name="Message Listener"):
@@ -39,7 +39,5 @@ async def birthday_action(message: nextcord.Message) -> None:
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)
):
if message.author.id in birthday_list or bool(set(message.raw_mentions) & set(birthday_list)):
await message.add_reaction("🎂")
+2 -7
View File
@@ -1,7 +1,6 @@
"""
File contains classes defining the Hydra configuraton structure.
"""
from dataclasses import dataclass
@@ -11,22 +10,18 @@ class Params:
Class defines the various configuration parameters.
"""
active_category_id: int
admin_channel_id: int # might want to rename this...
announcement_category_id: int
archive_channel_group_id: int
clan_council_elect_role_id: int
clan_council_role_id: int
clan_info_channel_id: int
clan_member_role_id: int
discord_token: str
event_host_points: int
event_host_role_id: int
event_participate_points: int
event_post_channel_id: int
general_channel_id: int
git_branch: str
guild_id: int
high_council_role_id: int
voting_parent_id: int
@dataclass
+2 -8
View File
@@ -8,9 +8,7 @@ from hydra.core.config_store import ConfigStore
from nextcord.ext import commands
import database_setup.create_database as create_database
from classes.nominate_highcouncil_views import NominateHighCouncilView
from classes.nominate_views import NominateView
from classes.voting_highcouncil_views import VoteHighcouncilView
from classes.voting_views import VoteView
from config import DakinyConfig
@@ -22,9 +20,7 @@ birthday_list: list[int] = []
logger = logging.getLogger("nextcord")
logger.setLevel(logging.DEBUG)
handler = logging.FileHandler(filename="nextcord.log", encoding="utf-8", mode="w")
handler.setFormatter(
logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s")
)
handler.setFormatter(logging.Formatter("%(asctime)s:%(levelname)s:%(name)s: %(message)s"))
logger.addHandler(handler)
@@ -55,10 +51,8 @@ def main(cfg: DakinyConfig):
"""When Discord is connected"""
assert bot.user is not None
create_database # pylint: disable=pointless-statement
bot.add_view(NominateView(cfg, bot.get_guild(cfg.params.guild_id)))
bot.add_view(NominateHighCouncilView(cfg, bot.get_guild(cfg.params.guild_id)))
bot.add_view(NominateView(cfg))
bot.add_view(VoteView(cfg, bot.get_guild(cfg.params.guild_id)))
bot.add_view(VoteHighcouncilView(cfg, bot.get_guild(cfg.params.guild_id)))
print(f"{bot.user.name} has connected to Discord!")
# Run Discord bot
+3 -25
View File
@@ -1,22 +1,18 @@
CREATE TABLE ElectionInformation(
ElectionCycle varchar(30) not null unique,
ThreadId bigint unsigned not null,
ElectionChannel bigint unsigned not null,
CouncilVoteMsgId bigint unsigned,
HighCouncilVoteMsgId bigint unsigned,
CouncilNominationMsgId bigint unsigned,
HighCouncilNominationMsgId bigint unsigned,
PRIMARY KEY (ElectionCycle)
);
CREATE TABLE CouncilMembers(
ElectionCycle varchar(30) not null,
MemberId bigint unsigned not null,
SpeechMsgId bigint unsigned not null,
SpeechMsgId bigint unsigned not null,
Elected boolean DEFAULT false,
HighCouncilElect boolean DEFAULT false,
DiscordManager boolean DEFAULT false,
ClanLeader boolean DEFAULT false,
DiscordManagerSpeechId bigint unsigned,
ClanLeaderSpeechId bigint unsigned,
FOREIGN KEY (ElectionCycle) REFERENCES ElectionInformation(ElectionCycle),
PRIMARY KEY (ElectionCycle, MemberId)
);
@@ -33,22 +29,4 @@ CREATE TABLE UserBirthday (
UserId bigint unsigned not null,
Birthday varchar(5) not null,
PRIMARY KEY (UserId)
);
CREATE TABLE Presences (
Id integer PRIMARY KEY,
Presence varchar(100) not null
);
CREATE TABLE ParticipationPoints (
Cycle varchar(30) not null,
UserId varchar(32) not null,
DiscordMember boolean not null,
Points int unsigned DEFAULT 0,
PRIMARY KEY (Cycle, UserId)
);
CREATE TABLE Messages (
Path varchar(100) PRIMARY KEY,
Id bigint unsigned not null
);
+2 -6
View File
@@ -1,7 +1,6 @@
"""
Simple file to set up the sqlalchemy connection to database.
"""
from sqlalchemy import create_engine
from sqlalchemy.ext.automap import automap_base
@@ -12,10 +11,7 @@ Base = automap_base()
Base.prepare(autoload_with=engine)
# define the classes
CouncilMembers = Base.classes.CouncilMembers
ElectionInformation = Base.classes.ElectionInformation
Messages = Base.classes.Messages
ParticipationPoints = Base.classes.ParticipationPoints
Presences = Base.classes.Presences
UserBirthday = Base.classes.UserBirthday
CouncilMembers = Base.classes.CouncilMembers
Votes = Base.classes.Votes
UserBirthday = Base.classes.UserBirthday
+1 -4
View File
@@ -2,7 +2,4 @@ nextcord==2.4.2
python-dotenv==1.0.0
pytz==2022.7.1
hydra-core==1.3.2
sqlalchemy<2.0
pandas==1.5.2
fastapi==0.111.0
GitPython==3.1.43
sqlalchemy<2.0
+15 -422
View File
@@ -2,25 +2,13 @@
File contains util functions.
"""
import os
from datetime import datetime
import nextcord
from sqlalchemy import select, update
from sqlalchemy import select
from sqlalchemy.orm import Session
from config import DakinyConfig
from database_setup.sql_data_classes import (
CouncilMembers,
Messages,
ParticipationPoints,
Votes,
engine,
)
NOMINEE_SPLITTER = "========================================\n\n"
REPO_DIR = "./dakiny-messages"
MESSAGES_DIR = f"{REPO_DIR}/messages"
from database_setup.sql_data_classes import CouncilMembers, Votes, engine
def get_datetime(day: str) -> datetime:
@@ -38,12 +26,9 @@ def get_datetime(day: str) -> datetime:
day = int(day.split("/")[1])
# ensure that year rolls over if current month is December and the next month is less than 12.
if datetime.today().month == 12 and month < 12:
if datetime.today().month == 12 & month < 12:
year += 1
if datetime.today().month == 1 and month > 1:
year -= 1
return datetime(year, month, day)
@@ -51,62 +36,16 @@ def get_election_cycle_name(config: DakinyConfig) -> str:
"""
Builds the unique name of the current election cycle
Args:
config (DakinyConfig): the config used to know the election cycle.
Returns:
str: the name of the election cycle
"""
return f"vote-{get_cycle_name(config)}"
def get_cycle_name(config: DakinyConfig) -> str:
"""
Builds the name of the current cycle/quarter
Args:
config (DakinyConfig): the config used to know the election cycle.
Returns:
str: the name of the cycle.
"""
cycle = get_current_cycle(config)
current_cycle = config.election_cycles[cycle]
return f"{get_datetime(current_cycle.term_start).year}-{current_cycle.name}"
return f"vote-{get_datetime(current_cycle.election_start).year}-{current_cycle.name}"
def get_dm_election_name(config: DakinyConfig) -> str:
"""
creates the name of the discord manager election
Args:
config (DakinyConfig): Config sent to get_election_cycle_name
Returns:
str: the name of the election
"""
return f"{get_election_cycle_name(config)}-dm"
def get_cl_election_name(config: DakinyConfig) -> str:
"""
creates the name of the clan leader election
Args:
config (DakinyConfig): Config sent to get_election_cycle_name
Returns:
str: the name of the election
"""
return f"{get_election_cycle_name(config)}-cl"
def get_current_election_cycle(config: DakinyConfig) -> int:
def get_current_cycle(config: DakinyConfig) -> int:
"""
Returns index for active election. An election is considered active between the election start and the term start.
@@ -114,7 +53,7 @@ def get_current_election_cycle(config: DakinyConfig) -> int:
config (DakinyConfig): Config object to use
Returns:
int: cycle index for active election. None for no active cycle
int: cycle index for active election. -1 for no active cycle
"""
today = datetime.today()
@@ -126,38 +65,15 @@ def get_current_election_cycle(config: DakinyConfig) -> int:
):
return cycle
return None
return -1
def get_current_cycle(config: DakinyConfig) -> int:
"""
Returns the index for the current cycle/quarter. Defined as between term start and end. Shouldn't return none
Args:
config (DakinyConfig): Config object to use
Returns:
int: cycle index for current cycle.
"""
today = datetime.today()
for cycle, election in enumerate(config.election_cycles):
if (
today.date() >= get_datetime(election.term_start).date()
and today.date() <= get_datetime(election.term_end).date()
):
return cycle
return None
def build_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str:
def build_votes_str(config: DakinyConfig) -> str:
"""
Creates a string that contains all nominees and their votes to be sent as a discord message.
Args:
config (DakinyConfig): Daking config object
guild (nextcord.Guild): The guild object. Used to get display names for sorting.
Returns:
str: String that contains all nominees and their votes.
@@ -166,345 +82,22 @@ def build_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str:
votes: dict[int, int] = {}
with Session(engine) as session:
stmt = select(CouncilMembers).where(
CouncilMembers.ElectionCycle == get_election_cycle_name(config)
)
stmt = select(CouncilMembers).where(CouncilMembers.ElectionCycle == get_election_cycle_name(config))
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: guild.get_member(nominee.MemberId).display_name,
):
for nominee in session.scalars(stmt):
votes[nominee.MemberId] = 0
stmt = select(Votes).where(
Votes.ElectionCycle == get_election_cycle_name(config)
)
stmt = select(Votes).where(Votes.ElectionCycle == get_election_cycle_name(config))
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
votes[vote.VotedForId] += 1
votes_str: str = "===================="
votes_str += user_vote_str(votes)
return votes_str
def build_highcouncil_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str:
"""
Creates a string that contains all high council nominees and their votes to be sent as a discord message.
Args:
config (DakinyConfig): Daking config object
Returns:
str: String that contains all nominees and their votes.
guild (nextcord.Guild): The guild object. Used to get display names for sorting.
"""
cl_votes: dict[int, int] = {}
dm_votes: dict[int, int] = {}
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == get_election_cycle_name(config))
.filter(CouncilMembers.ClanLeaderSpeechId.is_not(None))
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: guild.get_member(nominee.MemberId).display_name,
):
cl_votes[nominee.MemberId] = 0
stmt = select(Votes).where(Votes.ElectionCycle == get_cl_election_name(config))
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
cl_votes[vote.VotedForId] += 1
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionCycle == get_election_cycle_name(config))
.filter(CouncilMembers.DiscordManagerSpeechId.is_not(None))
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: guild.get_member(nominee.MemberId).display_name,
):
dm_votes[nominee.MemberId] = 0
stmt = select(Votes).where(Votes.ElectionCycle == get_dm_election_name(config))
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
dm_votes[vote.VotedForId] += 1
# build the votes string
votes_str: str = "Clan Leader Votes\n===================="
votes_str += user_vote_str(cl_votes)
votes_str += "\n\nDiscord Manager Votes\n===================="
votes_str += user_vote_str(dm_votes)
return votes_str
def user_vote_str(votes: dict[int, int]) -> str:
"""
Creates the string representing how many votes each user has.
Args:
votes (dict[int, int]): Dict of all users and how many votes they have.
Returns:
str: The string showing names and number of votes.
"""
ret_str: str = ""
for user_id, num_votes in votes.items():
ret_str += f"\n<@{user_id}>: {'☑️' * num_votes}"
return ret_str
votes_str += f"\n<@{user_id}>: "
for num in range(num_votes):
votes_str += "☑️"
def decorate_nomination(user_id: int, speech: str) -> str:
"""
Decorates a nomination speech so that it sits within multiple speeches better.
Args:
user_id (int): ID of user sending the speech
speech (str): Speech from the user
Returns:
str: Decorated speech
"""
msg = f"**Clan Council Nominee: <@{user_id}>**\n"
msg += NOMINEE_SPLITTER
msg += speech
return msg
def decorate_cl_nomination(user_id: int, speech: str) -> str:
"""
Decorates a nomination speech so that it sits within multiple speeches better.
Args:
user_id (int): ID of user sending the speech
speech (str): Speech from the user
Returns:
str: Decorated speech
"""
msg = f"**Clan Leader Nominee: <@{user_id}>**\n"
msg += NOMINEE_SPLITTER
msg += speech
return msg
def decorate_dm_nomination(user_id: int, speech: str) -> str:
"""
Decorates a nomination speech so that it sits within multiple speeches better.
Args:
user_id (int): ID of user sending the speech
speech (str): Speech from the user
Returns:
str: Decorated speech
"""
msg = f"**Discord Manager Nominee: <@{user_id}>**\n"
msg += NOMINEE_SPLITTER
msg += speech
return msg
def record_event_participation(config: DakinyConfig, users: list[str], is_member: bool):
"""
Adds participation points to all users listed. Checks if they don't exist in order to make a new entry, otherwise
adds points to the existing value.
Args:
config (DakinyConfig): Dakiny config for getting points
users (list[int]): users that participated and need points.
is_member (bool): whether the users are a discord member or not
"""
cycle = get_cycle_name(config)
participation_point_value = config.params.event_participate_points
with Session(engine) as session:
for user in users:
stmt = (
select(ParticipationPoints)
.where(ParticipationPoints.UserId == user)
.where(ParticipationPoints.Cycle == cycle)
)
points = session.scalar(stmt)
if points is None:
new_points = ParticipationPoints(
Cycle=cycle,
UserId=user,
DiscordMember=is_member,
Points=participation_point_value,
)
session.add(new_points)
else:
points.Points += participation_point_value
session.commit()
def record_event_host(config: DakinyConfig, user: int):
"""
Adds host points to the given user. Checks if they don't exist in order to make a new entry, otherwise adds points
to the existing value.
Args:
config (DakinyConfig): Dakiny config for getting points
user (int): Id of user that hosted.
"""
cycle = get_cycle_name(config)
host_point_value = config.params.event_host_points
with Session(engine) as session:
stmt = (
select(ParticipationPoints)
.where(ParticipationPoints.UserId == str(user))
.where(ParticipationPoints.Cycle == cycle)
)
points = session.scalar(stmt)
if points is None:
new_points = ParticipationPoints(
Cycle=cycle,
UserId=str(user),
DiscordMember=True,
Points=host_point_value,
)
session.add(new_points)
else:
points.Points += host_point_value
session.commit()
async def update_clan_messages(guild: nextcord.Guild):
"""
Updates all clan messages to contents in MESSAGES_DIR.
Does not to any git work. Is expected to be called after git is pulled.
for ever directory, sort the files by number then post
posting involves checking if there is a message for that particular file then editing it or creating a
new one. the message could be deleted, which would cause a new message to be sent and database updated.
Channel fetching/management is handled by another function.
Args:
guild (nextcord.Guild): Guild object that all channels and messages belong to.
"""
dirs = [
dir_name
for dir_name in os.listdir(MESSAGES_DIR)
if os.path.isdir(f"{MESSAGES_DIR}/{dir_name}")
]
with Session(engine) as session:
for dir_name in dirs:
dir_path = f"{MESSAGES_DIR}/{dir_name}"
channel = await get_clan_message_channel(session, dir_name, guild)
sorted_files = sorted(
os.listdir(dir_path),
key=lambda filename: int(filename.split(".")[0]),
)
for file_name in sorted_files:
message_name = f"{dir_name}/{file_name}"
file_path = f"{dir_path}/{file_name}"
with open(file_path, "r", encoding="utf8") as reader:
stmt = select(Messages.Id).where(Messages.Path == message_name)
msg_id = session.scalar(stmt)
if msg_id is None:
# don't have a message for this file
msg = await channel.send(reader.read(2000))
new_message = Messages(Path=message_name, Id=msg.id)
session.add(new_message)
session.commit()
else:
try:
msg: nextcord.Message = await channel.fetch_message(msg_id)
await msg.edit(reader.read(2000))
except nextcord.NotFound:
msg = await channel.send(reader.read(2000))
stmt = (
update(Messages)
.where(Messages.Path == message_name)
.values(Id=msg.id)
)
session.execute(stmt)
session.commit()
async def get_clan_message_channel(
session: Session, channel_name: str, guild: nextcord.Guild
) -> nextcord.TextChannel:
"""
Used to obtain a channel from the guild object. Searches database for an id to get the channel. If that is not
found, then searches through all channels in the guild to find one with a matching name. If that also doesn't
turn up a channel, then creates it. Adds an entry into database for last two paths.
Assumes that there is an open session to the database.
Args:
session (Session): Open connection to database.
channel_name (str): Name of channel to obtain.
guild (nextcord.Guild): Guild object to get channel from.
Returns:
nextcord.TextChannel: Channel object with name channel_name.
"""
stmt = select(Messages.Id).where(Messages.Path == channel_name)
channel_id = session.scalar(stmt)
if channel_id is None:
ret_channel = None
# channel is not tracked. see if it exists
channels = await guild.fetch_channels()
for channel in channels:
if (
channel.name == channel_name
and channel.__class__ is nextcord.TextChannel
):
ret_channel = channel
if ret_channel is None:
# channel doesn't exist, we must create it
ret_channel = await guild.create_text_channel(
channel_name,
reason="Creating channel as required for by design of Dakiny-Messages repo.",
)
# record channel information
new_item = Messages(Path=channel_name, Id=ret_channel.id)
session.add(new_item)
session.commit()
return ret_channel
try:
return await guild.fetch_channel(channel_id)
except nextcord.NotFound:
# the channel was deleted for some reason
ret_channel = await guild.create_text_channel(
channel_name,
reason="Creating channel as required for by design of Dakiny-Messages repo.",
)
stmt = (
update(Messages)
.where(Messages.Path == channel_name)
.values(Id=ret_channel.id)
)
session.execute(stmt)
session.commit()
return ret_channel
return votes_str