diff --git a/pterodactyl/pterodactyl.py b/pterodactyl/pterodactyl.py index fc49e3f..a0be73e 100644 --- a/pterodactyl/pterodactyl.py +++ b/pterodactyl/pterodactyl.py @@ -67,20 +67,21 @@ class Pterodactyl(commands.Cog): self.update_topic.cancel() self.task.cancel() self.retry_counter = 0 - await self.client._session.close() # pylint: disable=protected-access + await self.client._session.close() # pylint: disable=protected-access # noqa: SLF001 def get_task(self) -> asyncio.Task: from pterodactyl.websocket import establish_websocket_connection + task = self.bot.loop.create_task(establish_websocket_connection(self), name="Pterodactyl Websocket Connection") task.add_done_callback(self.error_callback) return task - def error_callback(self, fut) -> None: #NOTE - Thanks flame442 and zephyrkul for helping me figure this out + def error_callback(self, fut) -> None: # NOTE - Thanks flame442 and zephyrkul for helping me figure this out try: fut.result() except asyncio.CancelledError: logger.info("WebSocket task has been cancelled.") - except Exception as e: # pylint: disable=broad-exception-caught + except Exception as e: # pylint: disable=broad-exception-caught logger.error("WebSocket task has failed: %s", e, exc_info=e) self.task.cancel() if self.retry_counter < 5: @@ -141,23 +142,27 @@ class Pterodactyl(commands.Cog): if await config.api_endpoint() == "minecraft": status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port()) if status: - placeholders.update({ - "I": response['ip'], - "M": str(response['players']['max']), - "P": str(response['players']['online']), - "V": response['version'], - "D": response['motd']['clean'][0] if response['motd']['clean'] else "unset", - }) + placeholders.update( + { + "I": response["ip"], + "M": str(response["players"]["max"]), + "P": str(response["players"]["online"]), + "V": response["version"], + "D": response["motd"]["clean"][0] if response["motd"]["clean"] else "unset", + } + ) else: - placeholders.update({ - "I": response['ip'], - "M": "0", - "P": "0", - "V": "Server Offline", - "D": "Server Offline", - }) + placeholders.update( + { + "I": response["ip"], + "M": "0", + "P": "0", + "V": "Server Offline", + "D": "Server Offline", + } + ) for key, value in placeholders.items(): - topic = topic.replace('.$' + key, value) + topic = topic.replace(".$" + key, value) return topic async def get_chat_command(self, message: discord.Message) -> str: @@ -166,21 +171,21 @@ class Pterodactyl(commands.Cog): "C": str(message.author.color), "D": message.author.discriminator, "I": str(message.author.id), - "M": message.content.replace('"','').replace("\n", " "), + "M": message.content.replace('"', "").replace("\n", " "), "N": message.author.display_name, "U": message.author.name, "V": await config.invite() or "use [p]pterodactyl config invite to change me", } for key, value in placeholders.items(): - command = command.replace('.$' + key, value) + command = command.replace(".$" + key, value) return command async def get_player_list(self) -> Optional[Tuple[str, list]]: if await config.api_endpoint() == "minecraft": status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port()) - if status and 'list' in response['players']: - output_str = '\n'.join(response['players']['list']) - return output_str, response['players']['list'] + if status and "list" in response["players"]: + output_str = "\n".join(response["players"]["list"]) + return output_str, response["players"]["list"] return None async def get_player_list_embed(self, ctx: Union[commands.Context, discord.Interaction]) -> Optional[discord.Embed]: @@ -191,7 +196,7 @@ class Pterodactyl(commands.Cog): return embed return None - 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): ctx = await self.bot.get_context(ctx) @@ -236,7 +241,7 @@ class Pterodactyl(commands.Cog): self.task = self.get_task() @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 if service_name == "pterodactyl": logger.info("Configuration value set: api_key\nRestarting task...") self.task.cancel() @@ -245,7 +250,7 @@ class Pterodactyl(commands.Cog): slash_pterodactyl = app_commands.Group(name="pterodactyl", description="Pterodactyl allows you to manage your Pterodactyl Panel from Discord.") - @slash_pterodactyl.command(name = "command", description = "Send a command to the server console.") + @slash_pterodactyl.command(name="command", description="Send a command to the server console.") async def slash_pterodactyl_command(self, interaction: discord.Interaction, command: str) -> None: """Send a command to the server console. @@ -255,7 +260,7 @@ class Pterodactyl(commands.Cog): The command to send to the server.""" return await self.send_command(interaction, command) - @slash_pterodactyl.command(name = "players", description = "Retrieve a list of players on the server.") + @slash_pterodactyl.command(name="players", description="Retrieve a list of players on the server.") async def slash_pterodactyl_players(self, interaction: discord.Interaction) -> None: """Retrieve a list of players on the server.""" e = await self.get_player_list_embed(interaction) @@ -264,13 +269,8 @@ class Pterodactyl(commands.Cog): else: await interaction.response.send_message("No players online.", ephemeral=True) - @slash_pterodactyl.command(name = "power", description = "Send power actions to the server.") - @app_commands.choices(action=[ - Choice(name="Start", value="start"), - Choice(name="Stop", value="stop"), - Choice(name="Restart", value="restart"), - Choice(name="⚠️ Kill ⚠️", value="kill") - ]) + @slash_pterodactyl.command(name="power", description="Send power actions to the server.") + @app_commands.choices(action=[Choice(name="Start", value="start"), Choice(name="Stop", value="stop"), Choice(name="Restart", value="restart"), Choice(name="⚠️ Kill ⚠️", value="kill")]) async def slash_pterodactyl_power(self, interaction: discord.Interaction, action: app_commands.Choice[str]) -> None: """Send power actions to the server. @@ -284,11 +284,11 @@ class Pterodactyl(commands.Cog): return await self.power(interaction, action.value, "stopping...") return await self.power(interaction, action.value, f"{action.value}ing...") - @commands.group(autohelp = True, name = "pterodactyl", aliases = ["ptero"]) + @commands.group(autohelp=True, name="pterodactyl", aliases=["ptero"]) async def pterodactyl(self, ctx: commands.Context) -> None: """Pterodactyl allows you to manage your Pterodactyl Panel from Discord.""" - @pterodactyl.command(name = "players", aliases=["list", "online", "playerlist", "who"]) + @pterodactyl.command(name="players", aliases=["list", "online", "playerlist", "who"]) async def pterodactyl_players(self, ctx: commands.Context) -> None: """Retrieve a list of players on the server.""" e = await self.get_player_list_embed(ctx) @@ -297,43 +297,43 @@ class Pterodactyl(commands.Cog): else: await ctx.send("No players online.") - @pterodactyl.command(name = "command", aliases = ["cmd", "execute", "exec"]) + @pterodactyl.command(name="command", aliases=["cmd", "execute", "exec"]) @commands.admin() async def pterodactyl_command(self, ctx: commands.Context, *, command: str) -> None: """Send a command to the server console.""" return await self.send_command(ctx, command) - @pterodactyl.group(autohelp = True, name = "power") + @pterodactyl.group(autohelp=True, name="power") @commands.admin() async def pterodactyl_power(self, ctx: commands.Context) -> None: """Send power actions to the server.""" - @pterodactyl_power.command(name = "start") + @pterodactyl_power.command(name="start") async def pterodactyl_power_start(self, ctx: commands.Context) -> Optional[discord.Message]: """Start the server.""" return await self.power(ctx, "start", "starting...") - @pterodactyl_power.command(name = "stop") + @pterodactyl_power.command(name="stop") async def pterodactyl_power_stop(self, ctx: commands.Context) -> Optional[discord.Message]: """Stop the server.""" return await self.power(ctx, "stop", "stopping...") - @pterodactyl_power.command(name = "restart") + @pterodactyl_power.command(name="restart") async def pterodactyl_power_restart(self, ctx: commands.Context) -> Optional[discord.Message]: """Restart the server.""" return await self.power(ctx, "restart", "restarting...") - @pterodactyl_power.command(name = "kill") + @pterodactyl_power.command(name="kill") async def pterodactyl_power_kill(self, ctx: commands.Context) -> Optional[discord.Message]: """Kill the server.""" return await self.power(ctx, "kill", "stopping... (forcefully killed)", warning="**⚠️ Forcefully killing the server process can corrupt data in some cases. ⚠️**\n") - @pterodactyl.group(autohelp = True, name = "config", aliases = ["settings", "set"]) + @pterodactyl.group(autohelp=True, name="config", aliases=["settings", "set"]) @commands.is_owner() async def pterodactyl_config(self, ctx: commands.Context) -> None: """Configure Pterodactyl settings.""" - @pterodactyl_config.command(name = "url") + @pterodactyl_config.command(name="url") async def pterodactyl_config_base_url(self, ctx: commands.Context, *, base_url: str) -> None: """Set the base URL of your Pterodactyl Panel. @@ -346,7 +346,7 @@ class Pterodactyl(commands.Cog): self.retry_counter = 0 self.task = self.get_task() - @pterodactyl_config.command(name = "serverid") + @pterodactyl_config.command(name="serverid") async def pterodactyl_config_server_id(self, ctx: commands.Context, *, server_id: str) -> None: """Set the ID of your server.""" await config.server_id.set(server_id) @@ -356,45 +356,45 @@ class Pterodactyl(commands.Cog): self.retry_counter = 0 self.task = self.get_task() - @pterodactyl_config.group(name = "console") + @pterodactyl_config.group(name="console") async def pterodactyl_config_console(self, ctx: commands.Context): """Configure console settings.""" - @pterodactyl_config_console.command(name = "channel") + @pterodactyl_config_console.command(name="channel") async def pterodactyl_config_console_channel(self, ctx: commands.Context, channel: discord.TextChannel) -> None: """Set the channel to send console output to.""" await config.console_channel.set(channel.id) await ctx.send(f"Console channel set to {channel.mention}") - @pterodactyl_config_console.command(name = "commands") + @pterodactyl_config_console.command(name="commands") async def pterodactyl_config_console_commands(self, ctx: commands.Context, enabled: bool) -> None: """Enable or disable console commands.""" await config.console_commands_enabled.set(enabled) await ctx.send(f"Console commands set to {enabled}") - @pterodactyl_config.command(name = "invite") + @pterodactyl_config.command(name="invite") async def pterodactyl_config_invite(self, ctx: commands.Context, invite: str) -> None: """Set the invite link for your server.""" await config.invite.set(invite) await ctx.send(f"Invite link set to {invite}") - @pterodactyl_config.group(name = "topic") + @pterodactyl_config.group(name="topic") async def pterodactyl_config_topic(self, ctx: commands.Context): """Set the topic for the console and chat channels.""" - @pterodactyl_config_topic.command(name = "host", aliases = ["hostname", "ip"]) + @pterodactyl_config_topic.command(name="host", aliases=["hostname", "ip"]) async def pterodactyl_config_topic_host(self, ctx: commands.Context, host: str) -> None: """Set the hostname or IP address of your server.""" await config.topic_hostname.set(host) await ctx.send(f"Hostname/IP set to `{host}`") - @pterodactyl_config_topic.command(name = "port") + @pterodactyl_config_topic.command(name="port") async def pterodactyl_config_topic_port(self, ctx: commands.Context, port: int) -> None: """Set the port of your server.""" await config.topic_port.set(port) await ctx.send(f"Port set to `{port}`") - @pterodactyl_config_topic.command(name = "text") + @pterodactyl_config_topic.command(name="text") async def pterodactyl_config_topic_text(self, ctx: commands.Context, *, text: str) -> None: """Set the text for the console and chat channels. @@ -410,17 +410,17 @@ class Pterodactyl(commands.Cog): await config.topic.set(text) await ctx.send(f"Topic set to:\n{box(text, 'yaml')}") - @pterodactyl_config.group(name = "chat") + @pterodactyl_config.group(name="chat") async def pterodactyl_config_chat(self, ctx: commands.Context): """Configure chat settings.""" - @pterodactyl_config_chat.command(name = "channel") + @pterodactyl_config_chat.command(name="channel") async def pterodactyl_config_chat_channel(self, ctx: commands.Context, channel: discord.TextChannel) -> None: """Set the channel to send chat output to.""" await config.chat_channel.set(channel.id) await ctx.send(f"Chat channel set to {channel.mention}") - @pterodactyl_config_chat.command(name = "command") + @pterodactyl_config_chat.command(name="command") async def pterodactyl_config_chat_command(self, ctx: commands.Context, *, command: str) -> None: """Set the command that will be used to send messages from Discord. @@ -429,11 +429,11 @@ class Pterodactyl(commands.Cog): await config.chat_command.set(command) await ctx.send(f"Chat command set to:\n{box(command, 'json')}") - @pterodactyl_config.group(name = "regex") + @pterodactyl_config.group(name="regex") async def pterodactyl_config_regex(self, ctx: commands.Context) -> None: """Set regex patterns.""" - @pterodactyl_config_regex.command(name = "chat") + @pterodactyl_config_regex.command(name="chat") async def pterodactyl_config_regex_chat(self, ctx: commands.Context, *, regex: str) -> None: """Set the regex pattern to match chat messages on the server. @@ -441,7 +441,7 @@ class Pterodactyl(commands.Cog): await config.chat_regex.set(regex) await ctx.send(f"Chat regex set to:\n{box(regex, 'regex')}") - @pterodactyl_config_regex.command(name = "server") + @pterodactyl_config_regex.command(name="server") async def pterodactyl_config_regex_server(self, ctx: commands.Context, *, regex: str) -> None: """Set the regex pattern to match server messages on the server. @@ -449,7 +449,7 @@ class Pterodactyl(commands.Cog): await config.server_regex.set(regex) await ctx.send(f"Server regex set to:\n{box(regex, 'regex')}") - @pterodactyl_config_regex.command(name = "join") + @pterodactyl_config_regex.command(name="join") async def pterodactyl_config_regex_join(self, ctx: commands.Context, *, regex: str) -> None: """Set the regex pattern to match join messages on the server. @@ -457,7 +457,7 @@ class Pterodactyl(commands.Cog): await config.join_regex.set(regex) await ctx.send(f"Join regex set to:\n{box(regex, 'regex')}") - @pterodactyl_config_regex.command(name = "leave") + @pterodactyl_config_regex.command(name="leave") async def pterodactyl_config_regex_leave(self, ctx: commands.Context, *, regex: str) -> None: """Set the regex pattern to match leave messages on the server. @@ -465,7 +465,7 @@ class Pterodactyl(commands.Cog): await config.leave_regex.set(regex) await ctx.send(f"Leave regex set to:\n{box(regex, 'regex')}") - @pterodactyl_config_regex.command(name = "achievement") + @pterodactyl_config_regex.command(name="achievement") async def pterodactyl_config_regex_achievement(self, ctx: commands.Context, *, regex: str) -> None: """Set the regex pattern to match achievement messages on the server. @@ -473,41 +473,41 @@ class Pterodactyl(commands.Cog): await config.achievement_regex.set(regex) await ctx.send(f"Achievement regex set to:\n{box(regex, 'regex')}") - @pterodactyl_config.group(name = "messages", aliases = ['msg', 'msgs', 'message']) + @pterodactyl_config.group(name="messages", aliases=["msg", "msgs", "message"]) async def pterodactyl_config_messages(self, ctx: commands.Context): """Configure message settings.""" - @pterodactyl_config_messages.command(name = "startup") + @pterodactyl_config_messages.command(name="startup") async def pterodactyl_config_messages_startup(self, ctx: commands.Context, *, message: str) -> None: """Set the message that will be sent when the server starts.""" await config.startup_msg.set(message) await ctx.send(f"Startup message set to: {message}") - @pterodactyl_config_messages.command(name = "shutdown") + @pterodactyl_config_messages.command(name="shutdown") async def pterodactyl_config_messages_shutdown(self, ctx: commands.Context, *, message: str) -> None: """Set the message that will be sent when the server stops.""" await config.shutdown_msg.set(message) await ctx.send(f"Shutdown message set to: {message}") - @pterodactyl_config_messages.command(name = "join") + @pterodactyl_config_messages.command(name="join") async def pterodactyl_config_messages_join(self, ctx: commands.Context, *, message: str) -> None: """Set the message that will be sent when a user joins the server. This is only shown in embeds.""" await config.join_msg.set(message) await ctx.send(f"Join message set to: {message}") - @pterodactyl_config_messages.command(name = "leave") + @pterodactyl_config_messages.command(name="leave") async def pterodactyl_config_messages_leave(self, ctx: commands.Context, *, message: str) -> None: """Set the message that will be sent when a user leaves the server. This is only shown in embeds.""" await config.leave_msg.set(message) await ctx.send(f"Leave message set to: {message}") - @pterodactyl_config.command(name = "ip") + @pterodactyl_config.command(name="ip") async def pterodactyl_config_mask_ip(self, ctx: commands.Context, mask: bool) -> None: """Mask the IP addresses of users in console messages.""" await config.mask_ip.set(mask) await ctx.send(f"IP masking set to {mask}") - @pterodactyl_config.command(name = "api") + @pterodactyl_config.command(name="api") async def pterodactyl_config_api(self, ctx: commands.Context, endpoint: str) -> None: """Set the API endpoint for retrieving user avatars. @@ -516,11 +516,14 @@ class Pterodactyl(commands.Cog): await config.api_endpoint.set(endpoint) await ctx.send(f"API endpoint set to {endpoint}") - @pterodactyl_config_regex.group(name = "blacklist", aliases = ['block', 'blocklist'],) + @pterodactyl_config_regex.group( + name="blacklist", + aliases=["block", "blocklist"], + ) async def pterodactyl_config_regex_blacklist(self, ctx: commands.Context): """Blacklist regex patterns.""" - @pterodactyl_config_regex_blacklist.command(name = "add") + @pterodactyl_config_regex_blacklist.command(name="add") async def pterodactyl_config_regex_blacklist_add(self, ctx: commands.Context, name: str, *, regex: str) -> None: """Add a regex pattern to the blacklist.""" async with config.regex_blacklist() as blacklist: @@ -538,7 +541,7 @@ class Pterodactyl(commands.Cog): else: await msg.edit(content="Cancelled.") - @pterodactyl_config_regex_blacklist.command(name = "remove") + @pterodactyl_config_regex_blacklist.command(name="remove") async def pterodactyl_config_regex_blacklist_remove(self, ctx: commands.Context, name: str) -> None: """Remove a regex pattern from the blacklist.""" async with config.regex_blacklist() as blacklist: @@ -555,7 +558,7 @@ class Pterodactyl(commands.Cog): else: await ctx.send(f"Name `{name}` does not exist in the blacklist.") - @pterodactyl_config.command(name = 'view', aliases = ['show']) + @pterodactyl_config.command(name="view", aliases=["show"]) async def pterodactyl_config_view(self, ctx: commands.Context) -> None: """View the current configuration.""" base_url = await config.base_url() @@ -580,7 +583,7 @@ class Pterodactyl(commands.Cog): topic_text = await config.topic() topic_hostname = await config.topic_hostname() topic_port = await config.topic_port() - embed = discord.Embed(color = await ctx.embed_color(), title="Pterodactyl Configuration") + embed = discord.Embed(color=await ctx.embed_color(), title="Pterodactyl Configuration") embed.description = f"""**Base URL:** {base_url} **Server ID:** `{server_id}` **Console Channel:** <#{console_channel}> @@ -596,19 +599,19 @@ class Pterodactyl(commands.Cog): **Topic Hostname:** `{topic_hostname}` **Topic Port:** `{topic_port}` - **Topic Text:** {box(topic_text, 'yaml')} + **Topic Text:** {box(topic_text, "yaml")} - **Chat Command:** {box(chat_command, 'json')} - **Chat Regex:** {box(chat_regex, 're')} - **Server Regex:** {box(server_regex, 're')} - **Join Regex:** {box(join_regex, 're')} - **Leave Regex:** {box(leave_regex, 're')} - **Achievement Regex:** {box(achievement_regex, 're')}""" + **Chat Command:** {box(chat_command, "json")} + **Chat Regex:** {box(chat_regex, "re")} + **Server Regex:** {box(server_regex, "re")} + **Join Regex:** {box(join_regex, "re")} + **Leave Regex:** {box(leave_regex, "re")} + **Achievement Regex:** {box(achievement_regex, "re")}""" await ctx.send(embed=embed) if not len(regex_blacklist) == 0: - regex_blacklist_embed = discord.Embed(color = await ctx.embed_color(), title="Regex Blacklist") + regex_blacklist_embed = discord.Embed(color=await ctx.embed_color(), title="Regex Blacklist") for name, regex in regex_blacklist.items(): - regex_blacklist_embed.add_field(name=name, value=box(regex, 're'), inline=False) + regex_blacklist_embed.add_field(name=name, value=box(regex, "re"), inline=False) await ctx.send(embed=regex_blacklist_embed) def get_bool_str(self, inp: bool) -> str: