imcompiling
abb50d71b4
build_develop / build (push) Successful in 36s
Narrowing down where the issue is. Active cycle is being obtained, but nothing was happening with it. No more debug messages either.
136 lines
4.9 KiB
Python
136 lines
4.9 KiB
Python
"""Defines the daily loop that runs the Dakiny bot functions."""
|
|
|
|
import random
|
|
from datetime import UTC, datetime, time
|
|
from typing import TYPE_CHECKING
|
|
|
|
from nextcord.ext import commands, tasks
|
|
from pytz import timezone
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from dakiny_bot.app import birthday_list
|
|
from dakiny_bot.classes.election import Election
|
|
from dakiny_bot.DakinySettings import settings
|
|
from dakiny_bot.database.sql_data_classes import UserBirthday, engine
|
|
from dakiny_bot.logger import dakiny_logger
|
|
from dakiny_bot.utils import get_active_election_cycle
|
|
|
|
if TYPE_CHECKING:
|
|
import nextcord
|
|
|
|
KANDY = 239547993106284544
|
|
|
|
IGNORE_CAKEDAY = [266378407682506754]
|
|
|
|
|
|
class DailyLoop(commands.Cog, name="Daily Loop"):
|
|
"""Class for performing daily functions."""
|
|
|
|
def __init__(self, bot: commands.Bot) -> None:
|
|
self.bot = bot
|
|
self.current_election = None
|
|
self.update_birthdays()
|
|
|
|
# pylint doesn't recognize that daily_loop is subclassed from tasks.loop.
|
|
self.daily_loop.start() # pylint: disable=no-member
|
|
|
|
@tasks.loop(time=time(hour=0, tzinfo=timezone("UTC")))
|
|
async def daily_loop(self) -> None:
|
|
"""
|
|
Runs daily.
|
|
|
|
Checks if the current election is None. If so, checks if there is an active election. If there is
|
|
an active election, runs the check_date() function for the election to perform actions. Finally checks if the
|
|
election is over to delete the current election object.
|
|
"""
|
|
dakiny_logger.debug("Daily loop. Current election: %s.", self.current_election)
|
|
if self.current_election is None:
|
|
self.current_election = self.get_election()
|
|
dakiny_logger.debug("Current election: %s.", self.current_election)
|
|
|
|
if self.current_election is not None:
|
|
await self.current_election.check_date()
|
|
if self.current_election.election_over:
|
|
self.current_election = None
|
|
|
|
self.update_birthdays()
|
|
|
|
if KANDY in birthday_list:
|
|
# inform kandy that it is her "birthday"
|
|
kandy: nextcord.User = await self.bot.fetch_user(KANDY)
|
|
await kandy.send('Happy "Birthday"!')
|
|
|
|
await self.cakedays()
|
|
|
|
@daily_loop.before_loop
|
|
async def before_loop(self) -> None:
|
|
"""Actions run before the loop starts. Want to wait for bot to be ready and get the current cycle."""
|
|
await self.bot.wait_until_ready()
|
|
|
|
# Bot needs to know guilds to get the current cycle.
|
|
self.current_election = self.get_election()
|
|
|
|
def get_election(self) -> Election:
|
|
"""
|
|
Creates an object for the current election cycle if one is active.
|
|
|
|
An election is considered active between the election start and the term start.
|
|
|
|
Returns:
|
|
Election: Current election cycle, None if there is no election happening.
|
|
"""
|
|
cycle = get_active_election_cycle()
|
|
dakiny_logger.debug("Current cycle: %s.", cycle)
|
|
|
|
# prevent exception from using cycle as an Int in an array
|
|
if cycle is None:
|
|
dakiny_logger.info("Cycle was None. Returning without an election.")
|
|
return None
|
|
|
|
election = Election(cycle, self.bot.get_guild(settings.params.guild_id))
|
|
|
|
dakiny_logger.debug("Election active: %s", election.active)
|
|
|
|
return election if election.active else None
|
|
|
|
def update_birthdays(self) -> None:
|
|
"""Goes through database to collect all birthdays that are today. Adds user's id to birthday list."""
|
|
birthday_list.clear()
|
|
|
|
# kandy's 1% chance
|
|
if random.randint(0, 99) == 0: # noqa: S311 isn't cryptography related.
|
|
birthday_list.append(KANDY)
|
|
|
|
today = datetime.now(tz=UTC).date()
|
|
birthday = f"{today.month}/{today.day}"
|
|
|
|
stmt = select(UserBirthday.UserId).where(UserBirthday.Birthday == birthday)
|
|
with Session(engine) as session:
|
|
to_add = session.scalars(stmt)
|
|
birthday_list.extend(to_add.all())
|
|
|
|
async def cakedays(self) -> None:
|
|
"""
|
|
Loops through the guild's humans checking for joined_at dates that match today's date.
|
|
|
|
Sends a message saying happy cakeday.
|
|
"""
|
|
today = datetime.now(tz=UTC).date()
|
|
guild: nextcord.Guild = self.bot.get_guild(settings.params.guild_id)
|
|
announce_channel: nextcord.TextChannel = guild.get_channel(settings.channel_ids.general)
|
|
|
|
cakedays: list[nextcord.Member] = [
|
|
human
|
|
for human in guild.humans
|
|
if human.joined_at.month == today.month
|
|
and human.joined_at.day == today.day
|
|
and human.id not in IGNORE_CAKEDAY
|
|
]
|
|
|
|
for member in cakedays:
|
|
years = today.year - member.joined_at.year
|
|
year_str = "years" if years > 1 else "year"
|
|
msg = f"Happy Cake day! <@{member.id}> joined the discord {years} {year_str} ago!"
|
|
await announce_channel.send(msg)
|