Compare commits

...

2 commits

Author SHA1 Message Date
4ad73ec6ee
fix(pterodactyl): ruff fix
Some checks failed
Actions / Build Documentation (MkDocs) (push) Successful in 43s
Actions / Lint Code (Ruff & Pylint) (push) Failing after 47s
2025-01-26 14:10:48 +00:00
2543563af1
fix(hotreload): ruff fix 2025-01-26 14:10:48 +00:00
2 changed files with 87 additions and 84 deletions

View file

@ -52,7 +52,7 @@ class HotReload(commands.Cog):
async def get_paths(self) -> tuple[Path]: async def get_paths(self) -> tuple[Path]:
"""Retrieve user defined paths.""" """Retrieve user defined paths."""
cog_manager = self.bot._cog_mgr cog_manager = self.bot._cog_mgr # noqa: SLF001 # We have to use this private method because there is no public API to get user defined paths
cog_paths = await cog_manager.user_defined_paths() cog_paths = await cog_manager.user_defined_paths()
return (Path(path) for path in cog_paths) return (Path(path) for path in cog_paths)

View file

@ -67,10 +67,11 @@ class Pterodactyl(commands.Cog):
self.update_topic.cancel() self.update_topic.cancel()
self.task.cancel() self.task.cancel()
self.retry_counter = 0 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: def get_task(self) -> asyncio.Task:
from pterodactyl.websocket import establish_websocket_connection from pterodactyl.websocket import establish_websocket_connection
task = self.bot.loop.create_task(establish_websocket_connection(self), name="Pterodactyl Websocket Connection") task = self.bot.loop.create_task(establish_websocket_connection(self), name="Pterodactyl Websocket Connection")
task.add_done_callback(self.error_callback) task.add_done_callback(self.error_callback)
return task return task
@ -141,23 +142,27 @@ class Pterodactyl(commands.Cog):
if await config.api_endpoint() == "minecraft": if await config.api_endpoint() == "minecraft":
status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port()) status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port())
if status: if status:
placeholders.update({ placeholders.update(
"I": response['ip'], {
"M": str(response['players']['max']), "I": response["ip"],
"P": str(response['players']['online']), "M": str(response["players"]["max"]),
"V": response['version'], "P": str(response["players"]["online"]),
"D": response['motd']['clean'][0] if response['motd']['clean'] else "unset", "V": response["version"],
}) "D": response["motd"]["clean"][0] if response["motd"]["clean"] else "unset",
}
)
else: else:
placeholders.update({ placeholders.update(
"I": response['ip'], {
"I": response["ip"],
"M": "0", "M": "0",
"P": "0", "P": "0",
"V": "Server Offline", "V": "Server Offline",
"D": "Server Offline", "D": "Server Offline",
}) }
)
for key, value in placeholders.items(): for key, value in placeholders.items():
topic = topic.replace('.$' + key, value) topic = topic.replace(".$" + key, value)
return topic return topic
async def get_chat_command(self, message: discord.Message) -> str: async def get_chat_command(self, message: discord.Message) -> str:
@ -166,21 +171,21 @@ class Pterodactyl(commands.Cog):
"C": str(message.author.color), "C": str(message.author.color),
"D": message.author.discriminator, "D": message.author.discriminator,
"I": str(message.author.id), "I": str(message.author.id),
"M": message.content.replace('"','').replace("\n", " "), "M": message.content.replace('"', "").replace("\n", " "),
"N": message.author.display_name, "N": message.author.display_name,
"U": message.author.name, "U": message.author.name,
"V": await config.invite() or "use [p]pterodactyl config invite to change me", "V": await config.invite() or "use [p]pterodactyl config invite to change me",
} }
for key, value in placeholders.items(): for key, value in placeholders.items():
command = command.replace('.$' + key, value) command = command.replace(".$" + key, value)
return command return command
async def get_player_list(self) -> Optional[Tuple[str, list]]: async def get_player_list(self) -> Optional[Tuple[str, list]]:
if await config.api_endpoint() == "minecraft": if await config.api_endpoint() == "minecraft":
status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port()) status, response = await mcsrvstatus.get_status(await config.topic_hostname(), await config.topic_port())
if status and 'list' in response['players']: if status and "list" in response["players"]:
output_str = '\n'.join(response['players']['list']) output_str = "\n".join(response["players"]["list"])
return output_str, response['players']['list'] return output_str, response["players"]["list"]
return None return None
async def get_player_list_embed(self, ctx: Union[commands.Context, discord.Interaction]) -> Optional[discord.Embed]: 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 embed
return None 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): if isinstance(ctx, discord.Interaction):
ctx = await self.bot.get_context(ctx) ctx = await self.bot.get_context(ctx)
@ -265,12 +270,7 @@ class Pterodactyl(commands.Cog):
await interaction.response.send_message("No players online.", ephemeral=True) await interaction.response.send_message("No players online.", ephemeral=True)
@slash_pterodactyl.command(name="power", description="Send power actions to the server.") @slash_pterodactyl.command(name="power", description="Send power actions to the server.")
@app_commands.choices(action=[ @app_commands.choices(action=[Choice(name="Start", value="start"), Choice(name="Stop", value="stop"), Choice(name="Restart", value="restart"), Choice(name="⚠️ Kill ⚠️", value="kill")])
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: async def slash_pterodactyl_power(self, interaction: discord.Interaction, action: app_commands.Choice[str]) -> None:
"""Send power actions to the server. """Send power actions to the server.
@ -473,7 +473,7 @@ class Pterodactyl(commands.Cog):
await config.achievement_regex.set(regex) await config.achievement_regex.set(regex)
await ctx.send(f"Achievement regex set to:\n{box(regex, '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): async def pterodactyl_config_messages(self, ctx: commands.Context):
"""Configure message settings.""" """Configure message settings."""
@ -516,7 +516,10 @@ class Pterodactyl(commands.Cog):
await config.api_endpoint.set(endpoint) await config.api_endpoint.set(endpoint)
await ctx.send(f"API endpoint set to {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): async def pterodactyl_config_regex_blacklist(self, ctx: commands.Context):
"""Blacklist regex patterns.""" """Blacklist regex patterns."""
@ -555,7 +558,7 @@ class Pterodactyl(commands.Cog):
else: else:
await ctx.send(f"Name `{name}` does not exist in the blacklist.") 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: async def pterodactyl_config_view(self, ctx: commands.Context) -> None:
"""View the current configuration.""" """View the current configuration."""
base_url = await config.base_url() base_url = await config.base_url()
@ -596,19 +599,19 @@ class Pterodactyl(commands.Cog):
**Topic Hostname:** `{topic_hostname}` **Topic Hostname:** `{topic_hostname}`
**Topic Port:** `{topic_port}` **Topic Port:** `{topic_port}`
**Topic Text:** {box(topic_text, 'yaml')} **Topic Text:** {box(topic_text, "yaml")}
**Chat Command:** {box(chat_command, 'json')} **Chat Command:** {box(chat_command, "json")}
**Chat Regex:** {box(chat_regex, 're')} **Chat Regex:** {box(chat_regex, "re")}
**Server Regex:** {box(server_regex, 're')} **Server Regex:** {box(server_regex, "re")}
**Join Regex:** {box(join_regex, 're')} **Join Regex:** {box(join_regex, "re")}
**Leave Regex:** {box(leave_regex, 're')} **Leave Regex:** {box(leave_regex, "re")}
**Achievement Regex:** {box(achievement_regex, 're')}""" **Achievement Regex:** {box(achievement_regex, "re")}"""
await ctx.send(embed=embed) await ctx.send(embed=embed)
if not len(regex_blacklist) == 0: 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(): 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) await ctx.send(embed=regex_blacklist_embed)
def get_bool_str(self, inp: bool) -> str: def get_bool_str(self, inp: bool) -> str: