307 lines
9.2 KiB
Python
307 lines
9.2 KiB
Python
from datetime import datetime, timezone, timedelta
|
|
|
|
import sys
|
|
import asyncio
|
|
|
|
import discord
|
|
from discord import errors
|
|
from discord.ext import commands
|
|
|
|
from tortoise import transactions
|
|
|
|
from payphone import bot
|
|
from payphone.db import models, utils
|
|
from payphone.cogs import base
|
|
from payphone.cogs.mods import embeds, views
|
|
from payphone.core import checks, converters, constants
|
|
|
|
|
|
class Mods(base.Cog):
|
|
show_in_help = False
|
|
|
|
def __init__(self, bot_instance: bot.Payphone):
|
|
self.bot = bot_instance
|
|
|
|
@commands.command(extras={"reply_argument": "user"})
|
|
@checks.is_moderator()
|
|
@transactions.atomic()
|
|
async def ban(
|
|
self,
|
|
ctx: commands.Context,
|
|
user: discord.User | discord.Member | converters.DiscordUser,
|
|
time: converters.Timedelta | None = None,
|
|
*,
|
|
reason: str | None = "",
|
|
) -> None:
|
|
"""
|
|
Ban a user by their ID, pinging them, or replying to their message.
|
|
|
|
Options:
|
|
- `<user>`: The user to ban or their ID.
|
|
- `[<time>]`: The amount of time to ban the user for. For example, `1d 5h`.
|
|
- `[<reason>]`: The reason of their ban.
|
|
"""
|
|
|
|
user_id = None
|
|
|
|
if isinstance(user, (discord.User, discord.Member)):
|
|
user_id = user.id
|
|
|
|
else:
|
|
user_id = user
|
|
|
|
profile = await models.Profile.filter(user_id=user_id).limit(1).first()
|
|
|
|
row = await self.bot.ban(
|
|
user_id,
|
|
(datetime.now(timezone.utc) + time) if time is not None else None,
|
|
reason,
|
|
ctx.author,
|
|
)
|
|
|
|
message = f"✅ {f"**{profile.name}**" if profile is not None else f"`ID: {user_id}`"} was banned"
|
|
|
|
if row.expires_at:
|
|
message += f" until <t:{int(row.expires_at.timestamp())}:F> (<t:{int(row.expires_at.timestamp())}:R>)"
|
|
|
|
message += "."
|
|
|
|
await ctx.send(message)
|
|
|
|
try:
|
|
notification_message = await (
|
|
await self.bot.get_or_fetch_user(user_id)
|
|
).send(
|
|
embed=embeds.BanNotificationEmbed(self.bot, row),
|
|
view=views.BanNotificationView(self.bot, row),
|
|
)
|
|
|
|
row.notification_message_id = notification_message.id
|
|
await row.save()
|
|
|
|
except errors.DiscordException:
|
|
pass
|
|
|
|
@commands.command(extras={"reply_argument": "user"})
|
|
@checks.is_moderator()
|
|
@transactions.atomic()
|
|
async def unban(
|
|
self,
|
|
ctx: commands.Context,
|
|
user: discord.User | discord.Member | converters.DiscordUser,
|
|
) -> None:
|
|
"""
|
|
Unbans a user by their ID, pinging them, or replying to their message. The user will be
|
|
attempted to get notified of the unban.
|
|
|
|
Options:
|
|
- `<user>`: The user to unban or their ID.
|
|
"""
|
|
|
|
user_id = user if isinstance(user, int) else user.id
|
|
|
|
await self.bot.unban(user_id)
|
|
|
|
profile = await models.Profile.filter(user_id=user_id).limit(1).first()
|
|
|
|
await ctx.send(
|
|
f"✅ {f"**{profile.name}**" if profile is not None else f"`ID: {user_id}`"} was unbanned."
|
|
)
|
|
|
|
try:
|
|
await (await self.bot.get_or_fetch_user(user_id)).send(
|
|
"🔔 You have been unbanned from Payphone."
|
|
)
|
|
except errors.DiscordException:
|
|
pass
|
|
|
|
@commands.command(extras={"reply_argument": "user"})
|
|
@checks.is_moderator()
|
|
@transactions.atomic()
|
|
async def warn(
|
|
self,
|
|
ctx: commands.Context,
|
|
user: discord.User | discord.Member | converters.DiscordUser,
|
|
*,
|
|
reason: str,
|
|
) -> None:
|
|
"""
|
|
Warn a user by their ID or mention.
|
|
|
|
Options:
|
|
- `<user>`: The user's mention or their ID.
|
|
- `<reason>`: The reason of the warn.
|
|
"""
|
|
|
|
user_id = user if isinstance(user, int) else user.id
|
|
|
|
user = None
|
|
|
|
try:
|
|
user = await self.bot.get_or_fetch_user(user_id)
|
|
except errors.DiscordError:
|
|
await ctx.send(
|
|
"❌ Could not find this user. It may be because the user has left all servers with Payphone in them."
|
|
)
|
|
return
|
|
|
|
embed = discord.Embed(
|
|
color=constants.EMBED_COLOR,
|
|
title="You've been warned",
|
|
description=reason,
|
|
timestamp=datetime.now(timezone.utc),
|
|
author=self.bot.embed_author,
|
|
footer=discord.EmbedFooter(
|
|
text="Ignoring this warning will result in a ban."
|
|
),
|
|
thumbnail=self.bot.user.display_avatar.url,
|
|
)
|
|
|
|
profile = await models.Profile.filter(user_id=user_id).limit(1).first()
|
|
|
|
try:
|
|
await user.send(embed=embed)
|
|
except errors.DiscordException:
|
|
await ctx.send(
|
|
f"❌ Could not send the warning to {f"**{profile.name}**" if profile is not None else "the user"}, probably because they have their DMs closed."
|
|
)
|
|
return
|
|
|
|
await ctx.send(
|
|
f"✅ {f"**{profile.name}**" if profile is not None else "the user"} has been warned."
|
|
)
|
|
|
|
@commands.command()
|
|
@checks.is_moderator()
|
|
@transactions.atomic()
|
|
async def reports(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Shows the next report to review.
|
|
"""
|
|
|
|
report = (
|
|
await models.Report.filter(status=models.ReportStatus.ON_REVIEW)
|
|
.order_by("id")
|
|
.limit(1)
|
|
.first()
|
|
.select_related("call")
|
|
.prefetch_related("call__call_rss", "call__call_rss__profile")
|
|
)
|
|
|
|
if report is None:
|
|
await ctx.send("✅ You're up to date! No reports left to check.")
|
|
return
|
|
|
|
await ctx.send(
|
|
embed=embeds.ReportEmbed(self.bot, report),
|
|
view=views.ReportView(self.bot, report),
|
|
)
|
|
|
|
@commands.command(aliases=["mod"])
|
|
@commands.is_owner()
|
|
@transactions.atomic()
|
|
async def moderator(
|
|
self,
|
|
ctx: commands.Context,
|
|
user: discord.User | discord.Member,
|
|
toggle: bool | None = None,
|
|
) -> None:
|
|
"""
|
|
Toggles a user's moderator status. It can only be ran by Payphone's owner.
|
|
|
|
Options:
|
|
- `<user>`: The user to toggle the moderation status for.
|
|
- `[<toggle>]`: The moderator status to set to the user.
|
|
"""
|
|
|
|
user_row, profile_row = await utils.get_user(user)
|
|
|
|
if toggle is not None:
|
|
user_row.is_moderator = toggle
|
|
|
|
else:
|
|
user_row.is_moderator = not user_row.is_moderator
|
|
|
|
await user_row.save()
|
|
|
|
await ctx.send(
|
|
f"✅ **{profile_row.name}**'s moderator status was set to `{"on" if user_row.is_moderator else "off"}`."
|
|
)
|
|
|
|
@commands.command()
|
|
@commands.is_owner()
|
|
@transactions.atomic()
|
|
async def shutdown(
|
|
self, ctx: commands.Context, seconds: int = 60, *, reason: str | None
|
|
) -> None:
|
|
"""
|
|
Notifies all the in-call users of the scheduled shutdown and shuts down Payphone after x seconds.
|
|
|
|
Options:
|
|
- `[<seconds>]`: The amount of seconds to wait before shutdown.
|
|
- `[<reason>]`: The reason of the shutdown.
|
|
"""
|
|
|
|
oncall_channel_ids = list(
|
|
await self.bot.redis.smembers(constants.CHANNELS_ONCALL_SET)
|
|
)
|
|
searching_channel_ids = list(
|
|
await self.bot.redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
|
|
)
|
|
|
|
channel_ids = [int(c) for c in oncall_channel_ids] + [
|
|
int(c) for c in searching_channel_ids
|
|
]
|
|
|
|
await ctx.send(f"⚡ Notifying `{len(channel_ids)}` channels of the shutdown.")
|
|
|
|
for channel_id in channel_ids:
|
|
try:
|
|
await self.bot.get_partial_messageable(channel_id).send(
|
|
f"⚡ Payphone will be shutdown <t:{int((datetime.now(timezone.utc) + timedelta(seconds=seconds)).timestamp())}:R> due to a maintenance restart."
|
|
+ (" Reason: " + reason if reason is not None else "")
|
|
)
|
|
|
|
except errors.DiscordException:
|
|
pass
|
|
|
|
await ctx.send(
|
|
f"⚡ All channels were notified of the shutdown. Payphone will shut down <t:{int((datetime.now(timezone.utc) + timedelta(seconds=seconds)).timestamp())}:R>."
|
|
)
|
|
|
|
await asyncio.sleep(seconds)
|
|
|
|
oncall_channel_ids = list(
|
|
await self.bot.redis.smembers(constants.CHANNELS_ONCALL_SET)
|
|
)
|
|
searching_channel_ids = list(
|
|
await self.bot.redis.smembers(constants.CHANNELS_SEARCH_QUEUE)
|
|
)
|
|
|
|
channel_ids = [int(c) for c in oncall_channel_ids] + [
|
|
int(c) for c in searching_channel_ids
|
|
]
|
|
|
|
for channel_id in channel_ids:
|
|
try:
|
|
await self.bot.get_partial_messageable(channel_id).send(
|
|
"⚡ The call will be ended due to a scheduled shutdown."
|
|
)
|
|
|
|
except errors.DiscordException:
|
|
pass
|
|
|
|
await ctx.send("⚡ Payphone was shut down.")
|
|
|
|
sys.exit()
|
|
|
|
|
|
def setup(b: bot.Payphone) -> None:
|
|
"""Load the cog into the bot.
|
|
|
|
Args:
|
|
b (bot.Payphone): The bot instance to load the cog into.
|
|
"""
|
|
|
|
b.add_cog(Mods(b))
|