Compare commits
2 commits
0b60a2df66
...
4ad73ec6ee
Author | SHA1 | Date | |
---|---|---|---|
4ad73ec6ee | |||
2543563af1 |
2 changed files with 87 additions and 84 deletions
|
@ -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)
|
||||||
|
|
||||||
|
|
|
@ -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:
|
||||||
|
|
Loading…
Add table
Reference in a new issue