Empyrion_Serverbot/serverbot.py

import telegram_bot_api as api
import toml
import os
import psutil
import telnetlib
import subprocess
import shlex
import threading

SKRIPTPFAD = os.path.abspath(os.path.dirname(__file__))
CONFIGDATEI = "cfg.toml"


def load_config():
    configfile = os.path.join(SKRIPTPFAD, CONFIGDATEI)
    with open(configfile) as conffile:
        config = toml.loads(conffile.read())
    return config


CONFIG = load_config()


class User:
    def __init__(self, telegramid):
        self.telegramid = telegramid
        self.menue = None
        self.umenue = None


class EmpyrionBot(api.Bot):
    def __init__(self, token):
        super().__init__(token)
        self.threads = []
        self.update_aktiv = False

    def start(self, nachricht, users, key):
        pass

    def start_server(self, nachricht, users, key):
        print("starte server")
        os.chdir("C:\steamCMD\empyriondedicatedserver")
        subprocess.run(['EmpyrionLauncher.exe', '-startDedi', '-dedicated dedicated_HP2.yaml'])

    def stop_server(self, nachricht, users, key):
        puffer = ""
        password = "1234"
        with telnetlib.Telnet('localhost', 30004, 20) as tn:
            ausgabe = tn.read_until(b"password:", 10).decode("utf-8")
            print(ausgabe)
            puffer = f"{puffer} {ausgabe}"
            nachricht = self.send_message(key, puffer)
            tn.write(password.encode('ascii') + b"\n")
            ausgabe = tn.read_until(b"password:", 10).decode("utf-8")
            print(ausgabe)
            puffer = f"{puffer} {ausgabe}"
            self.edit_message_text(puffer, key, nachricht["result"]["message_id"])
            tn.write(password.encode('ascii') + b"\n")
            tn.write("Saveandexit 0".encode('ascii') + b"\n")
            ausgabe = tn.read_until(b"minutes", 10).decode("utf-8")
            print(ausgabe)
            puffer = f"{puffer} {ausgabe}"
            self.edit_message_text(puffer, key, nachricht["result"]["message_id"])

    def status_server(self, nachricht, users, key):
        text = [str(p) for p in psutil.process_iter() if "Empyrion" in p.name()]
        if text:
            self.send_message(key, "\n".join(text))
        else:
            self.send_message(key, "Kein Prozess aktiv")

    def update_server(self, nachricht, users, key):
        # Zu Suchender String zum analysieren des Errorlogpfades:
        # b"Redirecting stderr to 'C:\\steamCMD\\logs\\stderr.txt'"
        if self.update_aktiv:
            self.send_message(key, "Update schon aktiv")
        else:

            if users[key].umenue is None:
                users[key].menue = self.update_server
                users[key].umenue = 0
                self.send_message(key, "Update starten? <j>")
            elif users[key].umenue == 0:
                if nachricht["message"]["text"] == "j":
                    cmd = shlex.split(CONFIG["steamcmd"]["update"], posix=False)
                    self.threads.append(threading.Thread(target=update_empyrion_server,
                                                         args=(self, cmd, users, key)).start())
                else:
                    self.send_message(key, "Kein Starbefehl erkannt")
                users[key].menue = None
                users[key].umenue = None

    def abbrechen(self, _, users, key):
        users[key].menue = None
        users[key].umenue = None
        self.send_message(key, "Abgebrochen")

    # ---------------------------------------------------------------------------------------------------------------------
    # Ab hier kommen die Botkommandos
    # ---------------------------------------------------------------------------------------------------------------------
    def commands(self, nachricht, users, key):
        """Hier werden alle Verfügbaren Telegramkommdos angelegt"""
        kommando = nachricht["message"]["text"]
        if kommando == "/start":
            self.start(nachricht, users, key)
        elif kommando == "/start_server":
            pass
            self.start_server(nachricht, users, key)
        elif kommando == "/stop_server":
            pass
            self.stop_server(nachricht, users, key)
        elif kommando == "/status_server":
            self.status_server(nachricht, users, key)
        elif kommando == "/update_server":
            self.update_server(nachricht, users, key)
        elif kommando == "/abbrechen":
            self.abbrechen(nachricht, users, key)
        else:
            self.send_message(key, "Command not found")


def update_empyrion_server(bot, cmd, users, key):
    puffer = "Update gestartet \n"
    nachricht = bot.send_message(key, puffer)
    bot.update_aktiv = True
    p = subprocess.Popen(cmd, stdout=subprocess.PIPE)
    while p.poll() is None:
        output = p.stdout.readline().decode("utf-8")
        if output:
            if check_nachrichten_laenge(f"{puffer} {output}"):
                puffer = f"{puffer} {output}"
                bot.edit_message_text(puffer, key, nachricht["result"]["message_id"])
            else:
                puffer = output
                nachricht = bot.send_message(key, puffer)

    rc = p.poll()
    print(rc)
    bot.update_aktiv = False
    bot.send_message(key, f"Update abgeschlossen RC: {rc}")


def check_nachrichten_laenge(nachricht, max_laenge=4096):
    if len(nachricht) >= max_laenge:
        return False
    else:
        return True


def nachrichten_handler(nachricht, bot, users):
    """Handling der vorliegenden Nachricht"""
    key = nachricht["message"]["from"]["id"]
    if key not in users.keys():
        bot.send_message(key, "Permission denied")
    if "message" in nachricht:
        # Prüfen ob es sich um ein Botkommando handelt
        if "bot_command" in nachricht["message"].get("entities", [{}])[0].get("type", ""):
            bot.commands(nachricht, users, key)
        elif users[key].menue is not None:
            users[key].menue(nachricht, users, key)


def read_steamcmd_errorlog(pfad):
    with open(pfad) as file:
        return [line.strip() for line in file.readlines()]


def main():
    bot = EmpyrionBot(CONFIG["telegram"]["token"])
    users = {telegramid: User(telegramid) for telegramid in CONFIG["telegram"]["allowed_ids"].values()}
    while True:
        messages = bot.get_updates()
        for message in messages:
            nachrichten_handler(message, bot, users)
            print(message)


if __name__ == "__main__":
    main()