Rewrite debug command
This commit is contained in:
@@ -5,7 +5,7 @@ import function as func
|
||||
|
||||
from discord.ext import commands
|
||||
|
||||
class Nodes(commands.Cog):
|
||||
class Listeners(commands.Cog):
|
||||
"""Music Cog."""
|
||||
|
||||
def __init__(self, bot: commands.Bot):
|
||||
@@ -82,4 +82,4 @@ class Nodes(commands.Cog):
|
||||
})
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
await bot.add_cog(Nodes(bot))
|
||||
await bot.add_cog(Listeners(bot))
|
||||
|
||||
@@ -1,9 +1,6 @@
|
||||
import discord
|
||||
import voicelink
|
||||
import io
|
||||
import contextlib
|
||||
import textwrap
|
||||
import traceback
|
||||
import psutil
|
||||
import function as func
|
||||
|
||||
from typing import Tuple
|
||||
@@ -19,11 +16,18 @@ from function import (
|
||||
get_aliases,
|
||||
cooldown_check
|
||||
)
|
||||
from views import DebugModal, HelpView, EmbedBuilderView
|
||||
from views import DebugView, HelpView, EmbedBuilderView
|
||||
|
||||
class Admin(commands.Cog, name="settings"):
|
||||
def formatBytes(bytes: int, unit: bool = False):
|
||||
if bytes <= 1_000_000_000:
|
||||
return f"{bytes / (1024 ** 2):.1f}" + ("MB" if unit else "")
|
||||
|
||||
else:
|
||||
return f"{bytes / (1024 ** 3):.1f}" + ("GB" if unit else "")
|
||||
|
||||
class Settings(commands.Cog, name="settings"):
|
||||
def __init__(self, bot) -> None:
|
||||
self.bot = bot
|
||||
self.bot: commands.Bot = bot
|
||||
self.description = "This category is only available to admin permissions on the server."
|
||||
|
||||
def get_settings(self, ctx: commands.Context) -> Tuple[voicelink.Player, dict]:
|
||||
@@ -246,53 +250,41 @@ class Admin(commands.Cog, name="settings"):
|
||||
if interaction.user.id not in func.settings.bot_access_user:
|
||||
return await interaction.response.send_message("You are not able to use this command!")
|
||||
|
||||
def clear_code(content: str):
|
||||
if content.startswith("```") and content.endswith("```"):
|
||||
return "\n".join(content.split("\n")[1:])[:-3]
|
||||
else:
|
||||
return content
|
||||
memory = psutil.virtual_memory()
|
||||
disk = psutil.disk_usage('/')
|
||||
|
||||
modal = DebugModal(title="Debug Panel")
|
||||
await interaction.response.send_modal(modal)
|
||||
await modal.wait()
|
||||
available_memory, total_memory = memory.available, memory.total
|
||||
used_disk_space, total_disk_space = disk.used, disk.total
|
||||
embed = discord.Embed(title="📄 Debug Panel", color=func.settings.embed_color)
|
||||
embed.description = "```== System Info ==\n" \
|
||||
f"• CPU: {psutil.cpu_freq().current}Mhz ({psutil.cpu_percent()}%)\n" \
|
||||
f"• RAM: {formatBytes(total_memory - available_memory)}/{formatBytes(total_memory, True)} ({memory.percent}%)\n" \
|
||||
f"• DISK: {formatBytes(total_disk_space - used_disk_space)}/{formatBytes(total_disk_space, True)} ({disk.percent}%)```"
|
||||
|
||||
if modal.values is None:
|
||||
return
|
||||
embed.add_field(
|
||||
name="🤖 Bot Information",
|
||||
value=f"```• LATENCY: {self.bot.latency:.2f}ms\n" \
|
||||
f"• GUILDS: {len(self.bot.guilds)}\n" \
|
||||
f"• USERS: {sum([guild.member_count for guild in self.bot.guilds])}\n" \
|
||||
f"• PLAYERS: {len(self.bot.voice_clients)}```",
|
||||
inline=False
|
||||
)
|
||||
|
||||
e = None
|
||||
|
||||
local_variables = {
|
||||
"discord": discord,
|
||||
"commands": commands,
|
||||
"voicelink": voicelink,
|
||||
"bot": self.bot,
|
||||
"interaction": interaction,
|
||||
"channel": interaction.channel,
|
||||
"author": interaction.user,
|
||||
"guild": interaction.guild,
|
||||
"message": interaction.message,
|
||||
"input": None
|
||||
}
|
||||
|
||||
code = clear_code(modal.values)
|
||||
str_obj = io.StringIO() # Retrieves a stream of data
|
||||
try:
|
||||
with contextlib.redirect_stdout(str_obj):
|
||||
exec(
|
||||
f"async def func():\n{textwrap.indent(code, ' ')}", local_variables)
|
||||
obj = await local_variables["func"]()
|
||||
result = f"{str_obj.getvalue()}\n-- {obj}\n"
|
||||
except Exception as e:
|
||||
errormsg = ''.join(
|
||||
traceback.format_exception(e, e, e.__traceback__))
|
||||
return await interaction.followup.send(f"```py\n{errormsg}```")
|
||||
|
||||
string = result.split("\n")
|
||||
text = ""
|
||||
for index, i in enumerate(string, start=1):
|
||||
text += f"{'%03d' % index} | {i}\n"
|
||||
return await interaction.followup.send(f"```{text}```")
|
||||
node: voicelink.Node
|
||||
for name, node in voicelink.NodePool._nodes.items():
|
||||
total_memory = node.stats.used + node.stats.free
|
||||
embed.add_field(
|
||||
name=f"{name} Node - " + ("🟢 Connected" if node._available else "🔴 Disconnected"),
|
||||
value=f"```• ADDRESS: {node._host}:{node._port}\n" \
|
||||
f"• PLAYERS: {len(node._players)}\n" \
|
||||
f"• CPU: {node.stats.cpu_process_load:.1f}%\n" \
|
||||
f"• RAM: {formatBytes(node.stats.free)}/{formatBytes(total_memory, True)} ({(node.stats.free/total_memory) * 100:.1f}%)\n"
|
||||
f"• LATENCY: {node.latency:.2f}ms\n" \
|
||||
f"• UPTIME: {func.time(node.stats.uptime)}```",
|
||||
inline=True
|
||||
)
|
||||
|
||||
await interaction.response.send_message(embed=embed, view=DebugView(self.bot), ephemeral=True)
|
||||
|
||||
async def setup(bot: commands.Bot) -> None:
|
||||
await bot.add_cog(Admin(bot))
|
||||
await bot.add_cog(Settings(bot))
|
||||
@@ -9,3 +9,4 @@ beautifulsoup4==4.11.1
|
||||
websockets==10.4
|
||||
Flask==2.2.3
|
||||
Flask-SocketIO==5.3.2
|
||||
psutil==5.9.5
|
||||
|
||||
@@ -13,5 +13,5 @@ from .chapter import ChapterView
|
||||
from .playlist import PlaylistView, CreateView
|
||||
from .inbox import InboxView
|
||||
from .link import LinkView
|
||||
from .debug import DebugModal
|
||||
from .debug import DebugView
|
||||
from .embedBuilder import EmbedBuilderView
|
||||
|
||||
159
views/debug.py
159
views/debug.py
@@ -22,20 +22,169 @@ SOFTWARE.
|
||||
"""
|
||||
|
||||
import discord
|
||||
import function
|
||||
import io
|
||||
import contextlib
|
||||
import textwrap
|
||||
import traceback
|
||||
|
||||
class DebugModal(discord.ui.Modal):
|
||||
def __init__(self, *args, **kwargs) -> None:
|
||||
from discord.ext import commands
|
||||
|
||||
class ExceuteModal(discord.ui.Modal):
|
||||
def __init__(self, code: str, *args, **kwargs) -> None:
|
||||
super().__init__(*args, **kwargs)
|
||||
self.values = None
|
||||
self.code = code
|
||||
|
||||
self.add_item(
|
||||
discord.ui.TextInput(
|
||||
label="Code Runner",
|
||||
placeholder="Input Your Code",
|
||||
style=discord.TextStyle.long,
|
||||
default=self.code
|
||||
)
|
||||
)
|
||||
|
||||
async def on_submit(self, interaction: discord.Interaction):
|
||||
self.values = self.children[0].value
|
||||
self.stop()
|
||||
await interaction.response.defer()
|
||||
self.code = self.children[0].value
|
||||
self.stop()
|
||||
|
||||
class CogsDropdown(discord.ui.Select):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot
|
||||
|
||||
options = [discord.SelectOption(label="All", description="All the cogs")]
|
||||
|
||||
for name, cog in bot.cogs.items():
|
||||
options.append(discord.SelectOption(label=name.capitalize(), description=cog.description[:50]))
|
||||
|
||||
super().__init__(
|
||||
placeholder="Select a cog to reload...",
|
||||
min_values=1, max_values=1,
|
||||
options=options,
|
||||
)
|
||||
|
||||
async def callback(self, interaction: discord.Interaction) -> None:
|
||||
selected = self.values[0].lower()
|
||||
try:
|
||||
if selected == "all":
|
||||
for name in self.bot.cogs.copy().keys():
|
||||
await self.bot.reload_extension(f"cogs.{name.lower()}")
|
||||
else:
|
||||
await self.bot.reload_extension(f"cogs.{selected}")
|
||||
except Exception as e:
|
||||
return await interaction.response.send_message(f"Unable to reload `{selected}`! Reason: {e}", ephemeral=True)
|
||||
|
||||
await interaction.response.send_message(f"Reloaded `{selected}` sucessfully!", ephemeral=True)
|
||||
|
||||
class ExceutePanel(discord.ui.View):
|
||||
def __init__(self, bot, *, timeout = 180):
|
||||
self.bot: commands.Bot = bot
|
||||
|
||||
self.message: discord.WebhookMessage = None
|
||||
self.code: str = None
|
||||
self._error: Exception = None
|
||||
|
||||
super().__init__(timeout=timeout)
|
||||
|
||||
def toggle_button(self, name: str, status: bool):
|
||||
child: discord.ui.Button
|
||||
for child in self.children:
|
||||
if child.label == name:
|
||||
child.disabled = status
|
||||
break
|
||||
|
||||
def clear_code(self, content: str):
|
||||
"""Automatically removes code blocks from the code."""
|
||||
if content.startswith('```') and content.endswith('```'):
|
||||
return '\n'.join(content.split('\n')[1:-1])
|
||||
|
||||
return content.strip('` \n')
|
||||
|
||||
async def on_timeout(self) -> None:
|
||||
for child in self.children:
|
||||
child.disabled = True
|
||||
if self.message:
|
||||
await self.message.edit(view=self)
|
||||
|
||||
async def execute(self, interaction: discord.Interaction):
|
||||
modal = ExceuteModal(self.code, title="Enter Your Code")
|
||||
await interaction.response.send_modal(modal)
|
||||
await modal.wait()
|
||||
|
||||
if not (code := modal.code):
|
||||
return
|
||||
|
||||
self._error = None
|
||||
text = ""
|
||||
|
||||
local_variables = {
|
||||
"discord": discord,
|
||||
"bot": self.bot,
|
||||
"interaction": interaction,
|
||||
"input": None
|
||||
}
|
||||
|
||||
self.code = self.clear_code(code)
|
||||
str_obj = io.StringIO() #Retrieves a stream of data
|
||||
try:
|
||||
with contextlib.redirect_stdout(str_obj):
|
||||
exec(f"async def func():\n{textwrap.indent(self.code, ' ')}", local_variables)
|
||||
obj = await local_variables["func"]()
|
||||
result = f"{str_obj.getvalue()}\n-- {obj}\n"
|
||||
except Exception as e:
|
||||
text = f"{e.__class__.__name__}: {e}"
|
||||
self._error = e
|
||||
|
||||
if not self._error:
|
||||
text = "\n".join([f"{'%03d' % index} | {i}" for index, i in enumerate(result.split("\n"), start=1)])
|
||||
|
||||
self.toggle_button("Error", True if self._error is None else False)
|
||||
|
||||
if not self.message:
|
||||
self.message = await interaction.followup.send(f"```{text}```", view=self, ephemeral=True)
|
||||
else:
|
||||
await self.message.edit(content=f"```{text}```", view=self)
|
||||
|
||||
@discord.ui.button(label="End", emoji="🗑️", custom_id="end")
|
||||
async def end(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
if self.message:
|
||||
await self.message.delete()
|
||||
self.stop()
|
||||
|
||||
@discord.ui.button(label="Rerun", emoji="🔄", custom_id="rerun")
|
||||
async def rerun(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
await self.execute(interaction)
|
||||
|
||||
@discord.ui.button(label="Error", emoji="👾", custom_id="Error")
|
||||
async def error(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
result = ''.join(traceback.format_exception(self._error, self._error, self._error.__traceback__))
|
||||
await self.message.edit(content=f"```py\n{result}```")
|
||||
|
||||
class CogsView(discord.ui.View):
|
||||
def __init__(self, bot, *, timeout: float | None = 180):
|
||||
super().__init__(timeout=timeout)
|
||||
|
||||
self.add_item(CogsDropdown(bot))
|
||||
|
||||
class DebugView(discord.ui.View):
|
||||
def __init__(self, bot, *, timeout: float | None = 180):
|
||||
self.bot: commands.Bot = bot
|
||||
self.panel: ExceutePanel = ExceutePanel(bot)
|
||||
|
||||
super().__init__(timeout=timeout)
|
||||
|
||||
@discord.ui.button(label='Command', emoji="▶️", style=discord.ButtonStyle.green)
|
||||
async def run_command(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
await self.panel.execute(interaction)
|
||||
|
||||
@discord.ui.button(label='Cogs', emoji="🔃")
|
||||
async def reload_cog(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
return await interaction.response.send_message("Reload Cogs", view=CogsView(self.bot), ephemeral=True)
|
||||
|
||||
@discord.ui.button(label='Send Logs', emoji="📥", style=discord.ButtonStyle.red)
|
||||
async def send_error_logs(self, interaction: discord.Interaction, button: discord.ui.Button):
|
||||
if not function.error_log:
|
||||
return await interaction.response.send_message("Sorry there are not error logs!", ephemeral=True)
|
||||
|
||||
await interaction.response.send_message(file=function.gen_report(), ephemeral=True)
|
||||
Reference in New Issue
Block a user