675 lines
24 KiB
Python
675 lines
24 KiB
Python
import asyncio
|
|
|
|
import random
|
|
|
|
import discord
|
|
from discord.ext import commands
|
|
|
|
import aiohttp
|
|
|
|
import sentry_sdk
|
|
|
|
from payphone.db import models
|
|
from payphone.bot import Payphone
|
|
from payphone.cogs import base
|
|
from payphone.core import constants, censorship, messages, masks, checks, formatting
|
|
|
|
|
|
class Calls(base.Cog):
|
|
"""Calls in Payphone."""
|
|
|
|
def __init__(self, bot: Payphone) -> None:
|
|
"""Initializes the Calls cog.
|
|
|
|
Args:
|
|
bot (Payphone): The bot instance.
|
|
"""
|
|
|
|
self.bot = bot
|
|
|
|
@commands.command(aliases=["c"])
|
|
@commands.bot_has_permissions(
|
|
send_messages=True,
|
|
read_messages=True,
|
|
read_message_history=True,
|
|
use_external_emojis=True,
|
|
manage_webhooks=True,
|
|
manage_messages=True,
|
|
)
|
|
@checks.is_not_banned()
|
|
async def call(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Call and connect with other people.
|
|
"""
|
|
|
|
if await self.bot.call_handler.is_on_call(ctx.channel.id):
|
|
return await ctx.send(
|
|
f"☎️ You're already on a call! Use `{ctx.clean_prefix}hangup` to hang up."
|
|
)
|
|
|
|
if await self.bot.call_handler.is_searching(ctx.channel.id):
|
|
return await ctx.send(
|
|
f"☎️ You're already searching for a call. Use `{ctx.clean_prefix}hangup` to hang up the phone."
|
|
)
|
|
|
|
await self.bot.call_handler.set_searching(ctx.channel.id, True)
|
|
|
|
await ctx.send(
|
|
(
|
|
f"**TIP**: {random.choice(messages.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
|
|
f"☎️ {random.choice(messages.CALLING)}"
|
|
)
|
|
)
|
|
|
|
@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.
|
|
"""
|
|
|
|
if not await self.bot.call_handler.is_on_call(ctx.channel.id):
|
|
return await ctx.send(
|
|
f"☎️ You're not on a call! Use `{ctx.clean_prefix}call` to start one."
|
|
)
|
|
|
|
call = await self.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
|
|
|
|
participant = await call.get_participant(ctx.author.id)
|
|
|
|
if participant is None:
|
|
return await ctx.send(
|
|
"❌ You cannot skip a call you have not participated in.\n"
|
|
"-# You didn't send any messages in this call, so you can't skip it."
|
|
)
|
|
|
|
channels = await self.bot.call_handler.skip(ctx.channel.id)
|
|
|
|
for channel in channels:
|
|
if channel == ctx.channel.id:
|
|
await ctx.send(
|
|
(
|
|
f"**TIP**: {random.choice(messages.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
|
|
"☎️ You skipped the call. Searching for another one..."
|
|
)
|
|
)
|
|
|
|
else:
|
|
await self.bot.get_partial_messageable(channel).send(
|
|
f"☎️ {random.choice(messages.OTHER_PARTY_HANGUP)}"
|
|
)
|
|
|
|
@commands.command(aliases=["h"])
|
|
@checks.is_not_banned()
|
|
async def hangup(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Hang up the call.
|
|
"""
|
|
|
|
if not await self.bot.call_handler.is_on_call(ctx.channel.id):
|
|
if await self.bot.call_handler.is_searching(ctx.channel.id):
|
|
return await ctx.send(f"☎️ {random.choice(messages.NO_ANSWER)}")
|
|
|
|
return await ctx.send(
|
|
f"☎️ You're not on a call! Use `{ctx.clean_prefix}call` to start one."
|
|
)
|
|
|
|
call = await self.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
|
|
|
|
participant = await call.get_participant(ctx.author.id)
|
|
|
|
if participant is None:
|
|
return await ctx.send(
|
|
"❌ You cannot hang up a call you have not participated in.\n"
|
|
"-# You didn't send any messages in this call, so you can't hang up. Try saying hi!"
|
|
)
|
|
|
|
channels = await self.bot.call_handler.hangup(ctx.channel.id)
|
|
|
|
other_channel = channels[0] if channels[0] != ctx.channel.id else channels[1]
|
|
|
|
await asyncio.gather(
|
|
ctx.send(f"☎️ {random.choice(messages.HANGUP)}"),
|
|
self.bot.get_partial_messageable(other_channel).send(
|
|
f"☎️ {random.choice(messages.OTHER_PARTY_HANGUP)}"
|
|
),
|
|
)
|
|
|
|
@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_by_channel_id(ctx.channel.id)
|
|
|
|
if call is None:
|
|
return await ctx.send(
|
|
f"☎️ You're not on a call! Use `{ctx.clean_prefix}call` to start one."
|
|
)
|
|
|
|
other_channel = (
|
|
call.channels[0] if call.channels[0] != ctx.channel.id else call.channels[1]
|
|
)
|
|
|
|
await asyncio.gather(
|
|
self.bot.get_partial_messageable(other_channel).send(
|
|
f"📬 **{ctx.author.display_name}** wants you to send them a friend request on Discord! Their username is `{ctx.author.name}`."
|
|
),
|
|
ctx.send("📬 Your friend request has been sent!"),
|
|
)
|
|
|
|
@commands.command(aliases=["m"])
|
|
@commands.bot_has_permissions(
|
|
send_messages=True,
|
|
use_external_emojis=True,
|
|
)
|
|
@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:
|
|
if participant is not None:
|
|
message += f" You're currently being displayed as **{masks.get_mask_emoji(participant.mask)} {masks.get_mask_name(participant.mask)}**."
|
|
|
|
else:
|
|
message += " Participate in the call to see your mask."
|
|
|
|
message += (
|
|
" You can toggle your mask on or off using `p.mask on` and `p.mask off`."
|
|
)
|
|
|
|
return await ctx.send(message)
|
|
|
|
@commands.command(aliases=["r"])
|
|
@commands.bot_has_permissions(
|
|
send_messages=True,
|
|
)
|
|
@checks.is_not_banned()
|
|
async def report(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Report a message in a call.
|
|
"""
|
|
|
|
embed = discord.Embed(
|
|
color=constants.MASK_COLOR,
|
|
title="Reporting has been moved!",
|
|
description="📝 This command has been replaced! Payphone has switched to per-message reporting due to the high amount of reports.",
|
|
fields=[
|
|
discord.EmbedField(
|
|
name="Reporting on Desktop",
|
|
value="Right click the message you want to report and, in the opened menu, press `Apps > Report Message`.",
|
|
),
|
|
discord.EmbedField(
|
|
name="Reporting on Mobile",
|
|
value="Press and hold the message you want to report and, in the opened menu, press `Apps > Report Message`.",
|
|
),
|
|
],
|
|
)
|
|
|
|
await ctx.send(embed=embed)
|
|
|
|
@commands.Cog.listener()
|
|
async def on_message(self, message: discord.Message) -> None:
|
|
"""
|
|
Handle messages in calls.
|
|
"""
|
|
|
|
if not isinstance(message.channel, (discord.channel.TextChannel)):
|
|
return
|
|
|
|
ctx = await self.bot.get_context(message)
|
|
|
|
if message.author.bot or ctx.valid or message.author.system:
|
|
return
|
|
|
|
if message.type not in [discord.MessageType.default, discord.MessageType.reply]:
|
|
return
|
|
|
|
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
|
|
|
|
participant, was_mask_toggled = await call.set_participant(
|
|
message.author.id,
|
|
message.author.display_name,
|
|
message.author.display_avatar.url,
|
|
message.channel.id,
|
|
)
|
|
|
|
if participant.display_name.lower() in [
|
|
"everyone",
|
|
"here",
|
|
"wumpus",
|
|
"discord",
|
|
]:
|
|
await message.reply(
|
|
f"❌ You can't use {participant.display_name} as your name in calls.\n-# Restriction imposed by Discord."
|
|
)
|
|
return
|
|
|
|
other_channel = (
|
|
call.channels[0]
|
|
if call.channels[0] != message.channel.id
|
|
else call.channels[1]
|
|
)
|
|
|
|
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:
|
|
await message.channel.send(
|
|
(
|
|
"❌ Your message wasn't sent because it contained a blocked word or "
|
|
f"pattern: `{blocked_section}`"
|
|
),
|
|
delete_after=10,
|
|
)
|
|
await self.bot.get_partial_messageable(other_channel).send(
|
|
(
|
|
f"❌ **{message.author.display_name}** sent a message that was blocked.\n"
|
|
"-# If you see an inappropiate message, please report it using `Message > "
|
|
"Apps > Report Message`."
|
|
),
|
|
)
|
|
return
|
|
|
|
message_content = self.bot.clean_message(message)
|
|
|
|
notices = []
|
|
|
|
if message.stickers:
|
|
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:
|
|
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>)."
|
|
)
|
|
|
|
if message.attachments:
|
|
if participant.user.flag & models.UserFlag.MODERATOR:
|
|
for attachment in message.attachments:
|
|
message_content += (
|
|
f"[{constants.BRAILE_CHARACTER}]({attachment.url})"
|
|
)
|
|
|
|
else:
|
|
await message.channel.send(
|
|
f"❌ **{participant.display_name}** You can't send attachments in calls."
|
|
)
|
|
|
|
if not message_content:
|
|
return
|
|
|
|
notices.append("The original message contained attachments.")
|
|
|
|
embeds = []
|
|
|
|
if message.reference:
|
|
if await call.is_message_from_call(message.reference.message_id):
|
|
replied_message = message.reference.cached_message
|
|
|
|
if replied_message is None:
|
|
replied_message = await message.channel.fetch_message(
|
|
message.reference.message_id
|
|
)
|
|
|
|
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
|
|
):
|
|
try:
|
|
replied_message = await self.bot.get_partial_messageable(
|
|
other_channel
|
|
).fetch_message(original_message_id)
|
|
|
|
except discord.errors.NotFound:
|
|
replied_message = None
|
|
|
|
except discord.errors.Forbidden:
|
|
replied_message = None
|
|
|
|
await self.bot.get_partial_messageable(other_channel).send(
|
|
"❌ **Payphone** is missing permissions to read messages from this "
|
|
"channel. Make sure it's got `Read Message History` permissions."
|
|
)
|
|
|
|
if replied_message is not None:
|
|
replied_message_content = self.bot.clean_message(replied_message)
|
|
replied_participant = await call.get_participant(
|
|
replied_message.author.id
|
|
)
|
|
|
|
if replied_participant is None:
|
|
notices.append("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:
|
|
notices.append("In reply to a message from outside this call.")
|
|
|
|
else:
|
|
notices.append("In reply to a message from outside this call.")
|
|
|
|
if was_mask_toggled:
|
|
if participant.is_masked:
|
|
embeds.append(
|
|
discord.Embed(
|
|
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."
|
|
),
|
|
)
|
|
)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
for _ in range(3):
|
|
try:
|
|
webhook = discord.Webhook.from_url(
|
|
await self.bot.call_handler.get_channel_webhook_url(
|
|
other_channel
|
|
),
|
|
session=session,
|
|
bot_token=constants.TOKEN,
|
|
)
|
|
|
|
notices_text = (
|
|
("\n-# ⓘ " + "\n-# ⓘ ".join(notices)) if notices else ""
|
|
)
|
|
|
|
propagated_msg = await webhook.send(
|
|
content=formatting.trim(
|
|
message_content, 2000 - len(notices_text)
|
|
)
|
|
+ notices_text,
|
|
username=participant.display_name,
|
|
avatar_url=participant.display_avatar_url,
|
|
embeds=embeds,
|
|
wait=True,
|
|
)
|
|
|
|
await call.set_message(
|
|
message.id,
|
|
message.channel.id,
|
|
propagated_msg.id,
|
|
message.author.id,
|
|
)
|
|
|
|
break
|
|
|
|
except discord.errors.NotFound:
|
|
await self.bot.call_handler.refresh_channel_webhook_url(
|
|
other_channel
|
|
)
|
|
|
|
except discord.errors.Forbidden:
|
|
await self.bot.get_partial_messageable(other_channel).send(
|
|
"❌ **Payphone** is missing `Manage Webhooks` permissions. You won't see any messages while that permission is not set."
|
|
)
|
|
break
|
|
|
|
except discord.errors.DiscordException as exc:
|
|
sentry_sdk.capture_exception(exc)
|
|
|
|
@commands.Cog.listener()
|
|
async def on_raw_message_delete(
|
|
self, payload: discord.RawMessageDeleteEvent
|
|
) -> None:
|
|
"""
|
|
Handle deleted messages in calls.
|
|
"""
|
|
|
|
call = await self.bot.call_handler.get_call_by_channel_id(payload.channel_id)
|
|
|
|
if call is None:
|
|
return
|
|
|
|
propagation = await call.get_message_propagation(payload.message_id)
|
|
|
|
if propagation is None:
|
|
return
|
|
|
|
propagation_channel, propagation_id = propagation
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
webhook = discord.Webhook.from_url(
|
|
await self.bot.call_handler.get_channel_webhook_url(
|
|
propagation_channel
|
|
),
|
|
session=session,
|
|
bot_token=constants.TOKEN,
|
|
)
|
|
|
|
try:
|
|
await webhook.delete_message(propagation_id)
|
|
|
|
except discord.errors.NotFound:
|
|
pass
|
|
|
|
@commands.Cog.listener()
|
|
async def on_raw_message_edit(self, payload: discord.RawMessageUpdateEvent) -> None:
|
|
"""
|
|
Handle edited messages in calls.
|
|
"""
|
|
|
|
call = await self.bot.call_handler.get_call_by_channel_id(payload.channel_id)
|
|
|
|
if call is None:
|
|
return
|
|
|
|
blocked_section = censorship.censor(payload.data["content"])
|
|
|
|
if blocked_section:
|
|
try:
|
|
await self.bot.get_partial_messageable(
|
|
payload.channel_id
|
|
).get_partial_message(payload.message_id).reply(
|
|
(
|
|
"❌ Your message wasn't sent because it contained a blocked word or "
|
|
f"pattern: `{blocked_section}`"
|
|
),
|
|
delete_after=10,
|
|
)
|
|
|
|
except discord.errors.Forbidden:
|
|
await self.bot.get_partial_messageable(payload.channel_id).send(
|
|
(
|
|
f"❌ **{payload.data["member"]["nick"] or payload.data["member"]["user"]["global_name"] or payload.data["member"]["user"]["username"]}** "
|
|
"your message wasn't sent because it contained a blocked message or "
|
|
f"pattern: `{blocked_section}`\n-# Payphone doesn't have read message history "
|
|
"permission? Make sure to keep that one enabled."
|
|
),
|
|
delete_after=10,
|
|
)
|
|
|
|
return
|
|
|
|
propagation = await call.get_message_propagation(payload.message_id)
|
|
|
|
if propagation is None:
|
|
return
|
|
|
|
propagation_channel, propagation_id = propagation
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
webhook = discord.Webhook.from_url(
|
|
await self.bot.call_handler.get_channel_webhook_url(
|
|
propagation_channel
|
|
),
|
|
session=session,
|
|
bot_token=constants.TOKEN,
|
|
)
|
|
|
|
await webhook.edit_message(propagation_id, content=payload.data["content"])
|
|
|
|
@commands.message_command(
|
|
name="Report Message" if not constants.IS_DEBUG else "(Dev) Report Message",
|
|
guild_ids=[constants.PAYPHONE_HUB_ID] if not constants.IS_DEBUG else None,
|
|
)
|
|
async def report_message(
|
|
self, ctx: discord.ApplicationContext, message: discord.Message
|
|
) -> None:
|
|
"""
|
|
Report a message in a call.
|
|
"""
|
|
|
|
if await self.bot.is_banned(ctx.author.id):
|
|
return await ctx.send_response(
|
|
"❌ You're banned from using Payphone.", ephemeral=True
|
|
)
|
|
|
|
if message.author.bot:
|
|
message_id = await self.bot.call_handler.get_original_message_id(message.id)
|
|
author_id = await self.bot.call_handler.get_message_author_id(message_id)
|
|
found_message = bool(author_id)
|
|
|
|
else:
|
|
message_id = message.id
|
|
author_id = message.author.id
|
|
found_message = bool(await self.bot.call_handler.get_message(message.id))
|
|
|
|
if not found_message:
|
|
return await ctx.send_response(
|
|
(
|
|
"❌ More than 15 minutes have passed since this message was sent.\n"
|
|
"-# You'll also see this error if the message wasn't sent during a call."
|
|
),
|
|
ephemeral=True,
|
|
)
|
|
|
|
await ctx.send_response("✅ The message has been reported.", ephemeral=True)
|
|
|
|
replying_to = (
|
|
await self.bot.get_partial_messageable(
|
|
message.reference.channel_id
|
|
).fetch_message(message.reference.message_id)
|
|
if message.reference and not message.reference.cached_message
|
|
else (
|
|
message.reference.cached_message
|
|
if message.reference and message.reference.cached_message
|
|
else None
|
|
)
|
|
)
|
|
|
|
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,
|
|
"reply_message_content": (
|
|
replying_to.clean_content if replying_to else None
|
|
),
|
|
"reply_user_id": (replying_to.author.id if replying_to else None),
|
|
"reply_user_name": (
|
|
replying_to.author.display_name if replying_to else None
|
|
),
|
|
"reply_user_avatar_url": (
|
|
replying_to.author.display_avatar.url if replying_to else None
|
|
),
|
|
"reported_by_id": ctx.author.id,
|
|
},
|
|
)
|
|
|
|
@commands.command(
|
|
extras={
|
|
"show_in_help": False,
|
|
"help_name": "Cat Mask",
|
|
"help_error_message": "No.",
|
|
}
|
|
)
|
|
async def catmask(self, ctx: commands.Context) -> None:
|
|
"""First easter egg. How'd you find it?"""
|
|
|
|
await ctx.message.delete()
|
|
|
|
call = await self.bot.call_handler.get_call_by_channel_id(ctx.channel.id)
|
|
|
|
if not call:
|
|
return
|
|
|
|
participant, _ = await call.set_participant(
|
|
ctx.author.id,
|
|
ctx.author.display_name,
|
|
ctx.author.display_avatar.url,
|
|
ctx.channel.id,
|
|
)
|
|
|
|
await participant.set_mask(masks.get_random_mask(call.masks, masks.CAT_MASKS))
|
|
|
|
|
|
def setup(bot: Payphone) -> None:
|
|
"""Loads the Calls cog into the bot.
|
|
|
|
Args:
|
|
bot (Payphone): The bot instance.
|
|
"""
|
|
|
|
bot.add_cog(Calls(bot))
|