Compare commits

..
6 Commits
Author SHA1 Message Date
imcompiling 1f1e4ad20d Feat: use db models in voting views.
build_develop / build (push) Successful in 39s
fix pylance smells.
2025-07-31 13:50:34 -04:00
imcompiling 354823ca1f Feat: use models in voting highcouncil views.
build_develop / build (push) Successful in 40s
2025-07-31 13:48:11 -04:00
imcompiling 45f17e09d3 Feat: Update Nominate classes to use models.
build_develop / build (push) Successful in 44s
2025-07-24 14:28:39 -04:00
imcompiling 530ca5b172 Feat: Update file to use new models.
build_develop / build (push) Successful in 38s
2025-07-24 13:46:13 -04:00
imcompiling 3dfdc10334 Feat: Swap Election to use new models.
build_develop / build (push) Successful in 1m7s
Address many pylance errors. Rewrite a few functions, will require testing to ensure
functionality remains the same.
2025-07-24 12:39:16 -04:00
imcompiling 6c9353990c Fix: dakiny_settings file rename.
build_develop / build (push) Successful in 34s
2025-07-17 09:18:33 -04:00
9 changed files with 381 additions and 233 deletions
+2
View File
@@ -8,7 +8,9 @@
"CLANLEADER",
"Dakiny",
"DISCORDMANAGER",
"gitea",
"highcouncil",
"imcompiling",
"nextcord",
"ondelete"
]
+1
View File
@@ -0,0 +1 @@
"""Empty init file for classes module."""
+156 -96
View File
@@ -5,17 +5,17 @@ from datetime import UTC, datetime, timedelta
from pathlib import Path
import nextcord
from sqlalchemy import select, update
from sqlalchemy.orm import Session
from sqlmodel import col
from dakiny_bot.classes.nominate_highcouncil_views import NominateHighCouncilView
from dakiny_bot.classes.nominate_views import NominateView
from dakiny_bot.classes.voting_highcouncil_views import VoteHighcouncilView
from dakiny_bot.classes.voting_views import VoteView
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote
from dakiny_bot.logger import dakiny_logger
from dakiny_bot.utils import (
TEMPLATES_DIR,
@@ -26,8 +26,6 @@ from dakiny_bot.utils import (
get_datetime,
get_election_name,
)
from sqlalchemy import select, true, update
from sqlalchemy.orm import Session
REQUIRED_VOTES_TO_BE_ELECTED = 3
@@ -70,7 +68,7 @@ class Election:
list[str]: A list of message contents in the correct order. Must be a string because the message is too
long.
"""
with Path(f"{TEMPLATES_DIR}/election_announcement.txt", encoding="utf8").open() as file:
with Path(f"{TEMPLATES_DIR}/election_announcement.txt").open() as file:
message = file.read()
next_election = get_datetime(
@@ -98,7 +96,7 @@ class Election:
Returns:
str: Message to send.
"""
with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt", encoding="utf8").open() as file:
with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt").open() as file:
message = file.read()
return message.format(
@@ -116,7 +114,7 @@ class Election:
Returns:
str: Message to send.
"""
with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt", encoding="utf8").open() as file:
with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt").open() as file:
message = file.read()
return message.format(
@@ -131,7 +129,7 @@ class Election:
Returns:
str: Message to send
"""
with Path(f"{TEMPLATES_DIR}/vote_start.txt", encoding="utf8").open() as file:
with Path(f"{TEMPLATES_DIR}/vote_start.txt").open() as file:
message = file.read()
vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end)
@@ -171,7 +169,10 @@ class Election:
async def election_start_action(self) -> None:
"""Performs required actions for when the election cycle starts."""
# create new channel for election
parent: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.voting)
parent = self.guild.get_channel(settings.channel_ids.voting)
if not isinstance(parent, nextcord.TextChannel):
dakiny_logger.error("Voting channel (id: %s) is not a text channel.", settings.channel_ids.voting)
return
quarter_name: str = get_election_name()
thread_msg: nextcord.Message = await parent.send(content=quarter_name)
@@ -192,9 +193,9 @@ class Election:
# create new entry into database.
with Session(engine) as session:
new_cycle = ElectionInformation(
ElectionName=quarter_name,
ThreadId=thread.id,
CouncilNominationMsgId=nominate_msg.id,
name=quarter_name,
thread_id=thread.id,
council_nomination_msg_id=nominate_msg.id,
)
session.add(new_cycle)
session.commit()
@@ -207,64 +208,98 @@ class Election:
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.guild),
view=VoteView(self.guild),
)
with Session(engine) as session:
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
info = session.scalar(stmt)
info.CouncilVoteMsgId = vote_msg.id
stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
if (election_info := session.scalar(stmt)) is None:
dakiny_logger.error("Unable to find election %s", get_election_name())
return
if (election_thread := self.guild.get_thread(election_info.thread_id)) is None:
dakiny_logger.error("Recorded thread id %s does not result in a thread.", election_info.thread_id)
return
await election_thread.send(message)
vote_msg = await election_thread.send(
build_votes_str(self.guild),
view=VoteView(self.guild),
)
election_info.council_vote_msg_id = vote_msg.id
session.commit()
thread: nextcord.Thread = self.guild.get_thread(info.ThreadId)
nominate_msg: nextcord.Message = await thread.fetch_message(info.CouncilNominationMsgId)
if election_info.council_nomination_msg_id is None:
dakiny_logger.error("Election doesn't have a nomination message set.")
return
nominate_msg: nextcord.Message = await election_thread.fetch_message(
election_info.council_nomination_msg_id
)
await nominate_msg.edit(view=None)
async def voting_end_action(self) -> None:
"""Performs required actions for when voting time has ended."""
message = self.build_highcouncil_election_announce()
council_elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
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.guild))
if (council_elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
dakiny_logger.error("Clan council elect role not found (id: %s).", settings.role_ids.clan_council_elect)
return
election_name = get_election_name()
with Session(engine) as session:
stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name))
if (info := session.scalar(stmt)) is None:
dakiny_logger.error("Unable to find election %s", election_name)
return
if (election_thread := self.guild.get_thread(info.thread_id)) is None:
dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
return
nominate_msg: nextcord.Message = await election_thread.send(
message, view=NominateHighCouncilView(self.guild)
)
# remove vote button from vote message
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == election_name)
info = session.scalar(stmt)
vote_msg: nextcord.Message = await election_thread.fetch_message(info.CouncilVoteMsgId)
if not info.council_vote_msg_id:
dakiny_logger.error("Council vote message id not set.")
return
vote_msg: nextcord.Message = await election_thread.fetch_message(info.council_vote_msg_id)
await vote_msg.edit(view=None)
# add the highcouncil nomination message to election information
info.HighCouncilNominationMsgId = nominate_msg.id
info.high_council_nomination_msg_id = nominate_msg.id
session.commit()
# add the highcouncil role to everyone with 3 or more votes. set elected to true in database
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.Type == VoteType.COUNCIL)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.type == VoteType.COUNCIL))
)
vote_count = collections.Counter(session.scalars(stmt))
for user_id, votes in vote_count.items():
if user_id is None:
# Abstained votes, ignore
continue
if votes >= REQUIRED_VOTES_TO_BE_ELECTED:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == user_id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == user_id))
)
council_member = session.scalar(stmt)
council_member.Elected = True
if (council_member := session.scalar(stmt)) is None:
dakiny_logger.error(
"Council member (id: %s) for election %s not found.", user_id, election_name
)
return
council_member.elected = True
session.commit()
member: nextcord.Member = self.guild.get_member(user_id)
if (member := self.guild.get_member(user_id)) is None:
dakiny_logger.error("Member (id: %s) not found in guild.", user_id)
return
await member.add_roles(
council_elect_role,
@@ -274,42 +309,58 @@ class Election:
async def highcouncil_election_start_action(self) -> None:
"""Performs required actions for when the highcouncil is elected."""
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.guild),
view=VoteHighcouncilView(self.guild),
)
with Session(engine) as session:
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
info = session.scalar(stmt)
info.HighCouncilVoteMsgId = vote_msg.id
stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
if (info := session.scalar(stmt)) is None:
dakiny_logger.error("Unknown election name %s.", get_election_name())
return
if (election_thread := self.guild.get_thread(info.thread_id)) is None:
dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
return
await election_thread.send(message)
vote_msg = await election_thread.send(
build_highcouncil_votes_str(self.guild),
view=VoteHighcouncilView(self.guild),
)
info.high_council_vote_msg_id = vote_msg.id
session.commit()
# remove highcouncil nomination view
nomination_msg: nextcord.Message = await election_thread.fetch_message(info.HighCouncilNominationMsgId)
if info.high_council_nomination_msg_id is None:
dakiny_logger.error("Election doesn't have a nomination message set.")
return
nomination_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_nomination_msg_id)
await nomination_msg.edit(view=None)
async def term_start_action(self) -> None:
"""Marks the election as over."""
self.election_over = True
elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
council_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council)
if (elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
dakiny_logger.error("Cannot find Elect Role.")
return
if (council_role := self.guild.get_role(settings.role_ids.clan_council)) is None:
dakiny_logger.error("Cannot find Council Role.")
return
election_name = get_election_name()
with Session(engine) as session:
# role management
stmt = (
select(CouncilMembers.MemberId)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.Elected == true())
select(col(CouncilMember.member_id))
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.elected))
)
for member_id in session.scalars(stmt):
member = self.guild.get_member(member_id)
if (member := self.guild.get_member(member_id)) is None:
dakiny_logger.warning("Cannot find member (id: %s) to manage roles.", member_id)
continue
await member.remove_roles(
elect_role,
reason=f"{member.display_name} no longer needs the elect role.",
@@ -322,35 +373,36 @@ class Election:
# announce high council members
# ties are undefined behavior.
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.Type == VoteType.DISCORDMANAGER)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.type == VoteType.DISCORDMANAGER))
)
dm_votes = collections.Counter(session.scalars(stmt))
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.Type == VoteType.CLANLEADER)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.type == VoteType.CLANLEADER))
)
cl_votes = collections.Counter(session.scalars(stmt))
cl_votes.most_common()
try:
dm_winner = max(dm_votes, key=dm_votes.get)
dm_winner = dm_votes.most_common(1).pop()[0]
stmt = (
update(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == dm_winner)
update(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == dm_winner))
.values(DiscordManager=True)
)
session.execute(stmt)
except ValueError:
dm_winner = "No winner"
try:
cl_winner = max(cl_votes, key=cl_votes.get)
cl_winner = cl_votes.most_common(1).pop()[0]
stmt = (
update(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == cl_winner)
update(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == cl_winner))
.values(ClanLeader=True)
)
session.execute(stmt)
@@ -359,14 +411,21 @@ class Election:
session.commit()
election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name))
if (info := session.scalar(stmt)) is None:
dakiny_logger.error("Could not find information for this election (%s).", election_name)
return
if (election_thread := self.guild.get_thread(info.thread_id)) is None:
dakiny_logger.error("Recorded thread id %s does not result in a thread.", info.thread_id)
return
if info.high_council_vote_msg_id is None:
dakiny_logger.error("No recorded High Council Voting Message.")
return
# clear view from vote message
stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
ElectionInformation.ElectionName == get_election_name()
)
vote_msg_id = session.scalar(stmt)
vote_msg: nextcord.Message = await election_thread.fetch_message(vote_msg_id)
vote_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_vote_msg_id)
await vote_msg.edit(view=None)
highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n"
@@ -378,22 +437,23 @@ class Election:
await election_thread.edit(archived=True, locked=True)
await edit_clan_council_info(self.guild)
def get_election_info(self) -> ElectionInformation:
"""
Gets information for current election.
Returns:
ElectionInformation: The information from the database about the current election.
"""
with Session(engine) as session:
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
return session.scalar(stmt)
async def election_reminder(self) -> None:
"""Sends a message to general chat reminder clan about election."""
general: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.general)
general = self.guild.get_channel(settings.channel_ids.general)
message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over "
message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!"
if not isinstance(general, nextcord.TextChannel):
dakiny_logger.error(
"Recorded general channel (id: %s) is not a text channel.", settings.channel_ids.general
)
return
await general.send(content=message)
with Session(engine) as session:
stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
if (info := session.scalar(stmt)) is None:
dakiny_logger.error("Could not find election %s.", get_election_name())
return
message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over "
message += f"to <#{info.thread_id}> to cast your vote!"
await general.send(content=message)
@@ -1,17 +1,21 @@
"""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 sqlmodel import col
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.sql_data_classes import CouncilMembers, engine
from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember
from dakiny_bot.logger import dakiny_logger
from dakiny_bot.utils import (
NOMINEE_SPLITTER,
decorate_cl_nomination,
decorate_dm_nomination,
get_election_name,
)
from nextcord.interactions import Interaction
from sqlalchemy import select
from sqlalchemy.orm import Session
class NominateDMModal(nextcord.ui.Modal):
@@ -52,26 +56,35 @@ class NominateDMModal(nextcord.ui.Modal):
interaction (Interaction): Interaction with user.
"""
speech = decorate_dm_nomination(interaction.user.id, self.speech.value)
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
speech = decorate_dm_nomination(interaction.user.id, self.speech.value if self.speech.value else "")
with Session(engine) as session:
# user already in table
election_name = get_election_name()
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == self.usr_id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == self.usr_id))
)
council_elect = session.scalar(stmt)
if not self.usr_nominated:
if council_elect is None:
dakiny_logger.error("Did not find member (id: %s) in election %s.", self.usr_id, election_name)
return
if not self.usr_nominated or council_elect.discord_manager_speech_id is None:
msg = await interaction.channel.send(speech)
council_elect.DiscordManagerSpeechId = msg.id
council_elect.discord_manager_speech_id = 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))
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.discord_manager_speech_id)
await msg.edit(content=speech)
class NominateCLModal(nextcord.ui.Modal):
@@ -112,26 +125,35 @@ class NominateCLModal(nextcord.ui.Modal):
interaction (Interaction): Interaction with user.
"""
speech = decorate_cl_nomination(interaction.user.id, self.speech.value)
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
speech = decorate_cl_nomination(interaction.user.id, self.speech.value if self.speech.value else "")
with Session(engine) as session:
# user already in table
election_name = get_election_name()
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == self.usr_id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == self.usr_id))
)
council_elect = session.scalar(stmt)
if not self.usr_nominated:
if council_elect is None:
dakiny_logger.error("Did not find member (id: %s) in election %s.", self.usr_id, election_name)
return
if not self.usr_nominated or council_elect.clan_leader_speech_id is None:
msg = await interaction.channel.send(speech)
council_elect.ClanLeaderSpeechId = msg.id
council_elect.clan_leader_speech_id = 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))
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.clan_leader_speech_id)
await msg.edit(content=speech)
class NominateHighCouncilView(nextcord.ui.View):
@@ -157,8 +179,16 @@ class NominateHighCouncilView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
# enforce council elect only
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
if (
isinstance(interaction.user, nextcord.User)
or self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
ephemeral=True,
@@ -170,16 +200,15 @@ class NominateHighCouncilView(nextcord.ui.View):
speech_content = ""
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == interaction.user.id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == interaction.user.id))
)
obj = session.scalar(stmt)
council_elect = session.scalar(stmt)
# user needs to exist to get here anyways...
if obj.DiscordManagerSpeechId is not None:
if council_elect is not None and council_elect.discord_manager_speech_id is not None:
usr_nominated = True
msg = await interaction.channel.fetch_message(obj.DiscordManagerSpeechId)
msg = await interaction.channel.fetch_message(council_elect.discord_manager_speech_id)
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(NominateDMModal(interaction.user.id, usr_nominated, speech_content))
@@ -200,8 +229,16 @@ class NominateHighCouncilView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
# enforce council elect only
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
if (
isinstance(interaction.user, nextcord.User)
or self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
ephemeral=True,
@@ -213,16 +250,15 @@ class NominateHighCouncilView(nextcord.ui.View):
speech_content = ""
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == interaction.user.id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == interaction.user.id))
)
obj = session.scalar(stmt)
council_elect = session.scalar(stmt)
# user needs to exist to get here anyways...
if obj.ClanLeaderSpeechId is not None:
if council_elect is not None and council_elect.clan_leader_speech_id is not None:
usr_nominated = True
msg = await interaction.channel.fetch_message(obj.ClanLeaderSpeechId)
msg = await interaction.channel.fetch_message(council_elect.clan_leader_speech_id)
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(NominateCLModal(interaction.user.id, usr_nominated, speech_content))
+46 -20
View File
@@ -1,12 +1,16 @@
"""Contains various views to be used in the nomination process."""
import nextcord
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.sql_data_classes import CouncilMembers, engine
from dakiny_bot.utils import NOMINEE_SPLITTER, decorate_nomination, get_election_name
from nextcord.interactions import Interaction
from sqlalchemy import select
from sqlalchemy.orm import Session
from sqlmodel import col
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember
from dakiny_bot.logger import dakiny_logger
from dakiny_bot.utils import NOMINEE_SPLITTER, decorate_nomination, get_election_name
class NominateModal(nextcord.ui.Modal):
@@ -47,31 +51,45 @@ class NominateModal(nextcord.ui.Modal):
interaction (Interaction): Interaction with user.
"""
speech = decorate_nomination(interaction.user.id, self.speech.value)
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
speech = decorate_nomination(interaction.user.id, self.speech.value if self.speech.value else "")
election_name = get_election_name()
with Session(engine) as session:
if not self.usr_nominated:
msg = await interaction.channel.send(speech)
council_member = CouncilMembers(
ElectionName=election_name,
MemberId=interaction.user.id,
SpeechMsgId=msg.id,
council_member = CouncilMember(
election_name=election_name,
member_id=interaction.user.id,
speech_msg_id=msg.id,
)
session.add(council_member)
session.commit()
else:
# user already nominated, just need to update the election speech
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == self.usr_id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == self.usr_id))
)
obj = session.scalar(stmt)
council_elect = 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))
if council_elect is None:
dakiny_logger.error("Did not find member (id: %s) in election %s.", self.usr_id, election_name)
return
if council_elect.speech_msg_id is None:
msg = await interaction.channel.send(speech)
council_elect.speech_msg_id = msg.id
session.commit()
else:
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.speech_msg_id)
await msg.edit(content=speech)
class NominateView(nextcord.ui.View):
@@ -97,8 +115,16 @@ class NominateView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
return
# enforce members only
if self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
if (
isinstance(interaction.user, nextcord.User)
or self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles
):
await interaction.response.send_message(
f"You must have the <@&{settings.role_ids.clan_member}> role to vote.",
ephemeral=True,
@@ -110,14 +136,14 @@ class NominateView(nextcord.ui.View):
speech_content = ""
with Session(engine) as session:
stmt = (
select(CouncilMembers)
.where(CouncilMembers.ElectionName == election_name)
.where(CouncilMembers.MemberId == interaction.user.id)
select(CouncilMember)
.where(col(CouncilMember.election_name == election_name))
.where(col(CouncilMember.member_id == interaction.user.id))
)
obj = session.scalar(stmt)
if obj is not None:
if obj is not None and obj.speech_msg_id is not None:
usr_nominated = True
msg = await interaction.channel.fetch_message(obj.SpeechMsgId)
msg = await interaction.channel.fetch_message(obj.speech_msg_id)
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
await interaction.response.send_modal(NominateModal(interaction.user.id, usr_nominated, speech_content))
-4
View File
@@ -4,7 +4,3 @@ extend = "../../pyproject.toml"
# Exclude a variety of commonly ignored directories.
exclude = ["gitea_webhook.py"]
[tool.ruff.lint]
# ignore implicit-namespace-package in this directory.
ignore = ["INP001"]
+59 -46
View File
@@ -1,18 +1,17 @@
"""Contains various views to be used while voting for high council. Controls the logic when performing the vote."""
import nextcord
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from dakiny_bot.utils import VoteType, build_highcouncil_votes_str, get_election_name
from nextcord import SelectOption
from nextcord.interactions import Interaction
from sqlalchemy import exc, select, update
from sqlalchemy.orm import Session
from sqlmodel import col
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote
from dakiny_bot.logger import dakiny_logger
from dakiny_bot.utils import VoteType, build_highcouncil_votes_str, get_election_name
class VoteSelect(nextcord.ui.StringSelect):
@@ -37,20 +36,32 @@ class VoteSelect(nextcord.ui.StringSelect):
Args:
interaction (Interaction): Interaction with this callback.
"""
if (
interaction.user is None
or not isinstance(interaction.channel, nextcord.TextChannel)
or interaction.guild is None
):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
# if no guild, is problem because should only be in guild.
return
election_name: str = get_election_name()
with Session(engine) as session:
stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
ElectionInformation.ElectionName == election_name
stmt = select(col(ElectionInformation.high_council_vote_msg_id)).where(
col(ElectionInformation.name == election_name)
)
vote_msg_id = session.scalar(stmt)
if (vote_msg_id := session.scalar(stmt)) is None:
dakiny_logger.error("Unable to find a Highcouncil Vote Message ID in database.")
return
voted_for = None if self.values[0] == "Abstain" else int(self.values[0])
new_vote = Votes(
ElectionName=election_name,
VoterId=interaction.user.id,
VotedForId=voted_for,
Type=self.election_type,
new_vote = Vote(
election_name=election_name,
voter_id=interaction.user.id,
voted_for_id=voted_for,
type=self.election_type,
)
try:
@@ -61,10 +72,10 @@ class VoteSelect(nextcord.ui.StringSelect):
# user already has a vote in the election and election type
session.rollback()
stmt = (
update(Votes)
.where(Votes.ElectionName == election_name)
.where(Votes.VoterId == interaction.user.id)
.where(Votes.Type == self.election_type)
update(Vote)
.where(col(Vote.election_name == election_name))
.where(col(Vote.voter_id == interaction.user.id))
.where(col(Vote.type == self.election_type))
.values(VotedForId=voted_for)
)
session.execute(stmt)
@@ -103,6 +114,10 @@ class VoteHighcouncilView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if not isinstance(interaction.user, nextcord.Member):
# the user should only be a member of the guild.
return
# enforce council elect only
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
await interaction.response.send_message(
@@ -115,32 +130,29 @@ class VoteHighcouncilView(nextcord.ui.View):
with Session(engine) as session:
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.VoterId == interaction.user.id)
.where(Votes.Type == VoteType.CLANLEADER)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.voter_id == interaction.user.id))
.where(col(Vote.type == VoteType.CLANLEADER))
)
voted_for = session.scalar(stmt)
stmt = (
select(CouncilMembers.MemberId)
.where(CouncilMembers.ElectionName == election_name)
.filter(CouncilMembers.ClanLeaderSpeechId.is_not(None))
select(col(CouncilMember.member_id))
.where(col(CouncilMember.election_name == election_name))
.filter(col(CouncilMember.clan_leader_speech_id 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}"
),
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
)
if (member := self.guild.get_member(nominee)) is not None
]
options.append(
SelectOption(
@@ -175,6 +187,10 @@ class VoteHighcouncilView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if not isinstance(interaction.user, nextcord.Member):
# the user should only be a member of the guild.
return
# enforce council elect only
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
await interaction.response.send_message(
@@ -187,32 +203,29 @@ class VoteHighcouncilView(nextcord.ui.View):
with Session(engine) as session:
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.VoterId == interaction.user.id)
.where(Votes.Type == VoteType.DISCORDMANAGER)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.voter_id == interaction.user.id))
.where(col(Vote.type == VoteType.DISCORDMANAGER))
)
voted_for = session.scalar(stmt)
stmt = (
select(CouncilMembers.MemberId)
.where(CouncilMembers.ElectionName == election_name)
.filter(CouncilMembers.DiscordManagerSpeechId.is_not(None))
select(col(CouncilMember.member_id))
.where(col(CouncilMember.election_name == election_name))
.filter(col(CouncilMember.discord_manager_speech_id 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}"
),
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
)
if (member := self.guild.get_member(nominee)) is not None
]
options.append(
SelectOption(
+44 -30
View File
@@ -1,18 +1,17 @@
"""Contains various views to be used while voting. Controls the logic when performing the vote."""
import nextcord
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.sql_data_classes import (
CouncilMembers,
ElectionInformation,
Votes,
engine,
)
from dakiny_bot.utils import VoteType, build_votes_str, get_election_name
from nextcord import SelectOption
from nextcord.interactions import Interaction
from sqlalchemy import exc, select, update
from sqlalchemy.orm import Session
from sqlmodel import col
from dakiny_bot.dakiny_settings import settings
from dakiny_bot.database.db import engine
from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote
from dakiny_bot.logger import dakiny_logger
from dakiny_bot.utils import VoteType, build_votes_str, get_election_name
class VoteSelect(nextcord.ui.StringSelect):
@@ -36,18 +35,32 @@ class VoteSelect(nextcord.ui.StringSelect):
Args:
interaction (Interaction): Interaction with this callback.
"""
if (
interaction.user is None
or not isinstance(interaction.channel, nextcord.TextChannel)
or interaction.guild is None
):
# ignore none... Don't know what situation this is.
# if not textchannel, don't know how callback happened
# if no guild, is problem because should only be in guild.
return
election_name = get_election_name()
with Session(engine) as session:
stmt = select(ElectionInformation.CouncilVoteMsgId).where(ElectionInformation.ElectionName == election_name)
vote_msg_id = session.scalar(stmt)
stmt = select(col(ElectionInformation.council_vote_msg_id)).where(
col(ElectionInformation.name == election_name)
)
if (vote_msg_id := session.scalar(stmt)) is None:
dakiny_logger.error("Unable to find a Vote Message ID in database.")
return
voted_for = None if self.values[0] == "Abstain" else int(self.values[0])
new_vote = Votes(
ElectionName=election_name,
VoterId=interaction.user.id,
VotedForId=voted_for,
Type=VoteType.COUNCIL,
new_vote = Vote(
election_name=election_name,
voter_id=interaction.user.id,
voted_for_id=voted_for,
type=VoteType.COUNCIL,
)
try:
@@ -57,10 +70,10 @@ class VoteSelect(nextcord.ui.StringSelect):
except exc.IntegrityError:
session.rollback()
stmt = (
update(Votes)
.where(Votes.ElectionName == election_name)
.where(Votes.VoterId == interaction.user.id)
.where(Votes.Type == VoteType.COUNCIL)
update(Vote)
.where(col(Vote.election_name == election_name))
.where(col(Vote.voter_id == interaction.user.id))
.where(col(Vote.type == VoteType.COUNCIL))
.values(VotedForId=voted_for)
)
session.execute(stmt)
@@ -94,6 +107,10 @@ class VoteView(nextcord.ui.View):
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
interaction (nextcord.Interaction): Interaction of user who pressed the button
"""
if not isinstance(interaction.user, nextcord.Member):
# the user should only be a member of the guild.
return
# enforce members only
if self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
await interaction.response.send_message(
@@ -106,28 +123,25 @@ class VoteView(nextcord.ui.View):
with Session(engine) as session:
stmt = (
select(Votes.VotedForId)
.where(Votes.ElectionName == election_name)
.where(Votes.VoterId == interaction.user.id)
.where(Votes.Type == VoteType.COUNCIL)
select(col(Vote.voted_for_id))
.where(col(Vote.election_name == election_name))
.where(col(Vote.voter_id == interaction.user.id))
.where(col(Vote.type == VoteType.COUNCIL))
)
voted_for = session.scalar(stmt)
stmt = select(CouncilMembers.MemberId).where(CouncilMembers.ElectionName == election_name)
stmt = select(col(CouncilMember.member_id)).where(col(CouncilMember.election_name == election_name))
options = [
SelectOption(
label=(
f"{self.guild.get_member(nominee).display_name} ☑️"
if voted_for == nominee
else f"{self.guild.get_member(nominee).display_name}"
),
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
value=str(nominee),
)
for nominee in sorted(
session.scalars(stmt),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
)
if (member := self.guild.get_member(nominee)) is not None
]
options.append(
SelectOption(
+1 -1
View File
@@ -4,7 +4,7 @@ import nextcord
from nextcord.ext import commands
from dakiny_bot.app import birthday_list
from dakiny_bot.DakinySettings import settings
from dakiny_bot.dakiny_settings import settings
class SubmissionListener(commands.Cog, name="Message Listener"):