SeaCogs/pterodactyl/pterodactyl.py

250 lines
12 KiB
Python

import asyncio
import json
from typing import Mapping, Optional
import discord
import websockets
from pydactyl import PterodactylClient
from redbot.core import commands
from redbot.core.bot import Red
from redbot.core.utils.chat_formatting import box
from pterodactyl.config import config, register_config
from pterodactyl.logger import logger
class Pterodactyl(commands.Cog):
"""Pterodactyl allows you to manage your Pterodactyl Panel from Discord."""
def __init__(self, bot: Red):
self.bot = bot
self.client: Optional[PterodactylClient] = None
self.task: Optional[asyncio.Task] = None
self.websocket: Optional[websockets.WebSocketClientProtocol] = None
register_config(config)
async def cog_load(self) -> None:
self.task = self.get_task()
async def cog_unload(self) -> None:
self.task.cancel()
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
def error_callback(self, fut) -> None: #NOTE - Thanks flame442 and zephyrkul for helping me figure this out
try:
fut.result()
except asyncio.CancelledError:
pass
except Exception as e: # pylint: disable=broad-exception-caught
logger.error("WebSocket task has failed: %s", e, exc_info=e)
self.task.cancel()
self.task = self.get_task()
@commands.Cog.listener()
async def on_message_without_command(self, message: discord.Message) -> None:
if message.channel.id == await config.console_channel() and message.author.bot is False:
logger.debug("Received console command from %s: %s", message.author.id, message.content)
await message.channel.send(f"Received console command from {message.author.id}: {message.content[:1900]}")
try:
await self.websocket.send(json.dumps({"event": "send command", "args": [message.content]}))
except websockets.exceptions.ConnectionClosed as e:
logger.error("WebSocket connection closed: %s", e)
self.task.cancel()
self.task = self.get_task()
if message.channel.id == await config.chat_channel() and message.author.bot is False:
logger.debug("Received chat message from %s: %s", message.author.id, message.content)
channel = self.bot.get_channel(await config.console_channel())
if channel:
await channel.send(f"Received chat message from {message.author.id}: {message.content[:1900]}")
msg = json.dumps({"event": "send command", "args": [await self.get_chat_command(message.author.display_name, message.content, message.author.color)]})
logger.debug("Sending chat message to server:\n%s", msg)
try:
await self.websocket.send(msg)
except websockets.exceptions.ConnectionClosed as e:
logger.error("WebSocket connection closed: %s", e)
self.task.cancel()
self.task = self.get_task()
async def get_chat_command(self, username: str, message: str, color: discord.Color) -> str:
command: str = await config.chat_command()
command = command.replace(".$U", username).replace(".$M", message).replace(".$C", str(color))
return command
@commands.Cog.listener()
async def on_red_api_tokens_update(self, service_name: str, api_tokens: Mapping[str,str]):
if service_name == "pterodactyl":
logger.info("Configuration value set: api_key\nRestarting task...")
self.task.cancel()
self.task = self.get_task()
@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.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")
async def pterodactyl_config_base_url(self, ctx: commands.Context, *, base_url: str = None) -> None:
"""Set the base URL of your Pterodactyl Panel.
Please include the protocol (http/https).
Example: `https://panel.example.com`"""
if base_url is None:
base_url = await config.base_url()
return await ctx.send(f"Base URL is currently set to {base_url}")
await config.base_url.set(base_url)
await ctx.send(f"Base URL set to {base_url}")
logger.info("Configuration value set: base_url = %s\nRestarting task...", base_url)
self.task.cancel()
self.task = self.get_task()
@pterodactyl_config.command(name = "serverid")
async def pterodactyl_config_server_id(self, ctx: commands.Context, *, server_id: str) -> None:
"""Set the server ID for your Pterodactyl Panel."""
await config.server_id.set(server_id)
await ctx.send(f"Server ID set to {server_id}")
logger.info("Configuration value set: server_id = %s\nRestarting task...", server_id)
self.task.cancel()
self.task = self.get_task()
@pterodactyl_config.command(name = "consolechannel")
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.group(name = "chat")
async def pterodactyl_config_chat(self, ctx: commands.Context):
"""Configure chat settings."""
@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")
async def pterodactyl_config_chat_command(self, ctx: commands.Context, *, command: str = None) -> None:
"""Set the command that will be used to send messages from Discord.
Required placeholders: `.$U` (username), `.$M` (message), `.$C` (color)
See [documentation]() for more information."""
#TODO - fix this link
if command is None:
command = await config.chat_command()
return await ctx.send(f"Chat command is currently set to:\n{box(command, 'json')}")
await config.chat_command.set(command)
await ctx.send(f"Chat command set to:\n{box(command, 'json')}")
@pterodactyl_config.group(name = "regex")
async def pterodactyl_config_regex(self, ctx: commands.Context) -> None:
"""Set regex patterns."""
@pterodactyl_config_regex.command(name = "chat")
async def pterodactyl_config_regex_chat(self, ctx: commands.Context, *, regex: str = None) -> None:
"""Set the regex pattern to match chat messages on the server.
See [documentation]() for more information."""
#TODO - fix this link
if regex is None:
regex = await config.chat_regex()
return await ctx.send(f"Chat regex is currently set to:\n{box(regex, 'regex')}")
await config.chat_regex.set(regex)
await ctx.send(f"Chat regex set to:\n{box(regex, 'regex')}")
@pterodactyl_config_regex.command(name = "server")
async def pterodactyl_config_regex_server(self, ctx: commands.Context, *, regex: str = None) -> None:
"""Set the regex pattern to match server messages on the server.
See [documentation]() for more information."""
#TODO - fix this link
if regex is None:
regex = await config.server_regex()
return await ctx.send(f"Server regex is currently set to:\n{box(regex, 'regex')}")
await config.server_regex.set(regex)
await ctx.send(f"Server regex set to:\n{box(regex, 'regex')}")
@pterodactyl_config_regex.command(name = "join")
async def pterodactyl_config_regex_join(self, ctx: commands.Context, *, regex: str = None) -> None:
"""Set the regex pattern to match join messages on the server.
See [documentation]() for more information."""
#TODO - fix this link
if regex is None:
regex = await config.join_regex()
return await ctx.send(f"Join regex is currently set to:\n{box(regex, 'regex')}")
await config.join_regex.set(regex)
await ctx.send(f"Join regex set to:\n{box(regex, 'regex')}")
@pterodactyl_config_regex.command(name = "leave")
async def pterodactyl_config_regex_leave(self, ctx: commands.Context, *, regex: str = None) -> None:
"""Set the regex pattern to match leave messages on the server.
See [documentation]() for more information."""
#TODO - fix this link
if regex is None:
regex = await config.leave_regex()
return await ctx.send(f"Leave regex is currently set to:\n{box(regex, 'regex')}")
await config.leave_regex.set(regex)
await ctx.send(f"Leave regex set to:\n{box(regex, 'regex')}")
@pterodactyl_config_regex.command(name = "achievement")
async def pterodactyl_config_regex_achievement(self, ctx: commands.Context, *, regex: str = None) -> None:
"""Set the regex pattern to match achievement messages on the server.
See [documentation]() for more information."""
#TODO - fix this link
if regex is None:
regex = await config.achievement_regex()
return await ctx.send(f"Achievement regex is currently set to:\n{box(regex, 'regex')}")
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'])
async def pterodactyl_config_messages(self, ctx: commands.Context):
"""Configure message settings."""
@pterodactyl_config_messages.command(name = "startup")
async def pterodactyl_config_messages_startup(self, ctx: commands.Context, *, message: str = None) -> None:
"""Set the message that will be sent when the server starts."""
if message is None:
message = await config.startup_msg()
return await ctx.send(f"Startup message is currently set to: {message}")
await config.startup_msg.set(message)
await ctx.send(f"Startup message set to: {message}")
@pterodactyl_config_messages.command(name = "shutdown")
async def pterodactyl_config_messages_shutdown(self, ctx: commands.Context, *, message: str = None) -> None:
"""Set the message that will be sent when the server stops."""
if message is None:
message = await config.shutdown_msg()
return await ctx.send(f"Shutdown message is currently set to: {message}")
await config.shutdown_msg.set(message)
await ctx.send(f"Shutdown message set to: {message}")
@pterodactyl_config_messages.command(name = "join")
async def pterodactyl_config_messages_join(self, ctx: commands.Context, *, message: str = None) -> None:
"""Set the message that will be sent when a user joins the server. This is only shown in embeds."""
if message is None:
message = await config.join_msg()
return await ctx.send(f"Join message is currently set to: {message}")
await config.join_msg.set(message)
await ctx.send(f"Join message set to: {message}")
@pterodactyl_config_messages.command(name = "leave")
async def pterodactyl_config_messages_leave(self, ctx: commands.Context, *, message: str = None) -> None:
"""Set the message that will be sent when a user leaves the server. This is only shown in embeds."""
if message is None:
message = await config.leave_msg()
return await ctx.send(f"Leave message is currently set to: {message}")
await config.leave_msg.set(message)
await ctx.send(f"Leave message set to: {message}")