Reworked reload cog
Added the reload app command to reload specific cog Responses now use embeds and are ephemeral, in case of error, an limited traceback is attatched
This commit is contained in:
+84
-9
@@ -4,22 +4,97 @@ from discord.ext import commands
|
|||||||
from cogs.base_cog import Base_Cog
|
from cogs.base_cog import Base_Cog
|
||||||
import logging
|
import logging
|
||||||
from datetime import datetime
|
from datetime import datetime
|
||||||
|
from utils.datetime_tools import get_elapsed_time_milliseconds
|
||||||
|
import traceback
|
||||||
|
from utils.truncate_str import truncate_message_with_notice
|
||||||
|
|
||||||
class Reload_Command(Base_Cog):
|
class Reload_Command(Base_Cog):
|
||||||
def __init__(self, bot:commands.Bot):
|
def __init__(self, bot:commands.Bot):
|
||||||
self.__bot = bot
|
self.__bot = bot
|
||||||
|
self.__cached_cog_names:list[str] = None
|
||||||
super().__init__(logging.getLogger("cmds.reload"))
|
super().__init__(logging.getLogger("cmds.reload"))
|
||||||
|
|
||||||
@app_commands.command(name = "reload_all", description = "Reloads all cogs")
|
async def autocomplete_cog(self, ctx: discord.Interaction, cog_name_stw:str):
|
||||||
async def about(self, ctx: discord.Interaction):
|
"""Helpermethod to recommend the right cog name to the user"""
|
||||||
self._logger.info("Reloading all cogs ...")
|
# Cache the cog names if not cached yet
|
||||||
task_start = datetime.now().timestamp()
|
if not self.__cached_cog_names:
|
||||||
|
self.__cached_cog_names = list(self.__bot.extensions.keys())
|
||||||
cog_names = list(self.__bot.extensions.keys())
|
|
||||||
for extension_name in cog_names:
|
|
||||||
await self.__bot.reload_extension(extension_name)
|
|
||||||
|
|
||||||
await ctx.response.send_message("Cogs successfully reloaded.")
|
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):
|
||||||
|
task_start = datetime.now().timestamp()
|
||||||
|
cog_names = list(self.__bot.extensions.keys())
|
||||||
|
self._logger.debug(f"Reloading all {len(cog_names)} cogs ...")
|
||||||
|
try:
|
||||||
|
embed_cog_stats = "Reloaded the following cogs:\n"
|
||||||
|
for extension_name in cog_names:
|
||||||
|
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()} <--"
|
||||||
|
traceback_str = truncate_message_with_notice(traceback.format_exc(), 1800, "\nSee console for more details on the full traceback")
|
||||||
|
|
||||||
|
embed = discord.Embed(
|
||||||
|
title = "Reloading All Cogs",
|
||||||
|
description = f"{embed_cog_stats} \n\nDuring the reload the following exception occured:\n```{traceback_str}```",
|
||||||
|
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}")
|
||||||
|
finally:
|
||||||
|
await ctx.response.send_message(embed = embed, ephemeral = True)
|
||||||
|
|
||||||
|
@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):
|
||||||
|
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:
|
||||||
|
traceback_str = truncate_message_with_notice(traceback.format_exc(), 1800, "\nSee console for more details on the full traceback")
|
||||||
|
embed = discord.Embed(
|
||||||
|
title = f"Reloading `{cog_name.removeprefix('cogs.').capitalize()}` cog",
|
||||||
|
description = f"Reload failed with the following exception:\n```{traceback_str}```",
|
||||||
|
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)
|
||||||
|
|
||||||
|
# Clear the cache
|
||||||
|
self.__cached_cog_names = None
|
||||||
|
|
||||||
async def setup(bot:commands.Bot):
|
async def setup(bot:commands.Bot):
|
||||||
await bot.add_cog(Reload_Command(bot))
|
await bot.add_cog(Reload_Command(bot))
|
||||||
@@ -18,4 +18,9 @@ def get_elapsed_time_smal(timestamp:float) -> str:
|
|||||||
def get_elapsed_time_big(timestamp:float) -> str:
|
def get_elapsed_time_big(timestamp:float) -> str:
|
||||||
"""Convert an timestamp into an predefined elapsed time format (00days 00hrs 00min)"""
|
"""Convert an timestamp into an predefined elapsed time format (00days 00hrs 00min)"""
|
||||||
time_elapsed = datetime.fromtimestamp(timestamp)
|
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"
|
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,21 @@
|
|||||||
|
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
|
||||||
Reference in New Issue
Block a user