This repository has been archived on 2024-03-07. You can view files and clone it, but cannot push or open issues or pull requests.
GalacticCogs/pterodactyl/ptero.py

416 lines
23 KiB
Python

import asyncio
import aiohttp
import discord
from discord import ui
from discord.ext import commands
from redbot.core import Config, app_commands, commands
class Pterodactyl(commands.Cog):
"""Pterodactyl allows you to manage your Pterodactyl Panel from Discord."""
def __init__(self, bot):
self.bot = bot
self.config = Config.get_conf(self, identifier=457581387213637448123567)
self.config.register_guild(
base_url=None,
api_key=None,
server_id=None,
startup_jar=None,
startup_arguments=None,
power_action_in_progress=False
)
self.session: aiohttp.ClientSession = None
async def cog_load(self):
self.session = aiohttp.ClientSession()
async def cog_unload(self):
await self.session.close()
async def get_headers(self, guild: discord.Guild):
"""Returns the headers used to access the Pterodactyl API."""
if await self.config.guild(guild).api_key() is None:
raise LookupError("API Key not set.")
headers = {
"Authorization": f"Bearer {await self.config.guild(guild).api_key()}",
"Content-Type": "application/json",
"Accept": "application/json"
}
return headers
async def get_url(self, guild, endpoint = None):
"""Returns the base url for the server's API, or the url for a specific API endpoint if one is provided."""
if await self.config.guild(guild).server_id() is None:
raise LookupError("Server ID not set.")
elif await self.config.guild(guild).base_url() is None:
raise LookupError("Base URL not set.")
base_url = await self.config.guild(guild).base_url()
server_id = await self.config.guild(guild).server_id()
url = f"https://{base_url}/api/client/servers/{server_id}"
if endpoint:
url += '/' + endpoint
return url
@app_commands.command(name="update", description="Updates the server.")
@app_commands.guild_only()
async def update(self, interaction: discord.Interaction):
"""Updates the server using the arguments provided in the server's configuration."""
if await self.config.guild(interaction.guild).power_action_in_progress() is True:
await interaction.response.send_message(ephemeral=True, content="Power action already in progress!\nTry again later.")
return
session = self.session
await interaction.response.defer(ephemeral=True, thinking=True)
interaction_message = await interaction.original_response()
if await self.config.guild(interaction.guild).startup_jar() is None:
await interaction_message.edit(f"Something went wrong.\nError: `Startup jar not set.`", ephemeral=True)
raise LookupError("Startup jar not set.")
elif await self.config.guild(interaction.guild).startup_arguments() is None:
await interaction_message.edit(f"Something went wrong.\nError: `Startup arguments not set.`", ephemeral=True)
raise LookupError("Startup arguments not set.")
else:
startup_jar = await self.config.guild(interaction.guild).startup_jar()
startup_commands = await self.config.guild(interaction.guild).startup_arguments()
try:
headers = await self.get_headers(interaction.guild)
except LookupError as e:
await interaction_message.edit(f"Something went wrong.\nError: `{e}`", ephemeral=True)
return
async with session.get(await self.get_url(interaction.guild, "resources"), headers=headers) as response:
response_dict = await response.json()
async with session.get(await self.get_url(interaction.guild, "startup"), headers=headers) as response:
list_var_response_dict = await response.json()
updater_startup_vars = [
{
"key": "FLAGS",
"value": startup_commands
},
{
"key": "SERVER_JARFILE",
"value": startup_jar
}
]
old_startup_vars = [
{
"key": "FLAGS",
"value": list_var_response_dict['data'][4]['attributes']['server_value']
},
{
"key": "SERVER_JARFILE",
"value": list_var_response_dict['data'][0]['attributes']['server_value']
}
]
if response_dict['attributes']['current_state'] == "offline":
await self.config.guild(interaction.guild).power_action_in_progress.set(True)
for data in updater_startup_vars:
await session.put(await self.get_url(interaction.guild, "startup/variable"), headers=headers, json=data)
await session.post(await self.get_url(interaction.guild, "power"), headers=headers, json={"signal": "start"})
await interaction_message.edit(content="Updater started...")
await asyncio.sleep(1)
while True:
async with session.get(await self.get_url(interaction.guild, "resources"), headers=headers) as response:
response_dict = await response.json()
if response_dict['attributes']['current_state'] == "offline":
await interaction_message.edit(content="Updater finished.")
break
else:
await asyncio.sleep(0.5)
continue
for data in old_startup_vars:
await session.put(await self.get_url(interaction.guild, "startup/variable"), headers=headers, json=data)
await interaction_message.edit(content="Updater finished.\nUpdate process completed!")
await self.config.guild(interaction.guild).power_action_in_progress.set(False)
elif response_dict['attributes']['current_state'] == "running" or response_dict['attributes']['current_state'] == "starting":
passed_info = {
"headers": headers,
"updater_startup_vars": updater_startup_vars,
"old_startup_vars": old_startup_vars,
"interaction": interaction,
"guild": interaction.guild,
}
await interaction_message.edit(content="The server is already running! Are you sure you'd like to stop the server for updates?", view=self.UpdateButtons(timeout=180, passed_info=passed_info, session=session))
power = app_commands.Group(name='power', description="Controls the server's power state.")
@power.command(name='start', description="Starts the server.")
@app_commands.guild_only()
async def start(self, interaction: discord.Interaction):
"""Starts the server."""
if await self.config.guild(interaction.guild).power_action_in_progress() is True:
await interaction.response.send_message(ephemeral=True, content="Power action already in progress!\nTry again later.")
return
await interaction.response.defer(ephemeral=True, thinking=True)
interaction_message = await interaction.original_response()
headers = await self.get_headers(interaction.guild)
async with self.session.get(await self.get_url(interaction.guild, "resources"), headers=headers) as response:
response_json = await response.json()
current_status = response_json['attributes']['current_state']
if current_status == "offline":
passed_info = {
"headers": headers,
"interaction": interaction,
"guild": interaction.guild,
"signal": "start",
"target_signal": "running",
"message": "Server starting...",
"completed_message": "Server started!"
}
await interaction_message.edit(content="Are you sure you'd like to start the server?", view=self.PowerButtons(timeout=180, passed_info=passed_info, session=self.session))
else:
message = await interaction_message.edit(content="The server is already running!")
await message.delete(delay=3)
@power.command(name='restart', description="Restarts the server.")
@app_commands.guild_only()
async def restart(self, interaction: discord.Interaction):
"""Restarts the server."""
if await self.config.guild(interaction.guild).power_action_in_progress() is True:
await interaction.response.send_message(ephemeral=True, content="Power action already in progress!\nTry again later.")
return
await interaction.response.defer(ephemeral=True, thinking=True)
interaction_message = await interaction.original_response()
headers = await self.get_headers(interaction.guild)
async with self.session.get(await self.get_url(interaction.guild, "resources"), headers=headers) as response:
response_json = await response.json()
current_status = response_json['attributes']['current_state']
if current_status == "running":
passed_info = {
"headers": headers,
"interaction": interaction,
"guild": interaction.guild,
"signal": "restart",
"target_signal": "running",
"message": "Server restarting...",
"completed_message": "Server restarted!"
}
await interaction_message.edit(content="Are you sure you'd like to restart the server?", view=self.PowerButtons(timeout=180, passed_info=passed_info, session=self.session))
elif current_status == "offline":
message = await interaction_message.edit(content="The server is offline!")
await message.delete(delay=3)
elif current_status == "starting":
message = await interaction_message.edit(content="The server is already starting!")
await message.delete(delay=3)
@power.command(name='stop', description="Stops the server.")
@app_commands.guild_only()
async def stop(self, interaction: discord.Interaction):
"""Stops the server."""
if await self.config.guild(interaction.guild).power_action_in_progress() is True:
await interaction.response.send_message(ephemeral=True, content="Power action already in progress!\nTry again later.")
return
await interaction.response.defer(ephemeral=True, thinking=True)
interaction_message = await interaction.original_response()
headers = await self.get_headers(interaction.guild)
async with self.session.get(await self.get_url(interaction.guild, "resources"), headers=headers) as response:
response_json = await response.json()
current_status = response_json['attributes']['current_state']
if current_status == "running" or current_status == "starting":
passed_info = {
"headers": headers,
"interaction": interaction,
"guild": interaction.guild,
"signal": "stop",
"target_signal": "offline",
"message": "Server stopping...",
"completed_message": "Server stopped!"
}
await interaction_message.edit(content="Are you sure you'd like to stop the server?", view=self.PowerButtons(timeout=180, passed_info=passed_info, session=self.session))
elif current_status == "offline":
message = await interaction_message.edit(content="The server is already offline!")
await message.delete(delay=3)
class UpdateButtons(ui.View):
def __init__(self, timeout, passed_info, session: aiohttp.ClientSession):
super().__init__()
self.passed_info = passed_info
self.session = session
self.config = Config.get_conf(None, cog_name='Pterodactyl', identifier=457581387213637448123567)
@ui.button(label="Yes", style=discord.ButtonStyle.success)
async def yes_button(self, button:ui.Button, interaction:discord.Interaction):
await self.config.guild(self.passed_info['guild']).power_action_in_progress.set(True)
session = self.session
await session.post(await Pterodactyl.get_url(self, self.passed_info['guild'], "power"), headers=self.passed_info['headers'], json={"signal": "stop"})
await self.passed_info['interaction'].edit_original_response(content="Server stopping...", view=None)
while True:
async with session.get(await Pterodactyl.get_url(self, self.passed_info['guild'], "resources"), headers=self.passed_info['headers']) as response:
response_dict = await response.json()
if response_dict['attributes']['current_state'] == "offline":
await self.passed_info['interaction'].edit_original_response(content="\nServer stopped!")
break
else:
await asyncio.sleep(0.5)
continue
for data in self.passed_info['updater_startup_vars']:
await session.put(url=await Pterodactyl.get_url(self, self.passed_info['guild'], "startup/variable"), headers=self.passed_info['headers'], json=data)
await session.post(url=await Pterodactyl.get_url(self, self.passed_info['guild'], "power"), headers=self.passed_info['headers'], json={"signal": "start"})
await self.passed_info['interaction'].edit_original_response(content="Updater started...")
await asyncio.sleep(2)
while True:
async with session.get(await Pterodactyl.get_url(self, self.passed_info['guild'], "resources"), headers=self.passed_info['headers']) as response:
response_dict = await response.json()
if response_dict['attributes']['current_state'] == "offline":
await self.passed_info['interaction'].edit_original_response(content="Updater finished!")
break
else:
await asyncio.sleep(1)
continue
for data in self.passed_info['old_startup_vars']:
await session.put(await Pterodactyl.get_url(self, self.passed_info['guild'], "startup/variable"), headers=self.passed_info['headers'], json=data)
await asyncio.sleep(2)
await session.post(url=await Pterodactyl.get_url(self, self.passed_info['guild'], "power"), headers=self.passed_info['headers'], json={"signal": "start"})
await self.passed_info['interaction'].edit_original_response(content="Server starting...")
while True:
async with session.get(await Pterodactyl.get_url(self, self.passed_info['guild'], "resources"), headers=self.passed_info['headers']) as response:
response_dict = await response.json()
if response_dict['attributes']['current_state'] == "running":
await self.passed_info['interaction'].edit_original_response(content="Server started!\nUpdate process completed!")
break
else:
await asyncio.sleep(0.5)
continue
await self.config.guild(self.passed_info['guild']).power_action_in_progress.set(False)
@ui.button(label="No", style=discord.ButtonStyle.danger)
async def no_button(self, button:ui.Button, interaction:discord.Interaction):
message = await self.passed_info['interaction'].edit_original_response(content=f"Command cancelled.", view=None)
await message.delete(delay=3)
class PowerButtons(ui.View):
def __init__(self, timeout, passed_info, session: aiohttp.ClientSession):
super().__init__()
self.passed_info = passed_info
self.session = session
self.config = Config.get_conf(None, cog_name='Pterodactyl', identifier=457581387213637448123567)
@ui.button(label="Yes", style=discord.ButtonStyle.success)
async def yes_button(self, button:ui.Button, interaction:discord.Interaction):
await self.config.guild(self.passed_info['guild']).power_action_in_progress.set(True)
headers = self.passed_info['headers']
await self.session.post(await Pterodactyl.get_url(self, self.passed_info['guild'], "power"), headers=headers, json={"signal": self.passed_info['signal']})
message = await self.passed_info['interaction'].edit_original_response(content=self.passed_info['message'], view=None)
await asyncio.sleep(10)
while True:
async with self.session.get(await Pterodactyl.get_url(self, self.passed_info['guild'], "resources"), headers=headers) as response:
response_dict = await response.json()
if response_dict['attributes']['current_state'] == self.passed_info['target_signal']:
await message.edit(content=self.passed_info['completed_message'])
break
else:
await asyncio.sleep(1)
continue
await self.config.guild(self.passed_info['guild']).power_action_in_progress.set(False)
@ui.button(label="No", style=discord.ButtonStyle.danger)
async def no_button(self, button:ui.Button, interaction:discord.Interaction):
message = await self.passed_info['interaction'].edit_original_response(content=f"Command cancelled.", view=None)
await message.delete(delay=3)
get_group = app_commands.Group(name='get', description="Retrieves information from the Pterodactyl API.")
@get_group.command(name='url', description="Retrieves the URL for the specified endpoint.")
@app_commands.guild_only()
async def retrieve_url(self, interaction: discord.Interaction, endpoint: str = None):
"""Retrieves the URL for the specified endpoint."""
try:
if endpoint:
url = await self.get_url(interaction.guild, endpoint)
else:
url = await self.get_url(interaction.guild)
except LookupError as e:
await interaction.response.send_message(f"Something went wrong.\nError: `{e}`", ephemeral=True)
return
await interaction.response.send_message(url, ephemeral=True)
configure = app_commands.Group(name="config", description="Configures the Pterodactyl cog.")
@configure.command(name="api", description="Sets the information used to access the Pterodactyl API.")
@app_commands.guild_only()
async def configure_api(self, interaction: discord.Interaction):
"""Sets the information used to access the Pterdoactyl API."""
await interaction.response.send_modal(self.APIConfigModal(self.config))
@configure.command(name="update", description="Sets the startup arguments for the update command.")
@app_commands.guild_only()
async def configure_update(self, interaction: discord.Interaction):
"""Sets the startup arguments for the update command."""
await interaction.response.send_modal(self.StartupConfigModal(self.config))
class APIConfigModal(discord.ui.Modal, title="Pterodactyl Manager Configuration"):
def __init__(self, config):
super().__init__()
self.config = config
base_url = discord.ui.TextInput(
label="Base URL",
placeholder="Input your Pterodactyl Panel's Base URL here, without HTTPS or HTTP.",
style=discord.TextStyle.paragraph,
required=False,
max_length=300
)
api_key = discord.ui.TextInput(
label="API Key",
placeholder="Input your Pterodactyl Client API Key here.",
style=discord.TextStyle.short,
required=False,
max_length=300
)
server_id = discord.ui.TextInput(
label="Server ID",
placeholder="Input your Pterodactyl server's Server ID here.",
style=discord.TextStyle.short,
required=False,
max_length=300
)
async def on_submit(self, interaction: discord.Interaction):
message = ""
if self.base_url.value != "":
await self.config.guild(interaction.guild).base_url.set(self.base_url.value)
message += f"- Base URL set to\n - `{self.base_url.value}`\n"
if self.api_key.value != "":
await self.config.guild(interaction.guild).api_key.set(self.api_key.value)
trimmed_api_key = self.api_key.value[:16]
message += f"- API Key set to\n - `{trimmed_api_key}` - Trimmed for security\n"
if self.server_id.value != "":
await self.config.guild(interaction.guild).server_id.set(self.server_id.value)
message += f"- Server ID set to\n - `{self.server_id.value}`\n"
if message == "":
trimmed_api_key = str(await self.config.guild(interaction.guild).api_key())[:16]
send = f"No changes were made.\nCurrent configuration:\n- Base URL:\n - `{await self.config.guild(interaction.guild).base_url()}`\n- API Key:\n - `{trimmed_api_key}` - Trimmed for security\n- Server ID:\n - `{await self.config.guild(interaction.guild).server_id()}`"
else:
send = f"Configuration changed:\n{message}"
await interaction.response.send_message(send, ephemeral=True)
class StartupConfigModal(discord.ui.Modal, title="Pterodactyl Manager Configuration"):
def __init__(self, config):
super().__init__()
self.config = config
startup_jar = discord.ui.TextInput(
label="Startup .jar",
placeholder="Input the name of your updater .jar here.",
style=discord.TextStyle.short,
required=False,
max_length=300
)
startup_arguments = discord.ui.TextInput(
label="Startup Arguments",
placeholder="Input your startup arguments here.\nExample:\n-g -s server https://repository.link/here",
style=discord.TextStyle.paragraph,
required=False,
max_length=1000
)
async def on_submit(self, interaction: discord.Interaction):
message = ""
if self.startup_jar.value != "":
await self.config.guild(interaction.guild).startup_jar.set(self.startup_jar.value)
message += f"- Startup jar set to\n - `{self.startup_jar.value}`\n"
if self.startup_arguments.value != "":
await self.config.guild(interaction.guild).startup_arguments.set(self.startup_arguments.value)
message += f"- Startup arguments set to:\n```{self.startup_arguments.value}```\n"
if message == "":
send = f"No changes were made.\nCurrent configuration:\n- Startup jar: `{await self.config.guild(interaction.guild).startup_jar()}`\n- Startup arguments:\n```{await self.config.guild(interaction.guild).startup_arguments()}```"
else:
send = f"Configuration changed:\n{message}"
await interaction.response.send_message(send, ephemeral=True)