440 lines
15 KiB
Python
440 lines
15 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import discord
|
|
from discord.ext import commands, pages, tasks
|
|
from discord.utils import find
|
|
|
|
from tortoise import transactions, functions, expressions, exceptions
|
|
|
|
from payphone import bot
|
|
from payphone.db import models, utils
|
|
from payphone.cogs import base
|
|
from payphone.core import constants, formatting as fmt
|
|
|
|
|
|
class Users(base.Cog):
|
|
def __init__(self, bot_instance: bot.Payphone):
|
|
self.bot = bot_instance
|
|
|
|
self.update_leaderboard.add_exception_type(exceptions.ConfigurationError)
|
|
self.update_leaderboard.start()
|
|
|
|
def cog_unload(self) -> None:
|
|
self.update_leaderboard.cancel()
|
|
|
|
@tasks.loop(minutes=15)
|
|
async def update_leaderboard(self) -> None:
|
|
"""Updates the leaderboard."""
|
|
|
|
calls = (
|
|
await models.Call.annotate(messages_count=functions.Count("messages"))
|
|
.filter(
|
|
started_at__month=datetime.now(timezone.utc).month,
|
|
started_at__year=datetime.now(timezone.utc).year,
|
|
messages_count__gte=25,
|
|
)
|
|
.prefetch_related("call_rss", "call_rss__user")
|
|
)
|
|
|
|
# ID to count
|
|
users_count_map: dict[int, int] = {}
|
|
|
|
users_map: dict[int, models.User] = {
|
|
rs.user.id: rs.user for call in calls for rs in call.call_rss
|
|
}
|
|
|
|
for call in calls:
|
|
for rs in call.call_rss:
|
|
if rs.user_id in users_count_map:
|
|
users_count_map[rs.user_id] += 1
|
|
|
|
else:
|
|
users_count_map[rs.user_id] = 0
|
|
|
|
user_counts = list(users_count_map.items())
|
|
|
|
user_counts.sort(key=lambda i: i[1], reverse=True)
|
|
|
|
for i, user_map in enumerate(user_counts):
|
|
users_map[user_map[0]].leaderboard_position = i + 1
|
|
users_map[user_map[0]].leaderboard_calls_count = user_map[1]
|
|
|
|
await models.User.bulk_update(users_map.values(), ["leaderboard_position", "leaderboard_calls_count"], 500)
|
|
|
|
@commands.command()
|
|
@transactions.atomic()
|
|
async def mask(self, ctx: commands.Context, toggle: bool | None = None) -> None:
|
|
user, _ = await utils.get_user(ctx.author)
|
|
|
|
channel_id = ctx.channel.id
|
|
|
|
if toggle is None:
|
|
if user.is_masked:
|
|
channel = await models.Channel.get_or_none(
|
|
id=channel_id
|
|
).select_related("call")
|
|
call_rs = (
|
|
await models.CallUserChannelRelationship.get_or_none(
|
|
channel=channel, user=user, call=channel.call
|
|
)
|
|
if channel is not None
|
|
else None
|
|
)
|
|
|
|
message = "🕵️ You're masked!"
|
|
|
|
if call_rs is not None:
|
|
mask_weared = constants.Mask(call_rs.mask)
|
|
mask_name = mask_weared.name.lower().replace("_", "")
|
|
|
|
message += f" You're currently being displayed as **<:{mask_name}:{call_rs.mask}>**."
|
|
|
|
else:
|
|
message += " Your identity will be hidden during calls."
|
|
|
|
message += f" Use `{ctx.clean_prefix}mask off` to turn it off."
|
|
|
|
await ctx.send(message)
|
|
|
|
else:
|
|
await ctx.send(
|
|
f"🕵️ You're not masked! Use `{ctx.clean_prefix}mask on` to mask yourself."
|
|
)
|
|
|
|
else:
|
|
user.is_masked = toggle
|
|
await user.save()
|
|
|
|
past_user_call_rss = (
|
|
await models.CallUserChannelRelationship.filter(user=user)
|
|
.order_by("-id")
|
|
.limit(5)
|
|
.select_related("call")
|
|
)
|
|
|
|
def update(
|
|
rs: models.CallUserChannelRelationship,
|
|
) -> models.CallUserChannelRelationship:
|
|
rs.is_masked = user.is_masked
|
|
return rs
|
|
|
|
ongoing_call_user_rss = list(
|
|
map(
|
|
update,
|
|
filter(lambda rs: rs.call.ended_at is None, past_user_call_rss),
|
|
)
|
|
)
|
|
|
|
if len(ongoing_call_user_rss) > 0:
|
|
await models.CallUserChannelRelationship.bulk_update(
|
|
ongoing_call_user_rss, ["is_masked"]
|
|
)
|
|
|
|
message = (
|
|
"🕵️ From now on your profile will be "
|
|
+ ("hidden" if user.is_masked else "displayed")
|
|
+ "."
|
|
)
|
|
|
|
current_channel_ongoing_call_user_rs = find(
|
|
lambda rs: rs.channel_id == channel_id, ongoing_call_user_rss
|
|
)
|
|
|
|
if current_channel_ongoing_call_user_rs and user.is_masked:
|
|
message += f" You're currently being displayed as **{current_channel_ongoing_call_user_rs.mask_emoji} {current_channel_ongoing_call_user_rs.display_name}**."
|
|
|
|
message += (
|
|
f" Use `{ctx.clean_prefix}mask` during a call to see your mask."
|
|
)
|
|
|
|
else:
|
|
message += f" Use `{ctx.clean_prefix}mask on` to turn it back on."
|
|
|
|
await ctx.send(message)
|
|
|
|
@commands.command()
|
|
@transactions.atomic()
|
|
async def history(self, ctx: commands.Context) -> None:
|
|
user_id = ctx.author.id
|
|
|
|
call_rss = (
|
|
await models.CallUserChannelRelationship.filter(user_id=user_id)
|
|
.order_by("-id")
|
|
.select_related("profile")
|
|
.limit(30)
|
|
)
|
|
|
|
if len(call_rss) == 0:
|
|
await ctx.send(
|
|
f"☎️ You haven't participated in a call yet! Use `{ctx.clean_prefix}call` to start one."
|
|
)
|
|
return
|
|
|
|
call_ids = [rs.call_id for rs in call_rss]
|
|
|
|
profile = call_rss[0].profile
|
|
|
|
call_rss = (
|
|
await models.CallUserChannelRelationship.filter(call_id__in=call_ids)
|
|
.select_related("profile")
|
|
.order_by("-id")
|
|
.all()
|
|
)
|
|
calls_rss_map = {}
|
|
|
|
for call_rs in call_rss:
|
|
if call_rs.call_id not in calls_rss_map:
|
|
calls_rss_map[call_rs.call_id] = {
|
|
"user_rs": None,
|
|
"others_rs": [],
|
|
"first_message": None,
|
|
}
|
|
|
|
if call_rs.user_id == user_id:
|
|
calls_rss_map[call_rs.call_id]["user_rs"] = call_rs
|
|
|
|
else:
|
|
calls_rss_map[call_rs.call_id]["others_rs"].append(call_rs)
|
|
|
|
min_ids = await (
|
|
models.Message.filter(call_id__in=call_ids)
|
|
.annotate(min_id=functions.Min("id"))
|
|
.group_by("call_id")
|
|
.values_list("min_id", flat=True)
|
|
)
|
|
|
|
# Step 2: Fetch the actual messages using these IDs
|
|
first_messages = (
|
|
await models.Message.filter(id__in=min_ids)
|
|
.select_related("propagation_channel", "channel")
|
|
.all()
|
|
)
|
|
|
|
for msg in first_messages:
|
|
calls_rss_map[msg.call_id]["first_message"] = msg
|
|
|
|
embeds = []
|
|
|
|
logs = ""
|
|
logs_count = 0
|
|
|
|
for call_id, data in calls_rss_map.items():
|
|
logs += f"- `{call_id}`: "
|
|
|
|
participant_names = ", ".join(
|
|
(
|
|
call_rs.display_name
|
|
if call_rs.channel_id != data["user_rs"].channel_id
|
|
else call_rs.profile.name
|
|
)
|
|
for call_rs in data["others_rs"] + [data["user_rs"]]
|
|
)
|
|
|
|
link = None
|
|
|
|
first_message: models.Message | None = data["first_message"]
|
|
|
|
if first_message is not None:
|
|
if first_message.channel_id == data["user_rs"].channel_id:
|
|
link = f"https://discord.com/channels/{first_message.channel.server_id}/{first_message.channel_id}/{first_message.id}"
|
|
|
|
else:
|
|
link = f"https://discord.com/channels/{first_message.propagation_channel.server_id}/{first_message.propagation_channel.id}/{first_message.propagation_id}"
|
|
|
|
if link is not None:
|
|
logs += f"[{participant_names}]({link})"
|
|
|
|
else:
|
|
logs += participant_names
|
|
|
|
logs += "\n"
|
|
logs_count += 1
|
|
|
|
if logs.count("\n") == 10:
|
|
embeds.append(
|
|
discord.Embed(
|
|
title="Call History",
|
|
description="Last calls:\n" + logs,
|
|
thumbnail=profile.avatar_url,
|
|
author=discord.EmbedAuthor(
|
|
name=profile.name, icon_url=profile.avatar_url
|
|
),
|
|
footer=discord.EmbedFooter(text=f"Page {len(embeds) + 1} of 3"),
|
|
)
|
|
)
|
|
|
|
logs = ""
|
|
|
|
if logs != "":
|
|
embeds.append(
|
|
discord.Embed(
|
|
title="Call History",
|
|
description="Last calls:\n" + logs,
|
|
thumbnail=profile.avatar_url,
|
|
author=discord.EmbedAuthor(
|
|
name=profile.name, icon_url=profile.avatar_url
|
|
),
|
|
footer=discord.EmbedFooter(text=f"Page {len(embeds) + 1} of 3"),
|
|
)
|
|
)
|
|
|
|
paginator = pages.Paginator(
|
|
embeds,
|
|
show_indicator=False,
|
|
use_default_buttons=False,
|
|
custom_buttons=[
|
|
pages.PaginatorButton(
|
|
button_type="prev", label="⬅️", style=discord.ButtonStyle.blurple
|
|
),
|
|
pages.PaginatorButton(
|
|
button_type="next", label="➡️", style=discord.ButtonStyle.blurple
|
|
),
|
|
],
|
|
)
|
|
|
|
await paginator.send(ctx)
|
|
|
|
@commands.command()
|
|
@transactions.atomic()
|
|
async def me(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Shows your statistics.
|
|
"""
|
|
|
|
user, profile = await utils.get_user(ctx.author)
|
|
|
|
messages_count = await models.Message.filter(user=user).count()
|
|
|
|
channel_ids = (
|
|
await models.CallUserChannelRelationship.filter(user=user)
|
|
.distinct()
|
|
.values_list("channel_id", flat=True)
|
|
)
|
|
call_ids = (
|
|
await models.CallUserChannelRelationship.filter(user=user)
|
|
.distinct()
|
|
.values_list("call_id", flat=True)
|
|
)
|
|
|
|
people_met_count = await models.CallUserChannelRelationship.filter(
|
|
call_id__in=call_ids, channel_id__not_in=channel_ids
|
|
).count()
|
|
ban = (
|
|
await models.Ban.filter(
|
|
expressions.Q(
|
|
expires_at__gt=datetime.now(timezone.utc),
|
|
expires_at=None,
|
|
join_type="OR",
|
|
),
|
|
user_id=user.id,
|
|
)
|
|
.limit(1)
|
|
.first()
|
|
)
|
|
|
|
embed = discord.Embed(
|
|
color=constants.EMBED_COLOR,
|
|
timestamp=datetime.now(timezone.utc),
|
|
fields=[
|
|
discord.EmbedField(
|
|
name="Messages", value=str(messages_count), inline=True
|
|
),
|
|
discord.EmbedField(name="Calls", value=len(call_ids), inline=True),
|
|
discord.EmbedField(
|
|
name="People Met", value=people_met_count, inline=True
|
|
),
|
|
discord.EmbedField(
|
|
name="Joined At",
|
|
value=f"<t:{int(user.joined_at.timestamp())}:d> <t:{int(user.joined_at.timestamp())}:T>",
|
|
inline=True,
|
|
),
|
|
discord.EmbedField(
|
|
name="Leaderboard",
|
|
value=(
|
|
f"#{user.leaderboard_position} (Monthly)"
|
|
if user.leaderboard_position is not None
|
|
else "*No calls yet.*"
|
|
),
|
|
inline=True,
|
|
),
|
|
discord.EmbedField(
|
|
name="Banned",
|
|
value=(
|
|
"*Not banned.*"
|
|
if ban is None
|
|
else (
|
|
"*Permanent.*"
|
|
if ban.expires_at is None
|
|
else f"<t:{int(ban.expires_at.timestamp())}:F> (<t:{int(ban.expires_at.timestamp())}:R>)"
|
|
)
|
|
),
|
|
inline=True,
|
|
),
|
|
],
|
|
author=discord.EmbedAuthor(name=profile.name, icon_url=profile.avatar_url),
|
|
footer=discord.EmbedFooter(
|
|
text="Thank you for using Payphone! Leaderboard updates every 15 minutes."
|
|
),
|
|
thumbnail=profile.avatar_url,
|
|
)
|
|
|
|
await ctx.send(embed=embed)
|
|
|
|
@commands.command(aliases=["l", "leader"])
|
|
async def leaderboard(self, ctx: commands.Context) -> None:
|
|
"""
|
|
Shows the leaderboard for the month.
|
|
"""
|
|
|
|
top_users = (
|
|
await models.User.filter(leaderboard_position__lte=10)
|
|
.order_by("leaderboard_position")
|
|
.limit(10)
|
|
)
|
|
|
|
rows = [
|
|
[
|
|
f"#{pos + 1}",
|
|
fmt.trim(user.username, 13),
|
|
f"{user.leaderboard_calls_count if user.leaderboard_calls_count is not None else "?"} call{"s" if user.leaderboard_calls_count != 1 else ""}",
|
|
]
|
|
for pos, user in enumerate(top_users)
|
|
]
|
|
|
|
if not any(map(lambda u: u.id == ctx.author.id, top_users)):
|
|
user, _ = await utils.get_user(ctx.author)
|
|
|
|
if user.leaderboard_position is not None:
|
|
rows.append(
|
|
[
|
|
f"#{user.leaderboard_position}",
|
|
fmt.trim(user.username, 13),
|
|
f"{user.leaderboard_calls_count if user.leaderboard_calls_count is not None else "?"} call{"s" if user.leaderboard_calls_count != 1 else ""}",
|
|
]
|
|
)
|
|
|
|
table = fmt.monospace_table(rows)
|
|
|
|
embed = discord.Embed(
|
|
color=constants.EMBED_COLOR,
|
|
title=f"Monthly Leaderboard ({datetime.now(timezone.utc).strftime("%B")})",
|
|
description=f"```\n{table if len(rows) > 0 else "Come back in 15 minutes!"}\n```",
|
|
author=self.bot.embed_author,
|
|
footer=discord.EmbedFooter(
|
|
text="The leaderboard is updated every 15 minutes."
|
|
),
|
|
thumbnail=self.bot.user.display_avatar.url,
|
|
)
|
|
|
|
await ctx.send(embed=embed)
|
|
|
|
|
|
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(Users(b))
|