482 lines
14 KiB
Python
482 lines
14 KiB
Python
from typing import Literal
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import sys
|
|
|
|
import discord
|
|
from discord.ext import commands
|
|
from discord.ext.commands import errors
|
|
|
|
from tortoise import expressions
|
|
|
|
import redis.asyncio as redis
|
|
|
|
import sentry_sdk
|
|
|
|
from payphone.db import setup, models
|
|
from payphone.core import constants, logging, exceptions, formatting, help_command
|
|
|
|
|
|
class Payphone(commands.AutoShardedBot):
|
|
log = logging.logger
|
|
|
|
redis = redis.Redis(
|
|
host=constants.REDIS_HOST, port=constants.REDIS_PORT, decode_responses=True
|
|
)
|
|
|
|
def __init__(self):
|
|
intents = discord.Intents.default()
|
|
intents.guilds = True
|
|
intents.guild_messages = True
|
|
intents.guild_reactions = True
|
|
intents.message_content = True
|
|
|
|
super().__init__(
|
|
intents=intents,
|
|
command_prefix=self.get_prefix,
|
|
strip_after_prefix=True,
|
|
case_insensitive=True,
|
|
help_command=help_command.HelpCommand(),
|
|
)
|
|
|
|
self.shard_ids = [constants.SHARD_ID]
|
|
self.shard_count = constants.SHARD_COUNT
|
|
|
|
for cog in ["calls", "admin", "mods"]:
|
|
self.load_extension(f"payphone.cogs.{cog}.cog")
|
|
self.log.info("Loaded cog %s.", cog)
|
|
|
|
from payphone.core import calls
|
|
|
|
self.call_handler = calls.CallHandler(self)
|
|
|
|
self.owner_ids = [
|
|
657091104877051907, # Anastasia
|
|
1338608911208419471, # Rafael
|
|
]
|
|
|
|
async def on_ready(self) -> None:
|
|
"""Runs when the bot has finished initializing."""
|
|
|
|
self.log.info("Ready! Logged in as %s.", self.user.display_name)
|
|
|
|
async def set_presence(self, shard_id: int) -> None:
|
|
"""Sets the presence of the bot back to online.
|
|
|
|
Args:
|
|
shard_id (int): The ID of the shard to set the status for.
|
|
"""
|
|
|
|
await self.change_presence(
|
|
activity=discord.CustomActivity(
|
|
name=f"Listening to p.help | Shard {shard_id + 1}"
|
|
),
|
|
status=discord.Status.online,
|
|
shard_id=shard_id,
|
|
)
|
|
|
|
async def on_shard_connect(self, _shard_id: int) -> None:
|
|
"""Runs when the shard connects.
|
|
|
|
Args:
|
|
shard_id (int): The ID of the shard that connected.
|
|
"""
|
|
|
|
self.log.info("Shard has connected.")
|
|
|
|
async def on_shard_ready(self, shard_id: int) -> None:
|
|
"""Sets the activity for the shard.
|
|
|
|
Args:
|
|
shard_id (int): The ID of the initialized shard.
|
|
"""
|
|
|
|
self.log.info("Shard is ready.")
|
|
|
|
await self.set_presence(shard_id)
|
|
|
|
async def on_shard_resumed(self, shard_id: int) -> None:
|
|
"""Sets the activity for the shard.
|
|
|
|
Args:
|
|
shard_id (int): The ID of the resumed shard.
|
|
"""
|
|
|
|
self.log.info("Shard has resumed.")
|
|
|
|
await self.set_presence(shard_id)
|
|
|
|
async def get_or_fetch_channel(
|
|
self, channel_id: int
|
|
) -> discord.abc.GuildChannel | discord.abc.PrivateChannel | discord.Thread | None:
|
|
"""Get's a channel from the bot's cache, or fetches it from Discord if it's not available.
|
|
|
|
Args:
|
|
channel_id (int): The channel's ID.
|
|
|
|
Returns:
|
|
(abc.GuildChannel | abc.PrivateChannel | discord.Thread | None): The channel, or None
|
|
if not found.
|
|
"""
|
|
|
|
channel = self.get_channel(channel_id)
|
|
|
|
if channel is None:
|
|
try:
|
|
return await self.fetch_channel(channel_id)
|
|
|
|
except discord.errors.NotFound:
|
|
return None
|
|
|
|
return channel
|
|
|
|
async def get_prefix(
|
|
self, message: discord.Message | None = None, server_id: int | None = None
|
|
) -> str | list[str]:
|
|
"""Returns the prefix for the message.
|
|
|
|
The prefix is case-insensitive, meaning that it'll adapt to the message to be valid not
|
|
depending of the case. In case that the prefix is invalid, the default or set prefix in its
|
|
original case will be returned.
|
|
|
|
This method can also be used on a server's raw ID. Either `message` or `server_id` must be
|
|
set, but not both.
|
|
|
|
Args:
|
|
message (discord.Message | None): The message to check the prefix for, or None.
|
|
server_id (int | None): The server's ID to get the prefix from.
|
|
|
|
Returns:
|
|
str: The prefix for the message.
|
|
|
|
Raises:
|
|
ValueError: Either `message` or `server_id` must be set, but not both.
|
|
"""
|
|
|
|
default_prefixes = ["p.", "P.", self.user.mention, f"<@!{self.user.id}>"]
|
|
|
|
if message is not None and server_id is not None:
|
|
raise ValueError(
|
|
"Either `message` or `server_id` must be set, but not both."
|
|
)
|
|
|
|
if message is None and server_id is None:
|
|
raise ValueError("Either `message` or `server_id` must be set.")
|
|
|
|
if message is not None and message.guild is None:
|
|
return default_prefixes
|
|
|
|
if message is not None and message.content is None:
|
|
return default_prefixes
|
|
|
|
if (
|
|
message is not None
|
|
and message.content is not None
|
|
and message.content.startswith(self.user.mention)
|
|
):
|
|
return default_prefixes
|
|
|
|
server_id = server_id if server_id is not None else message.guild.id
|
|
|
|
key = f"prefix:{server_id}"
|
|
|
|
prefix: str | None = await self.redis.get(key)
|
|
|
|
if prefix is None:
|
|
server, _ = await models.Server.update_or_create(
|
|
(
|
|
{"name": message.guild.name}
|
|
if message is not None and message.guild is not None
|
|
else None
|
|
),
|
|
id=server_id,
|
|
)
|
|
|
|
await self.redis.set(key, server.prefix, ex=60 * 60)
|
|
|
|
prefix = server.prefix
|
|
|
|
if message is not None and message.content.lower().startswith(prefix.lower()):
|
|
return message.content[: len(prefix)]
|
|
|
|
return prefix
|
|
|
|
def clean_message(self, message: discord.Message) -> str:
|
|
"""Cleans a message.
|
|
|
|
Args:
|
|
message (discord.Message): The message to clean.
|
|
call (Call): The call the message is in.
|
|
|
|
Returns:
|
|
(str): The cleaned message.
|
|
"""
|
|
|
|
content = message.content
|
|
|
|
content = content.replace(
|
|
"@everyone", f"@{constants.INVISIBLE_CHARACTER}everyone"
|
|
).replace("@here", f"@{constants.INVISIBLE_CHARACTER}here")
|
|
|
|
for mention in message.mentions:
|
|
content = content.replace(
|
|
f"<@{mention.id}>", f"@**{mention.display_name}**"
|
|
)
|
|
|
|
for mention in message.role_mentions:
|
|
content = content.replace(f"<@{mention.id}>", f"@**{mention.name}**")
|
|
|
|
for mention in message.channel_mentions:
|
|
content = content.replace(f"<@{mention.id}>", f"#**{mention.name}**")
|
|
|
|
return content
|
|
|
|
async def is_moderator(self, user_id: int) -> bool:
|
|
"""Checks whether a user is a moderator.
|
|
|
|
Args:
|
|
user_id (int): The user to check's ID.
|
|
|
|
Returns:
|
|
(bool): Whether the user is a moderator.
|
|
"""
|
|
|
|
user = await models.User.get_cache_or_db(self.redis, user_id)
|
|
|
|
return user.is_moderator
|
|
|
|
async def is_banned(self, user_id: int) -> bool:
|
|
"""Checks whether a user is banned.
|
|
|
|
Checks whether a user is banned, and caches the result in Redis. The cache expires in 5
|
|
minutes.
|
|
|
|
Args:
|
|
user_id (int): The user to check's ID.
|
|
|
|
Returns:
|
|
bool: Whether the user is banned or not.
|
|
"""
|
|
|
|
key = f"banned:{user_id}"
|
|
|
|
is_banned = await self.redis.get(key)
|
|
|
|
if is_banned is not None:
|
|
return bool(int(is_banned))
|
|
|
|
ban = await models.Ban.filter(
|
|
expressions.Q(user_id=user_id),
|
|
expressions.Q(
|
|
expires_at=None,
|
|
expires_at__gt=datetime.now(timezone.utc),
|
|
join_type="OR",
|
|
),
|
|
).first()
|
|
|
|
await self.redis.set(
|
|
key, int(ban is not None), exat=ban.expires_at if ban else None
|
|
)
|
|
|
|
return is_banned
|
|
|
|
async def ban(
|
|
self,
|
|
user_id: int,
|
|
report: models.Report | None,
|
|
expires_at: datetime | None,
|
|
created_by_id: int | None,
|
|
) -> models.Ban:
|
|
"""Bans a user.
|
|
|
|
Args:
|
|
user_id (int): The user to ban's ID.
|
|
report (Report | None): The report that caused the ban.
|
|
expires_at (datetime | None): When the ban expires.
|
|
created_by_id (int | None): The ID of the user who created the ban.
|
|
|
|
Returns:
|
|
(Ban): The ban.
|
|
"""
|
|
|
|
ban = models.Ban(
|
|
user_id=user_id,
|
|
report=report,
|
|
expires_at=expires_at,
|
|
created_by_id=created_by_id,
|
|
)
|
|
|
|
await ban.save()
|
|
|
|
await self.redis.set(f"banned:{user_id}", 1, exat=expires_at)
|
|
|
|
return ban
|
|
|
|
async def unban(self, user_id: int, expired_by: int | None) -> None:
|
|
"""Unbans a user.
|
|
|
|
Args:
|
|
user_id (int): The user to unban's ID.
|
|
expired_by (int | None): The user who expired the ban.
|
|
"""
|
|
|
|
await models.Ban.filter(
|
|
(
|
|
expressions.Q(expires_at__gt=datetime.now(timezone.utc))
|
|
| expressions.Q(expires_at=None)
|
|
)
|
|
& expressions.Q(user_id=user_id)
|
|
).update(expires_at=datetime.now(timezone.utc), expired_by_id=expired_by)
|
|
|
|
await self.redis.delete(f"banned:{user_id}")
|
|
|
|
async def on_command_error(
|
|
self, context: commands.Context, exception: commands.CommandError
|
|
) -> None:
|
|
"""Handles each of the errors raised by commands.
|
|
|
|
Args:
|
|
context (commands.Context): The context of the command.
|
|
exception (commands.CommandError): The exception raised.
|
|
"""
|
|
|
|
if isinstance(exception, exceptions.ChannelDisabledError):
|
|
await context.send(
|
|
"❌ You cannot run commands in this channel.", delete_after=20.0
|
|
)
|
|
|
|
elif isinstance(exception, exceptions.BannedError):
|
|
await context.send("❌ You have been banned from Payphone.")
|
|
|
|
elif isinstance(exception, exceptions.NotOnCallError):
|
|
await context.send("❌ You need to be on a call to run this command.")
|
|
|
|
elif isinstance(exception, exceptions.NotAModeratorError):
|
|
await context.send("❌ You're not a bot moderator.")
|
|
|
|
elif isinstance(exception, errors.MissingRequiredArgument):
|
|
await context.send(
|
|
f"❌ `{exception.param.name}` is missing. Use `{context.clean_prefix}help {context.command.name}` to learn how to use it."
|
|
)
|
|
|
|
elif isinstance(exception, errors.BotMissingPermissions):
|
|
await context.send(
|
|
f"❌ Payphone's missing {formatting.text_join([f"`{perm.replace("_", " ").title()}`" for perm in exception.missing_permissions])} permissions to run this command."
|
|
)
|
|
|
|
elif isinstance(exception, errors.NoPrivateMessage):
|
|
await context.send("❌ This command can only be run in a server.")
|
|
|
|
elif isinstance(exception, errors.MissingPermissions):
|
|
await context.send("❌ You don't have permission to run this command.")
|
|
|
|
elif isinstance(exception, errors.CommandOnCooldown):
|
|
await context.send(
|
|
f"❌ You're going too fast! Wait `{int(exception.retry_after)}` seconds before trying again."
|
|
)
|
|
|
|
elif isinstance(exception, errors.NotOwner):
|
|
await context.send("❌ You're not the owner of this bot.")
|
|
|
|
elif isinstance(exception, exceptions.AccountTooYoungError):
|
|
await context.send(
|
|
"❌ Your account is too young to use Payphone. It must be at least "
|
|
f"{exception.days} days old to be able to run commands."
|
|
)
|
|
|
|
elif isinstance(exception, errors.CommandNotFound):
|
|
return
|
|
|
|
else:
|
|
sentry_sdk.capture_exception(exception)
|
|
return await super().on_command_error(context, exception)
|
|
|
|
async def on_error(self, *args, **kwargs):
|
|
sentry_sdk.capture_exception(sys.exc_info())
|
|
|
|
@property
|
|
def embed_author(self) -> discord.EmbedAuthor:
|
|
"""The author of the embeds."""
|
|
|
|
return discord.EmbedAuthor(
|
|
name=self.user.display_name,
|
|
icon_url=self.user.display_avatar.url,
|
|
)
|
|
|
|
@commands.check
|
|
async def is_channel_enabled(self, ctx: commands.Context) -> Literal[True]:
|
|
"""Checks whether the current channel is enabled to run a command or not.
|
|
|
|
Args:
|
|
ctx (commands.Context): The current context.
|
|
|
|
Raises:
|
|
exceptions.ChannelDisabledError: The current channel is disabled.
|
|
"""
|
|
|
|
if ctx.guild is None:
|
|
return True
|
|
|
|
if ctx.command.qualified_name in ["config enable", "config disable"]:
|
|
return True
|
|
|
|
server_id = ctx.guild.id
|
|
channel_id = ctx.channel.id
|
|
|
|
channels = await self.redis.hgetall(f"channels:{server_id}")
|
|
|
|
if str(channel_id) not in channels:
|
|
channel, _ = await models.Channel.get_or_create(
|
|
server_id=server_id, id=channel_id
|
|
)
|
|
|
|
if channel.is_enabled:
|
|
await self.redis.hset(f"channels:{server_id}", channel_id, 1)
|
|
return True
|
|
|
|
channels = await models.Channel.filter(server_id=server_id, is_enabled=True)
|
|
|
|
if not channels:
|
|
await self.redis.hset(f"channels:{server_id}", channel_id, 1)
|
|
return True
|
|
|
|
await self.redis.hset(f"channels:{server_id}", channel_id, 0)
|
|
raise exceptions.ChannelDisabledError()
|
|
|
|
if channels[str(channel_id)] == "1":
|
|
return True
|
|
|
|
raise exceptions.ChannelDisabledError()
|
|
|
|
|
|
def run():
|
|
"""Runs the bot."""
|
|
|
|
sentry_sdk.init(
|
|
constants.SENTRY_DSN,
|
|
environment="development" if constants.IS_DEBUG else "production",
|
|
debug=constants.IS_DEBUG,
|
|
server_name=f"Shard #{constants.SHARD_ID + 1}",
|
|
attach_stacktrace=True,
|
|
send_default_pii=True,
|
|
enable_tracing=True,
|
|
traces_sample_rate=1.0,
|
|
profiles_sample_rate=1.0,
|
|
release=constants.VERSION,
|
|
)
|
|
|
|
bot = Payphone()
|
|
|
|
try:
|
|
import uvloop
|
|
|
|
uvloop.install()
|
|
|
|
except ImportError:
|
|
pass
|
|
|
|
bot.loop.run_until_complete(setup.init())
|
|
|
|
bot.run(constants.TOKEN)
|