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

%first/%rand/%last image

Browse Source
This commit is contained in:
Klemek
2021-05-19 13:31:07 +02:00
parent 13447ff869
commit 516eb75b5c
8 changed files with 129 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/data_types/history.py
+48 -1
View File
@@ -3,13 +3,60 @@ import random
# Custom libs
from utils import mention, from_now, str_datetime, message_link
from utils import mention, from_now, str_datetime, message_link, SPLIT_TOKEN
MAX_RANDOM_TRIES = 10
class History:
def __init__(self):
self.messages = []
async def to_string_image(self, *, type: str) -> List[str]:
if len(self.messages) == 0:
return ["There was no messages matching your filters"]
message = None
intro = None
real_message = None
if type == "first":
self.messages.sort(key=lambda m: m.created_at)
index = 0
while real_message is None and index < len(self.messages):
message = self.messages[index]
real_message = await message.fetch()
index += 1
intro = f"First image out of {len(self.messages):,}"
elif type == "last":
self.messages.sort(key=lambda m: m.created_at, reverse=True)
index = 0
while real_message is None and index < len(self.messages):
message = self.messages[index]
real_message = await message.fetch()
index += 1
intro = f"Last image out of {len(self.messages):,}"
elif type == "random":
intro = f"Random image out of {len(self.messages):,}"
tries = 0
while real_message is None and tries < MAX_RANDOM_TRIES:
message = random.choice(self.messages)
real_message = await message.fetch()
tries += 1
if real_message is None:
return ["There was no messages matching your filters"]
image = "<Error>"
if len(real_message.attachments) > 0:
image = real_message.attachments[0].url
elif len(real_message.embeds) > 0:
image = real_message.embeds[0].url
return [
intro,
f"{str_datetime(message.created_at)} ({from_now(message.created_at)}) {mention(message.author)} sent:",
f"<{message_link(message)}>",
SPLIT_TOKEN,
image,
]
def to_string(self, *, type: str) -> List[str]:
if len(self.messages) == 0:
return ["There was no messages matching your filters"]
src/logs/message_log.py
+9 -2
View File
@@ -1,11 +1,11 @@
from typing import Union, Any
from typing import Optional, Union, Any
import discord
from datetime import datetime
from utils import is_extension, serialize
IMAGE_FORMAT = [".gif", ".gifv", ".png", ".jpg", ".jpeg", ".bmp"]
EMBED_IMAGES = ["image", "gifv", "gif"]
EMBED_IMAGES = ["image", "gifv"]
class MessageLog:
@@ -76,6 +76,13 @@ class MessageLog:
self.reactions[str(reaction.emoji)] = []
async for user in reaction.users():
self.reactions[str(reaction.emoji)] += [user.id]
async def fetch(self) -> Optional[discord.Message]:
try:
return await self.channel.channel.fetch_message(self.id)
except (discord.NotFound, discord.Forbidden, discord.HTTPException):
return None
def dict(self) -> dict:
return serialize(
src/scanners/first_scanner.py
+11 -7
View File
@@ -9,13 +9,17 @@ from utils import generate_help
class FirstScanner(HistoryScanner):
@staticmethod
def help() -> str:
return generate_help("first", "Read first message (add text to filter like %find)")
return generate_help(
"first",
"Read first message (add text to filter like %find)",
args=["image - pull an image instead of a message"],
)
def __init__(self):
super().__init__(help=FirstScanner.help(), allow_queries=True)
super().__init__(help=FirstScanner.help())
def allow_message(self, *_) -> bool:
return True
def get_results(self, intro: str) -> List[str]:
return self.history.to_string(type="first")
async def get_results(self, intro: str) -> List[str]:
if self.images_only:
return await self.history.to_string_image(type="first")
else:
return self.history.to_string(type="first")
src/scanners/history_scanner.py
+18 -26
View File
@@ -1,5 +1,5 @@
from abc import ABC, abstractmethod
from typing import Callable, List
from typing import List, Tuple, Optional
import discord
import re
@@ -11,20 +11,20 @@ from logs import ChannelLogs, MessageLog
class HistoryScanner(Scanner, ABC):
def __init__(self, *, help: str, valid_args: List[str] = [], allow_queries: bool = False):
def __init__(self, *, help: str):
super().__init__(
has_digit_args=True,
valid_args=["all", "everyone"] + valid_args,
valid_args=["all", "everyone"],
help=help,
intro_context="",
all_args=allow_queries,
all_args=True,
)
self.allow_queries = allow_queries
async def init(self, message: discord.Message, *args: str) -> bool:
self.history = History()
self.all_messages = "all" in args or "everyone" in args
if self.allow_queries:
self.images_only = "image" in args
if not self.images_only:
self.queries = [(query.lower(), query.strip("`") if re.match(r"^`.*`$", query) else None) for query in self.other_args]
else:
self.queries = []
@@ -37,30 +37,14 @@ class HistoryScanner(Scanner, ABC):
self.history,
self.raw_members,
all_messages=self.all_messages,
allow_message=lambda *args: self.allow_message(*args) and self.allow_message_query(*args),
queries=self.queries,
images_only=self.images_only,
)
@abstractmethod
def get_results(self, intro: str):
pass
@abstractmethod
def allow_message(self, channel: ChannelLogs, message: MessageLog) -> bool:
pass
def allow_message_query(self, channel: ChannelLogs, message: MessageLog) -> bool:
if not self.allow_queries or len(self.queries) == 0:
return True
else:
content = message.content.lower()
for query in self.queries:
if query[1] is not None:
if not re.match(query[1], message.content):
return False
elif not query[0] in content:
return False
return True
@staticmethod
def analyse_message(
channel: ChannelLogs,
@@ -69,7 +53,8 @@ class HistoryScanner(Scanner, ABC):
raw_members: List[int],
*,
all_messages: bool,
allow_message: Callable
queries: List[Tuple[str, Optional[str]]],
images_only: bool,
) -> bool:
impacted = False
# If author is included in the selection (empty list is all)
@@ -80,8 +65,15 @@ class HistoryScanner(Scanner, ABC):
or message.author in raw_members
)
and (message.content or message.attachment)
and allow_message(channel, message)
and (not images_only or message.image)
):
content = message.content.lower()
for query in queries:
if query[1] is not None:
if not re.match(query[1], message.content):
return False
elif not query[0] in content:
return False
impacted = True
history.messages += [message]
return impacted
src/scanners/last_scanner.py
+11 -7
View File
@@ -9,13 +9,17 @@ from utils import generate_help
class LastScanner(HistoryScanner):
@staticmethod
def help() -> str:
return generate_help("last", "Read last message (add text to filter like %find)")
return generate_help(
"last",
"Read last message (add text to filter like %find)",
args=["image - pull an image instead of a message"],
)
def __init__(self):
super().__init__(help=LastScanner.help(), allow_queries=True)
super().__init__(help=LastScanner.help())
def allow_message(self, *_) -> bool:
return True
def get_results(self, intro: str) -> List[str]:
return self.history.to_string(type="last")
async def get_results(self, intro: str) -> List[str]:
if self.images_only:
return await self.history.to_string_image(type="last")
else:
return self.history.to_string(type="last")
src/scanners/random_scanner.py
+11 -7
View File
@@ -9,13 +9,17 @@ from utils import generate_help
class RandomScanner(HistoryScanner):
@staticmethod
def help() -> str:
return generate_help("rand", "Read a random message (add text to filter like %find)")
return generate_help(
"rand",
"Read a random message (add text to filter like %find)",
args=["image - pull an image instead of a message"],
)
def __init__(self):
super().__init__(help=RandomScanner.help(), allow_queries=True)
super().__init__(help=RandomScanner.help())
def allow_message(self, *_) -> bool:
return True
def get_results(self, intro: str) -> List[str]:
return self.history.to_string(type="random")
async def get_results(self, intro: str) -> List[str]:
if self.images_only:
return await self.history.to_string_image(type="random")
else:
return self.history.to_string(type="random")
src/scanners/scanner.py
+18 -5
View File
@@ -4,6 +4,7 @@ from datetime import datetime
import logging
import re
import discord
import inspect
from utils import (
@@ -16,6 +17,7 @@ from utils import (
parse_time,
command_cache,
FilterLevel,
SPLIT_TOKEN,
)
from logs import (
GuildLogs,
@@ -241,8 +243,7 @@ class Scanner(ABC):
await progress.edit(content="```Computing results...```")
# Display results
t0 = datetime.now()
results = self.get_results(
get_intro(
intro = get_intro(
self.intro_context,
self.full,
self.channels,
@@ -252,7 +253,10 @@ class Scanner(ABC):
self.start_date,
self.stop_date,
)
)
if inspect.iscoroutinefunction(self.get_results):
results = await self.get_results(intro)
else:
results = self.get_results(intro)
logging.info(
f"scan {guild.id} > results in {delta(t0):,}ms"
)
@@ -265,7 +269,7 @@ class Scanner(ABC):
)
for r in results:
if r:
if len(response + "\n" + r) > 2000:
if isinstance(r, int) and r == SPLIT_TOKEN:
await message.channel.send(
response,
reference=message if first else None,
@@ -273,7 +277,16 @@ class Scanner(ABC):
)
first = False
response = ""
response += "\n" + r
elif isinstance(r, str):
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,
src/utils/utils.py
+3
View File
@@ -57,6 +57,9 @@ class FilterLevel(Enum):
ONLY = 2
SPLIT_TOKEN = 1152317803
# DISCORD API