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

Merge pull request #37 from Klemek/dev

Browse Source 
v1.14
This commit is contained in:
Klemek
2021-04-22 15:23:23 +02:00
committed by GitHub
parent 0399fd8e61 3f7abd9a15
commit 2fda54a6f5
21 changed files with 536 additions and 318 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+12 -3
View File
@@ -21,12 +21,15 @@
* %first - read first message * %first - read first message
* %rand - read a random message * %rand - read a random message
* %last - read last message * %last - read last message
* %find - find specific words or phrases
* %repeat - repeat last analysis (adding supplied arguments)
* %mobile - fix @invalid-user for last command but mentions users
* %gdpr - displays GDPR information * %gdpr - displays GDPR information
* %emojis - rank emotes by their usage * %emojis - rank emojis by their usage
* arguments: * arguments:
* <n> - top <n> emojis, default is 20 * <n> - top <n> emojis, default is 20
* all - list all common emojis in addition to this guild's * all - list all common emojis in addition to this guild's
* members - show top member for each emote * members - show top member for each emoji
* sort:usage/reaction - other sorting methods * sort:usage/reaction - other sorting methods
* %mentions - rank mentions by their usage * %mentions - rank mentions by their usage
* arguments: * arguments:
@@ -58,6 +61,7 @@
* all/everyone - include bots messages * all/everyone - include bots messages
* fast: only read cache * fast: only read cache
* fresh: does not read cache * fresh: does not read cache
* mobile/mention: mentions users (fix @invalid-user bug)
(Sample dates: 2020 / 2021-11 / 2021-06-28 / 2020-06-28T23:00 / today / week / 8days / 1y) (Sample dates: 2020 / 2021-11 / 2021-06-28 / 2020-06-28T23:00 / today / week / 8days / 1y)
``` ```
@@ -109,6 +113,11 @@ python3 src/main.py
## Changelog ## Changelog
* **v1.14**
* `mobile/mention` arg to fix mobile bug
* `%repeat`, `%mobile` to repeat commands
* more scan: `%find`
* bug fix
* **v1.13** * **v1.13**
* improved scan `%words` * improved scan `%words`
* remove old and unused logs at start and guild leaving * remove old and unused logs at start and guild leaving
@@ -143,7 +152,7 @@ python3 src/main.py
* more scans: `%scan`, `%freq`, `%compo`, `%pres` * more scans: `%scan`, `%freq`, `%compo`, `%pres`
* huge bug fix * huge bug fix
* **v1.5**: * **v1.5**:
* top <n> emotes * top <n> emojis
* bug fix * bug fix
* **v1.4**: * **v1.4**:
* integrate miniscord * integrate miniscord
src/data_types/__init__.py
+3 -3
View File
@@ -1,6 +1,6 @@
from .emote import Emote, get_emote_dict from .emoji import Emoji, get_emoji_dict
from .frequency import Frequency
from .composition import Composition from .composition import Composition
from .presence import Presence
from .counter import Counter from .counter import Counter
from .frequency import Frequency
from .history import History from .history import History
from .presence import Presence
src/data_types/composition.py
+11 -11
View File
@@ -8,9 +8,9 @@ class Composition:
def __init__(self): def __init__(self):
self.total_characters = 0 self.total_characters = 0
self.plain_text = 0 self.plain_text = 0
self.emote_msg = 0 self.emoji_msg = 0
self.emote_only = 0 self.emoji_only = 0
self.emotes = defaultdict(int) self.emojis = defaultdict(int)
self.edited = 0 self.edited = 0
self.everyone = 0 self.everyone = 0
self.answers = 0 self.answers = 0
@@ -23,8 +23,8 @@ class Composition:
self.spoilers = 0 self.spoilers = 0
def to_string(self, msg_count: int) -> List[str]: def to_string(self, msg_count: int) -> List[str]:
total_emotes = val_sum(self.emotes) total_emojis = val_sum(self.emojis)
top_emote = top_key(self.emotes) top_emoji = top_key(self.emojis)
ret = [ ret = [
f"- **avg. characters / message**: {self.total_characters/msg_count:.2f}", f"- **avg. characters / message**: {self.total_characters/msg_count:.2f}",
f"- **plain text messages**: {self.plain_text:,} ({percent(self.plain_text/msg_count)})" f"- **plain text messages**: {self.plain_text:,} ({percent(self.plain_text/msg_count)})"
@@ -42,14 +42,14 @@ class Composition:
f"- **answers**: {self.answers:,} ({percent(self.answers/msg_count)})" f"- **answers**: {self.answers:,} ({percent(self.answers/msg_count)})"
if self.answers > 0 if self.answers > 0
else "", else "",
f"- **emojis**: {total_emotes:,} (in {percent(self.emote_msg/msg_count)} of msg, avg. {precise(total_emotes/msg_count)}/msg)" f"- **emojis**: {total_emojis:,} (in {percent(self.emoji_msg/msg_count)} of msg, avg. {precise(total_emojis/msg_count)}/msg)"
if total_emotes > 0 if total_emojis > 0
else "", else "",
f"- **most used emoji**: {top_emote} ({plural(self.emotes[top_emote], 'time')}, {percent(self.emotes[top_emote]/total_emotes)})" f"- **most used emoji**: {top_emoji} ({plural(self.emojis[top_emoji], 'time')}, {percent(self.emojis[top_emoji]/total_emojis)})"
if total_emotes > 0 if total_emojis > 0
else "", else "",
f"- **emoji-only messages**: {self.emote_only:,} ({percent(self.emote_only/msg_count)})" f"- **emoji-only messages**: {self.emoji_only:,} ({percent(self.emoji_only/msg_count)})"
if self.emote_only > 0 if self.emoji_only > 0
else "", else "",
f"- **images**: {self.images:,} ({percent(self.images/msg_count)})" f"- **images**: {self.images:,} ({percent(self.images/msg_count)})"
if self.images > 0 if self.images > 0
src/data_types/counter.py
+21 -11
View File
@@ -14,14 +14,16 @@ class Counter:
def update_use(self, count: int, date: datetime, item: int = 0): def update_use(self, count: int, date: datetime, item: int = 0):
self.usages[item] += count self.usages[item] += count
if self.last_used is None or date > self.last_used: if count > 0 and (self.last_used is None or date > self.last_used):
self.last_used = date self.last_used = date
def score(self) -> float: def score(self) -> float:
# Score is compose of usages + reactions # Score is compose of usages + reactions
# When 2 emotes have the same score, # When 2 emojis have the same score,
# the days since last use is stored in the digits # the days since last use is stored in the digits
# (more recent first) # (more recent first)
if self.last_used is None:
return 0
return self.all_usages() + 1 / ( return self.all_usages() + 1 / (
100000 * ((datetime.today() - self.last_used).days + 1) 100000 * ((datetime.today() - self.last_used).days + 1)
) )
@@ -37,21 +39,29 @@ class Counter:
total_usage: int, total_usage: int,
counted: str = "time", counted: str = "time",
transform: Optional[Callable[[int], str]] = None, transform: Optional[Callable[[int], str]] = None,
ranking: bool = True,
top: bool = True,
) -> str: ) -> str:
# place # place
output = "" output = ""
if i == 0: if ranking:
output += ":first_place:" if i == 0:
elif i == 1: output += ":first_place: "
output += ":second_place:" elif i == 1:
elif i == 2: output += ":second_place: "
output += ":third_place:" elif i == 2:
output += ":third_place: "
else:
output += f"**#{i + 1}** "
else: else:
output += f"**#{i + 1}**" output += f"- "
sum = val_sum(self.usages) sum = val_sum(self.usages)
output += f" {name} - {plural(sum, counted)} ({percent(sum/total_usage)}, last {from_now(self.last_used)})" if sum > 0:
output += f"{name} - {plural(sum, counted)} ({percent(sum/total_usage)}, last {from_now(self.last_used)})"
else:
output += f"{name} - unused"
top_item = top_key(self.usages) top_item = top_key(self.usages)
if top_item != 0 and transform is not None: if sum > 0 and top and top_item != 0 and transform is not None:
if self.usages[top_item] == sum: if self.usages[top_item] == sum:
output += f" (all{transform(top_item)})" output += f" (all{transform(top_item)})"
else: else:
src/data_types/emote.py → src/data_types/emoji.py
+7 -7
View File
@@ -8,9 +8,9 @@ import discord
from utils import mention, plural, from_now, top_key, percent from utils import mention, plural, from_now, top_key, percent
class Emote: class Emoji:
""" """
Custom class to store emotes data Custom class to store emojis data
""" """
def __init__(self, emoji: Optional[discord.Emoji] = None): def __init__(self, emoji: Optional[discord.Emoji] = None):
@@ -34,7 +34,7 @@ class Emote:
def score(self, *, usage_weight: int = 1, react_weight: int = 1) -> float: def score(self, *, usage_weight: int = 1, react_weight: int = 1) -> float:
# Score is compose of usages + reactions # Score is compose of usages + reactions
# When 2 emotes have the same score, # When 2 emojis have the same score,
# the days since last use is stored in the digits # the days since last use is stored in the digits
# (more recent first) # (more recent first)
return ( return (
@@ -99,8 +99,8 @@ class Emote:
return output return output
def get_emote_dict(guild: discord.Guild) -> Dict[str, Emote]: def get_emoji_dict(guild: discord.Guild) -> Dict[str, Emoji]:
emotes = defaultdict(Emote) emojis = defaultdict(Emoji)
for emoji in guild.emojis: for emoji in guild.emojis:
emotes[str(emoji)] = Emote(emoji) emojis[str(emoji)] = Emoji(emoji)
return emotes return emojis
src/main.py
+49 -46
View File
@@ -6,23 +6,8 @@ if sys.version_info < (3, 7):
print("Please upgrade your Python version to 3.7.0 or higher") print("Please upgrade your Python version to 3.7.0 or higher")
sys.exit(1) sys.exit(1)
from utils import emojis, gdpr from utils import emojis, gdpr, command_cache
from scanners import ( import scanners
EmotesScanner,
FullScanner,
FrequencyScanner,
CompositionScanner,
PresenceScanner,
MentionsScanner,
MentionedScanner,
MessagesScanner,
ChannelsScanner,
ReactionsScanner,
FirstScanner,
RandomScanner,
LastScanner,
WordsScanner,
)
from logs import GuildLogs from logs import GuildLogs
logging.basicConfig( logging.basicConfig(
@@ -33,7 +18,7 @@ emojis.load_emojis()
bot = Bot( bot = Bot(
"Discord Analyst", "Discord Analyst",
"1.13", "1.14",
alias="%", alias="%",
) )
@@ -67,87 +52,105 @@ bot.register_command(
) )
bot.register_command( bot.register_command(
"words", "words",
lambda *args: WordsScanner().compute(*args), lambda *args: scanners.WordsScanner().compute(*args),
"words: (BETA) rank words by their usage", "words: (BETA) rank words by their usage",
WordsScanner.help(), scanners.WordsScanner.help(),
)
bot.register_command(
"repeat",
command_cache.repeat,
"repeat: repeat last analysis (adding supplied arguments)",
"```\n%repeat: repeat last analysis (adding supplied arguments)\n```",
)
bot.register_command(
"mobile",
lambda *args: command_cache.repeat(*args, add_args=["mobile"]),
"mobile: fix @invalid-user for last command but mentions users",
"```\n%mobile: fix @invalid-user for last command but mentions users\n```",
)
bot.register_command(
"find",
lambda *args: scanners.FindScanner().compute(*args),
"find: find specific words or phrases",
scanners.FindScanner.help(),
) )
bot.register_command( bot.register_command(
"last", "last",
lambda *args: LastScanner().compute(*args), lambda *args: scanners.LastScanner().compute(*args),
"last: read last message", "last: read last message",
LastScanner.help(), scanners.LastScanner.help(),
) )
bot.register_command( bot.register_command(
"rand(om)?", "rand(om)?",
lambda *args: RandomScanner().compute(*args), lambda *args: scanners.RandomScanner().compute(*args),
"rand: read a random message", "rand: read a random message",
RandomScanner.help(), scanners.RandomScanner.help(),
) )
bot.register_command( bot.register_command(
"first", "first",
lambda *args: FirstScanner().compute(*args), lambda *args: scanners.FirstScanner().compute(*args),
"first: read first message", "first: read first message",
FirstScanner.help(), scanners.FirstScanner.help(),
) )
bot.register_command( bot.register_command(
"mentioned", "mentioned",
lambda *args: MentionedScanner().compute(*args), lambda *args: scanners.MentionedScanner().compute(*args),
"mentioned: rank specific user mentions by their usage", "mentioned: rank specific user mentions by their usage",
MentionedScanner.help(), scanners.MentionedScanner.help(),
) )
bot.register_command( bot.register_command(
"(mentions?)", "(mentions?)",
lambda *args: MentionsScanner().compute(*args), lambda *args: scanners.MentionsScanner().compute(*args),
"mentions: rank mentions by their usage", "mentions: rank mentions by their usage",
MentionsScanner.help(), scanners.MentionsScanner.help(),
) )
bot.register_command( bot.register_command(
"(emojis?|emotes?)", "(emojis?|emotes?)",
lambda *args: EmotesScanner().compute(*args), lambda *args: scanners.EmojisScanner().compute(*args),
"emojis: rank emojis by their usage", "emojis: rank emojis by their usage",
EmotesScanner.help(), scanners.EmojisScanner.help(),
) )
bot.register_command( bot.register_command(
"(react(ions?)?)", "(react(ions?)?)",
lambda *args: ReactionsScanner().compute(*args), lambda *args: scanners.ReactionsScanner().compute(*args),
"react: rank users by their reactions", "react: rank users by their reactions",
ReactionsScanner.help(), scanners.ReactionsScanner.help(),
) )
bot.register_command( bot.register_command(
"(channels?|chan)", "(channels?|chan)",
lambda *args: ChannelsScanner().compute(*args), lambda *args: scanners.ChannelsScanner().compute(*args),
"chan: rank channels by their messages", "chan: rank channels by their messages",
ChannelsScanner.help(), scanners.ChannelsScanner.help(),
) )
bot.register_command( bot.register_command(
"(messages?|msg)", "(messages?|msg)",
lambda *args: MessagesScanner().compute(*args), lambda *args: scanners.MessagesScanner().compute(*args),
"msg: rank users by their messages", "msg: rank users by their messages",
MessagesScanner.help(), scanners.MessagesScanner.help(),
) )
bot.register_command( bot.register_command(
"pres(ence)?", "pres(ence)?",
lambda *args: PresenceScanner().compute(*args), lambda *args: scanners.PresenceScanner().compute(*args),
"pres: presence analysis", "pres: presence analysis",
PresenceScanner.help(), scanners.PresenceScanner.help(),
) )
bot.register_command( bot.register_command(
"compo(sition)?", "compo(sition)?",
lambda *args: CompositionScanner().compute(*args), lambda *args: scanners.CompositionScanner().compute(*args),
"compo: composition analysis", "compo: composition analysis",
CompositionScanner.help(), scanners.CompositionScanner.help(),
) )
bot.register_command( bot.register_command(
"freq(ency)?", "freq(ency)?",
lambda *args: FrequencyScanner().compute(*args), lambda *args: scanners.FrequencyScanner().compute(*args),
"freq: frequency analysis", "freq: frequency analysis",
FrequencyScanner.help(), scanners.FrequencyScanner.help(),
) )
bot.register_command( bot.register_command(
"(full|scan)", "(full|scan)",
lambda *args: FullScanner().compute(*args), lambda *args: scanners.FullScanner().compute(*args),
"scan: full analysis", "scan: full analysis",
FullScanner.help(), scanners.FullScanner.help(),
) )
bot.start() bot.start()
src/scanners/__init__.py
+12 -9
View File
@@ -1,14 +1,17 @@
from .emotes_scanner import EmotesScanner from .scanner import Scanner
from .frequency_scanner import FrequencyScanner
from .composition_scanner import CompositionScanner
from .presence_scanner import PresenceScanner
from .full_scanner import FullScanner
from .mentions_scanner import MentionsScanner
from .mentioned_scanner import MentionedScanner
from .messages_scanner import MessagesScanner
from .channels_scanner import ChannelsScanner from .channels_scanner import ChannelsScanner
from .reactions_scanner import ReactionsScanner from .composition_scanner import CompositionScanner
from .emojis_scanner import EmojisScanner
from .find_scanner import FindScanner
from .first_scanner import FirstScanner from .first_scanner import FirstScanner
from .frequency_scanner import FrequencyScanner
from .full_scanner import FullScanner
from .last_scanner import LastScanner from .last_scanner import LastScanner
from .mentioned_scanner import MentionedScanner
from .mentions_scanner import MentionsScanner
from .messages_scanner import MessagesScanner
from .presence_scanner import PresenceScanner
from .random_scanner import RandomScanner from .random_scanner import RandomScanner
from .reactions_scanner import ReactionsScanner
from .words_scanner import WordsScanner from .words_scanner import WordsScanner
src/scanners/channels_scanner.py
+1 -1
View File
@@ -30,7 +30,6 @@ class ChannelsScanner(Scanner):
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 10 self.top = 10
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -62,6 +61,7 @@ class ChannelsScanner(Scanner):
total_usage=usage_count, total_usage=usage_count,
counted="message", counted="message",
transform=lambda id: f" by {mention(id)}", transform=lambda id: f" by {mention(id)}",
top=len(self.members) != 1,
) )
for name in names for name in names
] ]
src/scanners/composition_scanner.py
+11 -11
View File
@@ -57,19 +57,19 @@ class CompositionScanner(Scanner):
impacted = True impacted = True
compo.total_characters += len(message.content) compo.total_characters += len(message.content)
emotes_found = emojis.regex.findall(message.content) emojis_found = emojis.regex.findall(message.content)
without_emote = message.content without_emoji = message.content
for name in emotes_found: for name in emojis_found:
if name in emojis.unicode_list or re.match( if name in emojis.unicode_list or re.match(
r"(<a?:[\w\-\~]+:\d+>|:[\w\\-\~]+:)", name r"(<a?:[\w\-\~]+:\d+>|:[\w\\-\~]+:)", name
): ):
compo.emotes[name] += 1 compo.emojis[name] += 1
i = without_emote.index(name) i = without_emoji.index(name)
without_emote = without_emote[:i] + without_emote[i + len(name) :] without_emoji = without_emoji[:i] + without_emoji[i + len(name) :]
if len(message.content.strip()) > 0 and len(without_emote.strip()) == 0: if len(message.content.strip()) > 0 and len(without_emoji.strip()) == 0:
compo.emote_only += 1 compo.emoji_only += 1
if len(emotes_found) > 0: if len(emojis_found) > 0:
compo.emote_msg += 1 compo.emoji_msg += 1
links_found = re.findall(r"https?:\/\/", message.content) links_found = re.findall(r"https?:\/\/", message.content)
compo.links += len(links_found) compo.links += len(links_found)
@@ -102,7 +102,7 @@ class CompositionScanner(Scanner):
compo.tts += 1 compo.tts += 1
if ( if (
len(emotes_found) == 0 len(emojis_found) == 0
and message.reference is None and message.reference is None
and not message.image and not message.image
and len(message.mentions) == 0 and len(message.mentions) == 0
src/scanners/emotes_scanner.py → src/scanners/emojis_scanner.py
+31 -29
View File
@@ -6,12 +6,12 @@ import discord
# Custom libs # Custom libs
from logs import ChannelLogs, MessageLog from logs import ChannelLogs, MessageLog
from data_types import Emote, get_emote_dict from data_types import Emoji, get_emoji_dict
from .scanner import Scanner from .scanner import Scanner
from utils import emojis, generate_help, plural, precise from utils import emojis, generate_help, plural, precise
class EmotesScanner(Scanner): class EmojisScanner(Scanner):
@staticmethod @staticmethod
def help() -> str: def help() -> str:
return generate_help( return generate_help(
@@ -31,13 +31,13 @@ class EmotesScanner(Scanner):
super().__init__( super().__init__(
has_digit_args=True, has_digit_args=True,
valid_args=["all", "members", "sort:usage", "sort:reaction", "everyone"], valid_args=["all", "members", "sort:usage", "sort:reaction", "everyone"],
help=EmotesScanner.help(), help=EmojisScanner.help(),
intro_context="Emoji usage", intro_context="Emoji usage",
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
guild = message.channel.guild guild = message.channel.guild
# get max emotes to view # get max emojis to view
self.top = 20 self.top = 20
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -47,8 +47,8 @@ class EmotesScanner(Scanner):
self.show_members = "members" in args and ( self.show_members = "members" in args and (
len(self.members) == 0 or len(self.members) > 1 len(self.members) == 0 or len(self.members) > 1
) )
# Create emotes dict from custom emojis of the guild # Create emojis dict from custom emojis of the guild
self.emotes = get_emote_dict(guild) self.emojis = get_emoji_dict(guild)
self.sort = None self.sort = None
if "sort:usage" in args: if "sort:usage" in args:
self.sort = "usage" self.sort = "usage"
@@ -58,36 +58,36 @@ class EmotesScanner(Scanner):
return True return True
def compute_message(self, channel: ChannelLogs, message: MessageLog): def compute_message(self, channel: ChannelLogs, message: MessageLog):
return EmotesScanner.analyse_message( return EmojisScanner.analyse_message(
message, message,
self.emotes, self.emojis,
self.raw_members, self.raw_members,
all_emojis=self.all_emojis, all_emojis=self.all_emojis,
all_messages=self.all_messages, all_messages=self.all_messages,
) )
def get_results(self, intro: str) -> List[str]: def get_results(self, intro: str) -> List[str]:
names = [name for name in self.emotes] names = [name for name in self.emojis]
names.sort( names.sort(
key=lambda name: self.emotes[name].score( key=lambda name: self.emojis[name].score(
usage_weight=(0 if self.sort == "reaction" else 1), usage_weight=(0 if self.sort == "reaction" else 1),
react_weight=(0 if self.sort == "usage" else 1), react_weight=(0 if self.sort == "usage" else 1),
), ),
reverse=True, reverse=True,
) )
names = names[: self.top] names = names[: self.top]
# Get the total of all emotes used # Get the total of all emojis used
usage_count = 0 usage_count = 0
reaction_count = 0 reaction_count = 0
for name in self.emotes: for name in self.emojis:
usage_count += self.emotes[name].usages usage_count += self.emojis[name].usages
reaction_count += self.emotes[name].reactions reaction_count += self.emojis[name].reactions
res = [intro] res = [intro]
allow_unused = self.full and len(self.members) == 0 allow_unused = self.full and len(self.members) == 0
if self.sort is not None: if self.sort is not None:
res += [f"(Sorted by {self.sort})"] res += [f"(Sorted by {self.sort})"]
res += [ res += [
self.emotes[name].to_string( self.emojis[name].to_string(
names.index(name), names.index(name),
name, name,
total_usage=usage_count, total_usage=usage_count,
@@ -96,7 +96,7 @@ class EmotesScanner(Scanner):
show_members=self.show_members or len(self.raw_members) == 0, show_members=self.show_members or len(self.raw_members) == 0,
) )
for name in names for name in names
if allow_unused or self.emotes[name].used() if allow_unused or self.emojis[name].used()
] ]
res += [ res += [
f"Total: {plural(usage_count,'time')} ({precise(usage_count/self.msg_count)}/msg)" f"Total: {plural(usage_count,'time')} ({precise(usage_count/self.msg_count)}/msg)"
@@ -108,7 +108,7 @@ class EmotesScanner(Scanner):
@staticmethod @staticmethod
def analyse_message( def analyse_message(
message: MessageLog, message: MessageLog,
emotes: Dict[str, Emote], emojis_dict: Dict[str, Emoji],
raw_members: List[int], raw_members: List[int],
*, *,
all_emojis: bool, all_emojis: bool,
@@ -122,27 +122,29 @@ class EmotesScanner(Scanner):
or message.author in raw_members or message.author in raw_members
): ):
impacted = True impacted = True
# Find all emotes un the current message in the form "<:emoji:123456789>" # Find all emojis un the current message in the form "<:emoji:123456789>"
# Filter for known emotes # Filter for known emojis
found = emojis.regex.findall(message.content) found = emojis.regex.findall(message.content)
# For each emote, update its usage # For each emoji, update its usage
for name in found: for name in found:
if name not in emotes: if name not in emojis_dict:
if not all_emojis or name not in emojis.unicode_list: if not all_emojis or name not in emojis.unicode_list:
continue continue
emotes[name].usages += 1 emojis_dict[name].usages += 1
emotes[name].update_use(message.created_at, [message.author]) emojis_dict[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 each reaction of this message, test if known emoji and update when it's the case
for name in message.reactions: for name in message.reactions:
if name not in emotes: if name not in emojis_dict:
if not all_emojis or name not in emojis.unicode_list: if not all_emojis or name not in emojis.unicode_list:
continue continue
if len(raw_members) == 0: if len(raw_members) == 0:
emotes[name].reactions += len(message.reactions[name]) emojis_dict[name].reactions += len(message.reactions[name])
emotes[name].update_use(message.created_at, message.reactions[name]) emojis_dict[name].update_use(
message.created_at, message.reactions[name]
)
else: else:
for member in raw_members: for member in raw_members:
if member in message.reactions[name]: if member in message.reactions[name]:
emotes[name].reactions += 1 emojis_dict[name].reactions += 1
emotes[name].update_use(message.created_at, [member]) emojis_dict[name].update_use(message.created_at, [member])
return impacted return impacted
src/scanners/find_scanner.py
+105
View File
@@ -0,0 +1,105 @@
from typing import Dict, List
from collections import defaultdict
import discord
import re
# Custom libs
from logs import ChannelLogs, MessageLog
from .scanner import Scanner
from data_types import Counter
from utils import (
generate_help,
plural,
precise,
mention,
)
class FindScanner(Scanner):
@staticmethod
def help() -> str:
return generate_help(
"find",
"Find specific words or phrases (you can use quotes to add spaces in queries)",
args=[
"all/everyone - include bots",
],
example='#mychannel1 #mychannel2 @user "I love you" "you too"',
)
def __init__(self):
super().__init__(
all_args=True,
valid_args=["all", "everyone"],
help=FindScanner.help(),
intro_context="Matches",
)
async def init(self, message: discord.Message, *args: str) -> bool:
self.matches = defaultdict(Counter)
self.all_messages = "all" in args or "everyone" in args
if len(self.other_args) == 0:
await message.channel.send(
"You need to add a query to find (you can use quotes to add spaces in queries)",
reference=message,
)
return False
return True
def compute_message(self, channel: ChannelLogs, message: MessageLog):
return FindScanner.analyse_message(
message,
self.matches,
self.other_args,
self.raw_members,
all_messages=self.all_messages,
)
def get_results(self, intro: str) -> List[str]:
matches = [match for match in self.matches]
matches.sort(key=lambda match: self.matches[match].score(), reverse=True)
usage_count = Counter.total(self.matches)
res = [intro]
res += [
self.matches[match].to_string(
matches.index(match),
f"`{match}`",
total_usage=self.msg_count,
ranking=False,
transform=lambda id: f" by {mention(id)}",
top=len(self.members) != 1,
)
for match in matches
]
if len(matches) > 1:
res += [
f"Total: {plural(usage_count,'time')} ({precise(usage_count/self.msg_count)}/msg)"
]
return res
special_cases = ["'s", "s"]
@staticmethod
def analyse_message(
message: MessageLog,
matches: Dict[str, Counter],
queries: List[str],
raw_members: List[int],
*,
all_messages: bool,
) -> bool:
impacted = False
# If author is included in the selection (empty list is all)
if (
(not message.bot or all_messages)
and len(raw_members) == 0
or message.author in raw_members
):
impacted = True
content = message.content.lower()
for query in queries:
matches[query].update_use(
content.count(query.lower()), message.created_at, message.author
)
return impacted
src/scanners/full_scanner.py
+3 -1
View File
@@ -5,7 +5,9 @@ import discord
# Custom libs # Custom libs
from .scanner import Scanner from .scanner import Scanner
from . import FrequencyScanner, CompositionScanner, PresenceScanner from .composition_scanner import CompositionScanner
from .frequency_scanner import FrequencyScanner
from .presence_scanner import PresenceScanner
from data_types import Frequency, Composition, Presence from data_types import Frequency, Composition, Presence
from logs import ChannelLogs, MessageLog from logs import ChannelLogs, MessageLog
from utils import generate_help from utils import generate_help
src/scanners/mentioned_scanner.py
+3 -3
View File
@@ -31,7 +31,6 @@ class MentionedScanner(Scanner):
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 10 self.top = 10
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -55,7 +54,6 @@ class MentionedScanner(Scanner):
names = [name for name in self.mentions] names = [name for name in self.mentions]
names.sort(key=lambda name: self.mentions[name].score(), reverse=True) names.sort(key=lambda name: self.mentions[name].score(), reverse=True)
names = names[: self.top] names = names[: self.top]
# Get the total of all emotes used
usage_count = Counter.total(self.mentions) usage_count = Counter.total(self.mentions)
res = [intro] res = [intro]
res += [ res += [
@@ -63,6 +61,8 @@ class MentionedScanner(Scanner):
names.index(name), names.index(name),
name, name,
total_usage=usage_count, total_usage=usage_count,
transform=lambda id: f" for {mention(id)}",
top=len(self.members) != 1,
) )
for name in names for name in names
] ]
@@ -87,6 +87,6 @@ class MentionedScanner(Scanner):
mention(member_id) mention(member_id)
) + message.content.count(alt_mention(member_id)) ) + message.content.count(alt_mention(member_id))
mentions[mention(message.author)].update_use( mentions[mention(message.author)].update_use(
count, message.created_at count, message.created_at, member_id
) )
return impacted return impacted
src/scanners/mentions_scanner.py
+11 -7
View File
@@ -42,7 +42,6 @@ class MentionsScanner(Scanner):
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 10 self.top = 10
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -67,7 +66,6 @@ class MentionsScanner(Scanner):
names = [name for name in self.mentions] names = [name for name in self.mentions]
names.sort(key=lambda name: self.mentions[name].score(), reverse=True) names.sort(key=lambda name: self.mentions[name].score(), reverse=True)
names = names[: self.top] names = names[: self.top]
# Get the total of all emotes used
usage_count = Counter.total(self.mentions) usage_count = Counter.total(self.mentions)
res = [intro] res = [intro]
res += [ res += [
@@ -75,6 +73,8 @@ class MentionsScanner(Scanner):
names.index(name), names.index(name),
name, name,
total_usage=usage_count, total_usage=usage_count,
transform=lambda id: f" by {mention(id)}",
top=len(self.members) != 1,
) )
for name in names for name in names
] ]
@@ -105,24 +105,28 @@ class MentionsScanner(Scanner):
count = message.content.count(name) + message.content.count( count = message.content.count(name) + message.content.count(
alt_mention(member_id) alt_mention(member_id)
) )
mentions[name].update_use(count, message.created_at) mentions[name].update_use(count, message.created_at, message.author)
if all_mentions: if all_mentions:
for role_id in message.role_mentions: for role_id in message.role_mentions:
name = role_mention(role_id) name = role_mention(role_id)
mentions[name].update_use( mentions[name].update_use(
message.content.count(name), message.created_at message.content.count(name), message.created_at, message.author
) )
for channel_id in message.channel_mentions: for channel_id in message.channel_mentions:
name = channel_mention(channel_id) name = channel_mention(channel_id)
mentions[name].update_use( mentions[name].update_use(
message.content.count(name), message.created_at message.content.count(name), message.created_at, message.author
) )
if "@everyone" in message.content: if "@everyone" in message.content:
mentions["@\u200beveryone"].update_use( mentions["@\u200beveryone"].update_use(
message.content.count("@everyone"), message.created_at message.content.count("@everyone"),
message.created_at,
message.author,
) )
if "@here" in message.content: if "@here" in message.content:
mentions["@\u200bhere"].update_use( mentions["@\u200bhere"].update_use(
message.content.count("@here"), message.created_at message.content.count("@here"),
message.created_at,
message.author,
) )
return impacted return impacted
src/scanners/messages_scanner.py
+1 -1
View File
@@ -30,7 +30,6 @@ class MessagesScanner(Scanner):
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 10 self.top = 10
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -62,6 +61,7 @@ class MessagesScanner(Scanner):
total_usage=usage_count, total_usage=usage_count,
counted="message", counted="message",
transform=lambda id: f" in {channel_mention(id)}", transform=lambda id: f" in {channel_mention(id)}",
top=self.channels != 1,
) )
for name in names for name in names
] ]
src/scanners/reactions_scanner.py
+1 -1
View File
@@ -29,7 +29,6 @@ class ReactionsScanner(Scanner):
) )
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
# get max emotes to view
self.top = 10 self.top = 10
for arg in args: for arg in args:
if arg.isdigit(): if arg.isdigit():
@@ -59,6 +58,7 @@ class ReactionsScanner(Scanner):
total_usage=usage_count, total_usage=usage_count,
counted="reaction", counted="reaction",
transform=lambda id: f" in {channel_mention(id)}", transform=lambda id: f" in {channel_mention(id)}",
top=self.channels != 1,
) )
for name in names for name in names
] ]
src/scanners/scanner.py
+197 -160
View File
@@ -14,6 +14,7 @@ from utils import (
ISO8601_REGEX, ISO8601_REGEX,
RELATIVE_REGEX, RELATIVE_REGEX,
parse_time, parse_time,
command_cache,
) )
from logs import ( from logs import (
GuildLogs, GuildLogs,
@@ -26,6 +27,8 @@ from logs import (
class Scanner(ABC): class Scanner(ABC):
VALID_ARGS = ["me", "here", "fast", "fresh", "mobile", "mention"]
def __init__( def __init__(
self, self,
*, *,
@@ -33,12 +36,16 @@ class Scanner(ABC):
valid_args: List[str] = [], valid_args: List[str] = [],
help: str, help: str,
intro_context: str, intro_context: str,
all_args: bool = False,
): ):
self.has_digit_args = has_digit_args self.has_digit_args = has_digit_args
self.valid_args = valid_args self.valid_args = valid_args
self.all_args = all_args
self.help = help self.help = help
self.intro_context = intro_context self.intro_context = intro_context
self.other_args = []
self.members = [] self.members = []
self.raw_members = [] self.raw_members = []
self.full = False self.full = False
@@ -48,191 +55,221 @@ class Scanner(ABC):
self.chan_count = 0 self.chan_count = 0
async def compute( async def compute(
self, client: discord.client, message: discord.Message, *args: str self,
client: discord.client,
message: discord.Message,
*args: str,
other_mentions: List[str] = [],
): ):
args = list(args) args = list(args)
guild = message.guild guild = message.guild
with GuildLogs(guild) as logs: progress = None
# If "%cmd help" redirect to "%help cmd" try:
if "help" in args: with GuildLogs(guild) as logs:
await client.bot.help(client, message, "help", args[0]) # If "%cmd help" redirect to "%help cmd"
return if len(args) > 1 and args[1] == "help":
await client.bot.help(client, message, "help", args[0])
return
# check args validity # check args validity
str_channel_mentions = [ str_channel_mentions = [
str(channel.id) for channel in message.channel_mentions str(channel.id) for channel in message.channel_mentions
] ]
str_mentions = [str(member.id) for member in message.mentions] str_mentions = [str(member.id) for member in message.mentions]
dates = [] dates = []
for i, arg in enumerate(args[1:]): for i, arg in enumerate(args[1:]):
skip_check = False skip_check = False
if re.match(r"^<@!?\d+>$", arg): if re.match(r"^<@!?\d+>$", arg):
arg = arg[3:-1] if "!" in arg else arg[2:-1] arg = arg[3:-1] if "!" in arg else arg[2:-1]
elif re.match(r"^<#!?\d+>$", arg): elif re.match(r"^<#!?\d+>$", arg):
arg = arg[3:-1] if "!" in arg else arg[2:-1] arg = arg[3:-1] if "!" in arg else arg[2:-1]
elif re.match(ISO8601_REGEX, arg) or re.match(RELATIVE_REGEX, arg): elif re.match(ISO8601_REGEX, arg) or re.match(RELATIVE_REGEX, arg):
dates += [parse_time(arg)] dates += [parse_time(arg)]
skip_check = True skip_check = True
if len(dates) > 2: if len(dates) > 2:
await message.channel.send( await message.channel.send(
f"Too many date arguments: `{arg}`", reference=message f"Too many date arguments: `{arg}`", reference=message
) )
return return
if ( if (
arg not in self.valid_args + ["me", "here", "fast", "fresh"] arg not in self.valid_args + Scanner.VALID_ARGS
and (not arg.isdigit() or not self.has_digit_args) and (not arg.isdigit() or not self.has_digit_args)
and arg not in str_channel_mentions and arg not in str_channel_mentions
and arg not in str_mentions and arg not in str_mentions
and not skip_check and arg not in other_mentions
): and not skip_check
and len(arg) > 0
):
if self.all_args:
self.other_args += [arg]
else:
await message.channel.send(
f"Unrecognized argument: `{arg}`", reference=message
)
return
self.start_date = None if len(dates) < 1 else min(dates)
self.stop_date = None if len(dates) < 2 else max(dates)
if self.start_date is not None and self.start_date > datetime.now():
await message.channel.send( await message.channel.send(
f"Unrecognized argument: `{arg}`", reference=message f"Start date is after today", reference=message
) )
return return
self.start_date = None if len(dates) < 1 else min(dates) # Get selected channels or all of them if no channel arguments
self.stop_date = None if len(dates) < 2 else max(dates) self.channels = no_duplicate(message.channel_mentions)
if self.start_date is not None and self.start_date > datetime.now(): # transform the "here" arg
await message.channel.send( if "here" in args:
f"Start date is after today", reference=message self.channels += [message.channel]
)
return
# Get selected channels or all of them if no channel arguments self.full = len(self.channels) == 0
self.channels = no_duplicate(message.channel_mentions) if self.full:
self.channels = guild.text_channels
# transform the "here" arg # Get selected members
if "here" in args: self.members = no_duplicate(message.mentions)
self.channels += [message.channel] self.raw_members = no_duplicate(message.raw_mentions)
self.full = len(self.channels) == 0 # transform the "me" arg
if self.full: if "me" in args:
self.channels = guild.text_channels self.members += [message.author]
self.raw_members += [message.author.id]
# Get selected members self.mention_users = "mention" in args or "mobile" in args
self.members = no_duplicate(message.mentions)
self.raw_members = no_duplicate(message.raw_mentions)
# transform the "me" arg if not await self.init(message, *args):
if "me" in args: return
self.members += [message.author]
self.raw_members += [message.author.id]
if not await self.init(message, *args): # Start computing data
return async with message.channel.typing():
progress = await message.channel.send(
# Start computing data "```Starting analysis...```",
async with message.channel.typing():
progress = await message.channel.send(
"```Starting analysis...```",
reference=message,
allowed_mentions=discord.AllowedMentions.none(),
)
total_msg, total_chan = await logs.load(
progress,
self.channels,
self.start_date,
self.stop_date,
fast="fast" in args,
fresh="fresh" in args,
)
if total_msg == CANCELLED:
await message.channel.send(
"Operation cancelled by user",
reference=message, reference=message,
allowed_mentions=discord.AllowedMentions.none(),
) )
elif total_msg == ALREADY_RUNNING: total_msg, total_chan = await logs.load(
await message.channel.send( progress,
"An analysis is already running on this server, please be patient.", self.channels,
reference=message, self.start_date,
self.stop_date,
fast="fast" in args,
fresh="fresh" in args,
) )
elif total_msg == NO_FILE: if total_msg == CANCELLED:
await message.channel.send(gdpr.TEXT)
else:
if self.start_date is not None and len(logs.channels) > 0:
self.start_date = max(
self.start_date,
min(
[
logs.channels[channel.id].start_date
for channel in self.channels
if channel.id in logs.channels
and logs.channels[channel.id].start_date is not None
]
),
)
if self.stop_date is None:
self.stop_date = datetime.utcnow()
self.msg_count = 0
self.total_msg = 0
self.chan_count = 0
t0 = datetime.now()
for channel in self.channels:
if channel.id in logs.channels:
channel_logs = logs.channels[channel.id]
count = sum(
[
self.compute_message(channel_logs, message_log)
for message_log in channel_logs.messages
if (
self.start_date is None
or message_log.created_at >= self.start_date
)
and (
self.stop_date is None
or message_log.created_at <= self.stop_date
)
]
)
self.total_msg += len(channel_logs.messages)
self.msg_count += count
self.chan_count += 1 if count > 0 else 0
logging.info(f"scan {guild.id} > scanned in {delta(t0):,}ms")
if self.msg_count == 0:
await message.channel.send( await message.channel.send(
"There are no messages found matching the filters", "Operation cancelled by user",
reference=message, reference=message,
) )
else: elif total_msg == ALREADY_RUNNING:
await progress.edit(content="```Computing results...```") await message.channel.send(
# Display results "An analysis is already running on this server, please be patient.",
t0 = datetime.now() reference=message,
results = self.get_results(
get_intro(
self.intro_context,
self.full,
self.channels,
self.members,
self.msg_count,
self.chan_count,
self.start_date,
self.stop_date,
)
) )
logging.info(f"scan {guild.id} > results in {delta(t0):,}ms") elif total_msg == NO_FILE:
response = "" await message.channel.send(gdpr.TEXT)
first = True else:
for r in results: if self.start_date is not None and len(logs.channels) > 0:
if r: self.start_date = max(
if len(response + "\n" + r) > 2000: self.start_date,
await message.channel.send( min(
response, [
reference=message if first else None, logs.channels[channel.id].start_date
allowed_mentions=discord.AllowedMentions.none(), for channel in self.channels
) if channel.id in logs.channels
first = False and logs.channels[channel.id].start_date
response = "" is not None
response += "\n" + r ]
if len(response) > 0: ),
await message.channel.send(
response,
reference=message if first else None,
allowed_mentions=discord.AllowedMentions.none(),
) )
if self.stop_date is None:
self.stop_date = datetime.utcnow()
self.msg_count = 0
self.total_msg = 0
self.chan_count = 0
t0 = datetime.now()
for channel in self.channels:
if channel.id in logs.channels:
channel_logs = logs.channels[channel.id]
count = sum(
[
self.compute_message(channel_logs, message_log)
for message_log in channel_logs.messages
if (
self.start_date is None
or message_log.created_at >= self.start_date
)
and (
self.stop_date is None
or message_log.created_at <= self.stop_date
)
]
)
self.total_msg += len(channel_logs.messages)
self.msg_count += count
self.chan_count += 1 if count > 0 else 0
logging.info(f"scan {guild.id} > scanned in {delta(t0):,}ms")
if self.msg_count == 0:
await message.channel.send(
"There are no messages found matching the filters",
reference=message,
)
else:
await progress.edit(content="```Computing results...```")
# Display results
t0 = datetime.now()
results = self.get_results(
get_intro(
self.intro_context,
self.full,
self.channels,
self.members,
self.msg_count,
self.chan_count,
self.start_date,
self.stop_date,
)
)
logging.info(
f"scan {guild.id} > results in {delta(t0):,}ms"
)
response = ""
first = True
allowed_mentions = (
discord.AllowedMentions.all()
if self.mention_users
else discord.AllowedMentions.none()
)
for r in results:
if r:
if len(response + "\n" + r) > 2000:
await message.channel.send(
response,
reference=message if first else None,
allowed_mentions=allowed_mentions,
)
first = False
response = ""
response += "\n" + r
if len(response) > 0:
await message.channel.send(
response,
reference=message if first else None,
allowed_mentions=allowed_mentions,
)
command_cache.cache(self, message, args)
# Delete custom progress message # Delete custom progress message
await progress.delete() await progress.delete()
except Exception as error:
logging.exception(error)
await message.channel.send(
"An unexpected error happened while computing your command, we're sorry for the inconvenience.",
reference=message,
)
if progress is not None:
await progress.delete()
@abstractmethod @abstractmethod
async def init(self, message: discord.Message, *args: str) -> bool: async def init(self, message: discord.Message, *args: str) -> bool:
src/scanners/words_scanner.py
+4 -8
View File
@@ -8,11 +8,7 @@ import re
from logs import ChannelLogs, MessageLog from logs import ChannelLogs, MessageLog
from .scanner import Scanner from .scanner import Scanner
from data_types import Counter from data_types import Counter
from utils import ( from utils import generate_help, plural, precise, mention
generate_help,
plural,
precise,
)
class WordsScanner(Scanner): class WordsScanner(Scanner):
@@ -67,15 +63,15 @@ class WordsScanner(Scanner):
words = [word for word in self.words] words = [word for word in self.words]
words.sort(key=lambda word: self.words[word].score(), reverse=True) words.sort(key=lambda word: self.words[word].score(), reverse=True)
words = words[: self.top] words = words[: self.top]
# Get the total of all emotes used
usage_count = Counter.total(self.words) usage_count = Counter.total(self.words)
print(len(self.words))
res = [intro.format(self.letters)] res = [intro.format(self.letters)]
res += [ res += [
self.words[word].to_string( self.words[word].to_string(
words.index(word), words.index(word),
f"`{word}`", f"`{word}`",
total_usage=usage_count, total_usage=usage_count,
transform=lambda id: f" by {mention(id)}",
top=len(self.members) != 1,
) )
for word in words for word in words
] ]
@@ -122,5 +118,5 @@ class WordsScanner(Scanner):
words[word] = words[word + case] words[word] = words[word + case]
del words[word + case] del words[word + case]
break break
words[word].update_use(1, message.created_at) words[word].update_use(1, message.created_at, message.author)
return impacted return impacted
src/utils/command_cache.py
+45
View File
@@ -0,0 +1,45 @@
from typing import List
import logging
import discord
from scanners import Scanner
command_cache = {}
def cache(scanner: Scanner, message: discord.Message, args: List[str]):
id = message.channel.id
command_cache[id] = (
type(scanner),
list(args),
[str(channel.id) for channel in message.channel_mentions]
+ [str(member.id) for member in message.mentions],
)
async def repeat(
client: discord.client,
message: discord.Message,
*args: str,
add_args: List[str] = [],
):
if len(args) > 1 and args[1] == "help":
await client.bot.help(client, message, "help", args[0])
return
id = message.channel.id
if id not in command_cache:
await message.channel.send(
"No command to repeat on this channel (type %help for more info)",
reference=message,
)
return
(
scannerType,
original_args,
original_mentions,
) = command_cache[id]
args = original_args + add_args + list(args[1:]) + ["fast"]
logging.info(f"repeating {args}")
await scannerType().compute(
client, message, *args, other_mentions=original_mentions
)
src/utils/gdpr.py
+2 -2
View File
@@ -49,10 +49,10 @@ async def process(client: discord.client, message: discord.Message, *args: str):
args = list(args) args = list(args)
if len(args) == 1: if len(args) == 1:
await message.channel.send(TEXT) await message.channel.send(TEXT)
elif args[1] == "help":
await client.bot.help(client, message, "help", args[0])
elif len(args) > 2: elif len(args) > 2:
await message.channel.send(f"Too many arguments", reference=message) await message.channel.send(f"Too many arguments", reference=message)
elif args[1] == "help":
await message.channel.send(HELP, reference=message)
elif args[1] in ["agree", "accept"]: elif args[1] in ["agree", "accept"]:
GuildLogs.init_log(message.channel.guild) GuildLogs.init_log(message.channel.guild)
await message.channel.send(AGREE_TEXT, reference=message) await message.channel.send(AGREE_TEXT, reference=message)
src/utils/utils.py
+6 -4
View File
@@ -18,6 +18,7 @@ COMMON_HELP_ARGS = [
"<date2> - filter before <date2>", "<date2> - filter before <date2>",
"fast - only read cache", "fast - only read cache",
"fresh - does not read cache (long)", "fresh - does not read cache (long)",
"mobile/mention - mentions users (fix @invalid-user bug)",
] ]
@@ -180,15 +181,16 @@ def parse_iso_datetime(str_date: str) -> datetime:
return dateutil.parser.parse(str_date) return dateutil.parser.parse(str_date)
RELATIVE_REGEX = r"(yesterday|today|\d*h(ours?)?|\d*d(ays?)?|\d*w(eeks?)?|\d*m(onths?)?|\d*y(ears?)?)" RELATIVE_REGEX = r"(yesterday|today|\d*hours?|\d+h(ours?)?|\d*days?|\d+d(ays?)?|\d*weeks?|\d+w(eeks?)?|\d*months?|\d+m(onths?)?|\d*years?|\d+y(ears?)?)"
def parse_relative_time(src: str) -> datetime: def parse_relative_time(src: str) -> datetime:
timezone_delta = datetime.utcnow() - datetime.now() today = datetime.utcnow().date()
today = datetime(today.year, today.month, today.day)
if src == "today": if src == "today":
return datetime.today() + timezone_delta return today
elif src == "yesterday": elif src == "yesterday":
return datetime.today() - relativedelta(days=1) + timezone_delta return today - relativedelta(days=1)
else: else:
m = re.match("(\d*)(\w+)", src) m = re.match("(\d*)(\w+)", src)
delta = None delta = None