592 lines
19 KiB
Python
592 lines
19 KiB
Python
import io
|
|
|
|
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
|
|
|
|
import colorthief
|
|
|
|
import httpx
|
|
|
|
from payphone import bot
|
|
from payphone.db import models, utils
|
|
from payphone.cogs import base
|
|
from payphone.core import constants
|
|
|
|
|
|
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.Cog.listener()
|
|
@transactions.atomic()
|
|
async def on_user_update(self, _before: discord.User, after: discord.User) -> None:
|
|
"""Updates a user's Profile.
|
|
|
|
Args:
|
|
_before (discord.User): The user before the change.
|
|
after (discord.User): The user after the change.
|
|
"""
|
|
|
|
profile = await models.Profile.get_or_none(user_id=after.id, server_id=None)
|
|
|
|
is_updated = False
|
|
|
|
if profile is None:
|
|
return
|
|
|
|
if profile.avatar_url != after.display_avatar.url:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(after.display_avatar.url)
|
|
|
|
if response.status_code not in range(200, 300):
|
|
return
|
|
|
|
f = io.BytesIO(response.content)
|
|
|
|
(r, g, b) = colorthief.ColorThief(f).get_color(quality=1)
|
|
|
|
profile.color = (r << 16) | (g << 8) | b
|
|
|
|
is_updated = True
|
|
|
|
if profile.name != after.display_name:
|
|
profile.name = after.display_name
|
|
is_updated = True
|
|
|
|
if is_updated:
|
|
await profile.save()
|
|
|
|
@commands.Cog.listener()
|
|
@transactions.atomic()
|
|
async def on_member_update(
|
|
self, _before: discord.Member, after: discord.Member
|
|
) -> None:
|
|
"""Updates a member's profile.
|
|
|
|
Args:
|
|
_before (discord.Member): The member's profile before the update.
|
|
after (discord.Member): The member's profile after the update.
|
|
"""
|
|
|
|
profile = await models.Profile.get_or_none(
|
|
user_id=after.id, server_id=after.guild.id
|
|
)
|
|
|
|
is_updated = False
|
|
|
|
if profile is None:
|
|
return
|
|
|
|
if profile.avatar_url != after.display_avatar.url:
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.get(after.display_avatar.url)
|
|
|
|
if response.status_code not in range(200, 300):
|
|
return
|
|
|
|
f = io.BytesIO(response.content)
|
|
|
|
(r, g, b) = colorthief.ColorThief(f).get_color(quality=1)
|
|
|
|
profile.color = (r << 16) | (g << 8) | b
|
|
|
|
is_updated = True
|
|
|
|
if profile.name != after.display_name:
|
|
profile.name = after.display_name
|
|
is_updated = True
|
|
|
|
if is_updated:
|
|
await profile.save()
|
|
|
|
@commands.command()
|
|
@transactions.atomic()
|
|
async def mask(self, ctx: commands.Context, toggle: bool | None = None) -> None:
|
|
"""
|
|
Hide your identity during calls with a mask.
|
|
|
|
Options:
|
|
- `[<toggle>]`: On or off. If not set, your current status and mask will be displayed.
|
|
"""
|
|
|
|
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:
|
|
message += f" You're currently being displayed as **{call_rs.mask_emoji} {call_rs.display_name}**."
|
|
|
|
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 toggle 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 toggle:
|
|
message += f" Your profile will be 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:
|
|
if toggle:
|
|
message += f" Use `{ctx.clean_prefix}mask off` to turn it off."
|
|
|
|
else:
|
|
message += f" Use `{ctx.clean_prefix}mask on` to turn it on."
|
|
|
|
await ctx.send(message)
|
|
|
|
@commands.command()
|
|
@transactions.atomic()
|
|
async def history(self, ctx: commands.Context) -> None:
|
|
"""
|
|
See your calls history. Links will take you to the beginning of the call.
|
|
"""
|
|
|
|
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)
|
|
.exclude(propagation_id=None)
|
|
.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"- `ID: {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(
|
|
color=profile.color,
|
|
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 user 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()
|
|
)
|
|
|
|
reviewed_reports = (
|
|
(await models.Report.filter(reviewed_by_id=user.id).count())
|
|
if user.is_moderator
|
|
else None
|
|
)
|
|
bans_made = (
|
|
(await models.Ban.filter(banned_by_id=user.id).count())
|
|
if user.is_moderator
|
|
else None
|
|
)
|
|
bans_unmade = (
|
|
(await models.Ban.filter(unbanned_by_id=user.id).count())
|
|
if user.is_moderator
|
|
else None
|
|
)
|
|
|
|
embed = discord.Embed(
|
|
color=profile.color or 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"Until <t:{int(ban.expires_at.timestamp())}:F> (<t:{int(ban.expires_at.timestamp())}:R>)"
|
|
)
|
|
),
|
|
inline=True,
|
|
),
|
|
]
|
|
+ (
|
|
[
|
|
discord.EmbedField(
|
|
name="Reviewed Reports", value=reviewed_reports, inline=True
|
|
),
|
|
discord.EmbedField(name="Bans Made", value=bans_made, inline=True),
|
|
discord.EmbedField(
|
|
name="Bans Unmade", value=bans_unmade, inline=True
|
|
),
|
|
]
|
|
if user.is_moderator
|
|
else []
|
|
),
|
|
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", "ranking"])
|
|
async def leaderboard(self, ctx: commands.Context, board: str = "users") -> None:
|
|
"""
|
|
Shows the leaderboard for the month.
|
|
|
|
Options:
|
|
- `[<board>]`: The board to show (users/servers). Defaults to users.
|
|
"""
|
|
|
|
users_board_options = ["u", "user", "users"]
|
|
server_board_options = ["s", "server", "servers", "g", "guild", "guilds"]
|
|
|
|
board = board.lower()
|
|
|
|
if board not in users_board_options + server_board_options:
|
|
await ctx.send(
|
|
f"❌ This leaderboard does not exist. The leaderboards are `servers` and `users`, for example `{ctx.clean_prefix}leaderboard users`."
|
|
)
|
|
return
|
|
|
|
table = ""
|
|
table_name = "users" if board in users_board_options else "servers"
|
|
|
|
if board in users_board_options:
|
|
top_users = (
|
|
await models.User.filter(leaderboard_position__range=[1, 5])
|
|
.prefetch_related("profiles")
|
|
.order_by("leaderboard_position")
|
|
.limit(5)
|
|
)
|
|
user_rows = [
|
|
f"**#{pos + 1} - {user.profiles[0].name}**{" (You!)" if user.id == ctx.author.id else ""}\nParticipated in {user.leaderboard_calls_count} calls."
|
|
for pos, user in enumerate(top_users)
|
|
]
|
|
|
|
table = "__**Top Users:**__\n\n" + "\n\n".join(user_rows)
|
|
|
|
else:
|
|
top_servers = (
|
|
await models.Server.filter(leaderboard_position__range=[1, 5])
|
|
.order_by("leaderboard_position")
|
|
.limit(5)
|
|
)
|
|
server_rows = [
|
|
f"**#{pos + 1} - {server.name or "Updating..."}{f" · .gg/{server.vanity_code}" if server.vanity_code else ""}**{" (Here!)" if ctx.guild is not None and server.id == ctx.guild.id else ""}\nParticipated in {server.leaderboard_calls_count} calls."
|
|
for pos, server in enumerate(top_servers)
|
|
]
|
|
|
|
table = "__**Top Servers:**__\n\n" + "\n\n".join(server_rows)
|
|
|
|
embed = discord.Embed(
|
|
color=constants.EMBED_COLOR,
|
|
title=f"Monthly Leaderboard ({datetime.now(timezone.utc).strftime("%B")})",
|
|
description=(
|
|
f"Based on the amount of calls {table_name} have participated in with more than 25 "
|
|
"messages sent (to prevent spam).\n\n"
|
|
)
|
|
+ table,
|
|
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))
|