101 lines
3.1 KiB
Python
101 lines
3.1 KiB
Python
"""J.A.R.V.I.S. LockdownCog."""
|
|
from contextlib import suppress
|
|
from datetime import datetime
|
|
|
|
from discord.ext import commands
|
|
from discord_slash import cog_ext
|
|
from discord_slash import SlashContext
|
|
from discord_slash.utils.manage_commands import create_option
|
|
|
|
from jarvis.db.models import Lock
|
|
from jarvis.utils.cachecog import CacheCog
|
|
from jarvis.utils.permissions import admin_or_permissions
|
|
|
|
|
|
class LockdownCog(CacheCog):
|
|
"""J.A.R.V.I.S. LockdownCog."""
|
|
|
|
def __init__(self, bot: commands.Bot):
|
|
super().__init__(bot)
|
|
|
|
@cog_ext.cog_subcommand(
|
|
base="lockdown",
|
|
name="start",
|
|
description="Locks a server",
|
|
options=[
|
|
create_option(
|
|
name="reason",
|
|
description="Lockdown Reason",
|
|
option_type=3,
|
|
required=True,
|
|
),
|
|
create_option(
|
|
name="duration",
|
|
description="Lockdown duration in minutes (default 10)",
|
|
option_type=4,
|
|
required=False,
|
|
),
|
|
],
|
|
)
|
|
@admin_or_permissions(manage_channels=True)
|
|
async def _lockdown_start(
|
|
self,
|
|
ctx: SlashContext,
|
|
reason: str,
|
|
duration: int = 10,
|
|
) -> None:
|
|
await ctx.defer(hidden=True)
|
|
if duration <= 0:
|
|
await ctx.send("Duration must be > 0", hidden=True)
|
|
return
|
|
elif duration >= 300:
|
|
await ctx.send("Duration must be < 5 hours", hidden=True)
|
|
return
|
|
channels = ctx.guild.channels
|
|
roles = ctx.guild.roles
|
|
updates = []
|
|
for channel in channels:
|
|
for role in roles:
|
|
with suppress(Exception):
|
|
await self._lock_channel(channel, role, ctx.author, reason)
|
|
updates.append(
|
|
Lock(
|
|
channel=channel.id,
|
|
guild=ctx.guild.id,
|
|
admin=ctx.author.id,
|
|
reason=reason,
|
|
duration=duration,
|
|
active=True,
|
|
created_at=datetime.utcnow(),
|
|
)
|
|
)
|
|
if updates:
|
|
Lock.objects().insert(updates)
|
|
await ctx.send(f"Server locked for {duration} minute(s)")
|
|
|
|
@cog_ext.cog_subcommand(
|
|
base="lockdown",
|
|
name="end",
|
|
description="Unlocks a server",
|
|
)
|
|
@commands.has_permissions(administrator=True)
|
|
async def _lockdown_end(
|
|
self,
|
|
ctx: SlashContext,
|
|
) -> None:
|
|
channels = ctx.guild.channels
|
|
roles = ctx.guild.roles
|
|
update = False
|
|
locks = Lock.objects(guild=ctx.guild.id, active=True)
|
|
if not locks:
|
|
await ctx.send("No lockdown detected.", hidden=True)
|
|
return
|
|
await ctx.defer()
|
|
for channel in channels:
|
|
for role in roles:
|
|
with suppress(Exception):
|
|
await self._unlock_channel(channel, role, ctx.author)
|
|
update = True
|
|
if update:
|
|
Lock.objects(guild=ctx.guild.id, active=True).update(set__active=False)
|
|
await ctx.send("Server unlocked")
|