hofei
/
Empyrion_Serverbot
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update + Stop funktioniert. Start nicht

Browse Source
master
Hofei 4 years ago
parent 7b5a1f987e
commit 87f9757903
2 changed files with 136 additions and 31 deletions
Show Stats Download Patch File Download Diff File

1
requirements.txt
Unescape Escape View File

@ -1,2 +1,3 @@
toml~=0.10.2
psutil~=5.8.0
git+https://github.com/Hofei90/telegram_api.git#master

164
serverbot.py
Unescape Escape View File

@ -2,6 +2,10 @@ 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"
@ -15,66 +19,166 @@ def load_config():
CONFIG = load_config()
USERS = CONFIG["telegram"]["allowed_ids"]
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):
pass
def start_server(self):
pass
def stop_server(self):
pass
def status_server(self):
def start(self, nachricht, users, key):
pass
def update_server(self):
pass
def start_server(self, nachricht, users, key):
print("start server")
puffer = "Starte Server\n"
nachricht = self.send_message(key, puffer)
cmd = shlex.split(CONFIG["steamcmd"]["start"], posix=False)
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}"
self.edit_message_text(puffer, key, nachricht["result"]["message_id"])
else:
puffer = output
nachricht = self.send_message(key, puffer)
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):
[print(p) for p in psutil.process_iter() if "Empyrion" in p.name()]
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:
def abbrechen(self):
pass
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, bot, telegramid):
def commands(self, nachricht, users, key):
"""Hier werden alle Verfügbaren Telegramkommdos angelegt"""
kommando = nachricht["message"]["text"]
if kommando == "/start":
self.start()
self.start(nachricht, users, key)
elif kommando == "/start_server":
self.status_server()
self.start_server(nachricht, users, key)
elif kommando == "/stop_server":
self.stop_server()
pass
self.stop_server(nachricht, users, key)
elif kommando == "/status_server":
self.status_server()
self.status_server(nachricht, users, key)
elif kommando == "/update_server":
self.update_server()
self.update_server(nachricht, users, key)
elif kommando == "/abbrechen":
self.abbrechen()
self.abbrechen(nachricht, users, key)
else:
bot.send_message(telegramid, "Command not found")
def nachrichten_handler(nachricht, bot):
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"""
telegramid = nachricht["message"]["from"]["id"]
if telegramid not in USERS.value():
bot.send_message(telegramid, "Permission denied")
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, bot, telegramid)
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__":

Write Preview
Loading…
Cancel
Save