352 lines
11 KiB
Python
352 lines
11 KiB
Python
from datetime import datetime
|
|
|
|
import enum
|
|
|
|
from tortoise import models, fields
|
|
|
|
from discord import webhook
|
|
|
|
import aiohttp
|
|
|
|
from payphone.core import config, constants
|
|
|
|
|
|
class User(models.Model):
|
|
id: int = fields.BigIntField(primary_key=True)
|
|
username: str = fields.CharField(max_length=32, db_index=True)
|
|
is_masked: bool = fields.BooleanField(default=False)
|
|
is_moderator: bool = fields.BooleanField(default=False)
|
|
profiles: fields.BackwardFKRelation["Profile"]
|
|
call_rss: fields.BackwardFKRelation["CallUserChannelRelationship"]
|
|
leaderboard_position: int | None = fields.IntField(
|
|
default=None, null=True, db_index=True
|
|
)
|
|
leaderboard_calls_count: int | None = fields.IntField(default=None, null=True)
|
|
joined_at: datetime = fields.DatetimeField(auto_now_add=True)
|
|
|
|
|
|
class Call(models.Model):
|
|
started_at = fields.DatetimeField(auto_now_add=True)
|
|
ended_at = fields.DatetimeField(null=True, default=None)
|
|
|
|
channels: fields.BackwardFKRelation["Channel"]
|
|
call_rss: fields.BackwardFKRelation["CallUserChannelRelationship"]
|
|
report: fields.BackwardOneToOneRelation["Report"]
|
|
messages: fields.BackwardFKRelation["Message"]
|
|
|
|
|
|
class Profile(models.Model):
|
|
user: fields.ForeignKeyRelation["User"] = fields.ForeignKeyField(
|
|
"models.User", related_name="profiles", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
user_id: int
|
|
server: fields.ForeignKeyNullableRelation["Server"] = fields.ForeignKeyField(
|
|
"models.Server",
|
|
related_name="profiles",
|
|
null=True,
|
|
on_delete=fields.OnDelete.CASCADE,
|
|
)
|
|
avatar_url: str = fields.TextField()
|
|
name: str = fields.CharField(max_length=32)
|
|
color: int | None = fields.IntField(null=True, default=None)
|
|
|
|
call_rss: fields.BackwardFKRelation["CallUserChannelRelationship"]
|
|
|
|
@property
|
|
def jump_url(self) -> str:
|
|
"""The URL that'll open the user in Discord.
|
|
|
|
Returns:
|
|
str: The URL.
|
|
"""
|
|
|
|
return f"discord://-/users/{self.user_id}"
|
|
|
|
@property
|
|
def link(self) -> str:
|
|
"""A markdown link that opens the user in Discord.
|
|
|
|
Returns:
|
|
str: The markdown link.
|
|
"""
|
|
|
|
return f"[{self.name}]({self.jump_url})"
|
|
|
|
class Meta:
|
|
unique_together = [("user_id", "server_id")]
|
|
|
|
|
|
class Server(models.Model):
|
|
id: int = fields.BigIntField(primary_key=True)
|
|
prefix: str = fields.CharField(max_length=5, default="p.")
|
|
channels: fields.BackwardFKRelation["Channel"]
|
|
profiles: fields.BackwardFKRelation["Profile"]
|
|
|
|
|
|
class Channel(models.Model):
|
|
id = fields.BigIntField(primary_key=True)
|
|
server: fields.ForeignKeyRelation[Server] = fields.ForeignKeyField(
|
|
"models.Server", related_name="channels", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
call: fields.ForeignKeyNullableRelation["Call"] = fields.ForeignKeyField(
|
|
"models.Call", null=True, related_name="channels"
|
|
)
|
|
enabled: bool | None = fields.BooleanField(null=True, default=None)
|
|
webhook_id: int | None = fields.BigIntField(null=True)
|
|
webhook_url: str | None = fields.TextField(null=True)
|
|
|
|
call_rss: fields.BackwardFKRelation["CallUserChannelRelationship"]
|
|
messages: fields.BackwardFKRelation["Message"]
|
|
propagated_messages: fields.BackwardFKRelation["Message"]
|
|
|
|
def webhook(self, session: aiohttp.ClientSession) -> webhook.Webhook | None:
|
|
"""Initializes the channel's webhook instance.
|
|
|
|
Args:
|
|
session (aiohttp.ClientSession): An initialized aiohttp client session.
|
|
|
|
Returns:
|
|
webhook.Webhook | None: The webhook, or None if not initialized.
|
|
"""
|
|
|
|
if self.webhook_url is None:
|
|
return None
|
|
|
|
return webhook.Webhook.from_url(
|
|
self.webhook_url, session=session, bot_token=config.TOKEN
|
|
)
|
|
|
|
class Meta:
|
|
index = [("server_id", "enabled")]
|
|
|
|
|
|
class CallUserChannelRelationship(models.Model):
|
|
call: fields.ForeignKeyRelation["Call"] = fields.ForeignKeyField(
|
|
"models.Call", related_name="call_rss", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
user: fields.ForeignKeyRelation["User"] = fields.ForeignKeyField(
|
|
"models.User", related_name="call_rss", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
profile: fields.ForeignKeyRelation["Profile"] = fields.ForeignKeyField(
|
|
"models.Profile", related_name="call_rss", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
channel: fields.ForeignKeyRelation["Channel"] = fields.ForeignKeyField(
|
|
"models.Channel", related_name="call_rss", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
is_masked: bool = fields.BooleanField()
|
|
mask: int = fields.BigIntField()
|
|
created_at: datetime = fields.DatetimeField(auto_now_add=True, db_index=True)
|
|
|
|
@property
|
|
def mask_name(self) -> str:
|
|
"""Returns the mask's name.
|
|
|
|
Returns:
|
|
str: The mask's name.
|
|
"""
|
|
|
|
return constants.Mask(self.mask).name.replace("_", " ").lower().title()
|
|
|
|
@property
|
|
def display_name(self) -> str:
|
|
"""Returns the displayed name.
|
|
|
|
Returns:
|
|
str: The displayed name.
|
|
"""
|
|
|
|
if self.is_masked:
|
|
return self.mask_name
|
|
|
|
return self.profile.name
|
|
|
|
@property
|
|
def display_avatar(self) -> str:
|
|
"""Returns the displayed avatar URL.
|
|
|
|
Returns:
|
|
str: The displayed avatar URL.
|
|
"""
|
|
|
|
if self.is_masked:
|
|
mask = constants.Mask(self.mask)
|
|
return f"https://payphone.nyeki.dev/assets/masks/{mask.name.lower().replace("_", "-")}.png"
|
|
|
|
return self.profile.avatar_url
|
|
|
|
@property
|
|
def mask_emoji(self) -> str:
|
|
"""Returns the mask's emoji.
|
|
|
|
Returns:
|
|
str: The mask's emoji (<:name:id>).
|
|
"""
|
|
|
|
mask = constants.Mask(self.mask)
|
|
mask_name = mask.name.lower().replace("_", "")
|
|
|
|
return f"<:{mask_name}:{self.mask}>"
|
|
|
|
class Meta:
|
|
unique_together = [("call_id", "user_id", "channel_id")]
|
|
|
|
|
|
class Message(models.Model):
|
|
id = fields.BigIntField(primary_key=True)
|
|
call: fields.ForeignKeyRelation["Call"] = fields.ForeignKeyField(
|
|
"models.Call", related_name="messages", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
channel: fields.ForeignKeyRelation["Channel"] = fields.ForeignKeyField(
|
|
"models.Channel", related_name="messages", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
user = fields.ForeignKeyRelation("models.User", on_delete=fields.OnDelete.CASCADE)
|
|
user_id: int
|
|
profile: fields.ForeignKeyRelation["Profile"] = fields.ForeignKeyField(
|
|
"models.Profile", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
replying_to: fields.ForeignKeyNullableRelation["Message"] = fields.ForeignKeyField(
|
|
"models.Message",
|
|
related_name="replies",
|
|
on_delete=fields.OnDelete.SET_NULL,
|
|
null=True,
|
|
default=None,
|
|
)
|
|
mask = fields.BigIntField(null=True)
|
|
was_deleted = fields.BooleanField(default=False)
|
|
propagation_id = fields.BigIntField(null=True, db_index=True)
|
|
propagation_channel: fields.ForeignKeyNullableRelation["Channel"] = (
|
|
fields.ForeignKeyField(
|
|
"models.Channel", related_name="propagated_messages", null=True
|
|
)
|
|
)
|
|
created_at = fields.DatetimeField(auto_now_add=True)
|
|
contents: fields.BackwardFKRelation["MessageContent"]
|
|
|
|
@property
|
|
def mask_name(self) -> str:
|
|
"""Returns the mask's name.
|
|
|
|
Returns:
|
|
str: The mask's name.
|
|
"""
|
|
|
|
return constants.Mask(self.mask).name.replace("_", " ").lower().title()
|
|
|
|
@property
|
|
def display_name(self) -> str:
|
|
"""Returns the displayed name.
|
|
|
|
Returns:
|
|
str: The displayed name.
|
|
"""
|
|
|
|
if self.mask is not None:
|
|
return self.mask_name
|
|
|
|
return self.profile.name
|
|
|
|
@property
|
|
def display_avatar(self) -> str:
|
|
"""Returns the displayed avatar URL.
|
|
|
|
Returns:
|
|
str: The displayed avatar URL.
|
|
"""
|
|
|
|
if self.mask is not None:
|
|
mask = constants.Mask(self.mask)
|
|
return f"https://payphone.nyeki.dev/assets/masks/{mask.name.lower().replace("_", "-")}.png"
|
|
|
|
return self.profile.avatar_url
|
|
|
|
@property
|
|
def mask_emoji(self) -> str:
|
|
"""Returns the mask's emoji.
|
|
|
|
Returns:
|
|
str: The mask's emoji (<:name:id>).
|
|
"""
|
|
|
|
mask = constants.Mask(self.mask)
|
|
mask_name = mask.name.lower().replace("_", "")
|
|
|
|
return f"<:{mask_name}:{self.mask}>"
|
|
|
|
def jump_url(self, is_this_channel: bool = True) -> str | None:
|
|
"""Returns the jump URL for this message.
|
|
|
|
Args:
|
|
is_this_channel (bool, optional): Whether the message sent in this channel or not.
|
|
Defaults to True.
|
|
|
|
Returns:
|
|
str | None: The URL, or None if the message was not propagated.
|
|
"""
|
|
|
|
if not is_this_channel:
|
|
return f"https://discord.com/channels/{self.channel.server_id}/{self.channel.id}/{self.id}"
|
|
|
|
if self.propagation_id is not None:
|
|
return f"https://discord.com/channels/{self.propagation_channel.server_id}/{self.propagation_channel.id}/{self.propagation_id}"
|
|
|
|
return None
|
|
|
|
|
|
class MessageContent(models.Model):
|
|
message = fields.ForeignKeyField(
|
|
"models.Message", related_name="contents", on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
content = fields.CharField(max_length=4096)
|
|
was_propagated = fields.BooleanField(default=True)
|
|
created_at = fields.DatetimeField(auto_now_add=True)
|
|
|
|
|
|
class ReportStatus(enum.IntEnum):
|
|
ACCEPTED = 0
|
|
DECLINED = 1
|
|
ON_REVIEW = 2
|
|
|
|
|
|
class Report(models.Model):
|
|
call: fields.OneToOneRelation["Call"] = fields.OneToOneField(
|
|
"models.Call", related_name="report"
|
|
)
|
|
reported_by: fields.ForeignKeyNullableRelation["User"] = fields.ForeignKeyField(
|
|
"models.User", null=True, on_delete=fields.OnDelete.CASCADE
|
|
)
|
|
reason: str = fields.CharField(max_length=500)
|
|
priority: int = fields.IntField()
|
|
status: ReportStatus = fields.IntEnumField(
|
|
ReportStatus, default=ReportStatus.ON_REVIEW, db_index=True
|
|
)
|
|
created_at: datetime = fields.DatetimeField(auto_now_add=True)
|
|
|
|
|
|
class BanStatus(enum.IntEnum):
|
|
VALID = 0
|
|
ON_REVIEW = 1
|
|
APPEAL_ACCEPTED = 2
|
|
APPEAL_DECLINED = 3
|
|
|
|
|
|
class Ban(models.Model):
|
|
user_id: int = fields.BigIntField()
|
|
report: fields.ForeignKeyNullableRelation["Report"] = fields.ForeignKeyField(
|
|
"models.Report",
|
|
related_name="bans",
|
|
null=True,
|
|
on_delete=fields.OnDelete.SET_NULL,
|
|
)
|
|
notification_message_id: int = fields.BigIntField(null=True, db_index=True)
|
|
reason: str = fields.CharField(max_length=500)
|
|
appeal: str | None = fields.CharField(max_length=500, null=True)
|
|
status: BanStatus = fields.IntEnumField(BanStatus, default=BanStatus.VALID)
|
|
banned_by: fields.ForeignKeyRelation["User"] = fields.ForeignKeyField(
|
|
"models.User", related_name="bans_created", null=True
|
|
)
|
|
created_at: datetime = fields.DatetimeField(auto_now_add=True, db_index=True)
|
|
expires_at: datetime | None = fields.DatetimeField(null=True)
|
|
|
|
class Meta:
|
|
indexes = [("user_id", "expires_at")]
|