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
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2024 Lars Winzer
|
||||
Copyright (c) 2024-2025 Lars Winzer
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
|
||||
+1
-3
@@ -1,5 +1,3 @@
|
||||
discord.py==2.4.0
|
||||
asyncpraw==7.8.1
|
||||
colorama==0.4.6
|
||||
aiohttp==3.11.11
|
||||
pillow==11.1.0
|
||||
python-dotenv==1.1.1
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
from discord.ext import commands
|
||||
import discord
|
||||
from utils.datetime_tools import get_elapsed_time_small, get_elapsed_time_milliseconds
|
||||
from logging import Logger
|
||||
from datetime import datetime
|
||||
|
||||
class PostIt_Bot(commands.Bot):
|
||||
"""Class to extend the built-in """
|
||||
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 ["debug", "post"]:
|
||||
# await self.load_extension(f"cogs.{cog_name}")
|
||||
# await self.tree.sync()
|
||||
pass
|
||||
|
||||
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}")
|
||||
+106
-122
@@ -1,141 +1,125 @@
|
||||
print(" ____ ____ ___________ __________")
|
||||
print(" / __ \/ __ \/ ___/_ __/ / _/_ __/")
|
||||
print(" / /_/ / / / /\__ \ / /_____ / / / / ")
|
||||
print(" / ____/ /_/ /___/ // /_____// / / / ")
|
||||
print(" /_/ \____//____//_/ /___/ /_/ ")
|
||||
print(" Copyright (c) 2024 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")
|
||||
|
||||
from datetime import datetime
|
||||
startup = datetime.now().timestamp()
|
||||
|
||||
# Initialize the logger
|
||||
from utils.logger.custom_logging import Custom_Logger
|
||||
Custom_Logger.initialize()
|
||||
|
||||
import logging
|
||||
app_logger = logging.getLogger("app")
|
||||
startup_logger = logging.getLogger("app.startup")
|
||||
|
||||
from utils.adv_configparser import Advanced_ConfigParser
|
||||
from utils.datetime_tools import get_elapsed_time_smal, get_elapsed_time_big, get_elapsed_time_milliseconds
|
||||
import discord
|
||||
from discord.ext import commands
|
||||
from pathlib import Path
|
||||
import re
|
||||
import sys
|
||||
import traceback
|
||||
from utils.portal import Portal
|
||||
import asyncio
|
||||
from typing import Union
|
||||
from platforms.reddit import Reddit_Adapter
|
||||
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
|
||||
|
||||
source_path = Path(__file__).resolve()
|
||||
base_path = source_path.parents[1]
|
||||
app_logger.info(f"Using the following path as entrypoint: '{base_path}'")
|
||||
async def main(event_loop:asyncio.AbstractEventLoop, close_event:asyncio.Event):
|
||||
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")
|
||||
|
||||
intents = discord.Intents.default()
|
||||
intents.messages = True
|
||||
# Get timestamp for begin of actual execution
|
||||
startup_time = datetime.now().timestamp()
|
||||
|
||||
class MyBot(commands.Bot):
|
||||
def __init__(self):
|
||||
super().__init__(command_prefix=None, help_command=None, intents=intents)
|
||||
self.__portal:Portal
|
||||
self.__first_on_ready = False
|
||||
# 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} ...")
|
||||
|
||||
def set_portal(self, portal:Portal):
|
||||
self.__portal = portal
|
||||
# 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}'")
|
||||
|
||||
async def setup_hook(self):
|
||||
# Register cogs to handle commands
|
||||
for cog_name in ["debug", "post"]:
|
||||
await self.load_extension(f"cogs.{cog_name}")
|
||||
await self.tree.sync()
|
||||
# Detect operating system and attach
|
||||
shutdown_event = asyncio.Event()
|
||||
def signal_handler(*args):
|
||||
app_logger.info("Shutdown signal recieved, shutting down")
|
||||
event_loop.call_soon_threadsafe(shutdown_event.set)
|
||||
close_event.set()
|
||||
|
||||
async def on_app_command_completion(self, interaction: discord.Interaction, command: Union[discord.app_commands.Command, discord.app_commands.ContextMenu]):
|
||||
"""Called when a `app_commands.Command` or `app_commands.ContextMenu` has successfully completed without error"""
|
||||
self.__portal.no_succeeded_commands += 1
|
||||
print("Command succeeded")
|
||||
match sys.platform:
|
||||
case "linux":
|
||||
app_startup_logger.info("Detected platform: Linux")
|
||||
event_loop.add_signal_handler(signal.SIGINT, signal_handler)
|
||||
event_loop.add_signal_handler(signal.SIGTERM, signal_handler)
|
||||
|
||||
async def on_interaction(self, interaction: discord.Interaction):
|
||||
"""Called when an interaction happened"""
|
||||
match interaction.type.name:
|
||||
case discord.InteractionType.application_command.name:
|
||||
print("Interaction with bot", interaction.command.name)
|
||||
self.__portal.no_executed_commands += 1
|
||||
case discord.InteractionType.ping.name:
|
||||
print("App got pinged by discord")
|
||||
case discord.InteractionType.autocomplete.name:
|
||||
print("Interaction with autocomplete")
|
||||
case discord.InteractionType.modal_submit.name:
|
||||
print("Modal interaction submitted")
|
||||
case discord.InteractionType.component.name:
|
||||
print("Component interaction")
|
||||
case "darwin:":
|
||||
app_startup_logger.info("Detected platform: MacOS (Darwin)")
|
||||
event_loop.add_signal_handler(signal.SIGINT, signal_handler)
|
||||
event_loop.add_signal_handler(signal.SIGTERM, signal_handler)
|
||||
|
||||
async def on_connect(self):
|
||||
"""A coroutine to be called to setup the bot, after the bot is logged in but before it has connected to the Websocket"""
|
||||
if not self.__first_on_ready:
|
||||
startup_logger.info(f"Beginning startup routine ...")
|
||||
routine_begin = datetime.now().timestamp()
|
||||
await self.change_presence(status = discord.Status.dnd, activity = discord.CustomActivity("Executing pre startup routine"))
|
||||
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)
|
||||
|
||||
# Create the adapters for the platforms
|
||||
task_start = datetime.now().timestamp()
|
||||
startup_logger.debug(f"Loading platforms config ...")
|
||||
platforms_config = Advanced_ConfigParser(Path.joinpath(base_path, "config", "platforms.ini"))
|
||||
portal.platforms_config = platforms_config
|
||||
startup_logger.info(f"Loaded platforms config after {get_elapsed_time_milliseconds(datetime.now().timestamp() - task_start)}")
|
||||
# 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")
|
||||
|
||||
# Create platforms adapter
|
||||
task_start = datetime.now().timestamp()
|
||||
startup_logger.debug(f"Creating reddit adapter ...")
|
||||
portal.reddit_adapter = Reddit_Adapter(platforms_config["REDDIT"]["CLIENT_ID"], platforms_config["REDDIT"]["CLIENT_SECRET"])
|
||||
startup_logger.info(f"Created reddit adapter after {get_elapsed_time_milliseconds(datetime.now().timestamp() - task_start)}")
|
||||
# Check if required environment variables are present
|
||||
check_ok = True
|
||||
required_variables = [
|
||||
"DISCORD_TOKEN",
|
||||
]
|
||||
|
||||
await self.change_presence(status = discord.Status.online, activity = None)
|
||||
startup_logger.info(f"Startup routine finished after {get_elapsed_time_milliseconds(datetime.now().timestamp() - routine_begin)}")
|
||||
self.__first_on_ready = True
|
||||
else:
|
||||
startup_logger.info("Startup routine allready executed, omitting this execution")
|
||||
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
|
||||
|
||||
async def on_ready(self):
|
||||
app_logger.info(f"Successfully logged in (after {get_elapsed_time_smal(datetime.now().timestamp() - startup)}) as {self.user}")
|
||||
if not check_ok:
|
||||
app_startup_logger.critical("Multiple required variables are missing. Aborting startup")
|
||||
quit(1)
|
||||
|
||||
bot = MyBot()
|
||||
bot_config = Advanced_ConfigParser(Path.joinpath(base_path, "config", "bot.ini"))
|
||||
if re.match(r'[A-Za-z\d]{24}\.[\w-]{6}\.[\w-]{27}', bot_config["DISCORD"]["TOKEN"]):
|
||||
app_logger.critical("Bot (config/bot.ini) configuration invalid, please set a valid token")
|
||||
quit(1)
|
||||
elif bot_config.compare_to_template() not in ("equal", "config_minus"):
|
||||
app_logger.critical("Bot (config/bot.ini) configuration is missing some parts. Make sure it at least has all the same keys as the template")
|
||||
quit(1)
|
||||
else:
|
||||
app_logger.info("Bot configuration valid, continuing with startup")
|
||||
# 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)
|
||||
|
||||
# Execute some housekeeping actions
|
||||
portal = Portal.instance()
|
||||
portal.bot_config = bot_config
|
||||
portal.STARTUP_TIMESTAMP = startup
|
||||
bot.set_portal(portal)
|
||||
bot_task = event_loop.create_task(bot_instance.start(os.getenv("DISCORD_TOKEN")))
|
||||
shutdown_task = event_loop.create_task(shutdown_event.wait())
|
||||
|
||||
# Setup handlers to handle states of command execution
|
||||
@bot.tree.error
|
||||
async def on_app_command_error(ctx:discord.Interaction, error):
|
||||
"""Executed when exception during command execution occurs"""
|
||||
print('Ignoring exception in command {}:'.format(ctx.command), file=sys.stderr)
|
||||
traceback.print_exception(type(error), error, error.__traceback__)
|
||||
# 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
|
||||
)
|
||||
|
||||
portal.no_failed_commands += 1
|
||||
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)
|
||||
|
||||
try:
|
||||
bot.run(bot_config["DISCORD"]["TOKEN"], log_handler = None)
|
||||
except discord.errors.LoginFailure:
|
||||
app_logger.critical("Improper token has been passed. Aborting startup")
|
||||
quit(1)
|
||||
finally:
|
||||
close_event.set()
|
||||
app_logger.info(f"Exiting. Application ran for {get_elapsed_time_big(datetime.now().timestamp() - startup_time)}")
|
||||
|
||||
app_logger.info("Quitting application ...")
|
||||
asyncio.run(bot.close())
|
||||
app_logger.info(f"Exiting. Application ran for {get_elapsed_time_big(datetime.now().timestamp() - startup)}")
|
||||
# Entry point for execution
|
||||
if __name__ == "__main__":
|
||||
event_loop = asyncio.get_event_loop()
|
||||
asyncio.set_event_loop(event_loop)
|
||||
close_event = asyncio.Event()
|
||||
|
||||
event_loop.run_until_complete(main(event_loop, close_event))
|
||||
|
||||
# Wait until program is ready to close
|
||||
event_loop.run_until_complete(close_event.wait())
|
||||
event_loop.close()
|
||||
|
||||
@@ -10,7 +10,7 @@ def get_elapsed_time_ms(timestamp:float) -> str:
|
||||
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_smal(timestamp:float) -> str:
|
||||
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"
|
||||
|
||||
Reference in New Issue
Block a user