47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
"""JARVIS warnings tasks."""
|
|
import asyncio
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from beanie.operators import LTE, NotIn
|
|
from jarvis_core.db.models import Warning
|
|
from interactions import Client
|
|
|
|
from jarvis_tasks.util import runat
|
|
|
|
queue = []
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def _unwarn(warn: Warning) -> None:
|
|
logger.debug(f"Deactivating warning {warn.id}")
|
|
warn.active = False
|
|
await warn.save()
|
|
queue.remove(warn.id)
|
|
|
|
|
|
async def unwarn(bot: Client) -> None:
|
|
"""
|
|
Deactivate warnings when they expire.
|
|
|
|
Args:
|
|
bot: Client instance
|
|
"""
|
|
logger.debug("Starting Task-unwarn")
|
|
while True:
|
|
max_ts = datetime.now(tz=timezone.utc) + timedelta(minutes=55)
|
|
warns = Warning.find(
|
|
Warning.active == True,
|
|
LTE(Warning.expires_at, max_ts),
|
|
NotIn(Warning.id, queue),
|
|
)
|
|
async for warn in warns:
|
|
if warn.id in queue:
|
|
logger.warn("Warning found despite filter")
|
|
continue
|
|
coro = _unwarn(warn)
|
|
when = warn.expires_at
|
|
asyncio.create_task(runat(when, coro, logger))
|
|
queue.append(warn.id)
|
|
# Check every hour
|
|
await asyncio.sleep(3600)
|