circa 2017, doubt it still works

.gitignore

@@ -0,0 +1,22 @@
+#Pycharm.
+.idea/
+*.pyc
+~*/
+
+## Windows shortcuts
+*.lnk
+
+# macOS nonsense
+*.DS_Store
+.AppleDouble
+.LSOverride
+
+# Windows nonsense
+## Windows image file caches
+Thumbs.db
+ehthumbs.db
+
+# Ew
+desktop.ini
+
+

bot.py

@@ -0,0 +1,214 @@
+import os
+import logging
+
+from discord.ext import commands
+from utils.config import Config
+from utils import amp
+from utils import checks
+
+config = Config()
+
+desc = "Server control bot for {}".format(config.name)
+
+bot = commands.Bot(command_prefix=config.command_prefix, description=desc, pm_help=False)
+
+def init_console_logger():
+    logger = logging.getLogger("consolelogger")
+    format = logging.Formatter("%(asctime)s %(message)s")
+    fileHandler = logging.FileHandler("commands.log")
+    fileHandler.setFormatter(format)
+    streamHandler = logging.StreamHandler()
+    streamHandler.setFormatter(format)
+    logger.setLevel(logging.INFO)
+    logger.addHandler(fileHandler)
+    logger.addHandler(streamHandler)
+init_console_logger()
+
+console_logger = logging.getLogger("consolelogger")
+
+class Server():
+    def __init__(self, bot):
+        self.bot = bot
+
+    @commands.command()
+    async def state(self):
+        """Gets the server's current state"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        await self.bot.say("The server is **{}**".format(state))
+
+    @checks.is_senior_admin()
+    @commands.command()
+    async def start(self):
+        """Starts the server"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "online":
+            await self.bot.say("The server is already running")
+            return
+        elif state == "starting":
+            await self.bot.say("The server is already starting")
+            return
+        else:
+            amp.control_power(amp.Power.START)
+            await self.bot.say("Starting the server...")
+
+    @checks.is_senior_admin()
+    @commands.command()
+    async def restart(self):
+        """Restarts the server"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "starting":
+            await self.bot.say("The server is already starting")
+            return
+        elif state == "shutting down":
+            await self.bot.say("The server is already shutting down")
+            return
+        else:
+            amp.control_power(amp.Power.RESTART)
+            await self.bot.say("Restarting the server...")
+
+    @checks.is_senior_admin()
+    @commands.command()
+    async def stop(self):
+        """Stops the server"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "offline":
+            await self.bot.say("The server is already stopped")
+            return
+        elif state == "shutting down":
+            await self.bot.say("The server is already shutting down")
+            return
+        else:
+            amp.control_power(amp.Power.STOP)
+            await self.bot.say("Stopping the server...")
+
+    @checks.is_senior_admin()
+    @commands.command()
+    async def kill(self):
+        """Kills the server (useful if it isn't responding)"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "offline":
+            await self.bot.say("The server is already stopped")
+            return
+        else:
+            amp.control_power(amp.Power.KILL)
+            await self.bot.say("Killing the server...")
+
+    @commands.command()
+    async def list(self):
+        """Gets the list of online players"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "offline":
+            await self.bot.say("The server is offline")
+            return
+        await self.bot.say(amp.get_player_list())
+
+    @checks.is_senior_admin()
+    @commands.command(pass_context=True)
+    async def sendcommand(self, ctx, *, command:str):
+        """Send a console command"""
+        try:
+            state = amp.get_server_state()
+        except KeyError:
+            amp.get_session_id()
+            state = amp.get_server_state()
+        if state == "offline":
+            await self.bot.say("The server is offline")
+            return
+        amp.send_console_command(command)
+        console_logger.info("[Console Command] {}: {}".format(ctx.message.author, ctx.message.content.replace("{}{} ".format(config.command_prefix, ctx.command), "")))
+        await self.bot.say("Command sent!")
+
+bot.add_cog(Server(bot))
+
+@bot.event
+async def on_command_error(error, ctx):
+    if isinstance(error, commands.CommandNotFound):
+        return
+    if isinstance(error, commands.DisabledCommand):
+        await bot.send_message(ctx.message.channel, "This command has been disabled")
+        return
+    if isinstance(error, checks.dev_only):
+        await bot.send_message(ctx.message.channel, "This command can only be ran by the server developers")
+        return
+    if isinstance(error, checks.admin_only):
+        await bot.send_message(ctx.message.channel, "This command can only be ran by the discord admins")
+        return
+    if isinstance(error, checks.senior_admin_only):
+        await bot.send_message(ctx.message.channel, "This command can only be ran by the server senior admins")
+        return
+
+    # In case the bot failed to send a message to the channel, the try except pass statement is to prevent another error
+    try:
+        await bot.send_message(ctx.message.channel, error)
+    except:
+        pass
+    print("An error occured while executing the command named {}: {}".format(ctx.command.qualified_name, error))
+
+@bot.event
+async def on_ready():
+    print("Connected! Logged in as {}/{}".format(bot.user, bot.user.id))
+    amp.get_session_id()
+
+@checks.is_dev()
+@bot.command(hidden=True, pass_context=True)
+async def debug(ctx, *, shit:str):
+    import asyncio
+    import requests
+    import random
+    py = "```py\n{}```"
+    """This is the part where I make 20,000 typos before I get it right"""
+    # "what the fuck is with your variable naming" - EJH2
+    # seth seriously what the fuck - Robin
+    try:
+        rebug = eval(shit)
+        if asyncio.iscoroutine(rebug):
+            rebug = await rebug
+        await bot.say(py.format(rebug))
+    except Exception as damnit:
+        await bot.say(py.format("{}: {}".format(type(damnit).__name__, damnit)))
+
+@checks.is_dev()
+@bot.command(hidden=True, pass_context=True)
+async def terminal(ctx, *, command:str):
+    """Runs terminal commands and shows the output via a message. Oooh spoopy!"""
+    xl = "```xl\n{}```"
+    try:
+        await bot.send_typing(ctx.message.channel)
+        await bot.say(xl.format(os.popen(command).read()))
+    except:
+        await bot.say("Error, couldn't send command")
+
+@checks.is_dev()
+@bot.command(hidden=True)
+async def shutdown():
+    """Shuts down the bot"""
+    await bot.say("Shutting down...")
+    amp.logout()
+    await bot.logout()
+
+print("Connecting...")
+bot.run(config._token)

