48 lines
1.4 KiB
Python
48 lines
1.4 KiB
Python
"""JARVIS Lockdown background task."""
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from beanie.operators import LTE
|
|
from jarvis_core.db.models import Lockdown
|
|
from interactions import Client
|
|
from interactions.models.discord.enums import Permissions
|
|
from interactions.models.discord.role import Role
|
|
|
|
from jarvis_tasks.util import runat
|
|
|
|
queue = []
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def _lift(role: Role, lock: Lockdown) -> None:
|
|
logger.debug(f"Lifting lockdown {lock.id}")
|
|
original_perms = Permissions(lock.original_perms)
|
|
await role.edit(permissions=original_perms)
|
|
lock.active = False
|
|
await lock.save()
|
|
queue.remove(lock.id)
|
|
|
|
|
|
async def lift(bot: Client) -> None:
|
|
"""
|
|
Lift a lockdown.
|
|
|
|
Args:
|
|
bot: Bot instance
|
|
"""
|
|
logger.debug("Starting Task-lift")
|
|
while True:
|
|
max_ts = datetime.now(tz=timezone.utc) + timedelta(seconds=55)
|
|
locks = Lockdown.find(Lockdown.active == True, LTE(Lockdown.created_at, max_ts))
|
|
async for lock in locks:
|
|
if lock.id in queue:
|
|
continue
|
|
guild = await bot.fetch_guild(lock.guild)
|
|
role = await guild.fetch_role(guild.id)
|
|
coro = _lift(role, lock)
|
|
when = lock.created_at + timedelta(minutes=lock.duration)
|
|
asyncio.create_task(runat(when, coro, logger))
|
|
queue.append(lock.id)
|
|
|
|
await asyncio.sleep(delay=60)
|