klemek/discord-analyst
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

moved sources to src folder

Browse Source
This commit is contained in:
klemek
2021-01-08 12:31:58 +01:00
parent 4db8ee0b52
commit 6d3d1118f1
11 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/bot.py
+26
View File
@@ -0,0 +1,26 @@
from miniscord import Bot
import logging
import emojis
from emotes import EmotesScanner
logging.basicConfig(
format="[%(asctime)s][%(levelname)s][%(module)s] %(message)s", level=logging.INFO
)
emojis.load_emojis()
bot = Bot(
"Discord Analyst", # name
"1.6", # version
alias="%", # respond to '|command' messages
)
bot.log_calls = True
bot.client.bot = bot # TODO place in miniscord
bot.register_command(
"emotes", # command text (regex)
lambda *args: EmotesScanner().compute(*args), # command function
"emotes: Emotes analysis", # short help
EmotesScanner.help(),
)
bot.start()
src/composition_scanner.py
View File
src/data_types.py
+79
View File
@@ -0,0 +1,79 @@
from typing import List, Optional
from datetime import datetime
from collections import defaultdict
import discord
# Custom libs
from utils import mention, plural, day_interval
class Emote:
"""
Custom class to store emotes data
"""
def __init__(self, emoji: Optional[discord.Emoji] = None):
self.emoji = emoji
self.usages = 0
self.reactions = 0
self.last_used = None
self.members = defaultdict(int)
def update_use(self, date: datetime, members_id: List[int]):
"""
Update last use date if more recent and last member
"""
if self.last_used is None or date > self.last_used:
self.last_used = date
for member_id in members_id:
self.members[member_id] += 1
def used(self) -> bool:
return self.usages > 0 or self.reactions > 0
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) -> int:
return (datetime.today() - self.emoji.created_at).days
def use_days(self) -> int:
# If never used, use creation date instead
if self.last_used is None:
return self.life_days()
else:
return (datetime.today() - self.last_used).days
def get_top_member(self) -> int:
return sorted(self.members.keys(), key=lambda id: self.members[id])[-1]
def to_string(self, i: int, name: str, show_life: bool, show_members: bool) -> str:
# place
output = ""
if i == 0:
output += ":first_place:"
elif i == 1:
output += ":second_place:"
elif i == 2:
output += ":third_place:"
else:
output += f"**#{i + 1}**"
output += f" {name} - "
if not self.used():
output += "never used "
else:
output += f"{plural(self.usages, 'time')} "
if self.reactions >= 1:
output += f"and {plural(self.usages, 'reaction')} "
if show_life and not self.default:
output += f"(in {plural(self.life_days(), 'day')}) "
if self.used():
output += f"(last used {day_interval(self.use_days())})"
if show_members:
output += f" (mostly by {mention(self.get_top_member())}: {self.members[self.get_top_member()]})"
return output
src/emoji.json
+1
View File
File diff suppressed because one or more lines are too long
src/emojis.py
+86
View File
@@ -0,0 +1,86 @@
import re
import json
import logging
EXTRA_EMOJI = {
"thumbup": "1f44d",
"thumbdown": "1f44e",
"timer": "23f2-fe0f",
"cowboy": "1f920",
"clown": "1f921",
"newspaper2": "1f5de-fe0f",
"french_bread": "1f956",
"nerd": "1f913",
"zipper_mouth": "1f910",
"salad": "1f957",
"rolling_eyes": "1f644",
"basketball_player": "26f9-fe0f-200d-2642-fe0f",
"thinking": "1f914",
"e_mail": "2709-fe0f",
"slight_frown": "1f641",
"skull_crossbones": "2620-fe0f",
"hand_splayed": "1f590-fe0f",
"speaking_head": "1f5e3-fe0f",
"cross": "271d-fe0f",
"crayon": "1f58d-fe0f",
"head_bandage": "1f915",
"rofl": "1f923",
"flag_white": "1f3f3-fe0f",
"slight_smile": "1f642",
"fork_knife_plate": "1f37d-fe0f",
"robot": "1f916",
"hugging": "1f917",
"biohazard": "2623-fe0f",
"notepad_spiral": "1f5d2-fe0f",
"lifter": "1f3cb-fe0f-200d-2642-fe0f",
"race_car": "1f3ce-fe0f",
"left_facing_fist": "1f91b",
"right_facing_fist": "1f91c",
"tools": "1f6e0-fe0f",
"umbrella2": "2602-fe0f",
"upside_down": "2b07-fe0f",
"first_place": "1f947",
"dagger": "1f5e1-fe0f",
"fox": "1f98a",
"menorah": "1f54e",
"desktop": "1f5a5-fe0f",
"motorcycle": "1f3cd-fe0f",
"levitate": "1f574-fe0f",
"cheese": "1f9c0",
"fingers_crossed": "1f91e",
"frowning2": "1f626",
"microphone2": "1f399-fe0f",
"flag_black": "1f3f4",
"chair": "1FA91",
"champagne_glass": "1F942",
"raised_hand": "270B",
"knife": "1F52A",
"postal_horn": "1F4EF",
"punch": "1F44A",
}
global_list = {}
unicode_list = []
regex = re.compile("(<a?:[\\w\\-\\~]+:\\d+>|:[\\w\\-\\~]+:)")
def load_emojis():
global global_list, unicode_list, regex
emoji_list = []
with open("emoji.json", mode="r") as f:
emoji_list = json.loads(f.readline().strip())
for emoji in EXTRA_EMOJI:
emoji_list += [{"short_name": emoji, "unified": EXTRA_EMOJI[emoji]}]
unicode_list_escaped = []
for emoji in emoji_list:
shortname = emoji["short_name"]
unified = emoji["unified"]
if unified is not None and shortname is not None:
unicode_escaped = "".join([f"\\U{c:0>8}" for c in unified.split("-")])
unicode = bytes(unicode_escaped, "ascii").decode("unicode-escape")
shortcode = shortname.replace("-", "_")
global_list[unicode] = f":{shortcode}:"
unicode_list += [unicode]
unicode_list_escaped += [unicode_escaped]
regex = re.compile(f"(<a?:\\w+:\\d+>|:\\w+:|{'|'.join(unicode_list_escaped)})")
logging.info(f"loaded {len(unicode_list)} emojis")
src/emotes_scanner.py
+120
View File
@@ -0,0 +1,120 @@
from typing import Dict, List
import discord
# Custom libs
from log_manager import ChannelLogs, MessageLog
from data_types import Emote
from scanner import Scanner
import emojis
class EmotesScanner(Scanner):
@staticmethod
def help() -> str:
return "```\n"
+"%emotes : Rank emotes by their usage\n"
+"arguments:\n"
+"* @member : filter for one or more member\n"
+"* #channel : filter for one or more channel\n"
+"* <n> : top <n> emojis, default is 20\n"
+"* all : list all common emojis in addition to this guild's\n"
+"* members : show top member for each emote\n"
+"Example: %emotes 10 all #mychannel1 #mychannel2 @user\n"
+"```"
def __init__(self):
super().__init__(
has_digit_args=True,
valid_args=["all", "members"],
help=EmotesScanner.help(),
intro_context="emotes usage",
)
self.top = 20
self.all_emojis = False
self.show_members = False
async def compute_args(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 20
for arg in args:
if arg.isdigit():
self.top = int(arg)
# check other args
self.all_emojis = "all" in args
self.show_members = "members" in args and (
len(self.members) == 0 or len(self.members) > 1
)
return True
def compute_message(self, channel: ChannelLogs, message: MessageLog):
return analyse_message(
message, self.emotes, self.raw_members, all_emojis=self.all_emojis
)
def get_results(self, intro: str) -> List[str]:
names = [name for name in self.emotes]
names.sort(key=lambda name: self.emotes[name].score(), reverse=True)
names = names[: self.top]
res = [intro]
allow_unused = self.full and len(self.members) == 0
res += [
self.emotes[name].to_string(
names.index(name), name, False, self.show_members
)
for name in names
if allow_unused or self.emotes[name].used()
]
# Get the total of all emotes used
usage_count = 0
reaction_count = 0
for name in self.emotes:
usage_count += self.emotes[name].usages
reaction_count += self.emotes[name].reactions
res += [
f"Total: {usage_count:,} times ({usage_count / self.msg_count:.4f} / message)"
]
if reaction_count > 0:
res[-1] += f" and {reaction_count:,} reactions"
return res
# ANALYSIS
def analyse_message(
message: MessageLog,
emotes: Dict[str, Emote],
raw_members: List[int],
*,
all_emojis: bool,
) -> bool:
impacted = False
# If author is included in the selection (empty list is all)
if not message.bot and (len(raw_members) == 0 or message.author in raw_members):
impacted = True
# Find all emotes un the current message in the form "<:emoji:123456789>"
# Filter for known emotes
found = emojis.regex.findall(message.content)
# For each emote, update its usage
for name in found:
if name not in emotes:
if not all_emojis or name not in emojis.unicode_list:
continue
emotes[name].usages += 1
emotes[name].update_use(message.created_at, [message.author])
# For each reaction of this message, test if known emote and update when it's the case
for name in message.reactions:
if name not in emotes:
if not all_emojis or name not in emojis.unicode_list:
continue
if len(raw_members) == 0:
emotes[name].reactions += len(message.reactions[name])
emotes[name].update_use(message.created_at, message.reactions[name])
else:
for member in raw_members:
if member in message.reactions[name]:
emotes[name].reactions += 1
emotes[name].update_use(message.created_at, [member])
return impacted
src/frequency_scanner.py
View File
src/log_manager.py
+271
View File
@@ -0,0 +1,271 @@
from typing import Union, List, Tuple
import os
import discord
import json
import gzip
from datetime import datetime
import logging
from utils import code_message, is_extension
LOG_DIR = "logs"
if not os.path.exists(LOG_DIR):
os.mkdir(LOG_DIR)
CHUNK_SIZE = 1000
FORMAT = 3
IMAGE_FORMAT = ["gif", "gifv", "png", "jpg", "jpeg", "bmp"]
EMBED_IMAGES = ["image", "gifv"]
current_analysis = []
class FakeMessage:
def __init__(self, id: int):
self.id = id
class MessageLog:
def __init__(self, message: Union[discord.Message, dict]):
if isinstance(message, discord.Message):
self.id = message.id
self.created_at = message.created_at
self.edited_at = message.edited_at
self.author = message.author.id
self.pinned = message.pinned
self.mention_everyone = message.mention_everyone
self.tts = message.tts
self.reference = (
message.reference.message_id if message.reference is not None else None
)
self.bot = message.author.bot or message.author.system
self.content = message.content
self.mentions = message.raw_mentions
self.role_mentions = message.raw_role_mentions
self.channel_mentions = message.raw_channel_mentions
self.image = False
for attachment in message.attachments:
if is_extension(attachment.filename, IMAGE_FORMAT):
self.image = True
break
if not self.image:
for embed in message.embeds:
if embed.type in EMBED_IMAGES:
self.image = True
break
self.reactions = {}
elif isinstance(message, dict):
self.id = int(message["id"])
self.created_at = datetime.fromisoformat(message["created_at"])
self.edited_at = (
datetime.fromisoformat(message["edited_at"])
if message["edited_at"] is not None
else None
)
self.author = int(message["author"])
self.pinned = message["pinned"]
self.mention_everyone = message["mention_everyone"]
self.tts = message["tts"]
self.reference = (
int(message["reference"]) if message["reference"] is not None else None
)
self.bot = message["bot"]
self.content = message["content"]
self.mentions = [int(m) for m in message["mentions"]]
self.role_mentions = [int(m) for m in message["role_mentions"]]
self.channel_mentions = [int(m) for m in message["channel_mentions"]]
self.image = message["image"]
self.reactions = message["reactions"]
async def load(self, message: discord.Message):
for reaction in message.reactions:
self.reactions[str(reaction.emoji)] = []
async for user in reaction.users():
self.reactions[str(reaction.emoji)] += [user.id]
def dict(self) -> dict:
message = dict(self.__dict__)
message["created_at"] = self.created_at.isoformat()
message["edited_at"] = (
self.edited_at.isoformat() if self.edited_at is not None else None
)
return message
class ChannelLogs:
def __init__(self, channel: Union[discord.TextChannel, dict]):
if isinstance(channel, discord.TextChannel):
self.id = channel.id
self.name = channel.name
self.last_message_id = None
self.format = FORMAT
self.messages = []
elif isinstance(channel, dict):
self.format = channel["format"] if "format" in channel else None
if self.format != FORMAT:
return
self.id = int(channel["id"])
self.name = channel["name"]
self.last_message_id = int(channel["last_message_id"])
self.messages = [MessageLog(message) for message in channel["messages"]]
async def load(self, channel: discord.TextChannel) -> Tuple[int, int]:
self.name = channel.name
self.channel = channel
try:
if self.last_message_id is not None: # append
while self.last_message_id != channel.last_message_id:
async for message in channel.history(
limit=CHUNK_SIZE,
after=FakeMessage(self.last_message_id),
oldest_first=True,
):
self.last_message_id = message.id
m = MessageLog(message)
await m.load(message)
self.messages.insert(0, m)
yield len(self.messages), False
else: # first load
last_message_id = None
done = 0
while done >= CHUNK_SIZE or last_message_id is None:
done = 0
async for message in channel.history(
limit=CHUNK_SIZE,
before=FakeMessage(last_message_id)
if last_message_id is not None
else None,
oldest_first=False,
):
done += 1
last_message_id = message.id
m = MessageLog(message)
await m.load(message)
self.messages += [m]
yield len(self.messages), False
self.last_message_id = channel.last_message_id
except discord.errors.HTTPException:
return # When an exception occurs (like Forbidden)
yield len(self.messages), True
def dict(self) -> dict:
channel = dict(self.__dict__)
channel.pop("channel", None)
channel["messages"] = [message.dict() for message in self.messages]
return channel
class GuildLogs:
def __init__(self, guild: discord.Guild):
self.guild = guild
self.log_file = os.path.join(LOG_DIR, f"{guild.id}.logz")
self.channels = {}
def dict(self) -> dict:
return {id: self.channels[id].dict() for id in self.channels}
async def load(
self, progress: discord.Message, target_channels: List[discord.TextChannel] = []
) -> Tuple[int, int]:
global current_analysis
if self.log_file in current_analysis:
return -1, -1
current_analysis += [self.log_file]
# read logs
t0 = datetime.now()
if os.path.exists(self.log_file):
channels = {}
try:
gziped_data = None
await code_message(progress, "Reading saved history (1/4)...")
with open(self.log_file, mode="rb") as f:
gziped_data = f.read()
await code_message(progress, "Reading saved history (2/4)...")
json_data = gzip.decompress(gziped_data)
await code_message(progress, "Reading saved history (3/4)...")
channels = json.loads(json_data)
await code_message(progress, "Reading saved history (4/4)...")
self.channels = {int(id): ChannelLogs(channels[id]) for id in channels}
# remove invalid format
self.channels = {
id: self.channels[id]
for id in self.channels
if self.channels[id].format == FORMAT
}
dt = (datetime.now() - t0).total_seconds()
logging.info(f"log {self.guild.id} > loaded in {dt} s")
except json.decoder.JSONDecodeError:
logging.error(f"log {self.guild.id} > invalid JSON")
except IOError:
logging.error(f"log {self.guild.id} > cannot read")
# load channels
t0 = datetime.now()
if len(target_channels) == 0:
target_channels = self.guild.text_channels
loading_new = 0
total_msg = 0
queried_msg = 0
total_chan = 0
max_chan = len(target_channels)
await code_message(
progress,
f"Reading history...\n0 messages in 0/{max_chan} channels\n(this might take a while)",
)
for channel in target_channels:
if channel.id not in self.channels:
loading_new += 1
self.channels[channel.id] = ChannelLogs(channel)
start_msg = len(self.channels[channel.id].messages)
async for count, done in self.channels[channel.id].load(channel):
if count > 0:
tmp_queried_msg = queried_msg + count - start_msg
tmp_msg = total_msg + count
warning_msg = "(this might take a while)"
if len(target_channels) > 5 and loading_new > 5:
warning_msg = (
"(most channels are new, this might take a looong while)"
)
elif loading_new > 0:
warning_msg = (
"(some channels are new, this might take a long while)"
)
dt = (datetime.now() - t0).total_seconds()
await code_message(
progress,
f"Reading history...\n{tmp_msg:,} messages in {total_chan + 1}/{max_chan} channels ({round(tmp_queried_msg/dt)}m/s)\n{warning_msg}",
)
if done:
total_chan += 1
total_msg += len(self.channels[channel.id].messages)
queried_msg += count - start_msg
dt = (datetime.now() - t0).total_seconds()
logging.info(
f"log {self.guild.id} > queried in {dt} s -> {queried_msg / dt} m/s"
)
# write logs
t0 = datetime.now()
await code_message(
progress,
f"Saving (1/3)...\n{total_msg:,} messages in {total_chan} channels",
)
json_data = bytes(json.dumps(self.dict()), "utf-8")
await code_message(
progress,
f"Saving (2/3)...\n{total_msg:,} messages in {total_chan} channels",
)
gziped_data = gzip.compress(json_data)
await code_message(
progress,
f"Saving (3/3)...\n{total_msg:,} messages in {total_chan} channels",
)
with open(self.log_file, mode="wb") as f:
f.write(gziped_data)
dt = (datetime.now() - t0).total_seconds()
logging.info(f"log {self.guild.id} > written in {dt} s")
await code_message(
progress, f"Analysing...\n{total_msg:,} messages in {total_chan} channels"
)
current_analysis.remove(self.log_file)
return total_msg, total_chan
src/other_scanner.py
View File
src/scanner.py
+131
View File
@@ -0,0 +1,131 @@
from abc import ABC, abstractmethod
from typing import List
from collections import defaultdict
import discord
from utils import no_duplicate, get_intro
from log_manager import GuildLogs, ChannelLogs, MessageLog
from data_types import Emote
class Scanner(ABC):
def __init__(
self,
*,
has_digit_args: bool,
valid_args: List[str],
help: str,
intro_context: str,
):
self.has_digit_args = has_digit_args
self.valid_args = valid_args
self.help = help
self.intro_context = intro_context
self.members = []
self.raw_members = []
self.full = False
self.channels = []
self.msg_count = 0
self.chan_count = 0
async def compute(
self, client: discord.client, message: discord.Message, *args: str
):
guild = message.guild
logs = GuildLogs(guild)
# If "%cmd help" redirect to "%help cmd"
if "help" in args:
await client.bot.help(client, message, "help", args[0])
return
# check args validity
str_channel_mentions = [channel.mention for channel in message.channel_mentions]
str_mentions = [member.mention for member in message.mentions]
for arg in args[1:]:
if (
arg not in self.valid_args
and (not arg.isdigit() or not self.has_digit_args)
and arg not in str_channel_mentions
and arg not in str_mentions
):
await message.channel.send(
f"{message.author.mention} unrecognized argument: `{arg}`"
)
return
# Create emotes dict from custom emojis of the guild
self.emotes = defaultdict(Emote)
for emoji in guild.emojis:
self.emotes[str(emoji)] = Emote(emoji)
# Get selected channels or all of them if no channel arguments
self.channels = no_duplicate(message.channel_mentions)
self.full = len(self.channels) == 0
if self.full:
self.channels = guild.text_channels
# Get selected members
self.members = no_duplicate(message.mentions)
self.raw_members = no_duplicate(message.raw_mentions)
if not await self.compute_args(message, *args):
return
# Start computing data
async with message.channel.typing():
progress = await message.channel.send("```Starting analysis...```")
total_msg, total_chan = await logs.load(progress, self.channels)
if total_msg == -1:
await message.channel.send(
f"{message.author.mention} An analysis is already running on this server, please be patient."
)
else:
self.msg_count = 0
self.chan_count = 0
for channel in self.channels:
channel_logs = logs.channels[channel.id]
count = sum(
[
self.compute_message(channel_logs, message_log)
for message_log in channel_logs.messages
]
)
self.msg_count += count
self.chan_count += 1 if count > 0 else 0
await progress.edit(content="```Computing results...```")
# Display results
results = self.get_results(
get_intro(
self.intro_context,
self.full,
self.channels,
self.members,
self.msg_count,
self.chan_count,
)
)
response = ""
for r in results:
if len(response + "\n" + r) > 2000:
await message.channel.send(response)
response = ""
response += "\n" + r
if len(response) > 0:
await message.channel.send(response)
# Delete custom progress message
await progress.delete()
@abstractmethod
async def compute_args(self, message: discord.Message, *args: str) -> bool:
pass
@abstractmethod
def compute_message(self, channel: ChannelLogs, message: MessageLog) -> bool:
pass
@abstractmethod
def get_results(self, intro: str):
pass
src/utils.py
+122
View File
@@ -0,0 +1,122 @@
from typing import List, Dict
import os
import logging
import discord
# DISCORD API
def debug(message: discord.Message, txt: str):
logging.info(f"{message.guild} > #{message.channel}: {txt}")
async def code_message(message: discord.Message, content: str):
await message.edit(content=f"```\n{content}\n```")
def mention(member_id: int) -> str:
return f"<@{member_id}>"
# FILE
def is_extension(filepath: str, ext_list: List[str]) -> bool:
filename, file_extension = os.path.splitext(filepath.lower())
return file_extension in ext_list
# LISTS
def no_duplicate(seq: list) -> list:
"""
Remove any duplicates on a list
:param seq: original list
:type seq: list
:return: same list with no duplicates
:rtype: list
"""
return list(dict.fromkeys(seq))
# MESSAGE FORMATTING
def aggregate(names: List[str]) -> str:
"""
Aggregate names with , and &
Example : "a, b, c & d"
"""
if len(names) == 0:
return ""
elif len(names) == 1:
return names[0]
else:
return ", ".join(names[:-1]) + " & " + names[-1]
def plural(count: int, word: str) -> str:
return str(count) + " " + word + ("s" if count != 1 else "")
def day_interval(interval: int) -> str:
if interval == 0:
return "today"
elif interval == 1:
return "yesterday"
else:
return f"{interval} days ago"
# APP SPECIFIC
def get_intro(
subject: str,
full: bool,
channels: List[discord.TextChannel],
members: List[discord.Member],
nmm: int, # number of messages impacted
nc: int, # number of impacted channels
) -> str:
"""
Get the introduction sentence of the response
"""
# Show all data (members, channels) when it's less than 5 units
if len(members) == 0:
# Full scan of the server
if full:
return f"{subject} in this server ({nc} channels, {nmm:,} messages):"
elif len(channels) < 5:
return f"{aggregate([c.mention for c in channels])} {subject} in {nmm:,} messages:"
else:
return f"These {len(channels)} channels {subject} in {nmm:,} messages:"
elif len(members) < 5:
if full:
return f"{aggregate([m.mention for m in members])} {subject} 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"{subject} in {nmm:,} messages:"
)
else:
return (
f"{aggregate([m.mention for m in members])} on these {len(channels)} channels "
f"{subject} in {nmm:,} messages:"
)
else:
if full:
return f"These {len(members)} members {subject} in {nmm:,} messages:"
elif len(channels) < 5:
return (
f"These {len(members)} members on {aggregate([c.mention for c in channels])} "
f"{subject} in {nmm:,} messages:"
)
else:
return (
f"These {len(members)} members on these {len(channels)} channels "
f"{subject} in {nmm:,} messages:"
)