config.ini

@@ -0,0 +1,30 @@
+[Bot]
+; Your server name
+Name = a minecraft server
+
+; The command prefix you would like to use
+Command_Prefix = srv!
+
+; The bot account token
+Token =
+
+; Developer only ID
+; In order for a user to use dev-only commands, they need to at least have this role. To get the id of a role, get a bot that can get it for you.
+Developer_Role_ID =
+
+; In order for a user to use admin-only commands, they need at least one of these roles or the developer role. To get the id of a role, get a bot that can get it for you.
+; Ex: Admin_Role_IDs = ID1 ID2 ID3
+Admin_Role_IDs =
+
+; In order for a user to use senior-admin+ only commands, they need to atleast have this role, the admin role, or the developer role. To get the id of a role, get a bot that can get it for you.
+Senior_Admin_Role_ID =
+
+[AMP]
+; MAKE SURE YOU ARE USING YOUR MINECRAFT INSTANCE AND NOT THE ADS INSTANCE
+AMP_Username =
+
+AMP_Password =
+
+; DO NOT LEAVE A TRAILING SLASH (Ex: https://example.com:1234/)
+; Ex: URL = https://example.com:1234
+URL =

run.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+python3.5 -V > /dev/null 2>&1 || {
+	echo >&2 "Python 3.5 doesn't seem to be installed! Do you have a weird installation?"
+	echo >&2 "If you have python 3.5, use it to run bot.py instead of this script."
+	exit 1; }
+
+python3.5 bot.py

runbot.bat

@@ -0,0 +1,38 @@
+@ECHO off
+
+CHCP 65001 > NUL
+CD /d "%~dp0"
+
+SETLOCAL ENABLEEXTENSIONS
+SET KEY_NAME="HKCU\Software\Microsoft\Windows\CurrentVersion\Explorer\Advanced"
+SET VALUE_NAME=HideFileExt
+
+FOR /F "usebackq tokens=1-3" %%A IN (`REG QUERY %KEY_NAME% /v %VALUE_NAME% 2^>nul`) DO (
+    SET ValueName=%%A
+    SET ValueType=%%B
+    SET ValueValue=%%C
+)
+
+IF x%ValueValue:0x0=%==x%ValueValue% (
+    ECHO Unhiding file extensions...
+    START CMD /c REG ADD HKCU\Software\Microsoft\Windows\CurrentVersion\Explorer\Advanced /v HideFileExt /t REG_DWORD /d 0 /f
+)
+ENDLOCAL
+
+
+IF EXIST %SYSTEMROOT%\py.exe (
+    CMD /k py.exe -3.6 bot.py
+    EXIT
+)
+
+python --version > NUL 2>&1
+IF %ERRORLEVEL% NEQ 0 GOTO nopython
+
+CMD /k python bot.py
+GOTO end
+
+:nopython
+ECHO ERROR: Python has either not been installed or not added to your PATH.
+
+:end
+PAUSE

runbot_mac.command

@@ -0,0 +1,8 @@
+#!/bin/bash
+
+cd "$(dirname "$BASH_SOURCE")" || {
+	echo "Python 3.5 doesn't seem to be installed!" >&2
+exit 1
+}
+
+python3.5 bot.py

utils/amp.py

@@ -0,0 +1,89 @@
+import requests
+
+from utils.config import Config
+
+config = Config()
+username = config.amp_username
+password = config.amp_password
+url = config.amp_panel_url
+
+headers = {"Accept":"text/javascript"}
+session_id = None
+
+class Paths:
+    LOGIN = url + "/API/Core/Login"
+    LOGOUT = url + "/API/Core/Logout"
+    GET_STATUS = url + "/API/Core/GetStatus"
+    class power:
+        START = url + "/API/Core/Start"
+        RESTART = url + "/API/Core/Restart"
+        STOP = url + "/API/Core/Stop"
+        KILL = url + "/API/Core/Kill"
+    PLAYER_LIST = url + "/API/Core/GetUserList"
+    SEND_CONSOLE_COMMAND = url + "/API/Core/SendConsoleMessage"
+
+class States:
+    OFFLINE = 0
+    STARTING = 10
+    ONLINE = 20
+    SHUTTING_DOWN = 30
+
+class Power:
+    START = "start"
+    RESTART = "restart"
+    STOP = "stop"
+    KILL = "kill"
+
+def get_session_id():
+    # The API only accepts json in the form of a string for some odd reason
+    data = str({"username":config.amp_username,"password":config.amp_password,"token":"","rememberMe":"false","SESSIONID":""})
+    response = requests.post(Paths.LOGIN, headers=headers, data=data).json()
+    global session_id
+    session_id = response["sessionID"]
+
+def logout():
+    data = str({"SESSIONID": session_id})
+    requests.post(Paths.LOGOUT, headers=headers, data=data)
+
+def get_server_state():
+    data = str({"SESSIONID":session_id})
+    response = requests.post(Paths.GET_STATUS, headers=headers, data=data).json()
+    state = response["State"]
+    if state == States.OFFLINE:
+        return "offline"
+    elif state == States.STARTING:
+        return "starting"
+    elif state == States.ONLINE:
+        return "online"
+    elif state == States.SHUTTING_DOWN:
+        return "shutting down"
+    else:
+        return str(state)
+
+def control_power(action):
+    path = None
+    if action == Power.START:
+        path = Paths.power.START
+    elif action == Power.RESTART:
+        path = Paths.power.RESTART
+    elif action == Power.STOP:
+        path = Paths.power.STOP
+    elif action == Power.KILL:
+        path = Paths.power.KILL
+    data = str({"SESSIONID":session_id})
+    requests.post(path, headers=headers, data=data)
+
+def get_player_list():
+    data = str({"SESSIONID":session_id})
+    response = requests.post(Paths.PLAYER_LIST, headers=headers, data=data).json()
+    players = []
+    for p in response["result"].values():
+        players.append(p)
+    if len(players) == 0:
+        return "```There are no players online```"
+    else:
+        return "```{}```".format(", ".join(players))
+
+def send_console_command(command):
+    data = str({"SESSIONID":session_id, "message":command})
+    requests.post(Paths.SEND_CONSOLE_COMMAND, headers=headers, data=data).json()

utils/checks.py

@@ -0,0 +1,37 @@
+from discord.ext import commands
+from utils.config import Config
+config = Config()
+
+class admin_only(commands.CommandError):
+    pass
+
+class senior_admin_only(commands.CommandError):
+    pass
+
+class dev_only(commands.CommandError):
+    pass
+
+def is_dev():
+    def predicate(ctx):
+        for role in ctx.message.author.roles:
+            if role.id == config.dev_role_id:
+                return True
+        else:
+            raise dev_only
+    return commands.check(predicate)
+
+def is_admin():
+    def predicate(ctx):
+        for role in ctx.message.author.roles:
+            if role.id in config.admin_role_ids or role.id == config.dev_role_id:
+                return True
+        raise admin_only
+    return commands.check(predicate)
+
+def is_senior_admin():
+    def predicate(ctx):
+        for role in ctx.message.author.roles:
+            if role.id == config.senior_admin_role_id or role.id in config.admin_role_ids or role.id == config.dev_role_id:
+                return True
+        raise senior_admin_only
+    return commands.check(predicate)

utils/config.py

@@ -0,0 +1,60 @@
+import os
+import configparser
+
+class Defaults:
+    name = "a minecraft server"
+    token = None
+    command_prefix = "srv!"
+    dev_role_id = None
+    admin_role_ids = []
+    senior_admin_role_id = None
+    amp_username = None
+    amp_password = None
+    amp_panel_url = None
+
+class Config:
+    def __init__(self):
+
+        self.config_file = "config.ini"
+
+        config = configparser.ConfigParser(interpolation=None)
+        config.read(self.config_file, encoding="utf-8")
+
+        sections = {"Bot", "AMP"}.difference(config.sections())
+        if sections:
+            print("Could not load a section in the config file, please obtain a new config file from the github repo")
+            os._exit(1)
+        self.name = config.get("Bot", "Name", fallback=Defaults.name)
+        self._token = config.get("Bot", "Token", fallback=Defaults.token)
+        self.command_prefix = config.get("Bot", "Command_Prefix", fallback=Defaults.command_prefix)
+        self.dev_role_id = config.get("Bot", "Developer_Role_ID", fallback=Defaults.dev_role_id)
+        self.admin_role_ids = config.get("Bot", "Admin_Role_IDs", fallback=Defaults.admin_role_ids)
+        self.senior_admin_role_id = config.get("Bot", "Senior_Admin_Role_ID", fallback=Defaults.senior_admin_role_id)
+        self.amp_username = config.get("AMP", "AMP_Username", fallback=Defaults.amp_username)
+        self.amp_password = config.get("AMP", "AMP_Password", fallback=Defaults.amp_password)
+        self.amp_panel_url = config.get("AMP", "URL", fallback=Defaults.amp_panel_url)
+
+        self.check()
+
+    def check(self):
+        if not self._token:
+            print("No token was specified in the config, please put your bot's token in the config.")
+            os._exit(1)
+
+        if len(self.admin_role_ids) is not 0:
+            try:
+                self.admin_role_ids = list(self.admin_role_ids.split())
+            except:
+                self.admin_role_ids = Defaults.admin_role_ids
+
+        if not self.amp_username:
+            print("No AMP username was specified in the config")
+            os._exit(1)
+
+        if not self.amp_password:
+            print("No AMP password was specified in the config")
+            os._exit(1)
+
+        if not self.amp_panel_url:
+            print("No AMP panel url was specified in the config")
+            os._exit(1)