imcompiling/dakiny-bot
1
0
Fork 0
Code Issues 22 Pull Requests 9 Actions Packages Projects Activity

Turned on all ruff rules. Formatted all files.

Browse Source
This commit is contained in:
imcompiling
2024-05-22 15:06:39 -04:00
parent 3ac164ab65
commit 796b052bfc
18 changed files with 161 additions and 469 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils.py
+20 -75
View File
@@ -149,9 +149,7 @@ def build_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str:
election_name = get_election_name(config)
with Session(engine) as session:
stmt = select(CouncilMembers).where(
CouncilMembers.ElectionCycle == election_name
)
stmt = select(CouncilMembers).where(CouncilMembers.ElectionCycle == election_name)
for nominee in sorted(
session.scalars(stmt),
@@ -159,11 +157,7 @@ def build_votes_str(config: DakinyConfig, guild: nextcord.Guild) -> str:
):
votes[nominee.MemberId] = 0
stmt = (
select(Votes)
.where(Votes.ElectionCycle == election_name)
.where(Votes.Type == VoteType.COUNCIL)
)
stmt = select(Votes).where(Votes.ElectionCycle == election_name).where(Votes.Type == VoteType.COUNCIL)
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
@@ -205,11 +199,7 @@ def build_highcouncil_votes_str(config: DakinyConfig, guild: nextcord.Guild) ->
):
cl_votes[nominee.MemberId] = 0
stmt = (
select(Votes)
.where(Votes.ElectionCycle == election_name)
.where(Votes.Type == VoteType.CLANLEADER)
)
stmt = select(Votes).where(Votes.ElectionCycle == election_name).where(Votes.Type == VoteType.CLANLEADER)
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
@@ -227,11 +217,7 @@ def build_highcouncil_votes_str(config: DakinyConfig, guild: nextcord.Guild) ->
):
dm_votes[nominee.MemberId] = 0
stmt = (
select(Votes)
.where(Votes.ElectionCycle == election_name)
.where(Votes.Type == VoteType.DISCORDMANAGER)
)
stmt = select(Votes).where(Votes.ElectionCycle == election_name).where(Votes.Type == VoteType.DISCORDMANAGER)
for vote in session.scalars(stmt):
if vote.VotedForId is not None:
@@ -318,9 +304,7 @@ def decorate_dm_nomination(user_id: int, speech: str) -> str:
return msg
def record_event_points(
config: DakinyConfig, users: Sequence[str], is_member: bool, add_points: int
):
def record_event_points(config: DakinyConfig, users: Sequence[str], is_member: bool, add_points: int):
"""
Adds event points to all users listed. Checks if they don't exist in order to make a new entry, otherwise
adds points to the existing value. Can be used for hosting or participating.
@@ -343,9 +327,7 @@ def record_event_points(
points = session.scalar(stmt)
if points is None:
add_points = get_allowed_points(
session, config.params.event_points_weekly_limit, user, add_points
)
add_points = get_allowed_points(session, config.params.event_points_weekly_limit, user, add_points)
new_points = ParticipationPoints(
Cycle=quarter_name,
UserId=user,
@@ -354,17 +336,13 @@ def record_event_points(
)
session.add(new_points)
else:
add_points = get_allowed_points(
session, config.params.event_points_weekly_limit, user, add_points
)
add_points = get_allowed_points(session, config.params.event_points_weekly_limit, user, add_points)
points.Points += add_points
session.commit()
def get_allowed_points(
session: Session, points_limit: int, user: str, points: int
) -> int:
def get_allowed_points(session: Session, points_limit: int, user: str, points: int) -> int:
"""
Gets the maximum number of points a user can be given. Searches how many points they have been given
this week and subtracts that from the weekly limit. If under, returns up to "points" number of points (will not
@@ -380,14 +358,8 @@ def get_allowed_points(
int: how many points are allowed to be given to the user.
"""
# check for limit
this_week = (
f"{datetime.today().isocalendar().year}-{datetime.today().isocalendar().week}"
)
stmt = (
select(EventPointsLimit)
.where(EventPointsLimit.UserId == user)
.where(EventPointsLimit.Week == this_week)
)
this_week = f"{datetime.today().isocalendar().year}-{datetime.today().isocalendar().week}"
stmt = select(EventPointsLimit).where(EventPointsLimit.UserId == user).where(EventPointsLimit.Week == this_week)
# if under limit, can add up to however many points would make the current limit equal to the
# allowed limit.
@@ -424,11 +396,7 @@ async def update_clan_messages(guild: nextcord.Guild, channel_updates: Sequence[
channel_updates: (Sequence[str]): a sequence of the channels to update.
"""
dirs = [
dir_name
for dir_name in channel_updates
if os.path.isdir(f"{MESSAGES_DIR}/{dir_name}")
]
dirs = [dir_name for dir_name in channel_updates if os.path.isdir(f"{MESSAGES_DIR}/{dir_name}")]
with Session(engine) as session:
for dir_name in dirs:
@@ -443,14 +411,10 @@ async def update_clan_messages(guild: nextcord.Guild, channel_updates: Sequence[
message_name = f"{dir_name}/{file_name}"
file_path = f"{dir_path}/{file_name}"
with open(file_path, "r", encoding="utf8") as reader:
await edit_clan_message(
session, channel, message_name, reader.read(2000)
)
await edit_clan_message(session, channel, message_name, reader.read(2000))
async def get_clan_message_channel(
session: Session, channel_name: str, guild: nextcord.Guild
) -> nextcord.TextChannel:
async def get_clan_message_channel(session: Session, channel_name: str, guild: nextcord.Guild) -> nextcord.TextChannel:
"""
Used to obtain a channel from the guild object. Searches database for an id to get the channel. If that is not
found, then searches through all channels in the guild to find one with a matching name. If that also doesn't
@@ -472,10 +436,7 @@ async def get_clan_message_channel(
# channel is not tracked. see if it exists
channels = await guild.fetch_channels()
for channel in channels:
if (
channel.name == channel_name
and channel.__class__ is nextcord.TextChannel
):
if channel.name == channel_name and channel.__class__ is nextcord.TextChannel:
ret_channel = channel
if ret_channel is None:
@@ -499,11 +460,7 @@ async def get_clan_message_channel(
channel_name,
reason="Creating channel as required for by design of Dakiny-Messages repo.",
)
stmt = (
update(Messages)
.where(Messages.Path == channel_name)
.values(Id=ret_channel.id)
)
stmt = update(Messages).where(Messages.Path == channel_name).values(Id=ret_channel.id)
session.execute(stmt)
session.commit()
return ret_channel
@@ -540,9 +497,7 @@ async def edit_clan_message(
await msg.edit(message_content)
except nextcord.NotFound:
msg = await channel.send(message_content)
stmt = (
update(Messages).where(Messages.Path == message_name).values(Id=msg.id)
)
stmt = update(Messages).where(Messages.Path == message_name).values(Id=msg.id)
session.execute(stmt)
session.commit()
@@ -555,9 +510,7 @@ async def edit_clan_council_info(guild: nextcord.Guild, config: DakinyConfig):
guild (nextcord.Guild): Guild object that to use to get channel.
config (DakinyConfig): Configs
"""
clan_info_channel: nextcord.TextChannel = await guild.fetch_channel(
config.params.clan_info_channel_id
)
clan_info_channel: nextcord.TextChannel = await guild.fetch_channel(config.params.clan_info_channel_id)
election = get_election_name(config)
with Session(engine) as session:
@@ -579,17 +532,11 @@ async def edit_clan_council_info(guild: nextcord.Guild, config: DakinyConfig):
.where(CouncilMembers.Elected == true())
)
council_members_strs = [
f"<@{id}>"
for id in session.scalars(stmt)
if id not in (discord_manager, clan_leader)
]
council_members_strs = [f"<@{id}>" for id in session.scalars(stmt) if id not in (discord_manager, clan_leader)]
quarter = get_quarter_name(config)
with open(
f"{TEMPLATES_DIR}/council_members.txt", mode="r", encoding="utf8"
) as file:
with open(f"{TEMPLATES_DIR}/council_members.txt", mode="r", encoding="utf8") as file:
message = file.read()
message = message.format(
quarter=quarter,
@@ -598,6 +545,4 @@ async def edit_clan_council_info(guild: nextcord.Guild, config: DakinyConfig):
discord_manager=discord_manager,
)
await edit_clan_message(
session, clan_info_channel, "council_members.txt", message
)
await edit_clan_message(session, clan_info_channel, "council_members.txt", message)