Compare commits
7 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 29fe020056 | |||
| c1b7036465 | |||
| fc9873402f | |||
| 63466a62de | |||
| 79e335bf2c | |||
| 8c9368865a | |||
| ede0060c86 |
@@ -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
|
||||
@@ -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
|
||||
@@ -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
@@ -4,8 +4,9 @@ config/*.ini
|
||||
pastebin.txt
|
||||
.DS_Store
|
||||
test*.py
|
||||
.dockerignore
|
||||
docker-*
|
||||
.vscode/settings.json
|
||||
.vscode
|
||||
|
||||
|
||||
# Byte-compiled / optimized / DLL files
|
||||
|
||||
@@ -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>
|
||||
@@ -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
|
||||
@@ -0,0 +1,3 @@
|
||||
discord.py==2.4.0
|
||||
colorama==0.4.6
|
||||
python-dotenv==1.1.1
|
||||
+25
@@ -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}")
|
||||
@@ -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")
|
||||
@@ -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))
|
||||
@@ -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
@@ -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())
|
||||
@@ -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"
|
||||
@@ -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")
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user