Files
payphone/payphone/cogs/calls/cog.py
T

1237 lines
39 KiB
Python

from datetime import datetime, timezone, timedelta
import secrets
import discord
from discord import errors
from discord.ext import commands, tasks
from discord.utils import find
from tortoise import transactions, exceptions
import asyncpg.exceptions
import aiohttp
import orjson
from payphone import bot
from payphone.db import utils as dbutils, models
from payphone.cogs import base
from payphone.cogs.calls import embeds
from payphone.core import constants, checks, censorship, formatting, utils
from payphone.core.redis import redis
class Calls(base.Cog):
def __init__(self, bot_instance: bot.Payphone):
self.bot = bot_instance
self.hangup_calls.add_exception_type(exceptions.ConfigurationError)
# self.hangup_calls.start()
self.flush_messages.add_exception_type(exceptions.ConfigurationError)
self.flush_messages.add_exception_type(TypeError)
self.flush_messages.add_exception_type(RuntimeError)
self.flush_messages.start()
def cog_unload(self) -> None:
self.hangup_calls.cancel()
return super().cog_unload()
@tasks.loop(seconds=15)
async def hangup_calls(self) -> None:
"""Hangs up inactive calls.
Hangs up inactive calls. Calls are hung up when they either:
- Have had no messages and more than one minute has passed since the start of the call.
- Have had no messages for the past 15 minutes.
In case they are hung up, both channels are attempted to be notified. In case of failure,
that channel is not notified and hung up silently.
"""
calls = await models.Call.filter(ended_at=None).prefetch_related("channels")
hung_up_channel_ids = []
for call in calls:
message = (
await models.Message.filter(call=call).order_by("-id").limit(1).first()
)
if (
message is None
and call.started_at < datetime.now(timezone.utc) - timedelta(minutes=1)
) or (
message is not None
and message.created_at
< datetime.now(timezone.utc) - timedelta(minutes=15)
):
for channel in call.channels:
hung_up_channel_ids.append(channel.id)
try:
await self.bot.get_partial_messageable(channel.id).send(
"☎️ The call was hung up because nobody replied."
)
except errors.DiscordException:
pass
hung_up_channel_ids = [id for id in hung_up_channel_ids if id is not None]
if len(hung_up_channel_ids) > 0:
await models.Channel.filter(id__in=hung_up_channel_ids).update(call_id=None)
await redis.srem(constants.CHANNELS_ONCALL_SET, *hung_up_channel_ids)
@tasks.loop(seconds=15)
async def flush_messages(self, all_calls: bool = False) -> None:
"""Fetches all the pending messages from Redis and flushes them to the DB.
Args:
all_calls (bool, optional): Whether to flush the messages for all calls of all shards.
"""
call_ids = await redis.smembers(constants.CALL_IDS_TO_FLUSH)
# Redis returns bytes.
call_ids = [
int(call_id)
for call_id in call_ids
if all_calls
or (int(call_id) % self.bot.shard_count) - self.bot.shard_ids[0] == 0
]
if len(call_ids) == 0:
return
await redis.srem(constants.CALL_IDS_TO_FLUSH, *call_ids)
pipe = redis.pipeline()
for call_id in call_ids:
pipe.hgetall(constants.CALL_MESSAGES_TEMPLATE % call_id)
pipe.hgetall(constants.CALL_MESSAGE_CONTENTS_TEMPLATE % call_id)
for call_id, (messages, message_contents) in zip(
call_ids, utils.chunks(await pipe.execute(), 2)
):
message_rows: list[models.Message] = []
for message_json in messages.values():
message_rows.append(models.Message(**orjson.loads(message_json)))
message_content_rows: list[models.MessageContent] = []
for message_content_json in message_contents.values():
message_content_rows.append(
models.MessageContent(**orjson.loads(message_content_json))
)
message_rows.sort(key=lambda m: m.id)
message_content_rows.sort(key=lambda mc: mc.message_id)
pipe = redis.pipeline()
pipe.delete(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % call_id,
constants.CALL_MESSAGE_CONTENTS_TEMPLATE % call_id,
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
constants.CALL_MESSAGES_TEMPLATE % call_id,
constants.CALL_MESSAGE_REPLIES_TEMPLATE % call_id,
)
await pipe.execute()
for _ in range(3):
try:
await models.Message.bulk_create(message_rows, batch_size=20)
break
except (
asyncpg.exceptions.PostgresError,
asyncpg.exceptions.InterfaceError,
):
continue
for _ in range(3):
try:
await models.MessageContent.bulk_create(
message_content_rows, batch_size=20
)
break
except (
asyncpg.exceptions.PostgresError,
asyncpg.exceptions.InterfaceError,
):
continue
@commands.command(aliases=["c"])
@commands.guild_only()
@checks.is_unbanned()
@transactions.atomic()
async def call(self, ctx: commands.Context) -> None:
"""
Call and connect with other people.
"""
if ctx.channel.is_nsfw():
await ctx.send("❌ Calls cannot be started in NSFW channels.")
return
user_id = ctx.author.id
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
if is_on_call:
await ctx.reply(
f"☎️ You're already on a call! Use `{ctx.clean_prefix}hangup` to hangup."
)
return
is_searching = await redis.sismember(
constants.CHANNELS_SEARCH_QUEUE, str(channel_id)
)
if is_searching:
await ctx.reply(
f"☎️ You're already searching for a call! Use `{ctx.clean_prefix}hangup` to stop searching."
)
return
last_channel_calls = [
i[1]
for i in (
await models.CallUserChannelRelationship.filter(channel_id=channel_id)
.limit(5)
.distinct()
.order_by("-id")
.values_list("id", "call_id")
)
]
last_user_calls = [
i[1]
for i in (
await models.CallUserChannelRelationship.filter(user_id=user_id)
.limit(5)
.limit(5)
.distinct()
.order_by("-id")
.values_list("id", "call_id")
)
]
last_calls = set(last_user_calls + last_channel_calls)
last_calls_rss = await models.CallUserChannelRelationship.filter(
call_id__in=last_calls
)
last_calls_rss_channels_map = {}
for call_rs in last_calls_rss:
if call_rs.user_id == user_id or call_rs.channel_id == channel_id:
last_calls_rss_channels_map[call_rs.call_id] = call_rs.channel_id
last_calls_other_channels_ids = set()
for call_rs in last_calls_rss:
if last_calls_rss_channels_map[call_rs.call_id] != call_rs.channel_id:
last_calls_other_channels_ids.add(call_rs.channel_id)
other_channel_id: int = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
i = 0
while True:
# For some weird reason, sometimes it returns None and sometimes b"None".
if other_channel_id is None or other_channel_id == b"None":
other_channel_id = None
break
# Redis returns the ID as a bytes string.
other_channel_id = int(other_channel_id)
if i == 5:
break
i += 1
if other_channel_id in last_calls_other_channels_ids:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
other_channel_id = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
if other_channel_id is None:
break
continue
break
channel = await dbutils.get_channel(ctx.channel)
if other_channel_id is not None:
await redis.sadd(
constants.CHANNELS_ONCALL_SET, str(channel_id), str(other_channel_id)
)
msg_phone_answered = f"☎️ {secrets.choice(constants.MSGS_PHONE_ANSWERED)}"
try:
await self.bot.get_partial_messageable(other_channel_id).send(
msg_phone_answered
)
call = await models.Call.create(is_nsfw=ctx.channel.is_nsfw())
user, profile = await dbutils.get_user(ctx.author)
await models.CallUserChannelRelationship.create(
call=call,
channel_id=channel_id,
user=user,
profile=profile,
is_masked=user.is_masked,
mask=secrets.choice(constants.MASKS_LIST).value,
)
await models.Channel.filter(
id__in=[other_channel_id, channel.id]
).update(call=call)
except errors.DiscordException:
await redis.srem(
constants.CHANNELS_ONCALL_SET,
str(channel_id),
str(other_channel_id),
)
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
msg_phone_answered = ""
await ctx.send(
(
f"**TIP**: {secrets.choice(constants.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
f"☎️ {secrets.choice(constants.MSGS_PHONE_CALLING)}\n"
)
+ msg_phone_answered
)
else:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await ctx.send(
(
f"**TIP**: {secrets.choice(constants.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
f"☎️ {secrets.choice(constants.MSGS_PHONE_CALLING)}\n"
)
)
@commands.command(aliases=["s"])
@commands.guild_only()
@checks.is_unbanned()
@transactions.atomic()
async def skip(self, ctx: commands.Context) -> None:
"""
Hang up the current call and search for another.
"""
if ctx.channel.is_nsfw():
await ctx.send("❌ Calls cannot be started in NSFW channels.")
return
user_id = ctx.author.id
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
if not is_on_call:
await ctx.send("❌ You're not in a call!")
return
call = await models.Call.get(channels__id=channel_id).prefetch_related(
"channels"
)
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
call.ended_at = datetime.now(timezone.utc)
await call.save()
channel = find(lambda c: c.id == channel_id, call.channels)
other_channel = find(lambda c: c.id != channel_id, call.channels)
if other_channel is not None:
other_channel.call = None
await other_channel.save()
try:
await self.bot.get_partial_messageable(other_channel.id).send(
f"☎️ {secrets.choice(constants.MSGS_PHONE_HANGUP_YOU)}"
)
except errors.DiscordException:
pass
await redis.srem(
constants.CHANNELS_ONCALL_SET,
*(
[str(channel_id)]
+ ([str(other_channel.id)] if other_channel is not None else [])
),
)
last_channel_calls = [
i[1]
for i in (
await models.CallUserChannelRelationship.filter(channel_id=channel_id)
.limit(5)
.distinct()
.order_by("-id")
.values_list("id", "call_id")
)
]
last_user_calls = [
i[1]
for i in (
await models.CallUserChannelRelationship.filter(user_id=user_id)
.limit(5)
.limit(5)
.distinct()
.order_by("-id")
.values_list("id", "call_id")
)
]
last_calls = set(last_user_calls + last_channel_calls)
last_calls_rss = await models.CallUserChannelRelationship.filter(
call_id__in=last_calls
)
last_calls_rss_channels_map = {}
for call_rs in last_calls_rss:
if call_rs.user_id == user_id or call_rs.channel_id == channel_id:
last_calls_rss_channels_map[call_rs.call_id] = call_rs.channel_id
last_calls_other_channels_ids = set()
for call_rs in last_calls_rss:
if last_calls_rss_channels_map[call_rs.call_id] != call_rs.channel_id:
last_calls_other_channels_ids.add(call_rs.channel_id)
other_channel_id: int = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
i = 0
while True:
# For some weird reason, sometimes it returns None and sometimes b"b'None'".
if other_channel_id is None or other_channel_id == b"b'None'":
other_channel_id = None
break
# Redis returns the ID as a bytes string.
other_channel_id = int(other_channel_id)
if i == 5:
break
i += 1
if other_channel_id in last_calls_other_channels_ids:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(other_channel_id))
other_channel_id = await redis.spop(constants.CHANNELS_SEARCH_QUEUE)
if other_channel_id is None:
break
continue
break
if other_channel_id is not None:
await redis.sadd(
constants.CHANNELS_ONCALL_SET, str(channel_id), str(other_channel_id)
)
msg_phone_answered = f"☎️ {secrets.choice(constants.MSGS_PHONE_ANSWERED)}"
try:
await self.bot.get_partial_messageable(other_channel_id).send(
msg_phone_answered
)
call = await models.Call.create(is_nsfw=ctx.channel.is_nsfw())
user, profile = await dbutils.get_user(ctx.author)
await models.CallUserChannelRelationship.create(
call=call,
channel_id=channel_id,
user=user,
profile=profile,
is_masked=user.is_masked,
mask=secrets.choice(constants.MASKS_LIST).value,
)
await models.Channel.filter(
id__in=[other_channel_id, channel.id]
).update(call=call)
except errors.DiscordException:
await redis.srem(
constants.CHANNELS_ONCALL_SET,
str(channel_id),
str(other_channel_id),
)
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
msg_phone_answered = ""
await ctx.send(
(
f"**TIP**: {secrets.choice(constants.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
f"☎️ You skipped the call.\n"
)
+ (
msg_phone_answered
if msg_phone_answered
else f"☎️ {secrets.choice(constants.MSGS_PHONE_CALLING)}\n"
)
)
else:
await redis.sadd(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
channel.call = None
await channel.save()
await ctx.send(
(
f"**TIP**: {secrets.choice(constants.TIPS).replace("{prefix}", ctx.clean_prefix)}\n"
f"☎️ You skipped the call.\n"
f"☎️ {secrets.choice(constants.MSGS_PHONE_CALLING)}\n"
)
)
@commands.command(aliases=["hang", "h"], extras={"help_name": "Hang Up"})
@commands.guild_only()
@checks.is_unbanned()
@transactions.atomic()
async def hangup(self, ctx: commands.Context) -> None:
"""
Hang up the phone.
"""
channel_id = ctx.channel.id
is_on_call = await redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
if is_on_call:
channel = await dbutils.get_channel(ctx.channel)
call = await models.Call.get(id=channel.call_id).prefetch_related(
"channels"
)
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
other_channel: models.Channel = find(
lambda c: c.id != channel_id, call.channels
)
await redis.srem(
constants.CHANNELS_ONCALL_SET,
*(
[str(channel_id)]
+ ([str(other_channel.id)] if other_channel is not None else [])
),
)
if other_channel is not None:
try:
await (await self.bot.get_or_fetch_channel(other_channel.id)).send(
f"☎️ {secrets.choice(constants.MSGS_PHONE_HANGUP_YOU)}"
)
except errors.DiscordException:
pass
other_channel.call = None
await ctx.send(f"☎️ {secrets.choice(constants.MSGS_PHONE_HANGUP_ME)}")
channel.call = None
await models.Channel.bulk_update(
[channel] + ([other_channel] if other_channel is not None else []),
["call_id"],
)
call.ended_at = datetime.now(timezone.utc)
await call.save()
is_searching = await redis.sismember(
constants.CHANNELS_SEARCH_QUEUE, str(channel_id)
)
if is_searching:
await redis.srem(constants.CHANNELS_SEARCH_QUEUE, str(channel_id))
await ctx.send(f"☎️ {secrets.choice(constants.MSGS_PHONE_NO_ANSWER)}")
if not is_searching and not is_on_call:
await ctx.send(f"❌ {secrets.choice(constants.MSGS_PHONE_NOT_IN_CALL)}")
@commands.command()
@commands.guild_only()
@checks.is_unbanned()
async def report(
self, ctx: commands.Context, call_id: int | None = None, *, reason: str
) -> None:
"""
Report a call by it's ID, or the current/last call if no ID is specified.
Options:
- `[<call_id>]`: The ID of the call to report.
- `<reason>`: The reason why you're reporting the call.
"""
user, _ = await dbutils.get_user(ctx.author)
channel_id = ctx.channel.id
if call_id is None:
call_rs = (
await models.CallUserChannelRelationship.filter(channel_id=channel_id)
.order_by("-id")
.limit(1)
.first()
)
if call_rs is None:
await ctx.send("❌ No calls occurred on this channel.")
return
call_id = call_rs.call_id
call = await models.Call.get_or_none(id=call_id).prefetch_related(
"channels", "report", "call_rss"
)
if call is None:
await ctx.send(f"❌ There is no call with ID `{call_id}`.")
if call.report is not None:
await ctx.send(
"✅ This call has already been reported. Our moderator team will review it as soon as possible."
)
return
if not any(map(lambda rs: rs.user_id == user.id, call.call_rss)):
await ctx.send("❌ You have not participated in this call.")
return
reviewed_reports = await models.Report.filter(
reported_by_id=ctx.author.id,
status__in=[models.ReportStatus.ACCEPTED, models.ReportStatus.DECLINED],
)
priority = 0.5
if len(reviewed_reports) > 3:
accepted_reports = len(
[
report
for report in reviewed_reports
if report.status == models.ReportStatus.ACCEPTED
]
)
priority = accepted_reports / len(reviewed_reports)
await models.Report.create(
call_id=call_id, reported_by_id=user.id, reason=reason, priority=priority
)
message = "✅ The call has been reported."
if call.ended_at is None:
call.ended_at = datetime.now(timezone.utc)
await call.save()
await redis.sadd(constants.CALL_IDS_TO_FLUSH, call.id)
updated_channels = []
for channel in call.channels:
channel.call = None
updated_channels.append(channel)
await models.Channel.bulk_update(updated_channels, ["call_id"])
await redis.srem(
constants.CHANNELS_ONCALL_SET,
str(updated_channels[0]),
str(updated_channels[1]),
)
message += f"\n☎️ {secrets.choice(constants.MSGS_PHONE_HANGUP_ME)}"
other_channel = find(lambda c: c.id != channel_id, updated_channels)
await redis.srem(
constants.CHANNELS_ONCALL_SET, str(other_channel.id), str(channel_id)
)
try:
await self.bot.get_partial_messageable(other_channel.id).send(
f"☎️ {secrets.choice(constants.MSGS_PHONE_HANGUP_YOU)}"
)
except errors.DiscordException:
pass
await ctx.send(message)
@commands.command(aliases=["fr"], extras={"help_name": "Friend Request"})
@commands.guild_only()
@checks.is_unbanned()
@checks.is_on_call()
async def friendreq(self, ctx: commands.Context) -> None:
"""
Ask the other party to send you a friend request.
"""
user_id = ctx.author.id
channel_id = ctx.channel.id
call = await models.Call.get(channels__id=channel_id).prefetch_related(
"channels", "call_rss", "call_rss__profile"
)
other_channel = find(lambda c: c.id != channel_id, call.channels)
call_rs: models.CallUserChannelRelationship = find(
lambda rs: rs.user_id == user_id, call.call_rss
)
try:
await self.bot.get_partial_messageable(other_channel.id).send(
f"**{call_rs.mask_emoji + " " if call_rs.is_masked else ""}{call_rs.display_name}** (`{ctx.author.name}`) wants you to add them as a friend on Discord."
)
except errors.DiscordException:
await ctx.send("⚠️ Your friend request couldn't be sent.")
return
await ctx.send("✅ Your friend request was sent!")
@commands.Cog.listener()
@commands.guild_only()
async def on_message(self, message: discord.Message) -> None:
"""Processes every message received.
Args:
message (discord.Message): The message received.
"""
if (
message.author.bot
or message.is_system()
or len(message.stickers) > 0
or not message.content
):
return
if (await self.bot.get_context(message)).valid:
return
user_id = message.author.id
channel_id = message.channel.id
is_on_call = await redis.sismember(
constants.CHANNELS_ONCALL_SET, str(channel_id)
)
if not is_on_call:
return
is_banned = await self.bot.is_banned(user_id)
if is_banned:
return
channel = await dbutils.get_channel(message.channel)
try:
call = await models.Call.get(id=channel.call_id).prefetch_related(
"channels"
)
except exceptions.DoesNotExist:
await redis.srem(constants.CHANNELS_ONCALL_SET, str(channel_id))
return
other_channel: models.Channel = find(
lambda c: c.id != channel_id, call.channels
)
if other_channel is None:
# BUG: The call wasn't properly set up.
return
user, profile = await dbutils.get_user(message.author)
call_rss = await models.CallUserChannelRelationship.filter(
call=call, channel=channel
).select_related("profile")
used_masks = [rs.mask for rs in call_rss]
call_rs, _ = await models.CallUserChannelRelationship.get_or_create(
{
"mask": secrets.choice(
[i for i in constants.MASKS_LIST if i not in used_masks]
),
"is_masked": user.is_masked,
"profile": profile,
},
call=call,
user=user,
channel=channel,
)
# Since .select_related() can't be used and profile was already fetched, we need to
# manually set it.
call_rs.profile = profile
blocked_word = censorship.moderate(message.content)
webhook_message = None
cleaned_content = formatting.clean(message, call_rss)
replying_to_id = None
if blocked_word:
try:
await self.bot.get_partial_messageable(channel_id).send(
f"Your message wasn't delivered!\nBlocked word: `{blocked_word}`."
)
except errors.DiscordException:
pass
prefix = await self.bot.get_prefix(server_id=other_channel.server_id)
try:
await self.bot.get_partial_messageable(other_channel.id).send(
f"**{call_rs.mask_emoji + " " if call_rs.is_masked else ""}{call_rs.display_name}** said something that's not allowed. If you think they're doing something inappropiate, use `{prefix}report`."
)
except errors.DiscordException:
pass
else:
async with aiohttp.ClientSession() as session:
webhook = other_channel.webhook(session)
embed = None
if message.reference:
embed, replying_message = await embeds.ReplyEmbed(
message.reference, call_id=call.id
)
if replying_message is not None:
replying_to_id = replying_message.id
webhook_message = await webhook.send(
cleaned_content,
username=call_rs.display_name,
avatar_url=call_rs.display_avatar,
embed=embed,
wait=True,
)
pipe = redis.pipeline()
pipe.hset(
constants.CALL_MESSAGES_TEMPLATE % call.id,
message.id,
orjson.dumps(
{
"id": message.id,
"call_id": call.id,
"channel_id": channel.id,
"user_id": user.id,
"profile_id": profile.id,
"mask": call_rs.mask if call_rs.is_masked else None,
"propagation_id": webhook_message.id if webhook_message else None,
"propagation_channel_id": (
other_channel.id if webhook_message else None
),
"replying_to_id": replying_to_id,
"created_at": datetime.now(timezone.utc),
}
),
)
pipe.hset(
constants.CALL_MESSAGE_CONTENTS_TEMPLATE % call.id,
constants.CALL_MESSAGE_CONTENTS_HASH_TEMPLATE % (message.id, 1),
orjson.dumps(
{
"message_id": message.id,
"content": cleaned_content,
"was_propagated": not bool(blocked_word),
"created_at": datetime.now(timezone.utc),
}
),
)
pipe.hset(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % call.id, message.id, 1
)
if webhook_message is not None:
pipe.hset(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call.id,
webhook_message.id,
message.id,
)
if replying_to_id is not None:
other_replies = await redis.hget(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % call.id, replying_to_id
)
if other_replies is None:
pipe.hset(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % call.id,
replying_to_id,
message.id,
)
else:
pipe.hset(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % call.id,
replying_to_id,
other_replies.decode() + f",{message.id}",
)
await pipe.execute()
@commands.Cog.listener()
@commands.guild_only()
@transactions.atomic()
async def on_raw_message_delete(
self, payload: discord.RawMessageDeleteEvent
) -> None:
"""Processes a message deletion received.
Args:
payload (discord.RawMessageDeleteEvent): The received payload.
"""
message_id = payload.message_id
channel_id = payload.channel_id
channel = await models.Channel.get_or_none(id=channel_id)
if channel is None:
return
if channel.call_id is None:
return
message_json = await redis.hget(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id, message_id
)
if message_json is None:
return
message_dict = orjson.loads(message_json)
message = models.Message(**message_dict)
message_dict["was_deleted"] = True
await redis.hset(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id,
message.id,
orjson.dumps(message_dict),
)
other_channel = await models.Channel.get(id=message.propagation_channel_id)
if other_channel.call_id != message.call_id:
return
async with aiohttp.ClientSession() as session:
webhook = other_channel.webhook(session)
try:
await webhook.delete_message(message.propagation_id)
except errors.DiscordException:
# This message was already deleted.
pass
message_replies = await redis.hget(
constants.CALL_MESSAGE_REPLIES_TEMPLATE % channel.call_id, message.id
)
if message_replies is None:
return
for message_id in message_replies.decode().split(","):
try:
await webhook.edit_message(message_id, embed=None)
except errors.DiscordException:
pass
@commands.Cog.listener()
@commands.guild_only()
@transactions.atomic()
async def on_raw_message_edit(self, payload: discord.RawMessageUpdateEvent) -> None:
"""Processes a message edit received.
Args:
payload (discord.RawMessageUpdateEvent): The event payload.
"""
if (
payload.cached_message is not None
and (
payload.cached_message.author.bot or payload.cached_message.is_system()
)
or "content" not in payload.data
):
return
message_id = payload.message_id
channel_id = payload.channel_id
content = payload.data["content"]
channel = await models.Channel.get_or_none(id=channel_id)
if channel is None:
return
if channel.call_id is None:
return
message_json = await redis.hget(
constants.CALL_MESSAGES_TEMPLATE % channel.call_id, message_id
)
if message_json is None:
return
message = models.Message(**orjson.loads(message_json))
other_channel = await models.Channel.get(id=message.propagation_channel_id)
if other_channel.call_id != message.call_id:
return
original_message = await self.bot.get_partial_messageable(
channel_id
).fetch_message(message.id)
if original_message is None:
return
call_rss = await models.CallUserChannelRelationship.filter(
call_id=channel.call_id
)
cleaned_content = formatting.clean(original_message, call_rss)
blocked_word = censorship.moderate(content)
iteration = int(
await redis.hget(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % channel.call_id,
message.id,
)
)
pipe = redis.pipeline()
pipe.hincrby(
constants.CALL_MESSAGE_CONTENT_ITERATIONS_TEMPLATE % channel.call_id,
message.id,
)
pipe.hset(
constants.CALL_MESSAGE_CONTENTS_TEMPLATE % channel.call_id,
constants.CALL_MESSAGE_CONTENTS_HASH_TEMPLATE % (message.id, iteration + 1),
orjson.dumps(
{
"message_id": message.id,
"content": cleaned_content,
"was_propagated": not bool(blocked_word),
"created_at": datetime.now(timezone.utc),
}
),
)
await pipe.execute()
async with aiohttp.ClientSession() as session:
webhook = other_channel.webhook(session)
try:
await webhook.edit_message(
message.propagation_id,
content=(
cleaned_content
if blocked_word is None
else "*This message was edited and censored.*"
),
)
except errors.DiscordException:
# This message was deleted.
pass
# TODO: Edit replies.
@commands.Cog.listener()
async def on_raw_reaction_add(
self, payload: discord.RawReactionActionEvent
) -> None:
if payload.user_id == self.bot.user.id:
return
channel = await models.Channel.get_or_none(id=payload.channel_id)
if channel is None:
return
if channel.call_id is None:
return
call_id = channel.call_id
other_channel_id = None
other_channel_message_id = await redis.hget(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
payload.message_id,
)
if other_channel_message_id is None:
other_channel_message_json = await redis.hget(
constants.CALL_MESSAGES_TEMPLATE % call_id, payload.message_id
)
if other_channel_message_json is None:
return
other_channel_message_dict = orjson.loads(other_channel_message_json)
if other_channel_message_dict["propagation_id"] is None:
# The message wasn't propagated
return
other_channel_id = other_channel_message_dict["propagation_channel_id"]
other_channel_message_id = other_channel_message_dict["propagation_id"]
else:
other_channel = await models.Channel.get_or_none(
call_id=call_id, id__not=payload.channel_id
)
if other_channel is not None:
other_channel_id = other_channel.id
else:
return
other_channel_message_id = int(other_channel_message_id)
channel = await self.bot.get_or_fetch_channel(other_channel_id)
if channel is None:
return
await channel.get_partial_message(other_channel_message_id).add_reaction(
payload.emoji
)
@commands.Cog.listener()
async def on_raw_reaction_remove(
self, payload: discord.RawReactionActionEvent
) -> None:
if payload.user_id == self.bot.user.id:
return
channel = await models.Channel.get_or_none(id=payload.channel_id)
if channel is None:
return
if channel.call_id is None:
return
call_id = channel.call_id
other_channel_id = None
other_channel_message_id = await redis.hget(
constants.CALL_MESSAGE_FROM_PROPAGATION_TEMPLATE % call_id,
payload.message_id,
)
if other_channel_message_id is None:
other_channel_message_json = await redis.hget(
constants.CALL_MESSAGES_TEMPLATE % call_id, payload.message_id
)
if other_channel_message_json is None:
return
other_channel_message_dict = orjson.loads(other_channel_message_json)
if other_channel_message_dict["propagation_id"] is None:
# The message wasn't propagated
return
other_channel_id = other_channel_message_dict["propagation_channel_id"]
other_channel_message_id = other_channel_message_dict["propagation_id"]
else:
other_channel = await models.Channel.get_or_none(
call_id=call_id, id__not=payload.channel_id
)
if other_channel is not None:
other_channel_id = other_channel.id
else:
return
other_channel_message_id = int(other_channel_message_id)
channel = await self.bot.get_or_fetch_channel(other_channel_id)
if channel is None:
return
await channel.get_partial_message(other_channel_message_id).remove_reaction(
payload.emoji, self.bot.user
)
def setup(b: bot.Payphone) -> None:
"""Load the cog into the bot.
Args:
b (bot.Payphone): The bot instance to load the cog into.
"""
b.add_cog(Calls(b))