cswimr/SeaCogs
cswimr/SeaCogs
1
1
Fork 1
Code Issues 10 Pull requests 4 Projects 1 Packages 1 Wiki Activity Actions

chore(pterodactyl): clean up some dirty code
All checks were successful
Actions / Build Documentation (MkDocs) (push) Successful in 30s
Details
Actions / Lint Code (Ruff & Pylint) (push) Successful in 54s
Details

Browse source
This commit is contained in:
Seaswimmer 2024-08-26 15:03:37 -04:00
parent b4a54d214d
commit 1c0f12660b
Signed by: cswimr
GPG key ID: 3813315477F26F82
1 changed files with 27 additions and 55 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

82
pterodactyl/pterodactyl.py
View file

@ -9,7 +9,7 @@ from pydactyl import PterodactylClient
from redbot.core import app_commands, commands from redbot.core import app_commands, commands
from redbot.core.app_commands import Choice from redbot.core.app_commands import Choice
from redbot.core.bot import Red from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import box, error, humanize_list from redbot.core.utils.chat_formatting import bold, box, error, humanize_list
from redbot.core.utils.views import ConfirmView from redbot.core.utils.views import ConfirmView
from pterodactyl import mcsrvstatus from pterodactyl import mcsrvstatus
@ -20,8 +20,9 @@ from pterodactyl.logger import logger
class Pterodactyl(commands.Cog): class Pterodactyl(commands.Cog):
"""Pterodactyl allows you to manage your Pterodactyl Panel from Discord.""" """Pterodactyl allows you to manage your Pterodactyl Panel from Discord."""
__author__ = ["SeaswimmerTheFsh"] __author__ = ["[cswimr](https://www.coastalcommits.com/cswimr)"]
__version__ = "2.0.0" __git__ = "https://www.coastalcommits.com/cswimr/SeaCogs"
__version__ = "2.0.1"
__documentation__ = "https://seacogs.coastalcommits.com/pterodactyl/" __documentation__ = "https://seacogs.coastalcommits.com/pterodactyl/"
def __init__(self, bot: Red): def __init__(self, bot: Red):
@ -39,9 +40,9 @@ class Pterodactyl(commands.Cog):
n = "\n" if "\n\n" not in pre_processed else "" n = "\n" if "\n\n" not in pre_processed else ""
text = [ text = [
f"{pre_processed}{n}", f"{pre_processed}{n}",
f"Cog Version: **{self.__version__}**", f"{bold('Cog Version:')} [{self.__version__}]({self.__git__})",
f"Author: {humanize_list(self.__author__)}", f"{bold('Author:')} {humanize_list(self.__author__)}",
f"Documentation: {self.__documentation__}", f"{bold('Documentation:')} {self.__documentation__}",
] ]
return "\n".join(text) return "\n".join(text)
@ -175,76 +176,47 @@ class Pterodactyl(commands.Cog):
async def power(self, ctx: Union[discord.Interaction, commands.Context], action: str, action_ing: str, warning: str = '') -> None: async def power(self, ctx: Union[discord.Interaction, commands.Context], action: str, action_ing: str, warning: str = '') -> None:
if isinstance(ctx, discord.Interaction): if isinstance(ctx, discord.Interaction):
author = ctx.user ctx = await self.bot.get_context(ctx)
else:
author = ctx.author
current_status = await config.current_status() current_status = await config.current_status()
if current_status == action_ing: if current_status == action_ing:
if isinstance(ctx, discord.Interaction): return await ctx.send(f"Server is already {action_ing}.", ephemeral=True)
return await ctx.response.send_message(f"Server is already {action_ing}.", ephemeral=True)
return await ctx.send(f"Server is already {action_ing}.")
if current_status in ["starting", "stopping"] and action != "kill": if current_status in ["starting", "stopping"] and action != "kill":
if isinstance(ctx, discord.Interaction): return await ctx.send("Another power action is already in progress.", ephemeral=True)
return await ctx.response.send_message("Another power action is already in progress.", ephemeral=True)
return await ctx.send("Another power action is already in progress.")
view = ConfirmView(author, disable_buttons=True) view = ConfirmView(ctx.author, disable_buttons=True)
if isinstance(ctx, discord.Interaction): message = await ctx.send(f"{warning}Are you sure you want to {action} the server?", view=view)
await ctx.response.send_message(f"{warning}Are you sure you want to {action} the server?", view=view)
else:
message = await ctx.send(f"{warning}Are you sure you want to {action} the server?", view=view)
await view.wait() await view.wait()
if view.result is True: if view.result is True:
if isinstance(ctx, discord.Interaction): await message.edit(content=f"Sending websocket command to {action} server...", view=None)
await ctx.edit_original_response(content=f"Sending websocket command to {action} server...", view=None)
else:
await message.edit(content=f"Sending websocket command to {action} server...", view=None)
await self.websocket.send(json.dumps({"event": "set state", "args": [action]})) await self.websocket.send(json.dumps({"event": "set state", "args": [action]}))
if isinstance(ctx, discord.Interaction): await message.edit(content=f"Server {action_ing}", view=None)
await ctx.edit_original_response(content=f"Server {action_ing}", view=None)
else:
await message.edit(content=f"Server {action_ing}", view=None)
else: else:
if isinstance(ctx, discord.Interaction): await message.edit(content="Cancelled.", view=None)
await ctx.edit_original_response(content="Cancelled.", view=None)
else:
await message.edit(content="Cancelled.", view=None)
async def send_command(self, ctx: Union[discord.Interaction, commands.Context], command: str): async def send_command(self, ctx: Union[discord.Interaction, commands.Context], command: str):
channel = self.bot.get_channel(await config.console_channel()) channel = self.bot.get_channel(await config.console_channel())
if isinstance(ctx, discord.Interaction): if isinstance(ctx, discord.Interaction):
if channel: ctx = await self.bot.get_context(ctx)
await channel.send(f"Received console command from {ctx.user.id}: {command[:1900]}", allowed_mentions=discord.AllowedMentions.none()) if channel:
try: await channel.send(f"Received console command from {ctx.author.id}: {command[:1900]}", allowed_mentions=discord.AllowedMentions.none())
await self.websocket.send(json.dumps({"event": "send command", "args": [command]})) try:
await ctx.response.send_message(f"Command sent to server. {box(command, 'json')}", ephemeral=True) await self.websocket.send(json.dumps({"event": "send command", "args": [command]}))
except websockets.exceptions.ConnectionClosed as e: await ctx.send(f"Command sent to server. {box(command, 'json')}")
logger.error("WebSocket connection closed: %s", e) except websockets.exceptions.ConnectionClosed as e:
await ctx.response.send_message(error("WebSocket connection closed.")) logger.error("WebSocket connection closed: %s", e)
self.task.cancel() await ctx.send(error("WebSocket connection closed."))
self.retry_counter = 0 self.task.cancel()
self.task = self.get_task() self.retry_counter = 0
else: self.task = self.get_task()
if channel:
await channel.send(f"Received console command from {ctx.author.id}: {command[:1900]}", allowed_mentions=discord.AllowedMentions.none())
try:
await self.websocket.send(json.dumps({"event": "send command", "args": [command]}))
await ctx.send(f"Command sent to server. {box(command, 'json')}")
except websockets.exceptions.ConnectionClosed as e:
logger.error("WebSocket connection closed: %s", e)
await ctx.send(error("WebSocket connection closed."))
self.task.cancel()
self.retry_counter = 0
self.task = self.get_task()
@commands.Cog.listener() @commands.Cog.listener()
async def on_red_api_tokens_update(self, service_name: str, api_tokens: Mapping[str,str]): # pylint: disable=unused-argument async def on_red_api_tokens_update(self, service_name: str, api_tokens: Mapping[str,str]): # pylint: disable=unused-argument