7 Commits

Author SHA1 Message Date
lars-winzer 29fe020056 Add: Cogs to be loaded on bot startup
Additionally: Fix class docstring and remove unused import
2025-12-08 16:32:25 +01:00
lars-winzer c1b7036465 Add: BaseCog, PostCommand and ReloadCommand
Reused BaseCog and PostCommand from old codebase

Used ReloadCommand by Apollo: https://github.com/official-Cromatin/Apollo/blob/3fe5972653da3d858e2601c126bacd244c803ccc/src/cogs/reload.py
2025-12-08 16:31:45 +01:00
lars-winzer fc9873402f Fix: Startup of application after 79e335bf2c broke it 2025-12-08 16:14:56 +01:00
lars-winzer 63466a62de Fix: Typo for sys.platform match case of "darwin" 2025-12-08 15:59:11 +01:00
lars-winzer 79e335bf2c Update: Entry point to now use asyncio.run
As of >3.10 .get_event_loop() is deprecated
2025-12-08 15:58:29 +01:00
lars-winzer 8c9368865a Rewrote main.py, Updated LICENSE
- Pushed bot class into its own file
- Now uses an .env file
- Listens for SIGINT and SIGTERM signals and acts accordingly
2025-08-12 17:56:52 +02:00
lars-winzer ede0060c86 Clean codebase for rewrite 2025-08-12 17:21:23 +02:00
17 changed files with 414 additions and 69 deletions
-15
View File
@@ -1,15 +0,0 @@
name: Code Analysis (Bandit)
on: push
jobs:
bandit:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: mdegis/bandit-action@v1.0
with:
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
path: "src/"
level: high
confidence: high
exit_zero: true
-44
View File
@@ -1,44 +0,0 @@
name: Release Multi-Arch Docker Image
on:
release:
types: [published]
permissions:
contents: read
packages: write
jobs:
build-and-push-docker:
runs-on: ubuntu-latest
steps:
- name: Checkout code
uses: actions/checkout@v4
- name: Set up QEMU
uses: docker/setup-qemu-action@v3
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
- name: Log in to GitHub Container Registry
uses: docker/login-action@v3
with:
registry: ghcr.io
username: ${{ github.actor }}
password: ${{ secrets.GITHUB_TOKEN }}
- name: Transform 'github.repository' to lowercase
id: lowercase_repo
shell: bash
run: echo "repository=${GITHUB_REPOSITORY,,}" >> $GITHUB_OUTPUT
- name: Build and push multi-arch image
uses: docker/build-push-action@v5
with:
context: .
push: true
platforms: linux/amd64,linux/arm64
tags: |
ghcr.io/${{ steps.lowercase_repo.outputs.repository }}:${{ github.event.release.tag_name }}
ghcr.io/${{ steps.lowercase_repo.outputs.repository }}:latest
-9
View File
@@ -1,9 +0,0 @@
name: Linting (Ruff)
on: [push, pull_request]
jobs:
ruff:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: astral-sh/ruff-action@v3
+2 -1
View File
@@ -4,8 +4,9 @@ config/*.ini
pastebin.txt
.DS_Store
test*.py
.dockerignore
docker-*
.vscode/settings.json
.vscode
# Byte-compiled / optimized / DLL files
+3
View File
@@ -0,0 +1,3 @@
[DISCORD]
TOKEN = <PLACE YOUR DISCORD TOKEN FROM THE DEV PORTAL HERE>
OWNER_ID = <PLACE THE USER ID OF THE OWNER HERE>
+6
View File
@@ -0,0 +1,6 @@
; Create an script on the reddit developer page (https://www.reddit.com/prefs/apps)
[REDDIT]
CLIENT_ID = <PASTE YOUR CLIENT ID HERE>
CLIENT_SECRET = <PASTE YOUR CLIENT SECRET HERE>
USER_AGENT = Small discord bot to embed posts (given by url) into an standardized format
EMBED_COLOR = 0xff4500
+3
View File
@@ -0,0 +1,3 @@
discord.py==2.4.0
colorama==0.4.6
python-dotenv==1.1.1
+25
View File
@@ -0,0 +1,25 @@
from discord.ext import commands
import discord
from utils.datetime_tools import get_elapsed_time_small
from logging import Logger
from datetime import datetime
class PostIt_Bot(commands.Bot):
"""Class to extend the built-in bot class"""
def __init__(self, startup_time:float, program_version:str, app_logger:Logger):
intents = discord.Intents.default()
intents.messages = True
super().__init__(command_prefix=None, help_command=None, intents=intents)
self.startup_time = startup_time
self.PROGRAM_VERSION = program_version
self.__logger = app_logger
async def setup_hook(self):
# Register cogs to handle commands
for cog_name in ["post", "reload"]:
await self.load_extension(f"cogs.{cog_name}")
await self.tree.sync()
async def on_ready(self):
self.__logger.info(f"Successfully logged in (after {get_elapsed_time_small(datetime.now().timestamp() - self.startup_time)}) as {self.user}")
View File
+15
View File
@@ -0,0 +1,15 @@
"""Contains the BaseCog"""
from discord.ext import commands
import logging
class BaseCog(commands.Cog):
def __init__(self, bot:commands.Bot, logger:logging.Logger):
self._bot = bot
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")
+28
View File
@@ -0,0 +1,28 @@
"""Contains the implementation for the /post command"""
from .base import BaseCog
import logging
from discord import app_commands, Interaction
from discord.ext import commands
class PostCommand(BaseCog):
def __init__(self, bot):
logger = logging.getLogger("cmds.post")
super().__init__(bot, logger)
@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:Interaction, url:str, custom_note:str = None, use_title:bool = True, quality:app_commands.Choice[int] = 95):
print("Hallo")
async def setup(bot:commands.Bot):
await bot.add_cog(PostCommand(bot))
+113
View File
@@ -0,0 +1,113 @@
import discord
from discord import app_commands
from discord.ext import commands
from .base import BaseCog
import logging
from datetime import datetime
from utils.datetime_tools import get_elapsed_time_milliseconds
class ReloadCommand(BaseCog):
def __init__(self, bot:commands.Bot):
self.__bot = bot
self.__cached_cog_names:list[str] = None
self.__reload_running = False
super().__init__(bot, logging.getLogger("cmds.reload"))
async def autocomplete_cog(self, ctx: discord.Interaction, cog_name_stw:str):
"""Helpermethod to recommend the right cog name to the user"""
# Cache the cog names if not cached yet
if not self.__cached_cog_names:
self.__cached_cog_names = list(self.__bot.extensions.keys())
if cog_name_stw == "":
# Return first 25 cog names, if provided name is empty
matching_cogs = sorted(self.__cached_cog_names)
else:
matching_cogs = []
for cog_name in self.__cached_cog_names:
if cog_name.startswith(cog_name_stw.lower()):
matching_cogs.append(cog_name)
choices = []
for matching_cog in matching_cogs[:25]:
choices.append(app_commands.Choice(name = matching_cog.removeprefix("cogs.").capitalize(), value = matching_cog))
return choices
@app_commands.command(name = "reload_all", description = "Reloads all cogs")
async def reload_all(self, ctx: discord.Interaction):
if self.__reload_running:
await ctx.response.send_message("Reload is allready being executed", ephemeral = True)
else:
self.__reload_running = True
task_start = datetime.now().timestamp()
cog_names = list(self.__bot.extensions.keys())
stepped_over = 0
self._logger.debug(f"Reloading all {len(cog_names)} cogs / extensions ...")
try:
embed_cog_stats = "Reloaded the following cogs:\n"
for extension_name in cog_names:
if "impls" in extension_name:
stepped_over += 1
else:
start_reload = datetime.now().timestamp()
await self.__bot.reload_extension(extension_name)
embed_cog_stats += f"- {extension_name.removeprefix('cogs.').capitalize()} (`{get_elapsed_time_milliseconds(datetime.now().timestamp() - start_reload)}`)\n"
except Exception as error:
embed_cog_stats += f"- {extension_name.removeprefix('cogs.').capitalize()} <--"
embed = discord.Embed(
title = "Reloading All Cogs",
description = f"{embed_cog_stats} \n\nDuring the reload an exception occured",
color = 0xED4337)
raise error
else:
elapsed_time = get_elapsed_time_milliseconds(datetime.now().timestamp() - task_start)
embed = discord.Embed(
title = "Reloading All Cogs",
description = f"{embed_cog_stats}Total time spend: `{elapsed_time}`",
color = 0x4BB543)
self._logger.info(f"Cogs successfully reloaded after {elapsed_time} (Ignored )")
finally:
await ctx.response.send_message(embed = embed, ephemeral = True)
self.__reload_running = False
@app_commands.command(name = "reload", description = "Reload specific cog")
@app_commands.autocomplete(cog_name = autocomplete_cog)
async def reload(self, ctx: discord.Interaction, cog_name:str):
if self.__reload_running:
await ctx.response.send_message("Reload is allready being executed", ephemeral = True)
else:
self.__reload_running = True
task_start = datetime.now().timestamp()
self._logger.debug(f"Reloading {cog_name} cog ...")
try:
await self.__bot.reload_extension(cog_name)
except Exception as error:
embed = discord.Embed(
title = f"Reloading `{cog_name.removeprefix('cogs.').capitalize()}` cog",
description = "Reload failed with an exception",
color = 0xED4337)
raise error
else:
embed = discord.Embed(
title = f"Reloading `{cog_name.removeprefix('cogs.').capitalize()}` cog",
description = f"Total time spend: `{get_elapsed_time_milliseconds(datetime.now().timestamp() - task_start)}`",
color = 0x4BB543)
self._logger.info(f"Cog {cog_name} successfully reloaded after {get_elapsed_time_milliseconds(datetime.now().timestamp() - task_start)}")
finally:
await ctx.response.send_message(embed = embed, ephemeral = True)
self.__reload_running = False
# Clear the cache
self.__cached_cog_names = None
async def setup(bot:commands.Bot):
await bot.add_cog(ReloadCommand(bot))
+115
View File
@@ -0,0 +1,115 @@
import asyncio
from datetime import datetime
from utils.logger.custom_logging import Custom_Logger
import logging
from pathlib import Path
import sys, signal
import os
import dotenv
from bot import PostIt_Bot
from utils.datetime_tools import get_elapsed_time_milliseconds, get_elapsed_time_big
import discord
async def main():
PROGRAM_VERSION = "1.0"
print(" ____ ____ ___________ __________")
print(" / __ \/ __ \/ ___/_ __/ / _/_ __/")
print(" / /_/ / / / /\__ \ / /_____ / / / / ")
print(" / ____/ /_/ /___/ // /_____// / / / ")
print(f" /_/ \____//____//_/ /___/ /_/ v{PROGRAM_VERSION}")
print(" Copyright (c) 2024-2025 Lars Winzer")
print()
print(" Source: https://github.com/official-Cromatin/Post-It")
print(" Report an Issue: https://github.com/official-Cromatin/Post-It/issues/new?assignees=&labels=bug&projects=&template=issue_report.yml")
print("\n")
# Get timestamp for begin of actual execution
startup_time = datetime.now().timestamp()
# Initialize the logger
Custom_Logger.initialize()
app_logger = logging.getLogger("app")
app_startup_logger = logging.getLogger("app.startup")
app_startup_logger.info(f"Starting Post-It v{PROGRAM_VERSION} ...")
# Detect entrypoint
source_path = Path(__file__).resolve()
base_path = source_path.parents[1]
app_startup_logger.info(f"Using the following path as entrypoint: '{base_path}'")
# Detect operating system and attach
shutdown_event = asyncio.Event()
def signal_handler(*args):
app_logger.info("Shutdown signal recieved, shutting down")
shutdown_event.set()
match sys.platform:
case "linux":
app_startup_logger.info("Detected platform: Linux")
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
case "darwin":
app_startup_logger.info("Detected platform: MacOS (Darwin)")
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
case "win32":
app_startup_logger.info("Detected platform: Windows (Win32)")
signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)
case _:
app_startup_logger.fatal(f"Detected unsupported platform: {sys.platform}")
quit(1)
# Load environment variables
if (os.getenv("skip_dotenv", False)):
app_startup_logger.warning("Skipped import of dotenv file")
else:
dotenv.load_dotenv(base_path / ".env")
app_startup_logger.warning("Imported dotenv file")
# Check if required environment variables are present
check_ok = True
required_variables = [
"DISCORD_TOKEN",
]
for variable_name in required_variables:
if os.getenv(variable_name) is None:
app_startup_logger.error(f"Environment variable '{variable_name}' is missing")
check_ok = False
if not check_ok:
app_startup_logger.critical("Multiple required variables are missing. Aborting startup")
quit(1)
# Create bot instance and start bot
app_startup_logger.info(f"Preperations complete after {get_elapsed_time_milliseconds(datetime.now().timestamp() - startup_time)}, launching bot")
try:
bot_instance = PostIt_Bot(startup_time, PROGRAM_VERSION, app_logger)
bot_task = asyncio.create_task(bot_instance.start(os.getenv("DISCORD_TOKEN")))
shutdown_task = asyncio.create_task(shutdown_event.wait())
# Wait if either the bot disconnects or the shutdown event is detected
finished_task, _ = await asyncio.wait(
[shutdown_task, bot_task],
return_when = asyncio.FIRST_COMPLETED
)
if shutdown_task in finished_task:
await bot_instance.close()
app_logger.info("Bot closed connection successfully")
except discord.errors.LoginFailure:
app_logger.critical("Improper token has been passed. Aborting startup")
quit(1)
finally:
app_logger.info(f"Exiting. Application ran for {get_elapsed_time_big(datetime.now().timestamp() - startup_time)}")
# Entry point for execution
if __name__ == "__main__":
asyncio.run(main())
+26
View File
@@ -0,0 +1,26 @@
from datetime import datetime
def get_elapsed_time(timestamp:float) -> str:
"""Convert an timestamp into an predefined elapsed time format (00min 00sec), without miliseconds"""
time_elapsed = datetime.fromtimestamp(timestamp)
return f"{time_elapsed.minute // 60}min {time_elapsed.second:02}sec"
def get_elapsed_time_ms(timestamp:float) -> str:
"""Convert an timestamp into an predefined elapsed time format (00min 00sec 000ms), with miliseconds"""
time_elapsed = datetime.fromtimestamp(timestamp)
return f"{time_elapsed.minute // 60}min {time_elapsed.second:02}sec {time_elapsed.microsecond // 1000}ms"
def get_elapsed_time_small(timestamp:float) -> str:
"""Convert an timestamp into an predefined elapsed time format (00sec 000ms)"""
time_elapsed = datetime.fromtimestamp(timestamp)
return f"{time_elapsed.second:02}sec {time_elapsed.microsecond // 1000:03}ms"
def get_elapsed_time_big(timestamp:float) -> str:
"""Convert an timestamp into an predefined elapsed time format (00days 00hrs 00min)"""
time_elapsed = datetime.fromtimestamp(timestamp)
return f"{time_elapsed.day - 1}days {time_elapsed.hour - 1}hrs {1 if time_elapsed.minute == 0 else time_elapsed.minute}min"
def get_elapsed_time_milliseconds(timestamp:float) -> str:
"""Convert an timestamp into an predefined elapsed time format (0000ms)"""
time_elapsed = datetime.fromtimestamp(timestamp)
return f"{time_elapsed.second * 1000 + time_elapsed.microsecond // 1000}ms"
+37
View File
@@ -0,0 +1,37 @@
import logging
from utils.logger.formatter import Colored_Formatter
class Custom_Logger:
console_handler: logging.StreamHandler
colored_formatter: Colored_Formatter
@classmethod
def initialize(cls):
# Create the handler for logging to the console
cls.console_handler = logging.StreamHandler()
cls.console_handler.setLevel(logging.DEBUG)
# Create the colorful formatter
cls.colored_formatter = Colored_Formatter()
cls.console_handler.setFormatter(cls.colored_formatter)
# Create the loggers for the different sections of the app
discordpy_logger = logging.getLogger("discord")
discordpy_logger.addHandler(cls.console_handler)
discordpy_logger.setLevel(logging.INFO)
app_logger = logging.getLogger("app")
app_logger.addHandler(cls.console_handler)
app_logger.setLevel(logging.DEBUG)
utils_logger = logging.getLogger("utils")
utils_logger.addHandler(cls.console_handler)
utils_logger.setLevel(logging.DEBUG)
commands_logger = logging.getLogger("cmds")
commands_logger.addHandler(cls.console_handler)
commands_logger.setLevel(logging.DEBUG)
logging.getLogger('discord.app_commands.tree').setLevel(logging.DEBUG)
app_logger.debug("Logging successfully initialized")
+11
View File
@@ -0,0 +1,11 @@
from functools import wraps
import discord
def log_command_execution(func):
"""Wraps an command function to print execution logs to the console"""
@wraps(func)
async def wrapper(self, interaction:discord.Interaction, *args, **kwargs):
self._logger.debug(f"User {interaction.user} in channel {interaction.channel} on guild {interaction.guild} executed the {func.__name__} command")
return await func(self, interaction, *args, **kwargs)
return wrapper
+30
View File
@@ -0,0 +1,30 @@
from logging import Formatter
from colorama import Style, Fore
class Colored_Formatter(Formatter):
def __init__(self, fmt=None, datefmt='%Y-%m-%d %H:%M:%S', style='%'):
super().__init__(fmt, datefmt, style)
def format(self, record):
# Formatieren des Zeitstempels fett
date = f"{Style.BRIGHT}{Fore.BLACK}{self.formatTime(record, self.datefmt)}{Style.RESET_ALL}"
# Färben des Log-Levels mit fester Breite
levelname_color = {
'DEBUG': Fore.CYAN,
'INFO': Fore.BLUE,
'WARNING': Fore.YELLOW,
'ERROR': Fore.RED,
'CRITICAL': f"{Style.BRIGHT}{Fore.RED}"
}
levelname = f"{levelname_color.get(record.levelname, Fore.WHITE)}{record.levelname:<8}{Fore.RESET}{Style.RESET_ALL}"
# Färben des Logger-Namens in Magenta mit fester Breite
name = f"{Fore.MAGENTA}{record.name:<20}{Fore.RESET}"
# Die Nachricht bleibt unverändert
message = record.getMessage()
# Zusammenfügen der gefärbten Teile
formatted_record = f"{date} {levelname} {name} {message}"
return formatted_record