499 lines
17 KiB
Python
499 lines
17 KiB
Python
import asyncio
|
|
|
|
import random
|
|
|
|
import discord
|
|
from discord.ext import commands
|
|
|
|
import aiohttp
|
|
|
|
from payphone.db import models
|
|
from payphone.bot import Payphone
|
|
from payphone.cogs import base
|
|
from payphone.core import constants, censorship, messages, masks, checks
|
|
|
|
|
|
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"])
|
|
@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."
|
|
)
|
|
|
|
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."
|
|
)
|
|
|
|
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"])
|
|
@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.
|
|
"""
|
|
|
|
ctx = await self.bot.get_context(message)
|
|
|
|
if message.author.bot or ctx.valid:
|
|
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,
|
|
)
|
|
|
|
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:
|
|
return await asyncio.gather(
|
|
message.channel.send(
|
|
(
|
|
"❌ Your message wasn't sent because it contained a blocked word or "
|
|
f"pattern: `{blocked_section}`"
|
|
),
|
|
delete_after=10,
|
|
),
|
|
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 `Apps > "
|
|
"Report Message`."
|
|
),
|
|
),
|
|
)
|
|
|
|
message_content = self.bot.clean_message(message, call)
|
|
|
|
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>)."
|
|
)
|
|
|
|
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
|
|
):
|
|
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 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 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.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:
|
|
webhook = discord.Webhook.from_url(
|
|
await self.bot.call_handler.get_channel_webhook_url(other_channel),
|
|
session=session,
|
|
bot_token=constants.TOKEN,
|
|
)
|
|
|
|
propagated_msg = await webhook.send(
|
|
content=message_content,
|
|
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,
|
|
)
|
|
|
|
@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,
|
|
)
|
|
|
|
await webhook.delete_message(propagation_id)
|
|
|
|
@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:
|
|
return (
|
|
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,
|
|
)
|
|
)
|
|
|
|
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")
|
|
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
|
|
)
|
|
|
|
author_id = await self.bot.call_handler.get_message_author_id(message.id)
|
|
|
|
if author_id is None:
|
|
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)
|
|
|
|
message_id: int
|
|
|
|
if message.author.bot:
|
|
message_id = await self.bot.call_handler.get_original_message_id(message.id)
|
|
|
|
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,
|
|
},
|
|
)
|
|
|
|
|
|
def setup(bot: Payphone) -> None:
|
|
"""Loads the Calls cog into the bot.
|
|
|
|
Args:
|
|
bot (Payphone): The bot instance.
|
|
"""
|
|
|
|
bot.add_cog(Calls(bot))
|