685 lines
19 KiB
Python
685 lines
19 KiB
Python
from dataclasses import dataclass
|
|
|
|
import orjson
|
|
|
|
from payphone.db import models
|
|
from payphone.bot import Payphone
|
|
from payphone.core import masks
|
|
|
|
|
|
@dataclass
|
|
class Participant:
|
|
"""A participant in Payphone."""
|
|
|
|
_bot: Payphone
|
|
|
|
id: int
|
|
name: str
|
|
avatar_url: str
|
|
mask: str
|
|
is_masked: bool
|
|
channel_id: int
|
|
call_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)
|
|
)
|
|
|
|
async def set_mask(self, mask: str) -> None:
|
|
"""Sets the mask of the participant.
|
|
|
|
Args:
|
|
mask (str): The mask to set.
|
|
"""
|
|
|
|
if mask == self.mask:
|
|
return
|
|
|
|
pipe = self._bot.redis.pipeline()
|
|
pipe.srem(f"calls:{self.call_id}:masks", self.mask)
|
|
pipe.sadd(f"calls:{self.call_id}:masks", mask)
|
|
await pipe.execute()
|
|
|
|
self.mask = mask
|
|
self.is_masked = True
|
|
|
|
data = orjson.dumps(
|
|
{
|
|
"name": self.name,
|
|
"avatar_url": self.avatar_url,
|
|
"mask": mask,
|
|
"is_masked": True,
|
|
"channel_id": self.channel_id,
|
|
}
|
|
)
|
|
|
|
await self._bot.redis.hset(f"calls:{self.call_id}:participants", self.id, data)
|
|
|
|
|
|
@dataclass
|
|
class Message:
|
|
id: int
|
|
call_id: int
|
|
channel_id: int
|
|
propagation_id: int
|
|
propagation_channel_id: int
|
|
author_id: int
|
|
|
|
|
|
@dataclass
|
|
class Call:
|
|
"""A call in Payphone."""
|
|
|
|
_bot: Payphone
|
|
|
|
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
|
|
) -> None:
|
|
"""Sets a message as part of a call.
|
|
|
|
Args:
|
|
id (int): The ID of the message to set.
|
|
channel_id (int): The ID of the channel to set the message in.
|
|
propagation_id (int): The ID of the propagation message.
|
|
author_id (int): The ID of the author of the message.
|
|
"""
|
|
|
|
pipe = self._bot.redis.pipeline()
|
|
|
|
pipe.hset("calls:messages", id, self.id)
|
|
pipe.hset("calls:messages:channels", id, channel_id)
|
|
pipe.hset("calls:messages:propagations", id, propagation_id)
|
|
pipe.hset("calls:messages:reversed", propagation_id, id)
|
|
pipe.hset("calls:messages:authors", id, author_id)
|
|
pipe.hset("calls:last-message", self.id, "")
|
|
pipe.hexpire("calls:messages", 15 * 60, id)
|
|
pipe.hexpire("calls:messages:channels", 15 * 60, id)
|
|
pipe.hexpire("calls:messages:propagations", 15 * 60, id)
|
|
pipe.hexpire("calls:messages:reversed", 15 * 60, propagation_id)
|
|
pipe.hexpire("calls:messages:authors", 15 * 60, id)
|
|
pipe.hexpire("calls:last-message", 15 * 60, self.id)
|
|
|
|
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.
|
|
|
|
Args:
|
|
id (int): The ID of the message to check.
|
|
|
|
Returns:
|
|
bool: True if the message is from the call, False otherwise.
|
|
"""
|
|
|
|
return bool(await self._bot.redis.hexists("calls:messages", id)) or bool(
|
|
await self._bot.redis.hexists("calls:messages:reversed", id)
|
|
)
|
|
|
|
async def get_message_propagation(self, id: int) -> tuple[int, int] | None:
|
|
"""Gets a message as part of a call.
|
|
|
|
Args:
|
|
id (int): The ID of the message to get.
|
|
|
|
Returns:
|
|
tuple[int, int]: The channel and propagation ID of the message.
|
|
"""
|
|
|
|
pipe = self._bot.redis.pipeline()
|
|
|
|
pipe.hget("calls:messages:channels", id)
|
|
pipe.hget("calls:messages:propagations", id)
|
|
|
|
channel_id, propagation_id = await pipe.execute()
|
|
|
|
if channel_id is None or propagation_id is None:
|
|
return
|
|
|
|
propagation_channel_id = (
|
|
self.channels[0]
|
|
if int(channel_id) != self.channels[0]
|
|
else self.channels[1]
|
|
)
|
|
|
|
return int(propagation_channel_id), int(propagation_id)
|
|
|
|
async def set_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, bool): The participant that was set and whether they changed their masked
|
|
status.
|
|
"""
|
|
|
|
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,
|
|
mask=mask,
|
|
is_masked=user.is_masked,
|
|
channel_id=channel_id,
|
|
user=user,
|
|
call_id=self.id,
|
|
), (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.
|
|
|
|
Args:
|
|
user_id (int): The ID of the user to get.
|
|
|
|
Returns:
|
|
Participant | None: The participant that was set.
|
|
"""
|
|
|
|
data = await self._bot.redis.hget(f"calls:{self.id}:participants", user_id)
|
|
|
|
if data is None:
|
|
return
|
|
|
|
data = orjson.loads(data)
|
|
|
|
return Participant(
|
|
_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"],
|
|
call_id=self.id,
|
|
)
|
|
|
|
async def get_participants(self) -> list["Participant"]:
|
|
"""Gets all participants in a call.
|
|
|
|
Returns:
|
|
list[Participant]: The participants in the call.
|
|
"""
|
|
|
|
participants = []
|
|
|
|
data = await self._bot.redis.hgetall(f"calls:{self.id}:participants")
|
|
|
|
for user_id, data in data.items():
|
|
data = orjson.loads(data)
|
|
|
|
participants.append(
|
|
Participant(
|
|
_bot=self._bot,
|
|
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"],
|
|
call_id=self.id,
|
|
)
|
|
)
|
|
|
|
return participants
|
|
|
|
|
|
class CallHandler:
|
|
"""Handles calls in Payphone."""
|
|
|
|
def __init__(self, bot: Payphone) -> None:
|
|
"""Initializes the CallHandler.
|
|
|
|
Args:
|
|
bot (Payphone): The bot instance.
|
|
"""
|
|
|
|
self.bot = bot
|
|
|
|
async def set_searching(self, channel_id: int, status: bool) -> None:
|
|
"""Sets the channel as searching for a call.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to set as searching.
|
|
status (bool): The status to set the channel to.
|
|
"""
|
|
|
|
if status:
|
|
await self.bot.redis.sadd("calls:channels:searching", channel_id)
|
|
|
|
else:
|
|
await self.bot.redis.srem("calls:channels:searching", channel_id)
|
|
|
|
async def connect(self) -> list[tuple[int, int]]:
|
|
"""Connects two channels together.
|
|
|
|
Returns:
|
|
list[tuple[int, int]]: A list of channel pairs.
|
|
"""
|
|
|
|
pairs = []
|
|
|
|
while True:
|
|
channels = await self.bot.redis.spop("calls:channels:searching", 2)
|
|
|
|
if len(channels) < 2:
|
|
if len(channels) == 1:
|
|
await self.set_searching(channels[0], True)
|
|
|
|
break
|
|
|
|
last_calls = await self.bot.redis.hmget("last-calls", *channels)
|
|
|
|
break_outer = False
|
|
|
|
for channel in channels:
|
|
if channel in last_calls:
|
|
next_channel = await self.bot.redis.spop("calls:channels:searching")
|
|
|
|
if next_channel is not None:
|
|
other_channel_id = channels[abs(channels.index(channel) - 1)]
|
|
await self.set_searching(other_channel_id, True)
|
|
|
|
channels[channels.index(channel)] = next_channel
|
|
|
|
else:
|
|
for channel in channels:
|
|
await self.set_searching(channel, True)
|
|
|
|
break_outer = True
|
|
break
|
|
|
|
if break_outer:
|
|
break
|
|
|
|
pairs.append(tuple(channels))
|
|
|
|
mapping = {}
|
|
reverse_mapping = {}
|
|
|
|
for pair in pairs:
|
|
last_call_id = await self.bot.redis.incr("calls:calls:count")
|
|
|
|
mapping[pair[0]] = last_call_id
|
|
mapping[pair[1]] = last_call_id
|
|
|
|
reverse_mapping[last_call_id] = f"{pair[0]}:{pair[1]}"
|
|
|
|
pipe = self.bot.redis.pipeline(transaction=True)
|
|
|
|
if mapping:
|
|
pipe.hset("calls:channels:active", mapping=mapping)
|
|
pipe.hset("calls:calls:active", mapping=reverse_mapping)
|
|
|
|
for pair in pairs:
|
|
pipe.hmset("last-calls", {pair[0]: pair[1], pair[1]: pair[0]})
|
|
pipe.hset("calls:last-message", mapping[pair[0]], "")
|
|
pipe.hexpire("calls:last-message", 60, mapping[pair[0]])
|
|
|
|
await pipe.execute()
|
|
|
|
return pairs
|
|
|
|
async def expire(self) -> list[tuple[int, int]]:
|
|
"""
|
|
Ends all calls that have been inactive for more than 15 seconds.
|
|
"""
|
|
|
|
active_calls = await self.bot.redis.hgetall("calls:calls:active")
|
|
|
|
pairs = []
|
|
|
|
pipe = self.bot.redis.pipeline(transaction=True)
|
|
|
|
for call_id, channels in active_calls.items():
|
|
channels = [int(c) for c in channels.split(":")]
|
|
|
|
last_message = await self.bot.redis.hget("calls:last-message", call_id)
|
|
|
|
if last_message is not None:
|
|
continue
|
|
|
|
pipe.hdel("calls:last-message", call_id)
|
|
pipe.hdel("calls:channels:active", *channels)
|
|
pipe.hdel("calls:calls:active", call_id)
|
|
pipe.srem("calls:channels:searching", *channels)
|
|
|
|
pairs.append(tuple(channels))
|
|
|
|
await pipe.execute()
|
|
|
|
return pairs
|
|
|
|
async def is_searching(self, channel_id: int) -> bool:
|
|
"""Checks if a channel is searching for a call.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to check.
|
|
|
|
Returns:
|
|
bool: True if the channel is searching for a call, False otherwise.
|
|
"""
|
|
|
|
return bool(
|
|
await self.bot.redis.sismember("calls:channels:searching", channel_id)
|
|
)
|
|
|
|
async def is_on_call(self, channel_id: int) -> bool:
|
|
"""Checks if a channel is on a call.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to check.
|
|
|
|
Returns:
|
|
bool: True if the channel is on a call, False otherwise.
|
|
"""
|
|
|
|
return await self.bot.redis.hexists("calls:channels:active", channel_id)
|
|
|
|
async def get_call_by_channel_id(self, channel_id: int) -> Call | None:
|
|
"""Gets the call that a channel is in.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to get the call for.
|
|
|
|
Returns:
|
|
Call | None: The pair of channels that are connected.
|
|
"""
|
|
|
|
call_id = await self.bot.redis.hget("calls:channels:active", channel_id)
|
|
|
|
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(
|
|
":"
|
|
)
|
|
]
|
|
|
|
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.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to end the call in.
|
|
|
|
Returns:
|
|
tuple[int, int] | tuple[int]: The pair of channels that were connected.
|
|
"""
|
|
|
|
await self.set_searching(channel_id, False)
|
|
|
|
call_id = await self.bot.redis.hget("calls:channels:active", channel_id)
|
|
|
|
if call_id is None:
|
|
# Channel is searching.
|
|
return (channel_id,)
|
|
|
|
channels = [
|
|
int(c)
|
|
for c in (await self.bot.redis.hget("calls:calls:active", call_id)).split(
|
|
":"
|
|
)
|
|
]
|
|
|
|
pipe = self.bot.redis.pipeline(transaction=True)
|
|
|
|
pipe.srem("calls:channels:searching", *channels)
|
|
pipe.hdel("calls:channels:active", *channels)
|
|
pipe.hdel("calls:calls:active", call_id)
|
|
|
|
await pipe.execute()
|
|
|
|
return tuple(channels)
|
|
|
|
async def skip(self, channel_id: int) -> tuple[int, int]:
|
|
"""Skips a channel in the call queue.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to skip.
|
|
"""
|
|
|
|
await self.set_searching(channel_id, True)
|
|
|
|
call_id = await self.bot.redis.hget("calls:channels:active", channel_id)
|
|
|
|
if call_id is None:
|
|
return
|
|
|
|
channels = [
|
|
int(c)
|
|
for c in (await self.bot.redis.hget("calls:calls:active", call_id)).split(
|
|
":"
|
|
)
|
|
]
|
|
|
|
for channel in channels:
|
|
if channel != channel_id:
|
|
await self.set_searching(channel, False)
|
|
|
|
await self.bot.redis.hdel("calls:channels:active", *channels)
|
|
|
|
return tuple(channels)
|
|
|
|
async def get_channel_webhook_url(self, channel_id: int) -> str:
|
|
"""Gets the webhook for a channel.
|
|
|
|
Args:
|
|
channel_id (int): The ID of the channel to get the webhook for.
|
|
|
|
Returns:
|
|
str: The URL of the webhook.
|
|
"""
|
|
|
|
cached_url = await self.bot.redis.hget("calls:channels:webhooks", channel_id)
|
|
|
|
if cached_url:
|
|
return cached_url
|
|
|
|
channel = await self.bot.get_or_fetch_channel(channel_id)
|
|
|
|
if channel is None:
|
|
return
|
|
|
|
for webhook in await channel.webhooks():
|
|
if webhook.user == self.bot.user and webhook.token is not None:
|
|
await self.bot.redis.hset(
|
|
"calls:channels:webhooks", channel_id, webhook.url
|
|
)
|
|
return webhook.url
|
|
|
|
webhook = await channel.create_webhook(name="Payphone")
|
|
|
|
await self.bot.redis.hset("calls:channels:webhooks", channel_id, webhook.url)
|
|
|
|
return webhook.url
|
|
|
|
async def refresh_channel_webhook_url(self, channel_id: int) -> str:
|
|
"""Refreshes a channel's webhook.
|
|
|
|
Args:
|
|
channel_id (int): The channel ID of the webhook to refresh.
|
|
|
|
Returns:
|
|
str: The new webhook's URL.
|
|
"""
|
|
|
|
await self.bot.redis.hdel("calls:channels:webhooks", channel_id)
|
|
|
|
channel = await self.bot.get_or_fetch_channel(channel_id)
|
|
|
|
if channel is None:
|
|
return
|
|
|
|
for webhook in await channel.webhooks():
|
|
if webhook.name == "Payphone":
|
|
await webhook.delete()
|
|
|
|
return await self.get_channel_webhook_url(channel_id)
|
|
|
|
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:
|
|
Message | None: The message that was sent.
|
|
"""
|
|
|
|
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.
|
|
|
|
Args:
|
|
id (int): The ID of the message to get.
|
|
|
|
Returns:
|
|
int: The author ID of the message.
|
|
"""
|
|
|
|
pipe = self.bot.redis.pipeline()
|
|
|
|
pipe.hget("calls:messages", id)
|
|
pipe.hget("calls:messages:reversed", id)
|
|
|
|
call_id, message_id = await pipe.execute()
|
|
|
|
if call_id is None and message_id is None:
|
|
return
|
|
|
|
message_id = int(message_id) if message_id else id
|
|
|
|
author_id = await self.bot.redis.hget("calls:messages:authors", message_id)
|
|
|
|
return int(author_id) if author_id else None
|
|
|
|
async def get_original_message_id(self, id: int) -> int | None:
|
|
"""
|
|
Get the original message ID of a message in a call.
|
|
|
|
Args:
|
|
id (int): The ID of the message to get.
|
|
|
|
Returns:
|
|
int: The original message ID of the message.
|
|
"""
|
|
|
|
message_id = await self.bot.redis.hget("calls:messages:reversed", id)
|
|
|
|
if message_id:
|
|
return int(message_id)
|
|
|
|
return None
|