typing + transition to miniscord
This commit is contained in:
klemek
@@ -1,9 +1,9 @@
|
||||
from typing import Dict, List, Tuple
|
||||
from datetime import datetime
|
||||
import discord
|
||||
import re
|
||||
|
||||
# Custom libs
|
||||
import help
|
||||
from utils import debug, aggregate, no_duplicate
|
||||
|
||||
# CONSTANTS
|
||||
@@ -13,22 +13,18 @@ CHUNK_SIZE = 10000
|
||||
|
||||
# MAIN
|
||||
|
||||
async def compute(message, args):
|
||||
|
||||
async def compute(client: discord.client, message: discord.Message, *args: str):
|
||||
"""
|
||||
Computes the %emotes command
|
||||
|
||||
:param message: message sent
|
||||
:type message: :class:`discord.Message`
|
||||
:param args: arguments of the command
|
||||
:type args: list[:class:`str`]
|
||||
"""
|
||||
|
||||
guild = message.guild
|
||||
|
||||
# If "%emotes help" redirect to "%help emotes"
|
||||
if len(args) > 1 and args[1] == "help":
|
||||
await help.compute(message, ["%help", "emotes"])
|
||||
return
|
||||
# TODO in miniscord
|
||||
# # If "%emotes help" redirect to "%help emotes"
|
||||
# if len(args) > 1 and args[1] == "help":
|
||||
# await help.compute(message, ["%help", "emotes"])
|
||||
# return
|
||||
|
||||
# Create emotes dict from custom emojis of the guild
|
||||
emotes = {str(emoji): Emote(emoji) for emoji in guild.emojis}
|
||||
@@ -52,7 +48,9 @@ async def compute(message, args):
|
||||
progress = await message.channel.send(f"```starting analysis...```")
|
||||
# Analyse every channel selected
|
||||
for channel in channels:
|
||||
nm1, nmm1 = await analyse_channel(channel, emotes, members, progress, nm, nc)
|
||||
nm1, nmm1 = await analyse_channel(
|
||||
channel, emotes, members, progress, nm, nc
|
||||
)
|
||||
# If treatment was successful, increase numbers
|
||||
if nm1 >= 0:
|
||||
nm += nm1
|
||||
@@ -61,75 +59,52 @@ async def compute(message, args):
|
||||
# Delete custom progress message
|
||||
await progress.delete()
|
||||
# Display results
|
||||
await tell_results(get_intro(emotes, full, channels, members, nmm, nc),
|
||||
emotes, message.channel, nmm, allow_unused=full and len(members) == 0, show_life=False)
|
||||
await tell_results(
|
||||
get_intro(emotes, full, channels, members, nmm, nc),
|
||||
emotes,
|
||||
message.channel,
|
||||
nmm,
|
||||
allow_unused=full and len(members) == 0,
|
||||
show_life=False,
|
||||
)
|
||||
dt = (datetime.now() - t0).total_seconds()
|
||||
debug(message, f"response sent {dt} s -> {nm / dt} m/s")
|
||||
|
||||
|
||||
# CLASSES
|
||||
|
||||
|
||||
class Emote:
|
||||
"""
|
||||
Custom class to store emotes data
|
||||
|
||||
:param emoji: source discord emoji
|
||||
:type emoji: :class:`discord.Emoji`
|
||||
:ivar emoji: discord emoji
|
||||
:vartype emoji: discord.Emoji
|
||||
:ivar usages: number of usages in messages
|
||||
:vartype usages: int
|
||||
:ivar reactions: number of reactions below messages (always 0 if member specified)
|
||||
:vartype reactions: int
|
||||
:ivar last_used: date of last use
|
||||
:vartype last_used: datetime
|
||||
"""
|
||||
|
||||
def __init__(self, emoji):
|
||||
def __init__(self, emoji: discord.Emoji):
|
||||
self.emoji = emoji
|
||||
self.usages = 0
|
||||
self.reactions = 0
|
||||
self.last_used = None
|
||||
|
||||
def update_use(self, date):
|
||||
def update_use(self, date: datetime):
|
||||
"""
|
||||
Update last use date if more recent
|
||||
|
||||
:param date: date of use
|
||||
:type date: datetime
|
||||
"""
|
||||
if self.last_used is None or date > self.last_used:
|
||||
self.last_used = date
|
||||
|
||||
def used(self):
|
||||
"""
|
||||
:return: If this emote was used ever
|
||||
:rtype: bool
|
||||
"""
|
||||
def used(self) -> bool:
|
||||
return self.usages > 0 or self.reactions > 0
|
||||
|
||||
def score(self):
|
||||
"""
|
||||
:return: Score of this emote to be sorted
|
||||
:rtype: float
|
||||
"""
|
||||
def score(self) -> float:
|
||||
# Score is compose of usages + reactions
|
||||
# When 2 emotes have the same score, the days since last use is stored in the digits
|
||||
# (more recent first)
|
||||
return self.usages + self.reactions + 1 / (100000 * (self.use_days() + 1))
|
||||
|
||||
def life_days(self):
|
||||
"""
|
||||
:return: Days since creation
|
||||
:rtype: int
|
||||
"""
|
||||
def life_days(self) -> int:
|
||||
return (datetime.today() - self.emoji.created_at).days
|
||||
|
||||
def use_days(self):
|
||||
"""
|
||||
:return: Days since last use
|
||||
:rtype: int
|
||||
"""
|
||||
def use_days(self) -> int:
|
||||
# If never used, use creation date instead
|
||||
if self.last_used is None:
|
||||
return self.life_days()
|
||||
@@ -139,37 +114,33 @@ class Emote:
|
||||
|
||||
# ANALYSIS
|
||||
|
||||
async def analyse_channel(channel, emotes, members, progress, nm0, nc):
|
||||
"""
|
||||
Analyses a given channel
|
||||
|
||||
:param channel: channel to analyse
|
||||
:type channel: discord.TextChannel
|
||||
:param emotes: known emotes
|
||||
:type emotes: dict[str, Emote]
|
||||
:param members: selected members or empty for all
|
||||
:type members: list[discord.Member]
|
||||
:param progress: custom progress message (already sent)
|
||||
:type progress: discord.Message
|
||||
:param nm0: number of already analysed messages
|
||||
:type nm0: int
|
||||
:param nc: number of already analysed channels
|
||||
:type nc: int
|
||||
:return: nm, nc (-1 on exception)
|
||||
:rtype: int, int
|
||||
"""
|
||||
async def analyse_channel(
|
||||
channel: discord.TextChannel,
|
||||
emotes: Dict[str, Emote],
|
||||
members: List[discord.Member],
|
||||
progress: discord.Message,
|
||||
nm0: int, # number of already analysed messages
|
||||
nc: int, # number of already analysed channels
|
||||
) -> Tuple[int, int]:
|
||||
nm = 0
|
||||
nmm = 0
|
||||
try:
|
||||
messages = [None]
|
||||
while len(messages) >= CHUNK_SIZE or messages[-1] is None:
|
||||
messages = await channel.history(limit=CHUNK_SIZE, before=messages[-1]).flatten()
|
||||
messages = await channel.history(
|
||||
limit=CHUNK_SIZE, before=messages[-1]
|
||||
).flatten()
|
||||
for m in messages:
|
||||
# If author is not bot or included in the selection (empty list is all)
|
||||
if not m.author.bot and (len(members) == 0 or m.author in members):
|
||||
# Find all emotes un the current message in the form "<:emoji:123456789>"
|
||||
# Filter for known emotes
|
||||
found = [name for name in re.findall(r"(<:\w+:\d+>)", m.content) if name in emotes]
|
||||
found = [
|
||||
name
|
||||
for name in re.findall(r"(<:\w+:\d+>)", m.content)
|
||||
if name in emotes
|
||||
]
|
||||
# For each emote, update its usage
|
||||
for name in found:
|
||||
emotes[name].usages += 1
|
||||
@@ -191,7 +162,9 @@ async def analyse_channel(channel, emotes, members, progress, nm0, nc):
|
||||
emotes[name].reactions += 1
|
||||
emotes[name].update_use(m.created_at)"""
|
||||
nm += len(messages)
|
||||
await progress.edit(content=f"```{nm0 + nm:,} messages and {nc} channels analysed```")
|
||||
await progress.edit(
|
||||
content=f"```{nm0 + nm:,} messages and {nc} channels analysed```"
|
||||
)
|
||||
return nm, nmm
|
||||
except discord.errors.HTTPException:
|
||||
# When an exception occurs (like Forbidden) sent -1
|
||||
@@ -200,23 +173,16 @@ async def analyse_channel(channel, emotes, members, progress, nm0, nc):
|
||||
|
||||
# RESULTS
|
||||
|
||||
async def tell_results(intro, emotes, channel, nmm, *, allow_unused, show_life):
|
||||
"""
|
||||
Send the full results message
|
||||
|
||||
:param intro: introduction sentence (from get_intro)
|
||||
:type intro: str
|
||||
:param emotes: known emotes
|
||||
:type emotes: dict[str, Emote]
|
||||
:param channel: where to send the message (original message channel)
|
||||
:type channel: discord.TextChannel
|
||||
:param nmm: number of impacted messages
|
||||
:type nmm: int
|
||||
:param allow_unused: show unused emotes
|
||||
:type allow_unused: bool
|
||||
:param show_life: show emotes life span
|
||||
:type show_life: bool
|
||||
"""
|
||||
async def tell_results(
|
||||
intro: str, # introduction sentence (from get_intro)
|
||||
emotes: Dict[str, Emote],
|
||||
channel: discord.TextChannel,
|
||||
nmm: int, # number of impacted messages
|
||||
*,
|
||||
allow_unused: bool,
|
||||
show_life: bool,
|
||||
):
|
||||
names = [name for name in emotes]
|
||||
names.sort(key=lambda name: emotes[name].score(), reverse=True)
|
||||
res = [intro]
|
||||
@@ -226,7 +192,9 @@ async def tell_results(intro, emotes, channel, nmm, *, allow_unused, show_life):
|
||||
f"{get_reactions(emotes[name])}"
|
||||
f"{get_life(emotes[name], show_life)}"
|
||||
f"{get_last_used(emotes[name])}"
|
||||
for name in names if allow_unused or emotes[name].used()]
|
||||
for name in names
|
||||
if allow_unused or emotes[name].used()
|
||||
]
|
||||
res += [get_total(emotes, nmm)]
|
||||
response = ""
|
||||
for r in res:
|
||||
@@ -238,24 +206,16 @@ async def tell_results(intro, emotes, channel, nmm, *, allow_unused, show_life):
|
||||
await channel.send(response)
|
||||
|
||||
|
||||
def get_intro(emotes, full, channels, members, nmm, nc):
|
||||
def get_intro(
|
||||
emotes: Dict[str, Emote],
|
||||
full: bool,
|
||||
channels: List[discord.TextChannel],
|
||||
members: List[discord.Member],
|
||||
nmm: int, # number of messages impacted
|
||||
nc: int, # number of channels analysed
|
||||
) -> str:
|
||||
"""
|
||||
Get the introduction sentence of the response
|
||||
|
||||
:param emotes: known emotes
|
||||
:type emotes: dict[str, Emote]
|
||||
:param full: if the scan contained all channels
|
||||
:type full: bool
|
||||
:param channels: channels selected (ignored if full is True)
|
||||
:type channels: list[discord.TextChannel]
|
||||
:param members: members selected (empty for all)
|
||||
:type members: list[discord.Member]
|
||||
:param nmm: number of messages impacted
|
||||
:type nmm: int
|
||||
:param nc: number of channels analysed
|
||||
:type nc: int
|
||||
:return: the correct intro sentence
|
||||
:rtype: str
|
||||
"""
|
||||
# Show all data (members, channels) when it's less than 5 units
|
||||
if len(members) == 0:
|
||||
@@ -270,30 +230,33 @@ def get_intro(emotes, full, channels, members, nmm, nc):
|
||||
if full:
|
||||
return f"{aggregate([m.mention for m in members])} emotes usage in {nmm:,} messages:"
|
||||
elif len(channels) < 5:
|
||||
return f"{aggregate([m.mention for m in members])} on {aggregate([c.mention for c in channels])} " \
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
return (
|
||||
f"{aggregate([m.mention for m in members])} on {aggregate([c.mention for c in channels])} "
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
)
|
||||
else:
|
||||
return f"{aggregate([m.mention for m in members])} on these {len(channels)} channels " \
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
return (
|
||||
f"{aggregate([m.mention for m in members])} on these {len(channels)} channels "
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
)
|
||||
else:
|
||||
if full:
|
||||
return f"These {len(members)} members emotes usage in {nmm:,} messages:"
|
||||
elif len(channels) < 5:
|
||||
return f"These {len(members)} members on {aggregate([c.mention for c in channels])} " \
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
return (
|
||||
f"These {len(members)} members on {aggregate([c.mention for c in channels])} "
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
)
|
||||
else:
|
||||
return f"These {len(members)} members on these {len(channels)} channels " \
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
return (
|
||||
f"These {len(members)} members on these {len(channels)} channels "
|
||||
f"emotes usage in {nmm:,} messages:"
|
||||
)
|
||||
|
||||
|
||||
def get_place(i):
|
||||
def get_place(i: int) -> str:
|
||||
"""
|
||||
Get the correct rank displayed (1st to 3rd have an emoji)
|
||||
|
||||
:param i: index
|
||||
:type i: int
|
||||
:return: rank string
|
||||
:rtype: str
|
||||
"""
|
||||
if i == 0:
|
||||
return ":first_place:"
|
||||
@@ -304,13 +267,9 @@ def get_place(i):
|
||||
return f"**#{i + 1}**"
|
||||
|
||||
|
||||
def get_usage(emote):
|
||||
def get_usage(emote: Emote) -> str:
|
||||
"""
|
||||
Get the correct usage displayed
|
||||
|
||||
:type emote: Emote
|
||||
:return: usage description
|
||||
:rtype: str
|
||||
"""
|
||||
if emote.usages == 0 and emote.reactions == 0:
|
||||
return "never used "
|
||||
@@ -320,12 +279,9 @@ def get_usage(emote):
|
||||
return f"{emote.usages:,} times "
|
||||
|
||||
|
||||
def get_reactions(emote):
|
||||
def get_reactions(emote: Emote) -> str:
|
||||
"""
|
||||
Get the correct reactions displayed
|
||||
|
||||
:return: reactions description
|
||||
:rtype: str
|
||||
"""
|
||||
if emote.reactions == 0:
|
||||
return ""
|
||||
@@ -335,15 +291,9 @@ def get_reactions(emote):
|
||||
return f"and {emote.reactions:,} reactions "
|
||||
|
||||
|
||||
def get_life(emote, show_life):
|
||||
def get_life(emote: Emote, show_life: bool) -> str:
|
||||
"""
|
||||
Get the correct life span displayed
|
||||
|
||||
:type emote: Emote
|
||||
:param show_life: disable if False
|
||||
:type show_life: bool
|
||||
:return: life description
|
||||
:rtype: str
|
||||
"""
|
||||
if not show_life:
|
||||
return ""
|
||||
@@ -351,13 +301,9 @@ def get_life(emote, show_life):
|
||||
return f"(in {emote.life_days()} days) "
|
||||
|
||||
|
||||
def get_last_used(emote):
|
||||
def get_last_used(emote: Emote) -> str:
|
||||
"""
|
||||
Get the correct "last used" displayed
|
||||
|
||||
:type emote: Emote
|
||||
:return: last usage description
|
||||
:rtype: str
|
||||
"""
|
||||
if emote.usages == 0 and emote.reactions == 0:
|
||||
return ""
|
||||
@@ -369,16 +315,9 @@ def get_last_used(emote):
|
||||
return f"(last used {emote.use_days()} days ago)"
|
||||
|
||||
|
||||
def get_total(emotes, nmm):
|
||||
def get_total(emotes: Dict[str, Emote], nmm: int) -> str:
|
||||
"""
|
||||
Get the total of all emotes used
|
||||
|
||||
:param emotes: known emotes
|
||||
:type emotes: dict[str, Emote]
|
||||
:param nmm: number of messages impacted
|
||||
:type nmm: int
|
||||
:return: total sentence
|
||||
:rtype: str
|
||||
"""
|
||||
nu = 0
|
||||
nr = 0
|
||||
|
||||
Reference in New Issue
Block a user