Compare commits
20
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
a15a35fcda | ||
|
|
44baf01a51 | ||
|
|
abc2950b21 | ||
|
|
1cc46ce2fd | ||
|
|
bb2d6b62fe | ||
|
|
77ce54f383 | ||
|
|
f81ebca4de | ||
|
|
2e66b1c37c | ||
|
|
ae1f98abe3 | ||
|
|
490c30e88c | ||
|
|
35670358c4 | ||
|
|
7a59f3b453 | ||
|
|
e01ab64411 | ||
|
|
df482d80f0 | ||
|
|
0d06d98060 | ||
|
|
c267762efa | ||
|
|
e837cec46b | ||
|
|
0a856755c8 | ||
|
|
eb8cfee7fd | ||
|
|
6b5dc771d4 |
Vendored
-2
@@ -8,9 +8,7 @@
|
|||||||
"CLANLEADER",
|
"CLANLEADER",
|
||||||
"Dakiny",
|
"Dakiny",
|
||||||
"DISCORDMANAGER",
|
"DISCORDMANAGER",
|
||||||
"gitea",
|
|
||||||
"highcouncil",
|
"highcouncil",
|
||||||
"imcompiling",
|
|
||||||
"nextcord",
|
"nextcord",
|
||||||
"ondelete"
|
"ondelete"
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -1 +0,0 @@
|
|||||||
"""Empty init file for classes module."""
|
|
||||||
+96
-156
@@ -5,17 +5,17 @@ from datetime import UTC, datetime, timedelta
|
|||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import nextcord
|
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_highcouncil_views import NominateHighCouncilView
|
||||||
from dakiny_bot.classes.nominate_views import NominateView
|
from dakiny_bot.classes.nominate_views import NominateView
|
||||||
from dakiny_bot.classes.voting_highcouncil_views import VoteHighcouncilView
|
from dakiny_bot.classes.voting_highcouncil_views import VoteHighcouncilView
|
||||||
from dakiny_bot.classes.voting_views import VoteView
|
from dakiny_bot.classes.voting_views import VoteView
|
||||||
from dakiny_bot.dakiny_settings import settings
|
from dakiny_bot.dakiny_settings import settings
|
||||||
from dakiny_bot.database.db import engine
|
from dakiny_bot.database.sql_data_classes import (
|
||||||
from dakiny_bot.database.models import CouncilMember, ElectionInformation, Vote
|
CouncilMembers,
|
||||||
|
ElectionInformation,
|
||||||
|
Votes,
|
||||||
|
engine,
|
||||||
|
)
|
||||||
from dakiny_bot.logger import dakiny_logger
|
from dakiny_bot.logger import dakiny_logger
|
||||||
from dakiny_bot.utils import (
|
from dakiny_bot.utils import (
|
||||||
TEMPLATES_DIR,
|
TEMPLATES_DIR,
|
||||||
@@ -26,6 +26,8 @@ from dakiny_bot.utils import (
|
|||||||
get_datetime,
|
get_datetime,
|
||||||
get_election_name,
|
get_election_name,
|
||||||
)
|
)
|
||||||
|
from sqlalchemy import select, true, update
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
REQUIRED_VOTES_TO_BE_ELECTED = 3
|
REQUIRED_VOTES_TO_BE_ELECTED = 3
|
||||||
|
|
||||||
@@ -68,7 +70,7 @@ class Election:
|
|||||||
list[str]: A list of message contents in the correct order. Must be a string because the message is too
|
list[str]: A list of message contents in the correct order. Must be a string because the message is too
|
||||||
long.
|
long.
|
||||||
"""
|
"""
|
||||||
with Path(f"{TEMPLATES_DIR}/election_announcement.txt").open() as file:
|
with Path(f"{TEMPLATES_DIR}/election_announcement.txt", encoding="utf8").open() as file:
|
||||||
message = file.read()
|
message = file.read()
|
||||||
|
|
||||||
next_election = get_datetime(
|
next_election = get_datetime(
|
||||||
@@ -96,7 +98,7 @@ class Election:
|
|||||||
Returns:
|
Returns:
|
||||||
str: Message to send.
|
str: Message to send.
|
||||||
"""
|
"""
|
||||||
with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt").open() as file:
|
with Path(f"{TEMPLATES_DIR}/highcouncil_election.txt", encoding="utf8").open() as file:
|
||||||
message = file.read()
|
message = file.read()
|
||||||
|
|
||||||
return message.format(
|
return message.format(
|
||||||
@@ -114,7 +116,7 @@ class Election:
|
|||||||
Returns:
|
Returns:
|
||||||
str: Message to send.
|
str: Message to send.
|
||||||
"""
|
"""
|
||||||
with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt").open() as file:
|
with Path(f"{TEMPLATES_DIR}/highcouncil_vote.txt", encoding="utf8").open() as file:
|
||||||
message = file.read()
|
message = file.read()
|
||||||
|
|
||||||
return message.format(
|
return message.format(
|
||||||
@@ -129,7 +131,7 @@ class Election:
|
|||||||
Returns:
|
Returns:
|
||||||
str: Message to send
|
str: Message to send
|
||||||
"""
|
"""
|
||||||
with Path(f"{TEMPLATES_DIR}/vote_start.txt").open() as file:
|
with Path(f"{TEMPLATES_DIR}/vote_start.txt", encoding="utf8").open() as file:
|
||||||
message = file.read()
|
message = file.read()
|
||||||
vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end)
|
vote_end: datetime = get_datetime(settings.election_cycles[self.cycle].voting_end)
|
||||||
|
|
||||||
@@ -169,10 +171,7 @@ class Election:
|
|||||||
async def election_start_action(self) -> None:
|
async def election_start_action(self) -> None:
|
||||||
"""Performs required actions for when the election cycle starts."""
|
"""Performs required actions for when the election cycle starts."""
|
||||||
# create new channel for election
|
# create new channel for election
|
||||||
parent = self.guild.get_channel(settings.channel_ids.voting)
|
parent: nextcord.TextChannel = 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()
|
quarter_name: str = get_election_name()
|
||||||
|
|
||||||
thread_msg: nextcord.Message = await parent.send(content=quarter_name)
|
thread_msg: nextcord.Message = await parent.send(content=quarter_name)
|
||||||
@@ -193,9 +192,9 @@ class Election:
|
|||||||
# create new entry into database.
|
# create new entry into database.
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
new_cycle = ElectionInformation(
|
new_cycle = ElectionInformation(
|
||||||
name=quarter_name,
|
ElectionName=quarter_name,
|
||||||
thread_id=thread.id,
|
ThreadId=thread.id,
|
||||||
council_nomination_msg_id=nominate_msg.id,
|
CouncilNominationMsgId=nominate_msg.id,
|
||||||
)
|
)
|
||||||
session.add(new_cycle)
|
session.add(new_cycle)
|
||||||
session.commit()
|
session.commit()
|
||||||
@@ -208,98 +207,64 @@ class Election:
|
|||||||
Removes write permission from the general clan member role.
|
Removes write permission from the general clan member role.
|
||||||
"""
|
"""
|
||||||
message = self.build_vote_start()
|
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:
|
with Session(engine) as session:
|
||||||
stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
|
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
|
||||||
if (election_info := session.scalar(stmt)) is None:
|
info = session.scalar(stmt)
|
||||||
dakiny_logger.error("Unable to find election %s", get_election_name())
|
info.CouncilVoteMsgId = vote_msg.id
|
||||||
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()
|
session.commit()
|
||||||
|
|
||||||
if election_info.council_nomination_msg_id is None:
|
thread: nextcord.Thread = self.guild.get_thread(info.ThreadId)
|
||||||
dakiny_logger.error("Election doesn't have a nomination message set.")
|
nominate_msg: nextcord.Message = await thread.fetch_message(info.CouncilNominationMsgId)
|
||||||
return
|
|
||||||
|
|
||||||
nominate_msg: nextcord.Message = await election_thread.fetch_message(
|
|
||||||
election_info.council_nomination_msg_id
|
|
||||||
)
|
|
||||||
await nominate_msg.edit(view=None)
|
await nominate_msg.edit(view=None)
|
||||||
|
|
||||||
async def voting_end_action(self) -> None:
|
async def voting_end_action(self) -> None:
|
||||||
"""Performs required actions for when voting time has ended."""
|
"""Performs required actions for when voting time has ended."""
|
||||||
message = self.build_highcouncil_election_announce()
|
message = self.build_highcouncil_election_announce()
|
||||||
if (council_elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
|
council_elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
|
||||||
dakiny_logger.error("Clan council elect role not found (id: %s).", settings.role_ids.clan_council_elect)
|
election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
|
||||||
return
|
|
||||||
|
nominate_msg: nextcord.Message = await election_thread.send(message, view=NominateHighCouncilView(self.guild))
|
||||||
|
|
||||||
election_name = get_election_name()
|
election_name = get_election_name()
|
||||||
|
|
||||||
with Session(engine) as session:
|
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
|
# remove vote button from vote message
|
||||||
if not info.council_vote_msg_id:
|
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == election_name)
|
||||||
dakiny_logger.error("Council vote message id not set.")
|
info = session.scalar(stmt)
|
||||||
return
|
vote_msg: nextcord.Message = await election_thread.fetch_message(info.CouncilVoteMsgId)
|
||||||
vote_msg: nextcord.Message = await election_thread.fetch_message(info.council_vote_msg_id)
|
|
||||||
await vote_msg.edit(view=None)
|
await vote_msg.edit(view=None)
|
||||||
|
|
||||||
# add the highcouncil nomination message to election information
|
# add the highcouncil nomination message to election information
|
||||||
info.high_council_nomination_msg_id = nominate_msg.id
|
info.HighCouncilNominationMsgId = nominate_msg.id
|
||||||
session.commit()
|
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
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.type == VoteType.COUNCIL))
|
.where(Votes.Type == VoteType.COUNCIL)
|
||||||
)
|
)
|
||||||
vote_count = collections.Counter(session.scalars(stmt))
|
vote_count = collections.Counter(session.scalars(stmt))
|
||||||
|
|
||||||
for user_id, votes in vote_count.items():
|
for user_id, votes in vote_count.items():
|
||||||
if user_id is None:
|
|
||||||
# Abstained votes, ignore
|
|
||||||
continue
|
|
||||||
if votes >= REQUIRED_VOTES_TO_BE_ELECTED:
|
if votes >= REQUIRED_VOTES_TO_BE_ELECTED:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == user_id))
|
.where(CouncilMembers.MemberId == user_id)
|
||||||
)
|
)
|
||||||
if (council_member := session.scalar(stmt)) is None:
|
council_member = session.scalar(stmt)
|
||||||
dakiny_logger.error(
|
council_member.Elected = True
|
||||||
"Council member (id: %s) for election %s not found.", user_id, election_name
|
|
||||||
)
|
|
||||||
return
|
|
||||||
council_member.elected = True
|
|
||||||
session.commit()
|
session.commit()
|
||||||
|
|
||||||
if (member := self.guild.get_member(user_id)) is None:
|
member: nextcord.Member = self.guild.get_member(user_id)
|
||||||
dakiny_logger.error("Member (id: %s) not found in guild.", user_id)
|
|
||||||
return
|
|
||||||
|
|
||||||
await member.add_roles(
|
await member.add_roles(
|
||||||
council_elect_role,
|
council_elect_role,
|
||||||
@@ -309,58 +274,42 @@ class Election:
|
|||||||
async def highcouncil_election_start_action(self) -> None:
|
async def highcouncil_election_start_action(self) -> None:
|
||||||
"""Performs required actions for when the highcouncil is elected."""
|
"""Performs required actions for when the highcouncil is elected."""
|
||||||
message = self.build_highcouncil_vote_start()
|
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:
|
with Session(engine) as session:
|
||||||
stmt = select(ElectionInformation).where(col(ElectionInformation.name == get_election_name()))
|
stmt = select(ElectionInformation).where(ElectionInformation.ElectionName == get_election_name())
|
||||||
if (info := session.scalar(stmt)) is None:
|
info = session.scalar(stmt)
|
||||||
dakiny_logger.error("Unknown election name %s.", get_election_name())
|
info.HighCouncilVoteMsgId = vote_msg.id
|
||||||
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()
|
session.commit()
|
||||||
|
|
||||||
# remove highcouncil nomination view
|
# remove highcouncil nomination view
|
||||||
if info.high_council_nomination_msg_id is None:
|
nomination_msg: nextcord.Message = await election_thread.fetch_message(info.HighCouncilNominationMsgId)
|
||||||
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)
|
await nomination_msg.edit(view=None)
|
||||||
|
|
||||||
async def term_start_action(self) -> None:
|
async def term_start_action(self) -> None:
|
||||||
"""Marks the election as over."""
|
"""Marks the election as over."""
|
||||||
self.election_over = True
|
self.election_over = True
|
||||||
|
|
||||||
if (elect_role := self.guild.get_role(settings.role_ids.clan_council_elect)) is None:
|
elect_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council_elect)
|
||||||
dakiny_logger.error("Cannot find Elect Role.")
|
council_role: nextcord.Role = self.guild.get_role(settings.role_ids.clan_council)
|
||||||
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()
|
election_name = get_election_name()
|
||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
# role management
|
# role management
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(CouncilMember.member_id))
|
select(CouncilMembers.MemberId)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.elected))
|
.where(CouncilMembers.Elected == true())
|
||||||
)
|
)
|
||||||
|
|
||||||
for member_id in session.scalars(stmt):
|
for member_id in session.scalars(stmt):
|
||||||
if (member := self.guild.get_member(member_id)) is None:
|
member = self.guild.get_member(member_id)
|
||||||
dakiny_logger.warning("Cannot find member (id: %s) to manage roles.", member_id)
|
|
||||||
continue
|
|
||||||
await member.remove_roles(
|
await member.remove_roles(
|
||||||
elect_role,
|
elect_role,
|
||||||
reason=f"{member.display_name} no longer needs the elect role.",
|
reason=f"{member.display_name} no longer needs the elect role.",
|
||||||
@@ -373,36 +322,35 @@ class Election:
|
|||||||
# announce high council members
|
# announce high council members
|
||||||
# ties are undefined behavior.
|
# ties are undefined behavior.
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.type == VoteType.DISCORDMANAGER))
|
.where(Votes.Type == VoteType.DISCORDMANAGER)
|
||||||
)
|
)
|
||||||
dm_votes = collections.Counter(session.scalars(stmt))
|
dm_votes = collections.Counter(session.scalars(stmt))
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.type == VoteType.CLANLEADER))
|
.where(Votes.Type == VoteType.CLANLEADER)
|
||||||
)
|
)
|
||||||
cl_votes = collections.Counter(session.scalars(stmt))
|
cl_votes = collections.Counter(session.scalars(stmt))
|
||||||
cl_votes.most_common()
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
dm_winner = dm_votes.most_common(1).pop()[0]
|
dm_winner = max(dm_votes, key=dm_votes.get)
|
||||||
stmt = (
|
stmt = (
|
||||||
update(CouncilMember)
|
update(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == dm_winner))
|
.where(CouncilMembers.MemberId == dm_winner)
|
||||||
.values(DiscordManager=True)
|
.values(DiscordManager=True)
|
||||||
)
|
)
|
||||||
session.execute(stmt)
|
session.execute(stmt)
|
||||||
except ValueError:
|
except ValueError:
|
||||||
dm_winner = "No winner"
|
dm_winner = "No winner"
|
||||||
try:
|
try:
|
||||||
cl_winner = cl_votes.most_common(1).pop()[0]
|
cl_winner = max(cl_votes, key=cl_votes.get)
|
||||||
stmt = (
|
stmt = (
|
||||||
update(CouncilMember)
|
update(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == cl_winner))
|
.where(CouncilMembers.MemberId == cl_winner)
|
||||||
.values(ClanLeader=True)
|
.values(ClanLeader=True)
|
||||||
)
|
)
|
||||||
session.execute(stmt)
|
session.execute(stmt)
|
||||||
@@ -411,21 +359,14 @@ class Election:
|
|||||||
|
|
||||||
session.commit()
|
session.commit()
|
||||||
|
|
||||||
stmt = select(ElectionInformation).where(col(ElectionInformation.name == election_name))
|
election_thread: nextcord.Thread = self.guild.get_thread(self.get_election_info().ThreadId)
|
||||||
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
|
# clear view from vote message
|
||||||
vote_msg: nextcord.Message = await election_thread.fetch_message(info.high_council_vote_msg_id)
|
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)
|
||||||
await vote_msg.edit(view=None)
|
await vote_msg.edit(view=None)
|
||||||
|
|
||||||
highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n"
|
highcouncil_msg: str = f"Congratulations for our high council members!\nClan Leader: <@{cl_winner}>\n"
|
||||||
@@ -437,23 +378,22 @@ class Election:
|
|||||||
await election_thread.edit(archived=True, locked=True)
|
await election_thread.edit(archived=True, locked=True)
|
||||||
await edit_clan_council_info(self.guild)
|
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:
|
async def election_reminder(self) -> None:
|
||||||
"""Sends a message to general chat reminder clan about election."""
|
"""Sends a message to general chat reminder clan about election."""
|
||||||
general = self.guild.get_channel(settings.channel_ids.general)
|
general: nextcord.TextChannel = self.guild.get_channel(settings.channel_ids.general)
|
||||||
|
|
||||||
if not isinstance(general, nextcord.TextChannel):
|
message = f"Reminder to <@&{settings.role_ids.clan_member}>! Clan elections are happening! Head over "
|
||||||
dakiny_logger.error(
|
message += f"to <#{self.get_election_info().ThreadId}> to cast your vote!"
|
||||||
"Recorded general channel (id: %s) is not a text channel.", settings.channel_ids.general
|
|
||||||
)
|
|
||||||
return
|
|
||||||
|
|
||||||
with Session(engine) as session:
|
await general.send(content=message)
|
||||||
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,21 +1,17 @@
|
|||||||
"""Contains various views to be used in the high council nomination process."""
|
"""Contains various views to be used in the high council nomination process."""
|
||||||
|
|
||||||
import nextcord
|
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.dakiny_settings import settings
|
||||||
from dakiny_bot.database.db import engine
|
from dakiny_bot.database.sql_data_classes import CouncilMembers, engine
|
||||||
from dakiny_bot.database.models import CouncilMember
|
|
||||||
from dakiny_bot.logger import dakiny_logger
|
|
||||||
from dakiny_bot.utils import (
|
from dakiny_bot.utils import (
|
||||||
NOMINEE_SPLITTER,
|
NOMINEE_SPLITTER,
|
||||||
decorate_cl_nomination,
|
decorate_cl_nomination,
|
||||||
decorate_dm_nomination,
|
decorate_dm_nomination,
|
||||||
get_election_name,
|
get_election_name,
|
||||||
)
|
)
|
||||||
|
from nextcord.interactions import Interaction
|
||||||
|
from sqlalchemy import select
|
||||||
|
from sqlalchemy.orm import Session
|
||||||
|
|
||||||
|
|
||||||
class NominateDMModal(nextcord.ui.Modal):
|
class NominateDMModal(nextcord.ui.Modal):
|
||||||
@@ -56,35 +52,26 @@ class NominateDMModal(nextcord.ui.Modal):
|
|||||||
interaction (Interaction): Interaction with user.
|
interaction (Interaction): Interaction with user.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
|
speech = decorate_dm_nomination(interaction.user.id, self.speech.value)
|
||||||
# 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:
|
with Session(engine) as session:
|
||||||
# user already in table
|
# user already in table
|
||||||
election_name = get_election_name()
|
election_name = get_election_name()
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == self.usr_id))
|
.where(CouncilMembers.MemberId == self.usr_id)
|
||||||
)
|
)
|
||||||
council_elect = session.scalar(stmt)
|
council_elect = session.scalar(stmt)
|
||||||
|
|
||||||
if council_elect is None:
|
if not self.usr_nominated:
|
||||||
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)
|
msg = await interaction.channel.send(speech)
|
||||||
|
|
||||||
council_elect.discord_manager_speech_id = msg.id
|
council_elect.DiscordManagerSpeechId = msg.id
|
||||||
session.commit()
|
session.commit()
|
||||||
else:
|
else:
|
||||||
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.discord_manager_speech_id)
|
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.DiscordManagerSpeechId)
|
||||||
await msg.edit(content=speech)
|
await msg.edit(content=decorate_dm_nomination(interaction.user.id, self.speech.value))
|
||||||
|
|
||||||
|
|
||||||
class NominateCLModal(nextcord.ui.Modal):
|
class NominateCLModal(nextcord.ui.Modal):
|
||||||
@@ -125,35 +112,26 @@ class NominateCLModal(nextcord.ui.Modal):
|
|||||||
interaction (Interaction): Interaction with user.
|
interaction (Interaction): Interaction with user.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
|
speech = decorate_cl_nomination(interaction.user.id, self.speech.value)
|
||||||
# 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:
|
with Session(engine) as session:
|
||||||
# user already in table
|
# user already in table
|
||||||
election_name = get_election_name()
|
election_name = get_election_name()
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == self.usr_id))
|
.where(CouncilMembers.MemberId == self.usr_id)
|
||||||
)
|
)
|
||||||
council_elect = session.scalar(stmt)
|
council_elect = session.scalar(stmt)
|
||||||
|
|
||||||
if council_elect is None:
|
if not self.usr_nominated:
|
||||||
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)
|
msg = await interaction.channel.send(speech)
|
||||||
|
|
||||||
council_elect.clan_leader_speech_id = msg.id
|
council_elect.ClanLeaderSpeechId = msg.id
|
||||||
session.commit()
|
session.commit()
|
||||||
else:
|
else:
|
||||||
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.clan_leader_speech_id)
|
msg: nextcord.Message = await interaction.channel.fetch_message(council_elect.ClanLeaderSpeechId)
|
||||||
await msg.edit(content=speech)
|
await msg.edit(content=decorate_cl_nomination(interaction.user.id, self.speech.value))
|
||||||
|
|
||||||
|
|
||||||
class NominateHighCouncilView(nextcord.ui.View):
|
class NominateHighCouncilView(nextcord.ui.View):
|
||||||
@@ -179,16 +157,8 @@ class NominateHighCouncilView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce council elect only
|
||||||
if (
|
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
||||||
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(
|
await interaction.response.send_message(
|
||||||
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
|
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
|
||||||
ephemeral=True,
|
ephemeral=True,
|
||||||
@@ -200,15 +170,16 @@ class NominateHighCouncilView(nextcord.ui.View):
|
|||||||
speech_content = ""
|
speech_content = ""
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == interaction.user.id))
|
.where(CouncilMembers.MemberId == interaction.user.id)
|
||||||
)
|
)
|
||||||
council_elect = session.scalar(stmt)
|
obj = session.scalar(stmt)
|
||||||
|
|
||||||
if council_elect is not None and council_elect.discord_manager_speech_id is not None:
|
# user needs to exist to get here anyways...
|
||||||
|
if obj.DiscordManagerSpeechId is not None:
|
||||||
usr_nominated = True
|
usr_nominated = True
|
||||||
msg = await interaction.channel.fetch_message(council_elect.discord_manager_speech_id)
|
msg = await interaction.channel.fetch_message(obj.DiscordManagerSpeechId)
|
||||||
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
||||||
|
|
||||||
await interaction.response.send_modal(NominateDMModal(interaction.user.id, usr_nominated, speech_content))
|
await interaction.response.send_modal(NominateDMModal(interaction.user.id, usr_nominated, speech_content))
|
||||||
@@ -229,16 +200,8 @@ class NominateHighCouncilView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce council elect only
|
||||||
if (
|
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
||||||
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(
|
await interaction.response.send_message(
|
||||||
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
|
f"You must have the <@&{settings.role_ids.clan_council_elect}> role to vote.",
|
||||||
ephemeral=True,
|
ephemeral=True,
|
||||||
@@ -250,15 +213,16 @@ class NominateHighCouncilView(nextcord.ui.View):
|
|||||||
speech_content = ""
|
speech_content = ""
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == interaction.user.id))
|
.where(CouncilMembers.MemberId == interaction.user.id)
|
||||||
)
|
)
|
||||||
council_elect = session.scalar(stmt)
|
obj = session.scalar(stmt)
|
||||||
|
|
||||||
if council_elect is not None and council_elect.clan_leader_speech_id is not None:
|
# user needs to exist to get here anyways...
|
||||||
|
if obj.ClanLeaderSpeechId is not None:
|
||||||
usr_nominated = True
|
usr_nominated = True
|
||||||
msg = await interaction.channel.fetch_message(council_elect.clan_leader_speech_id)
|
msg = await interaction.channel.fetch_message(obj.ClanLeaderSpeechId)
|
||||||
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
||||||
|
|
||||||
await interaction.response.send_modal(NominateCLModal(interaction.user.id, usr_nominated, speech_content))
|
await interaction.response.send_modal(NominateCLModal(interaction.user.id, usr_nominated, speech_content))
|
||||||
|
|||||||
@@ -1,16 +1,12 @@
|
|||||||
"""Contains various views to be used in the nomination process."""
|
"""Contains various views to be used in the nomination process."""
|
||||||
|
|
||||||
import nextcord
|
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 nextcord.interactions import Interaction
|
||||||
from sqlalchemy import select
|
from sqlalchemy import select
|
||||||
from sqlalchemy.orm import Session
|
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):
|
class NominateModal(nextcord.ui.Modal):
|
||||||
@@ -51,45 +47,31 @@ class NominateModal(nextcord.ui.Modal):
|
|||||||
interaction (Interaction): Interaction with user.
|
interaction (Interaction): Interaction with user.
|
||||||
|
|
||||||
"""
|
"""
|
||||||
if interaction.user is None or not isinstance(interaction.channel, nextcord.TextChannel):
|
speech = decorate_nomination(interaction.user.id, self.speech.value)
|
||||||
# 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()
|
election_name = get_election_name()
|
||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
if not self.usr_nominated:
|
if not self.usr_nominated:
|
||||||
msg = await interaction.channel.send(speech)
|
msg = await interaction.channel.send(speech)
|
||||||
|
|
||||||
council_member = CouncilMember(
|
council_member = CouncilMembers(
|
||||||
election_name=election_name,
|
ElectionName=election_name,
|
||||||
member_id=interaction.user.id,
|
MemberId=interaction.user.id,
|
||||||
speech_msg_id=msg.id,
|
SpeechMsgId=msg.id,
|
||||||
)
|
)
|
||||||
session.add(council_member)
|
session.add(council_member)
|
||||||
session.commit()
|
session.commit()
|
||||||
else:
|
else:
|
||||||
# user already nominated, just need to update the election speech
|
# user already nominated, just need to update the election speech
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == self.usr_id))
|
.where(CouncilMembers.MemberId == self.usr_id)
|
||||||
)
|
)
|
||||||
council_elect = session.scalar(stmt)
|
obj = session.scalar(stmt)
|
||||||
|
|
||||||
if council_elect is None:
|
msg: nextcord.Message = await interaction.channel.fetch_message(obj.SpeechMsgId)
|
||||||
dakiny_logger.error("Did not find member (id: %s) in election %s.", self.usr_id, election_name)
|
await msg.edit(content=decorate_nomination(interaction.user.id, self.speech.value))
|
||||||
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):
|
class NominateView(nextcord.ui.View):
|
||||||
@@ -115,16 +97,8 @@ class NominateView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce members only
|
||||||
if (
|
if self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
|
||||||
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(
|
await interaction.response.send_message(
|
||||||
f"You must have the <@&{settings.role_ids.clan_member}> role to vote.",
|
f"You must have the <@&{settings.role_ids.clan_member}> role to vote.",
|
||||||
ephemeral=True,
|
ephemeral=True,
|
||||||
@@ -136,14 +110,14 @@ class NominateView(nextcord.ui.View):
|
|||||||
speech_content = ""
|
speech_content = ""
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(CouncilMember)
|
select(CouncilMembers)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.where(col(CouncilMember.member_id == interaction.user.id))
|
.where(CouncilMembers.MemberId == interaction.user.id)
|
||||||
)
|
)
|
||||||
obj = session.scalar(stmt)
|
obj = session.scalar(stmt)
|
||||||
if obj is not None and obj.speech_msg_id is not None:
|
if obj is not None:
|
||||||
usr_nominated = True
|
usr_nominated = True
|
||||||
msg = await interaction.channel.fetch_message(obj.speech_msg_id)
|
msg = await interaction.channel.fetch_message(obj.SpeechMsgId)
|
||||||
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
speech_content = msg.content.split(NOMINEE_SPLITTER)[1]
|
||||||
|
|
||||||
await interaction.response.send_modal(NominateModal(interaction.user.id, usr_nominated, speech_content))
|
await interaction.response.send_modal(NominateModal(interaction.user.id, usr_nominated, speech_content))
|
||||||
|
|||||||
@@ -4,3 +4,7 @@ extend = "../../pyproject.toml"
|
|||||||
|
|
||||||
# Exclude a variety of commonly ignored directories.
|
# Exclude a variety of commonly ignored directories.
|
||||||
exclude = ["gitea_webhook.py"]
|
exclude = ["gitea_webhook.py"]
|
||||||
|
|
||||||
|
[tool.ruff.lint]
|
||||||
|
# ignore implicit-namespace-package in this directory.
|
||||||
|
ignore = ["INP001"]
|
||||||
|
|||||||
@@ -1,17 +1,18 @@
|
|||||||
"""Contains various views to be used while voting for high council. Controls the logic when performing the vote."""
|
"""Contains various views to be used while voting for high council. Controls the logic when performing the vote."""
|
||||||
|
|
||||||
import nextcord
|
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 import SelectOption
|
||||||
from nextcord.interactions import Interaction
|
from nextcord.interactions import Interaction
|
||||||
from sqlalchemy import exc, select, update
|
from sqlalchemy import exc, select, update
|
||||||
from sqlalchemy.orm import Session
|
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):
|
class VoteSelect(nextcord.ui.StringSelect):
|
||||||
@@ -36,32 +37,20 @@ class VoteSelect(nextcord.ui.StringSelect):
|
|||||||
Args:
|
Args:
|
||||||
interaction (Interaction): Interaction with this callback.
|
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()
|
election_name: str = get_election_name()
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = select(col(ElectionInformation.high_council_vote_msg_id)).where(
|
stmt = select(ElectionInformation.HighCouncilVoteMsgId).where(
|
||||||
col(ElectionInformation.name == election_name)
|
ElectionInformation.ElectionName == election_name
|
||||||
)
|
)
|
||||||
if (vote_msg_id := session.scalar(stmt)) is None:
|
vote_msg_id = session.scalar(stmt)
|
||||||
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])
|
voted_for = None if self.values[0] == "Abstain" else int(self.values[0])
|
||||||
|
|
||||||
new_vote = Vote(
|
new_vote = Votes(
|
||||||
election_name=election_name,
|
ElectionName=election_name,
|
||||||
voter_id=interaction.user.id,
|
VoterId=interaction.user.id,
|
||||||
voted_for_id=voted_for,
|
VotedForId=voted_for,
|
||||||
type=self.election_type,
|
Type=self.election_type,
|
||||||
)
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -72,10 +61,10 @@ class VoteSelect(nextcord.ui.StringSelect):
|
|||||||
# user already has a vote in the election and election type
|
# user already has a vote in the election and election type
|
||||||
session.rollback()
|
session.rollback()
|
||||||
stmt = (
|
stmt = (
|
||||||
update(Vote)
|
update(Votes)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.voter_id == interaction.user.id))
|
.where(Votes.VoterId == interaction.user.id)
|
||||||
.where(col(Vote.type == self.election_type))
|
.where(Votes.Type == self.election_type)
|
||||||
.values(VotedForId=voted_for)
|
.values(VotedForId=voted_for)
|
||||||
)
|
)
|
||||||
session.execute(stmt)
|
session.execute(stmt)
|
||||||
@@ -114,10 +103,6 @@ class VoteHighcouncilView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce council elect only
|
||||||
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
||||||
await interaction.response.send_message(
|
await interaction.response.send_message(
|
||||||
@@ -130,29 +115,32 @@ class VoteHighcouncilView(nextcord.ui.View):
|
|||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.voter_id == interaction.user.id))
|
.where(Votes.VoterId == interaction.user.id)
|
||||||
.where(col(Vote.type == VoteType.CLANLEADER))
|
.where(Votes.Type == VoteType.CLANLEADER)
|
||||||
)
|
)
|
||||||
voted_for = session.scalar(stmt)
|
voted_for = session.scalar(stmt)
|
||||||
|
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(CouncilMember.member_id))
|
select(CouncilMembers.MemberId)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.filter(col(CouncilMember.clan_leader_speech_id is not None))
|
.filter(CouncilMembers.ClanLeaderSpeechId.is_not(None))
|
||||||
)
|
)
|
||||||
|
|
||||||
options = [
|
options = [
|
||||||
SelectOption(
|
SelectOption(
|
||||||
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
|
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),
|
value=str(nominee),
|
||||||
)
|
)
|
||||||
for nominee in sorted(
|
for nominee in sorted(
|
||||||
session.scalars(stmt),
|
session.scalars(stmt),
|
||||||
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
|
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
|
||||||
)
|
)
|
||||||
if (member := self.guild.get_member(nominee)) is not None
|
|
||||||
]
|
]
|
||||||
options.append(
|
options.append(
|
||||||
SelectOption(
|
SelectOption(
|
||||||
@@ -187,10 +175,6 @@ class VoteHighcouncilView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce council elect only
|
||||||
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
if self.guild.get_role(settings.role_ids.clan_council_elect) not in interaction.user.roles:
|
||||||
await interaction.response.send_message(
|
await interaction.response.send_message(
|
||||||
@@ -203,29 +187,32 @@ class VoteHighcouncilView(nextcord.ui.View):
|
|||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.voter_id == interaction.user.id))
|
.where(Votes.VoterId == interaction.user.id)
|
||||||
.where(col(Vote.type == VoteType.DISCORDMANAGER))
|
.where(Votes.Type == VoteType.DISCORDMANAGER)
|
||||||
)
|
)
|
||||||
voted_for = session.scalar(stmt)
|
voted_for = session.scalar(stmt)
|
||||||
|
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(CouncilMember.member_id))
|
select(CouncilMembers.MemberId)
|
||||||
.where(col(CouncilMember.election_name == election_name))
|
.where(CouncilMembers.ElectionName == election_name)
|
||||||
.filter(col(CouncilMember.discord_manager_speech_id is not None))
|
.filter(CouncilMembers.DiscordManagerSpeechId.is_not(None))
|
||||||
)
|
)
|
||||||
|
|
||||||
options = [
|
options = [
|
||||||
SelectOption(
|
SelectOption(
|
||||||
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
|
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),
|
value=str(nominee),
|
||||||
)
|
)
|
||||||
for nominee in sorted(
|
for nominee in sorted(
|
||||||
session.scalars(stmt),
|
session.scalars(stmt),
|
||||||
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
|
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
|
||||||
)
|
)
|
||||||
if (member := self.guild.get_member(nominee)) is not None
|
|
||||||
]
|
]
|
||||||
options.append(
|
options.append(
|
||||||
SelectOption(
|
SelectOption(
|
||||||
|
|||||||
@@ -1,17 +1,18 @@
|
|||||||
"""Contains various views to be used while voting. Controls the logic when performing the vote."""
|
"""Contains various views to be used while voting. Controls the logic when performing the vote."""
|
||||||
|
|
||||||
import nextcord
|
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 import SelectOption
|
||||||
from nextcord.interactions import Interaction
|
from nextcord.interactions import Interaction
|
||||||
from sqlalchemy import exc, select, update
|
from sqlalchemy import exc, select, update
|
||||||
from sqlalchemy.orm import Session
|
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):
|
class VoteSelect(nextcord.ui.StringSelect):
|
||||||
@@ -35,32 +36,18 @@ class VoteSelect(nextcord.ui.StringSelect):
|
|||||||
Args:
|
Args:
|
||||||
interaction (Interaction): Interaction with this callback.
|
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()
|
election_name = get_election_name()
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = select(col(ElectionInformation.council_vote_msg_id)).where(
|
stmt = select(ElectionInformation.CouncilVoteMsgId).where(ElectionInformation.ElectionName == election_name)
|
||||||
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 Vote Message ID in database.")
|
|
||||||
return
|
|
||||||
|
|
||||||
voted_for = None if self.values[0] == "Abstain" else int(self.values[0])
|
voted_for = None if self.values[0] == "Abstain" else int(self.values[0])
|
||||||
|
|
||||||
new_vote = Vote(
|
new_vote = Votes(
|
||||||
election_name=election_name,
|
ElectionName=election_name,
|
||||||
voter_id=interaction.user.id,
|
VoterId=interaction.user.id,
|
||||||
voted_for_id=voted_for,
|
VotedForId=voted_for,
|
||||||
type=VoteType.COUNCIL,
|
Type=VoteType.COUNCIL,
|
||||||
)
|
)
|
||||||
|
|
||||||
try:
|
try:
|
||||||
@@ -70,10 +57,10 @@ class VoteSelect(nextcord.ui.StringSelect):
|
|||||||
except exc.IntegrityError:
|
except exc.IntegrityError:
|
||||||
session.rollback()
|
session.rollback()
|
||||||
stmt = (
|
stmt = (
|
||||||
update(Vote)
|
update(Votes)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.voter_id == interaction.user.id))
|
.where(Votes.VoterId == interaction.user.id)
|
||||||
.where(col(Vote.type == VoteType.COUNCIL))
|
.where(Votes.Type == VoteType.COUNCIL)
|
||||||
.values(VotedForId=voted_for)
|
.values(VotedForId=voted_for)
|
||||||
)
|
)
|
||||||
session.execute(stmt)
|
session.execute(stmt)
|
||||||
@@ -107,10 +94,6 @@ class VoteView(nextcord.ui.View):
|
|||||||
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
button (nextcord.ui.Button): Unused. idk why its needed in the decorator
|
||||||
interaction (nextcord.Interaction): Interaction of user who pressed the button
|
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
|
# enforce members only
|
||||||
if self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
|
if self.guild.get_role(settings.role_ids.clan_member) not in interaction.user.roles:
|
||||||
await interaction.response.send_message(
|
await interaction.response.send_message(
|
||||||
@@ -123,25 +106,28 @@ class VoteView(nextcord.ui.View):
|
|||||||
|
|
||||||
with Session(engine) as session:
|
with Session(engine) as session:
|
||||||
stmt = (
|
stmt = (
|
||||||
select(col(Vote.voted_for_id))
|
select(Votes.VotedForId)
|
||||||
.where(col(Vote.election_name == election_name))
|
.where(Votes.ElectionName == election_name)
|
||||||
.where(col(Vote.voter_id == interaction.user.id))
|
.where(Votes.VoterId == interaction.user.id)
|
||||||
.where(col(Vote.type == VoteType.COUNCIL))
|
.where(Votes.Type == VoteType.COUNCIL)
|
||||||
)
|
)
|
||||||
voted_for = session.scalar(stmt)
|
voted_for = session.scalar(stmt)
|
||||||
|
|
||||||
stmt = select(col(CouncilMember.member_id)).where(col(CouncilMember.election_name == election_name))
|
stmt = select(CouncilMembers.MemberId).where(CouncilMembers.ElectionName == election_name)
|
||||||
|
|
||||||
options = [
|
options = [
|
||||||
SelectOption(
|
SelectOption(
|
||||||
label=(f"{member.display_name} ☑️" if voted_for == nominee else f"{member.display_name}"),
|
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),
|
value=str(nominee),
|
||||||
)
|
)
|
||||||
for nominee in sorted(
|
for nominee in sorted(
|
||||||
session.scalars(stmt),
|
session.scalars(stmt),
|
||||||
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(), # pyright: ignore[reportOptionalMemberAccess]
|
key=lambda nominee: self.guild.get_member(nominee).display_name.lower(),
|
||||||
)
|
)
|
||||||
if (member := self.guild.get_member(nominee)) is not None
|
|
||||||
]
|
]
|
||||||
options.append(
|
options.append(
|
||||||
SelectOption(
|
SelectOption(
|
||||||
|
|||||||
@@ -4,7 +4,6 @@ import nextcord
|
|||||||
from nextcord.ext import commands
|
from nextcord.ext import commands
|
||||||
|
|
||||||
from dakiny_bot.app import birthday_list
|
from dakiny_bot.app import birthday_list
|
||||||
from dakiny_bot.dakiny_settings import settings
|
|
||||||
|
|
||||||
|
|
||||||
class SubmissionListener(commands.Cog, name="Message Listener"):
|
class SubmissionListener(commands.Cog, name="Message Listener"):
|
||||||
@@ -17,15 +16,7 @@ class SubmissionListener(commands.Cog, name="Message Listener"):
|
|||||||
async def on_message(self, message: nextcord.Message) -> None:
|
async def on_message(self, message: nextcord.Message) -> None:
|
||||||
"""Listener for message creation. Adds score and attachments to submission."""
|
"""Listener for message creation. Adds score and attachments to submission."""
|
||||||
# don't want to process any messages from bot.
|
# don't want to process any messages from bot.
|
||||||
if message.author.id == self.bot.user.id: # pyright: ignore[reportOptionalMemberAccess]
|
if message.author.id == self.bot.user.id:
|
||||||
return
|
|
||||||
|
|
||||||
# Ban all members who message in this channel. Everyone in guild is a member, so shouldn't miss a ban
|
|
||||||
# due to message.author not being a nextcord.Member object.
|
|
||||||
if message.channel.id == settings.channel_ids.ban and isinstance(message.author, nextcord.Member):
|
|
||||||
await message.author.ban(
|
|
||||||
delete_message_seconds=30 * 60, reason="Probable spam. The ban channel was messaged."
|
|
||||||
)
|
|
||||||
return
|
return
|
||||||
|
|
||||||
await birthday_action(message)
|
await birthday_action(message)
|
||||||
@@ -40,7 +31,7 @@ def setup(bot: commands.Bot) -> None:
|
|||||||
|
|
||||||
async def birthday_action(message: nextcord.Message) -> None:
|
async def birthday_action(message: nextcord.Message) -> None:
|
||||||
"""
|
"""
|
||||||
Checks a message's author and mentions to see if they are in the birthday list.
|
Checks a message's auther and mentions to see if they are in the birthday list.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
message (nextcord.Message): The message to check for a birthday person.
|
message (nextcord.Message): The message to check for a birthday person.
|
||||||
|
|||||||
@@ -17,7 +17,6 @@ class ChannelIds:
|
|||||||
event_post: int
|
event_post: int
|
||||||
general: int
|
general: int
|
||||||
voting: int
|
voting: int
|
||||||
ban: int
|
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
|
|||||||
Reference in New Issue
Block a user