lars-winzer/Post-It
1
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Clean codebase for rewrite

Browse Source
This commit is contained in:
Lars Winzer
2025-08-12 17:21:23 +02:00
parent 435a2486a8
commit ede0060c86
9 changed files with 0 additions and 695 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cogs/base_cog.py
-14
View File
@@ -1,14 +0,0 @@
import discord
from discord import app_commands
from discord.ext import commands
import logging
class Base_Cog(commands.Cog):
def __init__(self, logger:logging.Logger):
self._logger = logger
async def cog_load(self):
self._logger.debug(f"Cog for the '{self.__class__.__name__}' command got loaded")
async def cog_unload(self):
self._logger.debug(f"Cog for the '{self.__class__.__name__}' command got unloaded")
src/cogs/debug.py
-47
View File
@@ -1,47 +0,0 @@
import discord
from discord import app_commands
from discord.ext import commands
from cogs.base_cog import Base_Cog
import logging
from utils.portal import Portal
from datetime import datetime
from utils.datetime_tools import get_elapsed_time_big
from utils.logger.decorator import log_command_execution
class Debug_Command(Base_Cog):
def __init__(self, bot:commands.Bot):
self.__bot = bot
super().__init__(logging.getLogger("cmds.debug"))
@app_commands.command(name = "debug", description = "Provides debug informations about the bot, useful for troubleshooting")
@log_command_execution
async def debug(self, ctx:discord.Interaction):
portal:Portal = Portal.instance()
embed = discord.Embed(title="Debug Information")
embed.add_field(name="Version",
value=f"`{portal.PROGRAM_VERSION}`",
inline=True)
embed.add_field(name="Uptime",
value=f"{get_elapsed_time_big(datetime.now().timestamp() - portal.STARTUP_TIMESTAMP)}",
inline=True)
embed.add_field(name="Bot Owner",
value=f"<@{portal.bot_config['DISCORD']['OWNER_ID']}>",
inline=True)
embed.add_field(name="Latency to Gateway",
value=f"{round(self.__bot.latency * 1000, 2)}ms",
inline=True)
embed.add_field(name="Application ID",
value=f"{self.__bot.application_id}",
inline=True)
embed.add_field(name="Number of Guilds",
value=f"{len(self.__bot.guilds)}",
inline=True)
embed.add_field(name = "Number of executed commands", value=f"Total: {portal.no_executed_commands}\nSucceeded: {portal.no_succeeded_commands}\nFailed: {portal.no_failed_commands}")
await ctx.response.send_message(embed=embed)
async def setup(bot:commands.Bot):
await bot.add_cog(Debug_Command(bot))
src/cogs/post.py
-157
View File
@@ -1,157 +0,0 @@
import discord
from discord import app_commands
from discord.ext import commands
from cogs.base_cog import Base_Cog
import logging
from urllib.parse import urlparse
from praw.models import Submission, Subreddit
from utils.portal import Portal
import aiohttp
from PIL import Image
from io import BytesIO
from datetime import datetime
from utils.datetime_tools import get_elapsed_time_milliseconds
class NoMediaFound(Exception):
pass
class Post_Command(Base_Cog):
def __init__(self, bot:commands.Bot):
self.__bot = bot
super().__init__(logging.getLogger("cmds.post"))
@app_commands.command(name = "post", description = "Post an embed in the Current Channel with a link to the content")
@app_commands.describe(url = "URL to the post", custom_note = "Describe the post with your own note", use_title = "Display the title of the post", quality = "Specifies the quality of the converted image, closer to 100 is better")
@app_commands.choices(quality = [
app_commands.Choice(name = "Poor (60)", value = 60),
app_commands.Choice(name = "Fair (70)", value = 70),
app_commands.Choice(name = "Good (80)", value = 80),
app_commands.Choice(name = "Very Good (85)", value = 85),
app_commands.Choice(name = "Excellent (90)", value = 90),
app_commands.Choice(name = "Superior (95)", value = 95),
app_commands.Choice(name = "Perfect (100)", value = 100)
])
async def post(self, ctx:discord.Interaction, url:str, custom_note:str = None, use_title:bool = True, quality:app_commands.Choice[int] = 95):
try:
domain_info = urlparse(url)
portal = Portal.instance()
toplevel_domain = '.'.join(domain_info.netloc.split('.')[-2:])
begin_process = datetime.now().timestamp()
match toplevel_domain:
case "reddit.com":
self._logger.debug(f"Recieved command by {ctx.user} ({ctx.user.id}) for reddit ({url})")
subm:Submission = await portal.reddit_adapter.fetch(url)
image_urls = []
# Check if submission has a gallery
if hasattr(subm, "media_metadata"):
for media_id, media in subm.media_metadata.items():
file_extension = media["m"].split("/")[1]
if file_extension not in ("jpg", "jpeg", "png", "webp", "heic", "heif"):
continue
image_urls.append(f"https://i.redd.it/{media_id}.{file_extension}")
else:
image_urls.append(subm.url)
image_count = len(image_urls)
if image_count == 0:
raise NoMediaFound
self._logger.debug(f"Found {image_count} image urls for the post")
progress_title = f"`{image_count}` images are going to be converted, it may take a while."
progress_temp = progress_title + f"\n`0` of `{image_count}` have already been loaded"
await ctx.response.send_message(progress_temp, ephemeral = True)
if isinstance(quality, app_commands.Choice):
quality_value = quality.value
else:
quality_value = quality
# Download and convert each image
image_files:list[discord.File] = []
begin_conversion = datetime.now().timestamp()
async with aiohttp.ClientSession() as session:
index = 0
for image_url in image_urls:
async with session.get(image_url) as response:
response.raise_for_status()
image_data = await response.read()
original_image = Image.open(BytesIO(image_data))
webp_buffer = BytesIO()
original_image.save(webp_buffer, format = "WEBP", quality = quality_value)
webp_buffer.seek(0)
image_file = discord.File(webp_buffer, filename = f"image_{index}.webp")
image_files.append(image_file)
index += 1
progress_temp = progress_title + f"\n`{index}` of `{image_count}` have already been loaded"
await ctx.edit_original_response(content = progress_temp)
self._logger.debug(f"Downloaded and converted {len(image_files)} images in {get_elapsed_time_milliseconds(datetime.now().timestamp() - begin_conversion)}")
author = subm.author.name if subm.author else "Author not found"
content = f":copyright: [{author}]({url})"
if use_title:
content += f"\n## {subm.title}"
if custom_note:
content += f"\n> {custom_note}"
await ctx.delete_original_response()
message = await ctx.followup.send(
content = content,
suppress_embeds = True,
files = image_files
)
self._logger.info(f"Successfully processed the command executed by {ctx.user.name} ({ctx.user.id}) after {get_elapsed_time_milliseconds(datetime.now().timestamp() - begin_process)} (ID of message: {message.id})")
# No domain for seperation found
case _:
if toplevel_domain == "":
toplevel_domain = "not_found"
embed = discord.Embed(
title = "Domain not found",
description = f"The requested domain `{toplevel_domain}` is currently not supported\nOpen [an issue](https://github.com/official-Cromatin/Post-It/issues/new?assignees=&labels=feature-request&projects=&template=feature_request.yml) to request support for it.\n\nCurrently supported plattforms:\n- Reddit",
color = 0xED4337)
await ctx.response.send_message(embed = embed, ephemeral = True)
except NoMediaFound:
self._logger.error(f"Aborted issued command by {ctx.user.name} ({ctx.user.id}). Post had no media attatched")
# Delete the original response, if existing
try:
await ctx.delete_original_response()
except discord.NotFound:
pass
embed = discord.Embed(
title = "No media found",
description = "The post had no media attatched, or was in an unsupported format\nVideos are not supported! (yet)",
color = 0xED4337
)
embed.set_footer(text = "Supported image formats: jpg, jpeg, png, webp, heic, heif")
except Exception as error:
self._logger.error(f"Could not complete command by {ctx.user.name} ({ctx.user.id})")
self._logger.exception(error, stack_info = True)
# Delete the original response, if existing
try:
await ctx.delete_original_response()
except discord.NotFound:
pass
# Explain to the user what the error was
embed = discord.Embed(
title = "Error while processing",
description = f"While we processed your request, the following exception occured: `{error}`",
color = 0xED4337
)
await ctx.followup.send(embed = embed, ephemeral = True)
async def setup(bot:commands.Bot):
await bot.add_cog(Post_Command(bot))
src/platforms/reddit.py
-45
View File
@@ -1,45 +0,0 @@
import asyncpraw
import asyncpraw.models
import logging
from utils.event_counter import Event_Counter
from utils.datetime_tools import get_elapsed_time_milliseconds
from datetime import datetime
class Reddit_Adapter(asyncpraw.Reddit):
"""A class that extends and abstracts the functionality of the `asyncpraw.Reddit` class by adding
logging and request tracking capabilities."""
VERSION = "1.0"
number_of_instances = 0
def __init__(self, client_id:str, client_secret:str):
"""Initializes the Reddit Adapter, while stating credentials for the login to the reddit api"""
self.__instance_number = self.__class__.number_of_instances
self.__class__.number_of_instances += 1
super().__init__(
client_id = client_id,
client_secret = client_secret,
user_agent="Small discord bot to embed posts (given by url) into an standardized format"
)
self.__events = Event_Counter(1000)
self.__logger = logging.getLogger(f"pltfm.reddit.{self.__instance_number}")
async def fetch(self, post_url:str) -> asyncpraw.models.Submission:
"""Fetches specified submission (post) and returns it"""
start_time = datetime.now().timestamp()
self.__events.increment()
subm = await self.submission(url = post_url)
self.__logger.debug(f"Submission for post (URL: {post_url}), successfully fetched after {get_elapsed_time_milliseconds(datetime.now().timestamp() - start_time)}")
return subm
def get_total_requests(self) -> int:
"""Returns the total number of requests made since the creation of the adapter"""
return self.__events.get_total_events()
def get_events_last_5m(self) -> int:
"""Returns the number of requests made in the last 5 minutes"""
return self.__events.get_count("5m")
def get_events_last_5m_10m_15m(self) -> tuple[int]:
"""Returns an tuple containing the number of requests made in the last 5, 10 and 15 minutes"""
return (self.__events.get_count("5m"), self.__events.get_count("10m"), self.__events.get_count("15m"))
src/utils/adv_configparser.py
-249
View File
@@ -1,249 +0,0 @@
import configparser
import logging
from os.path import isfile, split
from pathlib import Path
from shutil import copy
class Advanced_ConfigParser(configparser.ConfigParser):
"""Utility class to ease the use of the configparser libary"""
VERSION = "2.5"
number_of_instances = 0
def __init__(self, path:str, allow_template:bool = True, allow_update:bool = True) -> None:
"""
Initializes the Advanced_ConfigParser instance.
Args:
path (str): The path to the configuration file.
allow_template (bool, optional): Determines if an existing template can be used when the config file is missing. Default is True.
allow_update (bool, optional): Specifies if the config file should be updated based on a template. Default is True.
This constructor loads the configuration file if it exists, optionally creates one from a template if allowed,
and can update the configuration based on the template. It also initializes logging and tracks pending changes.
"""
super().__init__()
self.__instance_number = self.__class__.number_of_instances
self.__class__.number_of_instances += 1
self.__from_template = False
self.__logger = logging.getLogger(f"utils.config.{self.__instance_number}")
self.__logger.debug(f"New instance {self.__instance_number} of class created")
# Check existence of provided file
self.__template_path = self.__get_template_path(path)
if isfile(path):
self.__logger.debug(f"Opening existing file at path: '{path}'")
else:
# Check if it is allowed to use an existing template
if allow_template:
if isfile(self.__template_path):
self.__logger.info(f"Using existing template file {self.__template_path}")
copy(self.__template_path, path)
self.__from_template = True
else:
self.__logger.warning("No potential template found, starting with empty file")
# Load the configfile
self.__file_path = path
self.__pending_changes = 0
try:
self.read_file(open(path))
except:
pass
else:
self.__logger.info(f"Successfully opened config file at path {path}")
# Check for update if allowed
if allow_update:
self.check_for_update()
def __get_template_path(self, path:str) -> str | None:
"""Looks for a existing template and returns the path to it
If no config file was found, None is returned"""
path_to_file, file_name = split(path)
file_name = file_name.split(".")[0]
path_to_temp = Path(path_to_file) / f".{file_name}.template"
return path_to_temp
def __has_all_template_options(self, template:configparser.ConfigParser):
"""Check if config contains all sections and options from the template"""
for section in template.sections():
if section not in self:
return False
template_options = set(template.options(section))
config_options = set(config.options(section))
if not template_options.issubset(config_options):
return False
return True
def set(self, section: str, option: str, value: str | None = None) -> None:
old_value = self.get(section, option, fallback=None)
if old_value:
self.__logger.debug(f"New value in section '{section}' at option '{option}' added: '{value}'")
self.__pending_changes += 1
else:
if self.has_option(section, option):
self.__logger.debug(f"Value in section '{section}' at option '{option}' changed: '{old_value}' -> '{value}'")
if old_value != value:
self.__pending_changes += 1
return super().set(section, option, value)
def remove_option(self, section, option):
if self.has_option(section, option):
value = self.get(section, option)
self.__logger.debug(f"Option removed in section '{section}' for '{option}': '{value}'")
self.__pending_changes += 1
return super().remove_option(section, option)
def remove_section(self, section):
if self.has_section(section):
self.__logger.info(f"Section removed: '{section}'")
self.__pending_changes += 1
return super().remove_section(section)
def save(self):
"""Saves changes to disk"""
with open(self.__file_path, 'w') as configfile:
self.write(configfile)
self.__logger.debug(f"Contents ({self.__pending_changes} changes) have been saved to disk")
def get_config_file_path(self) -> str:
"""Returns the path to the config file"""
return self.__file_path
def get_template_file_path(self) -> str:
"""Returns the path to the template file, a config could be created from
Returns None if the is no template file"""
return self.__template_path
def compare_to_template(self) -> str:
"""Returns one of multiple possible keywords depending on the relation of both files
+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| Return Keyword | Description |
+-----------------+--------------------------------------------------------------------------------------------------------------------------+
| not_found | No template file found |
| equal | Both files contain the same sections and options |
| config_plus | Config contains additional sections and/or options compared to template |
| config_options | Config contains additional options but same sections |
| config_base | Same as "config_plus" but additionally config contains all sections and options from the template |
| config_minus | Config contains fewer sections and/or options as template |
+-----------------+--------------------------------------------------------------------------------------------------------------------------+"""
# Check existence of template
if self.__template_path == None:
return "not_found"
template_config = configparser.ConfigParser()
template_config.read_file(open(self.__template_path))
config_sections = set(self.sections())
template_sections = set(template_config.sections())
# Check if config contains all the same (or more) sections than template
same_or_more_sections = config_sections >= template_sections
# Check if both files have the same amount of sections
same_amount_of_sections = len(config_sections) == len(template_sections)
# Consolidate the options of both files
all_config_options = set()
all_template_options = set()
all_sections = config_sections.union(template_sections)
config_all_present = True
template_all_present = True
for section in all_sections:
try:
for option in self.options(section):
all_config_options.add(f"{section}.{option}")
except configparser.NoSectionError:
config_all_present = False
try:
for option in template_config.options(section):
all_template_options.add(f"{section}.{option}")
except configparser.NoSectionError:
template_all_present = False
# Check if both have the same options
same_or_more_options = all_config_options >= all_template_options
# Check if both have the same amount of options
same_amount_of_options = len(all_config_options) == len(all_template_options)
# Check for different cases
# Since both files have the same (or more) options (and the options contain the section name) and contain the same amount, they must be equal
if same_or_more_options and same_amount_of_options:
return "equal"
# Since the config contains not even the same sections as present in the template, it doesn't contain the same base options
if not same_or_more_sections:
return "config_minus"
# If the config contains more sections or options than the template
if same_or_more_sections:
if same_or_more_options:
# If config contains all sections and options from template but also has extras
return "config_base"
else:
# Config has extra sections and/or options, but not necessarily all from template
return "config_plus"
# If the config contains the same sections but additional options within those sections
if same_amount_of_sections and not same_amount_of_options:
return "config_options"
return "config_plus"
def created_from_template(self) -> bool:
"""Will ONLY return `True` if the config file has been created from template at the first run"""
return self.__from_template
def check_for_update(self, omit_save:bool = False):
"""Updates the already created config to the same level as the template,
no options are deleted or their values changed."""
if self.__template_path == None:
raise FileNotFoundError
template_config = configparser.ConfigParser()
template_config.read_file(open(self.__template_path))
updated_options = 0
for section in template_config.sections():
options = template_config.options(section)
for option in options:
if not self.has_option(section, option):
self.set(section, option, template_config.get(section, option))
updated_options += 1
if updated_options:
_, file_name = split(self.__file_path)
self.__logger.error(f"The confg updated at {updated_options} locations, you might want check the {file_name} for unchanged placeholders")
if not omit_save:
self.save()
if __name__ == "__main__":
# Dummy logger
console = logging.StreamHandler()
console.setLevel(logging.DEBUG)
logger = logging.getLogger("utils")
logger.addHandler(console)
logger.setLevel(logging.DEBUG)
file_path = Path(__file__).resolve()
base_path = file_path.parents[2]
print(f"Base path of the program: {base_path}")
config = Advanced_ConfigParser(Path.joinpath(base_path, "config", "bot.ini"))
print(config.compare_to_template())
# config["DEFAULT"]["Compression"] = "no"
# config.save()
src/utils/event_counter.py
-110
View File
@@ -1,110 +0,0 @@
from datetime import datetime
import re
class Event_Counter:
"""A class to track and count events, storing the event timestamps and allowing queries
for the number of events in specific time windows (e.g., last 5 minutes, 10 minutes)."""
REGEX_PATTERN = r'(\d+)\s*(sec|second|seconds|s|min|minute|minutes|m|h|hour|hours|d|day|days)'
TIME_MULTIPLIERS = {
'sec': 1, 'second': 1, 'seconds': 1, 's': 1,
'min': 60, 'minute': 60, 'minutes': 60, 'm': 60,
'h': 3600, 'hour': 3600, 'hours': 3600,
'd': 86400, 'day': 86400, 'days': 86400
}
def __init__(self, retention_duration:int = 900) -> None:
"""
Initializes the Event_Counter object with a specified retention duration.
Args:
retention_duration (int, optional):
The time period (in seconds) for which event data should be retained. Defaults to 900 (15 minutes).
uwu
"""
self.__events:list[float] = []
self.__total_events = 0
self.__retention_duration = retention_duration
def cleanup(self):
"""Cleans up events that are older than the retention duration.
Removes all events that occurred outside the defined retention window."""
offset_time = datetime.now().timestamp() - self.__retention_duration
while len(self.__events) > 0:
event_time = self.__events[0]
if event_time < offset_time:
self.__events.pop(0)
else:
break
@classmethod
def duration_to_seconds(cls, duration_str:str) -> int:
"""Converts a human-readable duration string (e.g., '5min 2sec') into its equivalent in seconds.
Args:
duration_str (str): The string representing the duration (e.g., "5min 2sec").
Returns:
int: The total duration in seconds derived from the input string."""
matches = re.findall(cls.REGEX_PATTERN, duration_str, flags=re.IGNORECASE)
total_seconds = 0
for value, unit in matches:
unit = unit.lower()
total_seconds += int(value) * cls.TIME_MULTIPLIERS[unit]
return total_seconds
def increment(self, count:int = 1, skip_cleanup:bool = False):
"""Increments the event count by a specified number of events.
Args:
count (int, optional): The number of events to add. Defaults to 1.
skip_cleanup (bool, optional): If True, skips the cleanup process. Defaults to False."""
timestamp = datetime.now().timestamp()
self.__total_events += count
self.__events.extend([timestamp] * count)
if skip_cleanup:
return
self.cleanup()
def get_count(self, time_window:str = "5M") -> int:
"""Returns the number of events that occurred within a specified time window.
Args:
time_window (str, optional):
The time window to query, e.g., "5M" for 5 minutes. Defaults to "5M".
Returns:
int: The count of events that occurred within the specified time window."""
self.cleanup()
offset_time = datetime.now().timestamp() - self.duration_to_seconds(time_window)
counter = 0
for event_time in self.__events:
if event_time > offset_time:
counter += 1
else:
break
return counter
def get_total_events(self) -> int:
"""Returns the total number of events tracked by the Event_Counter.
Returns:
int: The total number of events that have been recorded."""
return self.__total_events
if __name__ == "__main__":
from time import sleep
counter = Event_Counter(1)
counter.increment()
sleep(1)
counter.increment()
print(counter.get_count("5sEc"))
sleep(0)
print(counter.get_count("5sEc"))
src/utils/portal.py
-14
View File
@@ -1,14 +0,0 @@
from utils.singleton import Singleton
from utils.adv_configparser import Advanced_ConfigParser
from platforms.reddit import Reddit_Adapter
@Singleton
class Portal:
PROGRAM_VERSION = "0.3"
bot_config:Advanced_ConfigParser = None
platforms_config:Advanced_ConfigParser = None
STARTUP_TIMESTAMP:float = None
no_executed_commands:int = 0
no_succeeded_commands:int = 0
no_failed_commands:int = 0
reddit_adapter: Reddit_Adapter = None
src/utils/singleton.py
-38
View File
@@ -1,38 +0,0 @@
class Singleton:
"""
A non-thread-safe helper class to ease implementing singletons.
This should be used as a decorator -- not a metaclass -- to the
class that should be a singleton.
The decorated class can define one `__init__` function that
takes only the `self` argument. Also, the decorated class cannot be
inherited from. Other than that, there are no restrictions that apply
to the decorated class.
To get the singleton instance, use the `instance` method. Trying
to use `__call__` will result in a `TypeError` being raised.
"""
# Source: https://stackoverflow.com/questions/31875/is-there-a-simple-elegant-way-to-define-singletons
def __init__(self, decorated):
self._decorated = decorated
def instance(self):
"""
Returns the singleton instance. Upon its first call, it creates a
new instance of the decorated class and calls its `__init__` method.
On all subsequent calls, the already created instance is returned.
"""
try:
return self._instance
except AttributeError:
self._instance = self._decorated()
return self._instance
def __call__(self):
raise TypeError('Singletons must be accessed through `instance()`.')
def __instancecheck__(self, inst):
return isinstance(inst, self._decorated)
src/utils/truncate_str.py
-21
View File
@@ -1,21 +0,0 @@
def truncate_message_with_notice(output:str, max_length:int = 1000, suffix:str = "Output was truncated") -> str:
"""Trim a long output string to a specified length and append a suffix message if truncation occurs"""
total_max_length = max_length - len(suffix) # Adjust for the suffix length
# If the output is already shorter than the max length, return it as is
if len(output) <= max_length:
return output
# Split the output into lines and add lines until the length exceeds the allowed maximum
result = ""
lines = output.splitlines()
for line in lines:
if len(result) + len(line) + 1 > total_max_length: # +1 accounts for the line break
break
result += line + "\n"
# Append the suffix message
result += suffix
return result