54 lines
1.7 KiB
Python
54 lines
1.7 KiB
Python
"""JARVIS ban tasks."""
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from logging import Logger
|
|
|
|
from dis_snek import Snake
|
|
from dis_snek.client.errors import NotFound
|
|
from dis_snek.models.discord.guild import Guild
|
|
from dis_snek.models.discord.user import User
|
|
from jarvis_core.db import q
|
|
from jarvis_core.db.models import Ban, Unban
|
|
|
|
from jarvis_tasks.util import runat
|
|
|
|
|
|
async def _unban(bot: int, guild: Guild, user: User, ban: Ban, logger: Logger) -> None:
|
|
if guild and user:
|
|
logger.debug(f"Unbanning user {user.id} from guild {guild.id}")
|
|
try:
|
|
await guild.unban(user=user, reason="JARVIS tempban expired")
|
|
except NotFound:
|
|
logger.debug(f"User {user.id} not banned from guild {guild.id}")
|
|
ban.active = False
|
|
await ban.commit()
|
|
await Unban(
|
|
user=user.id,
|
|
guild=guild.id,
|
|
username=user.username,
|
|
discrim=user.discriminator,
|
|
admin=bot,
|
|
reason="Ban expired",
|
|
).commit()
|
|
|
|
|
|
async def unban(bot: Snake, logger: Logger) -> None:
|
|
"""
|
|
Unban users when ban expires.
|
|
|
|
Args:
|
|
bot: Snake instance
|
|
logger: Global logger
|
|
"""
|
|
while True:
|
|
max_ts = datetime.utcnow() + timedelta(minutes=9)
|
|
bans = Ban.find(q(type="temp", active=True, duration__lte=max_ts))
|
|
async for ban in bans:
|
|
guild = await bot.fetch_guild(ban.guild)
|
|
user = await bot.fetch_user(ban.user)
|
|
coro = _unban(bot.user.id, guild, user, ban, logger)
|
|
when = ban.created_at + timedelta(hours=ban.duration)
|
|
asyncio.create_task(runat(when, coro, logger))
|
|
|
|
# Check ever 10 minutes
|
|
await asyncio.sleep(600)
|