nyeki/payphone
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(shard): Lots of code.

Browse Source
This commit is contained in:
Nyeki.py
2024-11-15 23:24:11 -03:00
parent d9d752c730
commit 6377cb66c2
35 changed files with 2546 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
payphone/bot.py
+279 -15
View File
@@ -1,10 +1,17 @@
from typing import Literal
from datetime import datetime, timezone
import discord
from discord.ext import commands
from discord.ext.commands import errors
from tortoise import expressions
import redis.asyncio as redis
from payphone.db import setup, models
from payphone.core import constants, logging
from payphone.core import constants, logging, exceptions, formatting, help_command
class Payphone(commands.AutoShardedBot):
@@ -29,9 +36,10 @@ class Payphone(commands.AutoShardedBot):
command_prefix="p.",
strip_after_prefix=True,
case_insensitive=True,
help_command=help_command.HelpCommand(),
)
for cog in ["calls", "admin"]:
for cog in ["calls", "admin", "mods"]:
self.load_extension(f"payphone.cogs.{cog}.cog")
self.log.info("Loaded cog %s.", cog)
@@ -59,7 +67,7 @@ class Payphone(commands.AutoShardedBot):
shard_id=shard_id,
)
async def on_shard_connect(self, shard_id: int) -> None:
async def on_shard_connect(self, _shard_id: int) -> None:
"""Runs when the shard connects.
Args:
@@ -99,7 +107,8 @@ class Payphone(commands.AutoShardedBot):
channel_id (int): The channel's ID.
Returns:
(abc.GuildChannel | abc.PrivateChannel | discord.Thread | None): The channel, or None if not found.
(abc.GuildChannel | abc.PrivateChannel | discord.Thread | None): The channel, or None
if not found.
"""
channel = self.get_channel(channel_id)
@@ -109,6 +118,77 @@ class Payphone(commands.AutoShardedBot):
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, call) -> str:
"""Cleans a message.
@@ -138,29 +218,205 @@ class Payphone(commands.AutoShardedBot):
return content
async def get_user_flag(self, user_id: int) -> str:
"""Gets the flags of a user.
async def is_moderator(self, user_id: int) -> bool:
"""Checks whether a user is a moderator.
Args:
user_id (int): The ID of the user.
user_id (int): The user to check's ID.
Returns:
(str): The flags of the user.
(bool): Whether the user is a moderator.
"""
flag = await self.redis.hget(f"user:{user_id}", "flags")
user = await models.User.get_cache_or_db(self.redis, user_id)
if flag is None:
user, _ = await models.User.get_or_create(id=user_id)
return user.is_moderator
flag = user.flags
async def is_banned(self, user_id: int) -> bool:
"""Checks whether a user is banned.
await self.redis.hset(f"user:{user_id}", "flags", int(flag))
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) -> None:
"""Unbans a user.
Args:
user_id (int): The user to unban's ID.
"""
await models.Ban.filter(user_id=user_id).delete()
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, errors.CommandNotFound):
return
else:
flag = models.UserFlag(int(flag))
return await super().on_command_error(context, exception)
return flag
@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():
@@ -168,6 +424,14 @@ def run():
bot = Payphone()
try:
import uvloop
uvloop.install()
except ImportError:
pass
bot.loop.run_until_complete(setup.init())
bot.run(constants.TOKEN)
payphone/cogs/admin/cog.py
+73 -6
View File
@@ -5,11 +5,16 @@ from discord.ext import commands
from payphone.db import models
from payphone.bot import Payphone
from payphone.cogs import base
from payphone.cogs.admin import embeds, views
from payphone.core import checks
class Admin(commands.Cog):
class Admin(base.Cog):
"""Administration in Payphone."""
show_in_help = False
def __init__(self, bot: Payphone) -> None:
"""Initializes the Calls cog.
@@ -22,26 +27,69 @@ class Admin(commands.Cog):
@commands.command(aliases=["mod"])
@commands.is_owner()
async def moderator(
self, ctx: commands.Context, user: discord.User, toggle: bool
self, ctx: commands.Context, user: discord.User, toggle: bool = True
) -> None:
"""Gives the user the Moderator role."""
"""Gives the user the Moderator role.
Options:
- `<user>`: The user to give the role.
- `[toggle]`: Whether to give or remove the role.
await models.User.update_or_create(
Permissions:
- Bot Owner
"""
instance, _ = await models.User.update_or_create(
id=user.id, defaults={"is_moderator": toggle}
)
await instance.set_cache(self.bot.redis)
await ctx.send(
f"Toggled the moderator role {'on' if toggle else 'off'} for {user.mention}."
)
@commands.command(aliases=["ub"])
@checks.is_moderator()
async def unban(
self, ctx: commands.Context, user: discord.User | None = None
) -> None:
"""Unbans the user from the bot.
Options:
- `[user]`: The user to unban. If none is provided, the author will be unbanned.
Permissions:
- Bot Moderator
"""
user = user or ctx.author
await self.bot.unban(user.id)
await ctx.send(f"{user.mention} was unbanned from Payphone.")
try:
await user.send("🔔 You were unbanned from Payphone!")
except discord.Forbidden:
pass
@commands.command(aliases=["sup"])
@commands.is_owner()
async def supporter(
self, ctx: commands.Context, user: discord.User, toggle: bool = True
) -> None:
"""Gives the user the Supporter role."""
"""Gives the user the Supporter role.
Options:
- `<user>`: The user to give the role.
- `[toggle]`: Whether to give or remove the role.
await models.User.update_or_create(
Permissions:
- Bot Owner
"""
instance, _ = await models.User.update_or_create(
id=user.id,
defaults={
"is_supporter_until": (
@@ -49,11 +97,30 @@ class Admin(commands.Cog):
)
},
)
await instance.set_cache(self.bot.redis)
await ctx.send(
f"Gave the Supporter role to {ctx.author.mention} until in one month."
)
@commands.command()
@checks.is_moderator()
async def reports(self, ctx: commands.Context) -> None:
"""Shows the reports.
Permissions:
- Bot Moderator
"""
report = await models.Report.filter(resolved=False).first()
if report is None:
return await ctx.send("✅ No reports left, have this cookie 🍪 !!")
await ctx.send(
embed=embeds.report_embed(report), view=views.ReportView(self.bot, report)
)
def setup(bot: Payphone) -> None:
"""Loads the Admin cog into the bot.
payphone/cogs/admin/embeds.py
+33
View File
@@ -0,0 +1,33 @@
import discord
from payphone.db import models
from payphone.core import constants
def report_embed(report: models.Report) -> discord.Embed:
"""Creates an embed for a report.
Args:
report (models.Report): The report to create an embed for.
Returns:
(discord.Embed): The embed.
"""
return discord.Embed(
color=constants.PRIMARY_COLOR,
title="Message Report",
description=(
f"**ID:** {report.id}\n"
f"**User ID**: {report.user_id}\n"
f"**User Name**: {report.user_name}\n"
f"**Channel ID**: {report.channel_id}\n"
f"**Created At**: <t:{int(report.created_at.timestamp())}>\n"
),
fields=[
discord.EmbedField(
name="Message Content", value=report.message_content, inline=False
)
],
thumbnail=report.user_avatar_url,
)
payphone/cogs/admin/views.py
+197
View File
@@ -0,0 +1,197 @@
from datetime import datetime, timedelta, timezone
import discord
from payphone.db import models
from payphone.bot import Payphone
from payphone.core import constants
from payphone.cogs.admin import embeds
class ReportView(discord.ui.View):
"""The report view."""
def __init__(self, bot: Payphone, report: models.Report) -> None:
"""Initializes the ReportView.
Args:
bot (Payphone): The bot.
report (models.Report): The report.
"""
super().__init__(disable_on_timeout=True)
self.bot = bot
self.report = report
@discord.ui.select(
placeholder="Select an action",
options=[
discord.SelectOption(
value="warn",
label="Warn",
description="Send a warning to the user. This is done automatically.",
),
discord.SelectOption(
value="ban",
label="Ban",
description="Ban the user. The duration of the ban will be applied automatically.",
),
discord.SelectOption(
value="ban_permanently",
label="Permanent Ban",
description="Ban the user permanently. Use this when the user must not be allowed back.",
),
],
)
async def action_callback(
self, select: discord.SelectMenu, interaction: discord.Interaction
) -> None:
"""Handles the action select."""
user = await models.User.get_cache_or_db(self.bot.redis, self.report.user_id)
if select.values[0] == "warn":
user.warns_count += 1
await user.save(force_update=True)
await user.set_cache(self.bot.redis)
embed = discord.Embed(
color=constants.MASK_COLOR,
title="Warned",
description=f"You've been warned. You've received `{user.warns_count}` warnings in total.",
fields=[
discord.EmbedField(
name="Message Content",
value=f"You sent the following message in a call:\n> {self.report.message_content}",
),
discord.EmbedField(
name="User Name",
value=self.report.user_name,
),
discord.EmbedField(
name="User Avatar",
value=f"[Click to open your avatar URL](<{self.report.user_avatar_url}>)",
),
],
footer=discord.EmbedFooter(
text="⚠️ Repeating this action will result in a ban."
),
)
try:
await (await self.bot.get_or_fetch_user(self.report.user_id)).send(
embed=embed
)
except discord.Forbidden:
pass
else:
user.bans_count += 1
await user.save(force_update=True)
await user.set_cache(self.bot.redis)
expires_at = None
if select.values[0] == "ban":
expiry_times = {
1: datetime.now(timezone.utc) + timedelta(days=1),
2: datetime.now(timezone.utc) + timedelta(days=7),
3: datetime.now(timezone.utc) + timedelta(days=30),
}
expires_at = expiry_times.get(user.bans_count, None)
await self.bot.ban(
self.report.user_id,
self.report,
expires_at=expires_at,
created_by_id=interaction.user.id,
)
embed = discord.Embed(
color=constants.MODERATOR_COLOR,
title="Banned",
description=f"You've been banned. You've been banned `{user.bans_count}` times.",
fields=[
discord.EmbedField(
name="Message Content",
value=f"You sent the following message in a call:\n> {self.report.message_content}",
),
discord.EmbedField(
name="User Name",
value=self.report.user_name,
),
discord.EmbedField(
name="User Avatar",
value=f"[Click to open your avatar URL](<{self.report.user_avatar_url}>)",
),
discord.EmbedField(
name="Expires At",
value=(
expires_at.strftime("%Y-%m-%d %H:%M:%S")
+ f" <t:{int(expires_at.timestamp())}:R>"
if expires_at
else "Never"
),
),
],
footer=discord.EmbedFooter(
text=(
"⚠️ In case you get banned again, your ban will last longer. Multiple bans will result in a permanent ban."
if expires_at is not None
else "⚠️ You've been permanently banned."
)
),
)
try:
await (await self.bot.get_or_fetch_user(self.report.user_id)).send(
embed=embed
)
except discord.Forbidden:
pass
self.report.resolved = True
await self.report.save()
report = await models.Report.filter(resolved=False).first()
if report is None:
self.stop()
return await interaction.response.edit_message(
content="✅ No reports left, have this cookie 🍪 !!",
embeds=[],
view=None,
)
await interaction.response.edit_message(
embed=embeds.report_embed(report), view=ReportView(self.bot, report)
)
@discord.ui.button(label="Skip", style=discord.ButtonStyle.gray)
async def skip_callback(
self, _button: discord.ui.Button, interaction: discord.Interaction
) -> None:
"""Skips the report."""
self.report.resolved = True
await self.report.save()
report = await models.Report.filter(resolved=False).first()
if report is None:
self.stop()
return await interaction.response.edit_message(
content="✅ No reports left, have this cookie 🍪 !!",
embeds=[],
view=None,
)
await interaction.response.edit_message(
embed=embeds.report_embed(report), view=ReportView(self.bot, report)
)
payphone/cogs/base.py
+7
View File
@@ -0,0 +1,7 @@
from discord.ext import commands
class Cog(commands.Cog):
"""Base cog with Payphone-specific properties."""
show_in_help: bool = True
payphone/cogs/calls/cog.py
+159 -44
View File
@@ -9,10 +9,11 @@ import aiohttp
from payphone.db import models
from payphone.bot import Payphone
from payphone.core import constants, censorship, messages
from payphone.cogs import base
from payphone.core import constants, censorship, messages, masks, checks
class Calls(commands.Cog):
class Calls(base.Cog):
"""Calls in Payphone."""
def __init__(self, bot: Payphone) -> None:
@@ -24,7 +25,8 @@ class Calls(commands.Cog):
self.bot = bot
@commands.command()
@commands.command(aliases=["c"])
@checks.is_not_banned()
async def call(self, ctx: commands.Context) -> None:
"""
Call and connect with other people.
@@ -49,7 +51,8 @@ class Calls(commands.Cog):
)
)
@commands.command()
@commands.command(aliases=["s"])
@checks.is_not_banned()
async def skip(self, ctx: commands.Context) -> None:
"""
Skip the current call and search for another one.
@@ -76,7 +79,8 @@ class Calls(commands.Cog):
f"☎️ {random.choice(messages.OTHER_PARTY_HANGUP)}"
)
@commands.command()
@commands.command(aliases=["h"])
@checks.is_not_banned()
async def hangup(self, ctx: commands.Context) -> None:
"""
Hang up the call.
@@ -101,13 +105,14 @@ class Calls(commands.Cog):
),
)
@commands.command()
@commands.command(aliases=["fr", "friend"])
@checks.is_not_banned()
async def friendreq(self, ctx: commands.Context) -> None:
"""
Send a friend request to the other party.
"""
call = await self.bot.call_handler.get_call(ctx.channel.id)
call = await self.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
if call is None:
return await ctx.send(
@@ -125,26 +130,63 @@ class Calls(commands.Cog):
ctx.send("📬 Your friend request has been sent!"),
)
@commands.command(aliases=["m"])
@checks.is_not_banned()
async def mask(self, ctx: commands.Context, toggle: bool | None = None) -> None:
"""
Toggle your mask.
Options:
- `[toggle]`: Whether to toggle your mask on or off. If no toggle is specified, your mask will be displayed.
"""
call = await self.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
user = await models.User.get_cache_or_db(self.bot.redis, ctx.author.id)
if toggle is not None:
user.is_masked = toggle
await user.save(force_update=True)
await user.set_cache(self.bot.redis)
message = f"🎭 Your mask {"is" if toggle is None else "was set to"} {"`on`" if user.is_masked else "`off`"}."
if call is not None:
participant = await call.get_participant(ctx.author.id)
if user.is_masked:
message += f" You're currently being displayed as **{masks.get_mask_emoji(participant.mask)} {masks.get_mask_name(participant.mask)}**."
message += (
" You can toggle your mask on or off using `p.mask on` or `p.mask off`."
)
return await ctx.send(message)
@commands.Cog.listener()
async def on_message(self, message: discord.Message) -> None:
"""
Handle messages in calls.
"""
if message.author.bot or (await self.bot.get_context(message)).valid:
ctx = await self.bot.get_context(message)
if message.author.bot or ctx.valid:
return
call = await self.bot.call_handler.get_call(message.channel.id)
if await self.bot.is_banned(message.author.id):
return
call = await self.bot.call_handler.get_call_by_channel_id(message.channel.id)
if call is None:
return
is_new_participant = not bool(await call.get_participant(message.author.id))
await call.set_participant(
participant, was_mask_toggled = await call.set_participant(
message.author.id,
message.author.display_name,
message.author.display_avatar.url,
message.channel.id,
)
other_channel = (
@@ -153,9 +195,14 @@ class Calls(commands.Cog):
else call.channels[1]
)
user_flag = await self.bot.get_user_flag(message.author.id)
blocked_section = censorship.censor(message.content)
blocked_section = (
censorship.censor(
message.content,
participant.user.flag & models.UserFlag.SUPPORTER,
)
if not (participant.user.flag & models.UserFlag.MODERATOR)
else None
)
if blocked_section:
return await asyncio.gather(
@@ -178,18 +225,19 @@ class Calls(commands.Cog):
message_content = self.bot.clean_message(message, call)
if message.stickers:
if user_flag & (models.UserFlag.SUPPORTER | models.UserFlag.MODERATOR):
if participant.user.flag & (
models.UserFlag.SUPPORTER | models.UserFlag.MODERATOR
):
sticker = message.stickers[0]
if sticker.format != discord.StickerFormatType.lottie:
message_content = f"[Sticker: {sticker.name}]({sticker.url})"
else:
await message.channel.send(
"💖 Stickers are only allowed to donators! You can become one at [our Ko-Fi "
"page](https://ko-fi.com/payphonedev)."
return await message.channel.send(
"💖 Stickers and GIFs are only allowed to donators! You can become one at [our Ko-Fi "
"page](<https://ko-fi.com/payphonedev>)."
)
return
embeds = []
@@ -202,38 +250,91 @@ class Calls(commands.Cog):
message.reference.message_id
)
replied_message_content = self.bot.clean_message(replied_message, call)
embeds.append(
discord.Embed(
color=constants.PRIMARY_COLOR,
description=replied_message_content,
author=discord.EmbedAuthor(
name=replied_message.author.display_name,
icon_url=replied_message.author.display_avatar.url,
),
original_message_id = (
await self.bot.call_handler.get_original_message_id(
replied_message.id
)
)
if (
original_message_id is not None
and original_message_id != replied_message.id
):
replied_message = await self.bot.get_partial_messageable(
other_channel
).fetch_message(original_message_id)
replied_message_content = self.bot.clean_message(replied_message, call)
replied_participant = await call.get_participant(
replied_message.author.id
)
if replied_participant is None:
message_content += (
"\n-# ⓘ In reply to a message from outside this call."
)
else:
embeds.append(
discord.Embed(
color=constants.PRIMARY_COLOR,
description=replied_message_content,
author=discord.EmbedAuthor(
name=(
replied_participant.display_name
if replied_participant.channel_id
!= message.channel.id
else replied_participant.name
),
icon_url=(
replied_participant.display_avatar_url
if replied_participant.channel_id
!= message.channel.id
else replied_participant.avatar_url
),
),
)
)
else:
message_content += (
"\n-# ⓘ In reply to a message from outside this call."
)
if is_new_participant:
if user_flag & models.UserFlag.SUPPORTER:
embeds.append(
discord.Embed(
color=constants.SUPPORTER_COLOR,
description=f"🌟 **{message.author.display_name}** is a Payphone supporter!",
if was_mask_toggled:
if not participant.is_masked:
if participant.user.flag & models.UserFlag.SUPPORTER:
embeds.append(
discord.Embed(
color=constants.SUPPORTER_COLOR,
description=f"🌟 **{participant.display_name}** is a Payphone supporter!",
footer=random.choice(
[
discord.EmbedFooter(
text="You can also become a supporter at our Ko-Fi page."
)
]
+ [None] * 3
),
)
)
)
if user_flag & models.UserFlag.MODERATOR:
if participant.user.flag & models.UserFlag.MODERATOR:
embeds.append(
discord.Embed(
color=constants.MODERATOR_COLOR,
description=f"🛡️ **{participant.display_name}** is a Payphone moderator!",
)
)
else:
embeds.append(
discord.Embed(
color=constants.MODERATOR_COLOR,
description=f"🛡️ **{message.author.display_name}** is a Payphone moderator!",
color=constants.MASK_COLOR,
description=f"🎭 **{participant.display_name}** is wearing a mask! Their real identity is hidden.",
footer=discord.EmbedFooter(
text="You can set your mask using p.mask on."
),
)
)
@@ -246,8 +347,8 @@ class Calls(commands.Cog):
propagated_msg = await webhook.send(
content=message_content,
username=message.author.display_name,
avatar_url=message.author.display_avatar.url,
username=participant.display_name,
avatar_url=participant.display_avatar_url,
embeds=embeds,
wait=True,
)
@@ -267,7 +368,7 @@ class Calls(commands.Cog):
Handle deleted messages in calls.
"""
call = await self.bot.call_handler.get_call(payload.channel_id)
call = await self.bot.call_handler.get_call_by_channel_id(payload.channel_id)
if call is None:
return
@@ -296,7 +397,7 @@ class Calls(commands.Cog):
Handle edited messages in calls.
"""
call = await self.bot.call_handler.get_call(payload.channel_id)
call = await self.bot.call_handler.get_call_by_channel_id(payload.channel_id)
if call is None:
return
@@ -342,6 +443,11 @@ class Calls(commands.Cog):
Report a message in a call.
"""
if self.bot.is_banned(ctx.author.id):
return await ctx.send_response(
"❌ You're banned from using Payphone.", ephemeral=True
)
author_id = await self.bot.call_handler.get_message_author_id(message.id)
if author_id is None:
@@ -363,11 +469,20 @@ class Calls(commands.Cog):
else:
message_id = message.id
if message_id != message.id:
cached_message = await self.bot.call_handler.get_message(message_id)
message = await self.bot.get_partial_messageable(
cached_message.channel_id
).fetch_message(message_id)
await models.Report.update_or_create(
message_id=message_id,
defaults={
"channel_id": message.channel.id,
"user_id": author_id,
"user_name": message.author.display_name,
"user_avatar_url": message.author.display_avatar.url,
"message_content": message.content,
},
)
payphone/cogs/mods/__init__.py
View File
payphone/cogs/mods/cog.py
+139
View File
@@ -0,0 +1,139 @@
import discord
from discord.ext import commands
from payphone.db import models
from payphone.bot import Payphone
from payphone.cogs import base
class Mods(base.Cog):
"""Administration in Payphone."""
def __init__(self, bot: Payphone) -> None:
"""Initializes the Calls cog.
Args:
bot (Payphone): The bot instance.
"""
self.bot = bot
@commands.group(invoke_without_command=True)
@commands.guild_only()
@commands.has_guild_permissions(manage_guild=True)
async def config(
self, ctx: commands.Context, setting: str = "", value: str | None = None
) -> None:
"""
Set up Payphone for the current server.
Settings:
- `prefix`: Set a custom prefix.
- `enable`: Enable a (or all) channel(s) to run commands in it. Non-enabled channels will pass to be disabled.
- `disable`: Disables a channel to run commands in it. Non-disabled channels will stay enabled by default.
You can use `{prefix}help config <setting>` to get help about a specific configuration.
"""
await ctx.send_help("config")
@config.command(extras={"help_name": "Prefix"})
@commands.guild_only()
@commands.has_guild_permissions(manage_guild=True)
async def prefix(self, ctx: commands.Context, prefix: str = "p.") -> None:
"""
Set a custom bot prefix.
Options:
- `<prefix>`: The new custom prefix.
Permissions:
- Manage Server
"""
if len(prefix) > 10:
return await ctx.send(
"❌ The prefix must not be longer than 10 characters."
)
server, _ = await models.Server.update_or_create(
id=ctx.guild.id, defaults={"name": ctx.guild.name}
)
await server.set_prefix(self.bot.redis, prefix)
await ctx.send(f"✅ The prefix was set to `{prefix}`!")
@config.command(extras={"help_name": "Enable Channel"})
@commands.guild_only()
@commands.has_guild_permissions(manage_channels=True)
async def enable(
self, ctx: commands.Context, channel: discord.TextChannel | str | None
) -> None:
"""
Enable a channel to run commands in it. Non-enabled channels will pass to be disabled.
Options:
- `[<channel>]`: The channel to enable or "all". If no channel is specified, the current channel will be enabled.
Permissions:
- Manage Channels
"""
server_id = ctx.guild.id
channel = channel or ctx.channel
if isinstance(channel, str) and channel.lower() == "all":
await models.Channel.filter(server_id=server_id).update(is_enabled=None)
await ctx.bot.redis.delete(f"channels:{server_id}")
await ctx.send(
"✅ All channels were enabled! You can now run commands in any channel."
)
else:
await models.Channel.update_or_create(
id=channel.id, server_id=server_id, is_enabled=True
)
await ctx.bot.redis.hset(f"channels:{server_id}", channel.id, "1")
await ctx.send(
f"✅ {channel.mention} was enabled! Other non-enabled channels were disabled."
)
@config.command(extras={"help_name": "Disable Channel"})
@commands.guild_only()
@commands.has_guild_permissions(manage_channels=True)
async def disable(
self, ctx: commands.Context, channel: discord.TextChannel | None
) -> None:
"""
Disables a channel to run commands in it. Non-disabled channels will stay enabled by default.
Options:
- `[<channel>]`: The channel to disable. If no channel is specified, the current channel will be disabled.
Permissions:
- Manage Channels
"""
channel = channel or ctx.channel
await models.Channel.update_or_create(
id=channel.id, server_id=ctx.guild.id, is_enabled=False
)
await ctx.bot.redis.hset(f"channels:{ctx.guild.id}", channel.id, "0")
await ctx.send(
f"✅ {channel.mention} was disabled. Commands cannot be ran in this channel anymore."
)
def setup(bot: Payphone) -> None:
"""Loads the Mods cog into the bot.
Args:
bot (Payphone): The bot instance.
"""
bot.add_cog(Mods(bot))
payphone/core/calls.py
+146 -15
View File
@@ -2,7 +2,9 @@ from dataclasses import dataclass
import orjson
from payphone.db import models
from payphone.bot import Payphone
from payphone.core import masks
@dataclass
@@ -14,6 +16,44 @@ class Participant:
id: int
name: str
avatar_url: str
mask: str
is_masked: bool
channel_id: int
user: models.User | None = None
@property
def display_name(self) -> str:
"""Gets the display name of the participant.
Returns:
str: The display name of the participant.
"""
return self.name if not self.is_masked else masks.get_mask_name(self.mask)
@property
def display_avatar_url(self) -> str:
"""Gets the display avatar URL of the participant.
Returns:
str: The display avatar URL of the participant.
"""
return (
self.avatar_url
if not self.is_masked
else masks.get_mask_image_url(self.mask)
)
@dataclass
class Message:
id: int
call_id: int
channel_id: int
propagation_id: int
propagation_channel_id: int
author_id: int
@dataclass
@@ -24,6 +64,7 @@ class Call:
id: int
channels: tuple[int, int]
masks: list[str]
async def set_message(
self, id: int, channel_id: int, propagation_id: int, author_id: int
@@ -54,6 +95,38 @@ class Call:
await pipe.execute()
async def get_message(self, id: int) -> Message | None:
"""Gets a message as part of a call.
Args:
id (int): The ID of the message to get.
Returns:
Message | None: The message that was set.
"""
call_id = await self._bot.redis.hget("calls:messages", id)
if call_id is None:
return
channel_id = await self._bot.redis.hget("calls:messages:channels", id)
propagation_id = await self._bot.redis.hget("calls:messages:propagations", id)
author_id = await self._bot.redis.hget("calls:messages:authors", id)
return Message(
id=id,
call_id=int(call_id),
channel_id=int(channel_id),
propagation_id=int(propagation_id),
propagation_channel_id=(
self.channels[0]
if int(channel_id) != self.channels[0]
else self.channels[1]
),
author_id=int(author_id),
)
async def is_message_from_call(self, id: int) -> bool:
"""Checks if a message is from a call.
@@ -97,27 +170,55 @@ class Call:
return int(propagation_channel_id), int(propagation_id)
async def set_participant(
self, user_id: int, user_name: str, user_avatar_url: str
) -> Participant:
self, user_id: int, user_name: str, user_avatar_url: str, channel_id: int
) -> tuple[Participant, bool]:
"""Sets a participant in a call.
Args:
user_id (int): The ID of the user to set.
user_name (str): The name of the user to set.
user_avatar_url (str): The avatar URL of the user to set.
channel_id (int): The ID of the channel the user is in.
Returns:
Participant: The participant that was set.
(Participant, bool): The participant that was set and whether they changed their masked
status.
"""
data = orjson.dumps({"name": user_name, "avatar_url": user_avatar_url})
participant = await self.get_participant(user_id)
user = await models.User.get_cache_or_db(self._bot.redis, user_id)
mask = (
masks.get_random_mask(self.masks)
if participant is None
else participant.mask
)
data = orjson.dumps(
{
"name": user_name,
"avatar_url": user_avatar_url,
"mask": mask,
"is_masked": user.is_masked,
"channel_id": channel_id,
}
)
await self._bot.redis.hset(f"calls:{self.id}:participants", user_id, data)
await self._bot.redis.hexpire(f"calls:{self.id}:participants", 3600, user_id)
await self._bot.redis.sadd(f"calls:{self.id}:masks", mask)
await self._bot.redis.expire(f"calls:{self.id}:masks", 3600)
return Participant(
_bot=self._bot, id=user_id, name=user_name, avatar_url=user_avatar_url
)
_bot=self._bot,
id=user_id,
name=user_name,
avatar_url=user_avatar_url,
mask=mask,
is_masked=user.is_masked,
channel_id=channel_id,
user=user,
), (user.is_masked != participant.is_masked if participant else True)
async def get_participant(self, user_id: int) -> Participant | None:
"""Gets a participant in a call.
@@ -137,7 +238,13 @@ class Call:
data = orjson.loads(data)
return Participant(
_bot=self._bot, id=user_id, name=data["name"], avatar_url=data["avatar_url"]
_bot=self._bot,
id=user_id,
name=data["name"],
avatar_url=data["avatar_url"],
mask=data["mask"],
is_masked=data["is_masked"],
channel_id=data["channel_id"],
)
async def get_participants(self) -> list["Participant"]:
@@ -160,6 +267,9 @@ class Call:
id=int(user_id),
name=data["name"],
avatar_url=data["avatar_url"],
mask=data["mask"],
is_masked=data["is_masked"],
channel_id=data["channel_id"],
)
)
@@ -317,7 +427,7 @@ class CallHandler:
return await self.bot.redis.hexists("calls:channels:active", channel_id)
async def get_call(self, channel_id: int) -> Call | None:
async def get_call_by_channel_id(self, channel_id: int) -> Call | None:
"""Gets the call that a channel is in.
Args:
@@ -332,6 +442,20 @@ class CallHandler:
if call_id is None:
return
return await self.get_call(int(call_id))
async def get_call(self, call_id: int) -> Call | None:
"""Gets a call by its ID.
Args:
call_id (int): The ID of the call to get.
Returns:
Call | None: The call that was found.
"""
used_masks = await self.bot.redis.smembers(f"calls:{call_id}:masks")
channels = [
int(c)
for c in (await self.bot.redis.hget("calls:calls:active", call_id)).split(
@@ -339,7 +463,7 @@ class CallHandler:
)
]
return Call(_bot=self.bot, id=call_id, channels=channels)
return Call(_bot=self.bot, id=call_id, channels=channels, masks=used_masks)
async def hangup(self, channel_id: int) -> tuple[int, int] | tuple[int]:
"""Ends a call in a channel.
@@ -437,17 +561,24 @@ class CallHandler:
return webhook.url
async def get_message(self, message_id: int) -> None:
"""Gets a message as part of a call.
async def get_message(self, message_id: int) -> Message | None:
"""Gets a message sent in a call.
Args:
message_id (int): The ID of the message to get.
Returns:
None: The channel and propagation ID of the message.
Message | None: The message that was sent.
"""
return await self.bot.redis.hget("calls:messages", message_id)
call_id = await self.bot.redis.hget("calls:messages", message_id)
if call_id is None:
return
call = await self.get_call(int(call_id))
return await call.get_message(message_id)
async def get_message_author_id(self, id: int) -> int | None:
"""Gets the author of a message as part of a call.
@@ -471,9 +602,9 @@ class CallHandler:
message_id = int(message_id) if message_id else id
author_id = await self.bot.redis.hget("calls:messages:authors", id)
author_id = await self.bot.redis.hget("calls:messages:authors", message_id)
return int(author_id)
return int(author_id) if author_id else None
async def get_original_message_id(self, id: int) -> int | None:
"""
payphone/core/censorship.py
+8 -4
View File
@@ -17,11 +17,12 @@ URL_PATTERN = re.compile(r"https?://[^\s]+", re.IGNORECASE)
INVITES_PATTERN = re.compile(r"(discord)?.gg/[^\s]+", re.IGNORECASE)
def censor(message: str) -> str:
def censor(message: str, is_gif_allowed: bool = False) -> str:
"""Returns the first censored word or pattern found.
Args:
message (str): The message to moderate.
is_gif_allowed (bool, optional): If GIFs are allowed. Defaults to False.
Returns:
str | None: The blocked pattern found as it was found.
@@ -32,10 +33,13 @@ def censor(message: str) -> str:
if PATTERN.match(word):
return word
match = next(URL_PATTERN.finditer(message), None)
for match in URL_PATTERN.finditer(message):
string = message[match.start() : match.end()]
if match is not None:
return message[match.start() : match.end()]
if is_gif_allowed and string.startswith("https://tenor.com/view/"):
continue
return string
match = next(INVITES_PATTERN.finditer(message), None)
payphone/core/checks.py
+44
View File
@@ -0,0 +1,44 @@
from discord.ext import commands
from payphone.core import exceptions
def is_moderator():
"""Checks whether the user is a moderator or not."""
async def predicate(ctx: commands.Context) -> bool:
if ctx.author.bot:
return False
if await ctx.bot.is_moderator(ctx.author.id):
return True
raise exceptions.NotAModeratorError()
return commands.check(predicate)
def is_not_banned():
"""Checks that the user is not banned."""
async def predicate(ctx: commands.Context) -> bool:
if await ctx.bot.is_banned(ctx.author.id):
raise exceptions.BannedError()
return True
return commands.check(predicate)
def is_on_call():
"""Checks that the user is on a call."""
async def predicate(ctx: commands.Context) -> bool:
call = await ctx.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
if call is None:
raise exceptions.NotOnCallError()
return True
return commands.check(predicate)
payphone/core/constants.py
+1
View File
@@ -29,3 +29,4 @@ BRAILE_CHARACTER = "\u2800"
PRIMARY_COLOR = 0xE2D9D0
SUPPORTER_COLOR = 0xAB9EFF
MODERATOR_COLOR = 0xFF8787
MASK_COLOR = 0xFFC868
payphone/core/exceptions.py
+17
View File
@@ -0,0 +1,17 @@
from discord.ext.commands import errors
class NotAModeratorError(errors.CheckFailure):
"""Raised when the user is not a moderator."""
class BannedError(errors.CheckFailure):
"""Raised when the user is banned."""
class ChannelDisabledError(errors.CheckFailure):
"""Raised when the channel is disabled."""
class NotOnCallError(errors.CheckFailure):
"""Raised when the user is not on a call."""
payphone/core/formatting.py
+221
View File
@@ -0,0 +1,221 @@
from datetime import timedelta
import re
def trim(text: str, length: int = 100) -> str:
"""Trims a string if it's too long.
Args:
text (str): The original text.
length (int, optional): The max length of the string before trimming. Defaults to 100.
Returns:
str: The trimmed (or not) text.
"""
if len(text) > length:
return text[:length - 3] + "..."
return text
def text_join(texts: list[str], join_word: str = "and") -> str:
"""Joins a list with commas and the last item with an "and".
Args:
texts (list[str]): The list of texts to concatenate.
join_word (str): The word to use at the end of the joint.
Returns:
str: The formatted concatenation.
"""
if len(texts) > 1:
return f"{", ".join(texts[:-1])} {join_word} {texts[-1]}"
return texts[0]
def monospace_grid(values: list[str], space: int = 14) -> str:
"""Creates a monospace grid.
Args:
values (list[str]): All the values that the list must contain
space (int, optional): The width of each column. Defaults to 14.
Returns:
str: The formatted grid.
"""
return "```" + "".join([v + " " * (space - len(v)) for v in values]) + "```"
def monospace_table(
rows: list[list[str | tuple[str, str]]],
headers: list | None = None,
column_separator: str = "|",
header_separator: str = "-",
intersection: str = "+",
padding: int = 1,
) -> str:
"""Creates a monospace table.
Creates a monospace table. Row columns can be either a string with the column's content, or a
tuple with two items, being the first one aligned to the left and the second one aligned to the
right. Non-string values will be converted to string using `str(value)`.
Args:
rows (list[list[str | tuple[str, str]]]): The rows.
headers (list | None, optional): The headers.
column_separator (str, optional): The column separator. Defaults to "|".
header_separator (str, optional): The header separator. Defaults to "-".
intersection (str, optional): The intersection character where column and row separators
meet.
padding (int, optional): The padding that each column has at its sides.
Returns:
str: The monospace table.
"""
max_sizes = (
[
max(
*(
[
(
len(str(row[i]))
if not isinstance(row[i], tuple)
else (sum(len(str(item)) for item in row[i]) + 1)
)
for row in rows
]
+ ([len(headers[i])] if headers is not None else [])
)
)
for i in range(len(rows[0]))
]
if len(rows) >= 2
else (
[
(
len(str(col))
if not isinstance(col, tuple)
else (sum(len(str(item)) for item in col) + 1)
)
for col in rows[0]
]
if len(rows) > 0
else []
)
)
def format_row(row: list) -> str:
return column_separator.join(
[
(" " * padding)
+ (
(str(col) + (" " * (max_sizes[i] - len(str(col)))))
if not isinstance(col, tuple)
else (
str(col[0])
+ (" " * (max_sizes[i] - len(str(col[0])) - len(str(col[1]))))
+ str(col[1])
)
)
+ (" " * padding)
for i, col in enumerate(row)
]
)
formatted_rows = []
for row in rows:
formatted_rows.append(format_row(row))
table = "\n".join(formatted_rows)
if headers is not None:
header = format_row(headers)
separator = intersection.join(
[header_separator * (size + padding * 2) for size in max_sizes]
)
table = f"{header}\n{separator}\n{table}"
return table
def time_to_text(time_diff: timedelta) -> str:
"""Converts a timedelta to its string representation.
Args:
time_diff (timedelta): The timedelta.
Returns:
str: The string representation.
"""
years = time_diff.days // 365
days = time_diff.days % 365
hours, remainder = divmod(time_diff.seconds, 3600)
minutes, seconds = divmod(remainder, 60)
parts = []
if years > 0:
parts.append(f"{years} year{"s" if years != 1 else ""}")
if days > 0:
parts.append(f"{days} day{"s" if days != 1 else ""}")
if hours > 0:
parts.append(f"{hours} hour{"s" if hours != 1 else ""}")
if minutes > 0:
parts.append(f"{minutes} min{"s" if minutes != 1 else ""}")
if seconds > 0:
parts.append(f"{seconds} sec{"s" if seconds != 1 else ""}")
return " ".join(parts)
def parse_timedelta(text: str) -> timedelta:
"""Parses a timedelta string formatted as 30h 45m to a timedelta object.
Args:
text (str): The formatted string.
Raises:
ValueError: The string is invalid.
Returns:
datetime.timedelta: The timedelta object.
"""
pattern = re.compile(
r"(?:(\d+)y)?\s*" # years
r"(?:(\d+)mo)?\s*" # months
r"(?:(\d+)w)?\s*" # weeks
r"(?:(\d+)d)?\s*" # days
r"(?:(\d+)h)?\s*" # hours
r"(?:(\d+)m)?\s*" # minutes
r"(?:(\d+)s)?" # seconds
)
matches = pattern.match(text.strip())
if not matches:
raise ValueError("Invalid time string format")
years, months, weeks, days, hours, minutes, seconds = matches.groups()
years = int(years) if years else 0
months = int(months) if months else 0
weeks = int(weeks) if weeks else 0
days = int(days) if days else 0
hours = int(hours) if hours else 0
minutes = int(minutes) if minutes else 0
seconds = int(seconds) if seconds else 0
days += weeks * 7
days += months * 30
days += years * 365
return timedelta(days=days, hours=hours, minutes=minutes, seconds=seconds)
payphone/core/help_command.py
+152
View File
@@ -0,0 +1,152 @@
from typing import List, Dict
import discord
from discord.ext import commands
from payphone.cogs import base
from payphone.core import formatting, constants, utils
class HelpView(discord.ui.View):
"""The view sent with the help command."""
def __init__(self):
super().__init__(
discord.ui.Button(
label="Credits",
style=discord.ButtonStyle.gray,
url="https://payphone.nyeki.dev",
),
discord.ui.Button(
label="Donate",
style=discord.ButtonStyle.gray,
url="https://ko-fi.com/payphonedev",
),
discord.ui.Button(
label="Vote",
style=discord.ButtonStyle.gray,
url="https://top.gg/bot/1212940413795827763/vote",
),
discord.ui.Button(
label="Server",
style=discord.ButtonStyle.gray,
url="https://discord.gg/payphone",
),
timeout=None,
)
class HelpCommand(commands.HelpCommand):
async def send_bot_help(
self, mapping: Dict[base.Cog | None, List[commands.Command]]
) -> None:
bot_commands: List[commands.Command] = []
for cog, cog_commands in mapping.items():
if cog is not None and not cog.show_in_help:
continue
for command in cog_commands:
if isinstance(command, commands.Command):
bot_commands += [command]
embed = discord.Embed(
author=self.context.bot.embed_author,
title="Payphone, call and connect with other people.",
description=(
"Here is the list of all available commands. You can see more details about a "
f"specific command with `{self.context.clean_prefix}help <command>`."
),
thumbnail=self.context.bot.user.display_avatar.url,
fields=[
discord.EmbedField(
name="Commands",
value=formatting.monospace_grid(
[c.name for c in bot_commands], space=12
),
)
],
color=constants.PRIMARY_COLOR,
footer=discord.EmbedFooter(
text=f"Use {self.context.clean_prefix}help <command> to get more information about a specific command."
),
)
await self.context.send(embed=embed, view=HelpView())
async def send_command_help(self, command: commands.Command) -> None:
command_name = (
command.name
if isinstance(command, discord.SlashCommand)
else command.qualified_name
)
command_options = "".join(
[f" <{option.name}>" for option in command.options]
if isinstance(command, discord.SlashCommand)
else [
(
f" [{name}]"
if utils.is_optional(info.annotation)
else f" <{name}>"
)
for name, info in command.params.items()
if name not in ["self", "ctx"]
]
)
command_description = (
command.description if command.description else command.help
).replace("{prefix}", self.context.clean_prefix)
full_command_usage = (
self.context.clean_prefix + command.qualified_name + command_options
)
reply_argument = (
command.extras.get("reply_argument", None) if command.extras else None
)
embed = discord.Embed(
author=discord.EmbedAuthor(
name="Payphone", icon_url=self.context.bot.user.display_avatar.url
),
title=(
command.extras["help_name"]
if command.extras and "help_name" in command.extras
else command_name.title()
)
+ " Command",
description=command_description,
thumbnail=self.context.bot.user.display_avatar.url,
color=constants.PRIMARY_COLOR,
fields=[
discord.EmbedField(
name="Usage",
value=f"{f"`┌─ [<{reply_argument}>]`\n" if reply_argument else ""}`{full_command_usage}`",
)
]
+ (
[
discord.EmbedField(
name="Aliases",
value=", ".join(
[
f"`{self.context.clean_prefix}{alias}`"
for alias in command.aliases
]
),
)
]
if len(command.aliases) > 0
else []
),
footer=discord.EmbedFooter(
f"Use {self.context.clean_prefix}help <command> to get more information about a specific command."
),
)
await self.context.send(embed=embed)
async def send_group_help(self, group):
return await self.send_command_help(group)
async def send_error_message(self, error):
await self.context.send(f"❌ {error}")
payphone/core/masks.py
+109
View File
@@ -0,0 +1,109 @@
import random
from payphone.core import constants
MASKS = {
"ferret": 1278176917123760139 if constants.IS_DEBUG else 1278173356423909448,
"octopus": 1278176904410697880 if constants.IS_DEBUG else 1278173331421397072,
"cat": 1278176891290783824 if constants.IS_DEBUG else 1278173317034934282,
"monkey": 1278176881195094049 if constants.IS_DEBUG else 1278173297456185344,
"panda": 1278176868767629455 if constants.IS_DEBUG else 1278173272390762538,
"sheep": 1278176856910073946 if constants.IS_DEBUG else 1278173251914432563,
"horse": 1278176842116894760 if constants.IS_DEBUG else 1278173237582237708,
"bee": 1278176823846506577 if constants.IS_DEBUG else 1278173216652787826,
"wild boar": 1278176806498996235 if constants.IS_DEBUG else 1278173196658413649,
"duck": 1278176794591232102 if constants.IS_DEBUG else 1278173181483679918,
"chicken": 1278176781790216233 if constants.IS_DEBUG else 1278173167218589726,
"elephant": 1278176771413512213 if constants.IS_DEBUG else 1278173150063886346,
"cow": 1278176761565413483 if constants.IS_DEBUG else 1278173125024153733,
"deer": 1278176744691597312 if constants.IS_DEBUG else 1278173110105014406,
"turkey": 1278176730619576321 if constants.IS_DEBUG else 1278173091796877363,
"god": 1278176719932751903 if constants.IS_DEBUG else 1278173070846201856,
"beaver": 1278176709530882091 if constants.IS_DEBUG else 1278173051485425705,
"camel": 1278176699921465394 if constants.IS_DEBUG else 1278173035018584086,
"koala": 1278176687493746709 if constants.IS_DEBUG else 1278173021345021952,
"hippo": 1278176675846295603 if constants.IS_DEBUG else 1278173007419932804,
"ostrich": 1278176661317095494 if constants.IS_DEBUG else 1278172994669248523,
"rhino": 1278176650340864000 if constants.IS_DEBUG else 1278172980848885832,
"squirrel": 1278176637627924570 if constants.IS_DEBUG else 1278172965590274088,
"chamaleon": 1278176625392877648 if constants.IS_DEBUG else 1278172953305022496,
"frog": 1278176614512857088 if constants.IS_DEBUG else 1278172937043710094,
"skunk": 1278176604509442079 if constants.IS_DEBUG else 1278172923495976981,
"walrus": 1278176554723311678 if constants.IS_DEBUG else 1278172912368619603,
"hedgehog": 1278176541750067252 if constants.IS_DEBUG else 1278172899911405678,
"pig": 1278176531415564332 if constants.IS_DEBUG else 1278172885239861338,
"hen": 1278176518643646485 if constants.IS_DEBUG else 1278172871302184994,
"turtle": 1278176507428343891 if constants.IS_DEBUG else 1278172858199314455,
"mouse": 1278175957047578696 if constants.IS_DEBUG else 1278172846442414081,
"rabbit": 1278175944007487489 if constants.IS_DEBUG else 1278172831921999945,
"goat": 1278175933915856938 if constants.IS_DEBUG else 1278172815421472808,
"lion": 1278175922964664382 if constants.IS_DEBUG else 1278172803534684235,
"wolf": 1278175902047535165 if constants.IS_DEBUG else 1278172791052570687,
"snake": 1278175869331968061 if constants.IS_DEBUG else 1278172776259260547,
"llama": 1278175850008674325 if constants.IS_DEBUG else 1278172764913799168,
"crab": 1278175835634925620 if constants.IS_DEBUG else 1278172752322232330,
"owl": 1278175822947029152 if constants.IS_DEBUG else 1278172738309193749,
"bat": 1278175813572886550 if constants.IS_DEBUG else 1278172726426865724,
"zebra": 1278175799962505216 if constants.IS_DEBUG else 1278172715114565753,
"fox": 1278175788818104474 if constants.IS_DEBUG else 1278172702955274292,
"gorilla": 1278175775891259456 if constants.IS_DEBUG else 1278172687797194824,
"sloth": 1278175765661356175 if constants.IS_DEBUG else 1278172676837343242,
"puffer fish": 1278175753497743360 if constants.IS_DEBUG else 1278172665093558294,
"bear": 1278175741569142867 if constants.IS_DEBUG else 1278172653076746301,
"giraffe": 1278175732601983006 if constants.IS_DEBUG else 1278172640296702073,
"penguin": 1278175721956573236 if constants.IS_DEBUG else 1278172627629899808,
"parrot": 1278175706798489642 if constants.IS_DEBUG else 1278172600702337074,
}
def get_random_mask(not_int: list[str] = []) -> str:
"""Returns a random mask.
Args:
not_int (list[str], optional): Masks to exclude. Defaults to [].
Returns:
str: The mask.
"""
return random.choice([mask for mask in MASKS if mask not in not_int])
def get_mask_name(mask: str) -> str:
"""Returns the name of the mask.
Args:
mask (str): The mask.
Returns:
str: The name of the mask.
"""
return mask.title()
def get_mask_image_url(mask: str) -> str:
"""Returns the URL of the mask image.
Args:
mask (str): The mask.
Returns:
str: The URL of the mask image.
"""
return f"https://cdn.discordapp.com/emojis/{MASKS[mask]}.png"
def get_mask_emoji(mask: str) -> str:
"""Returns the emoji of the mask.
Args:
mask (str): The mask.
Returns:
str: The emoji of the mask.
"""
return f"<:{mask.replace(" ", "")}:{MASKS[mask]}>"
payphone/core/utils.py
+25
View File
@@ -0,0 +1,25 @@
import typing
from discord.ext import commands
def is_optional(type_hint) -> bool:
"""Checks if the type_hint is optional or not.
Args:
type_hint (typing.Type): The type hint.
Returns:
bool: Whether the type hint is optional or not.
"""
origin = typing.get_origin(type_hint)
if origin is typing.Union:
args = typing.get_args(type_hint)
return type(None) in args
if isinstance(type_hint, commands.Greedy):
return True
return False
payphone/db/migrations/models/6_20241114014251_update.py
+21
View File
@@ -0,0 +1,21 @@
from tortoise import BaseDBAsyncClient
async def upgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "report" ADD "user_avatar_url" TEXT NOT NULL;
ALTER TABLE "report" ADD "user_name" TEXT NOT NULL;
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
async def downgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "report" DROP COLUMN "user_avatar_url";
ALTER TABLE "report" DROP COLUMN "user_name";
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
payphone/db/migrations/models/7_20241114024101_update.py
+21
View File
@@ -0,0 +1,21 @@
from tortoise import BaseDBAsyncClient
async def upgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "user" ADD "warns_count" INT NOT NULL DEFAULT 0;
ALTER TABLE "user" ADD "bans_count" INT NOT NULL DEFAULT 0;
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
async def downgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "user" DROP COLUMN "warns_count";
ALTER TABLE "user" DROP COLUMN "bans_count";
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
payphone/db/migrations/models/8_20241115225636_update.py
+24
View File
@@ -0,0 +1,24 @@
from tortoise import BaseDBAsyncClient
async def upgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
CREATE TABLE IF NOT EXISTS "server" (
"id" BIGSERIAL NOT NULL PRIMARY KEY,
"prefix" TEXT NOT NULL,
"language" TEXT NOT NULL
);
COMMENT ON TABLE "server" IS 'A server.';;
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
async def downgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
DROP TABLE IF EXISTS "server";
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
payphone/db/migrations/models/9_20241115230145_update.py
+19
View File
@@ -0,0 +1,19 @@
from tortoise import BaseDBAsyncClient
async def upgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "server" ADD "name" TEXT NOT NULL;
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
async def downgrade(db: BaseDBAsyncClient) -> str:
return """
DROP INDEX "idx_channel_server__d940e4";
DROP INDEX "idx_ban_user_id_37a9cf";
ALTER TABLE "server" DROP COLUMN "name";
CREATE INDEX "idx_ban_user_id_37a9cf" ON "ban" ("user_id", "expires_at");;
CREATE INDEX "idx_channel_server__d940e4" ON "channel" ("server_id", "is_enabled");;"""
payphone/db/models.py
+100 -2
View File
@@ -4,6 +4,10 @@ import enum
from tortoise import models, fields
import redis.asyncio as redis
import orjson
class Report(models.Model):
"""
@@ -14,6 +18,8 @@ class Report(models.Model):
message_content = fields.TextField()
channel_id = fields.BigIntField()
user_id = fields.BigIntField()
user_name = fields.TextField()
user_avatar_url = fields.TextField()
created_at = fields.DatetimeField(auto_now_add=True)
resolved = fields.BooleanField(default=False, index=True)
resolved_at = fields.DatetimeField(null=True)
@@ -37,12 +43,15 @@ class User(models.Model):
id = fields.BigIntField(pk=True)
warns_count: int = fields.IntField(default=0)
bans_count: int = fields.IntField(default=0)
is_masked = fields.BooleanField(default=False)
is_moderator = fields.BooleanField(default=False)
is_supporter_until = fields.DatetimeField(null=True)
@property
def flags(self) -> UserFlag:
def flag(self) -> UserFlag:
"""
The flags of the user.
"""
@@ -60,6 +69,69 @@ class User(models.Model):
return flags
@classmethod
async def get_cache_or_db(cls, redis_client: redis.Redis, user_id: int) -> "User":
"""
Returns the user from the cache or the database.
Args:
redis_client (redis.Redis): The Redis client.
user_id (int): The user ID.
Returns:
(User): The user.
"""
cache = await redis_client.get(f"user:{user_id}")
instance = User(id=user_id)
if cache is not None:
cache = orjson.loads(cache)
instance.warns_count = cache["warns_count"]
instance.bans_count = cache["bans_count"]
instance.is_masked = cache["is_masked"]
instance.is_moderator = cache["is_moderator"]
instance.is_supporter_until = (
datetime.fromisoformat(cache["is_supporter_until"])
if cache["is_supporter_until"] is not None
else None
)
else:
instance, _ = await User.update_or_create(id=user_id)
await instance.set_cache(redis_client)
return instance
async def set_cache(self, redis_client: redis.Redis) -> None:
"""
Sets the user in the cache.
Args:
redis_client (redis.Redis): The Redis client.
"""
await redis_client.set(
f"user:{self.id}",
orjson.dumps(
{
"warns_count": self.warns_count,
"bans_count": self.bans_count,
"is_masked": self.is_masked,
"is_moderator": self.is_moderator,
"is_supporter_until": (
self.is_supporter_until.timestamp()
if self.is_supporter_until
else None
),
}
),
ex=60 * 15,
)
class Ban(models.Model):
"""
@@ -91,7 +163,7 @@ class Channel(models.Model):
id = fields.BigIntField(pk=True)
server_id = fields.BigIntField(index=True)
is_enabled = fields.BooleanField(default=True)
is_enabled = fields.BooleanField(null=True)
class Meta:
"""
@@ -101,3 +173,29 @@ class Channel(models.Model):
indexes = [
models.Index(fields=["server_id", "is_enabled"]),
]
class Server(models.Model):
"""
A server.
"""
id = fields.BigIntField(pk=True)
name = fields.TextField()
prefix = fields.TextField(default="p.")
language = fields.TextField(default="en")
async def set_prefix(self, redis_client: redis.Redis, prefix: str) -> None:
"""
Sets the prefix of the server.
Args:
redis_client (redis.Redis): The Redis client.
prefix (str): The prefix.
"""
self.prefix = prefix
await self.save()
await redis_client.set(f"prefix:{self.id}", prefix, ex=60 * 60 * 24)