Compare commits

..

No commits in common. "4ad73ec6ee8f4cadb1836305b9d70f9a86eac546" and "0b60a2df66bfd34790bb31b63e99ac44ab87daf7" have entirely different histories.

2 changed files with 84 additions and 87 deletions

View file

@ -52,7 +52,7 @@ class HotReload(commands.Cog):
async def get_paths(self) -> tuple[Path]:
"""Retrieve user defined paths."""
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_manager = self.bot._cog_mgr
cog_paths = await cog_manager.user_defined_paths()
return (Path(path) for path in cog_paths)

View file

@ -67,11 +67,10 @@ class Pterodactyl(commands.Cog):
self.update_topic.cancel()
self.task.cancel()
self.retry_counter = 0
await self.client._session.close() # pylint: disable=protected-access # noqa: SLF001
await self.client._session.close() # pylint: disable=protected-access
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
@ -142,27 +141,23 @@ 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"],
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:
@ -171,21 +166,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]:
@ -196,7 +191,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)
@ -270,7 +265,12 @@ class Pterodactyl(commands.Cog):
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")])
@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.
@ -473,7 +473,7 @@ 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."""
@ -516,10 +516,7 @@ 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."""
@ -558,7 +555,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()
@ -599,19 +596,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")
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: