Der Weg zum eigenen Telegram Bot mit Python
Vorwort
Vor längerer Zeit schon wurde ich von einem User gebeten, ob ich nicht meine Gedankengänge niederschreiben möchte, wie ich bei der Erstellung von meinen Telegram Bots vorgehe. Dieser Zusage möchte ich hiermit endlich nachkommen.
Als Programmiersprache kommt Python zum Einsatz. Grundverständnis der Sprache wird vorausgesetzt.
Es gibt viele Module zum erstellen eines Telegrambot für Python, jedoch habe ich mir selbst eine kleine API erstellt. Eines der verfügbaren Module im Internet habe ich bisher noch nie verwendet.
Anhand von 2 Beispielen möchte ich meine Gedankengänge aufzeigen, wie ich bei der Erstellung der Bots vorgehe.
Die telegram_bot_api
Zu finden gibt es das Pythonmodul hier: https://github.com/Hofei90/telegram_api
Es steht nicht auf PyPi zur Verfügung, dennoch lässt es sich mit folgenden Befehl installieren und für eigene Projekte verfügbar machen:
pip3 install git+https://github.com/Hofei90/telegram_api
Die offizielle Telegram API ist hier zu finden.
Mein Modul stellt das ganze in Python zur Verfügung, dass mit requests die gewünschte Methode an den Telegram Server gesendet wird. Jedoch sind nicht alle verfügbaren Methoden integriert, sondern nur die, die ich bisher selbst benötigt habe. Es lassen sich nicht vorhandene Methode leicht integrieren.
Hier die Erklärung, wie die Funktion (Methode) in dem Modul aufgebaut ist anhand von der Funktion send_message.
https://core.telegram.org/bots/api#sendmessage
Die Methode definiert unseren Funktionsname und wird noch angepasst, dass es PEP8 konform ist. Aus sendMessage von der Website wird also send_message().
Alle aufgelisteten Parameter in der Tabelle von der Webseite werden als Methodenparameter definiert. Beachte, da es sich hier um eine Methode innerhalb einer Klasse handelt, muss der erste Parameter immer self sein.
Anschließend gefolgt von allen Parametern, mit Typeannotation habe ich die verfügbaren erlaubten Types hinterlegt. Ist ein Parameter optional so erhält er noch ein None als Ausgangsdefinition.
Innerhalb der Funktion werden mit Hilfe von isinstance() die Parameter auf ihren Typ überprüft, ist dieser passend so wird der Wert des Parameter in das Dictionary params aufgenommen. Der Key muss dem Name des Parameter entsprechen, wie auf der Webseite.
Eine sinnvolle Ergänzung des Moduls wäre es, wenn Pflichtparameter einen nicht den geforderten Typen enthalten, eine Exception zu werfen. Dies ist aber noch nicht integriert.
Wurden alle Parameter überprüft, wird die URL für den Request zusammengesetzt.
Die URL beginnt mit dem statischen Teil der Telegram API "https://api.telegram.org/bot" gefolgt von dem Bottoken (später dazu mehr). Das Dictonary params wird dem Parameter params von request.get() übergeben, hier wird es passend zum Senden der Anfrage umgewandelt. Das Resultat von der Anfrage wird von der Methode zurückgegeben. Schon hat man eine Methode in die API integriert.
def send_message(self, chat_id: str or int, text: str, parse_mode: str = None,
disable_web_page_preview: bool = None, disable_notification: bool = None,
reply_to_message_id: int = None, reply_markup=None):
"""Funktion zur Übermittlung von Nachrichten
Args:
chat_id: ID des Chats zur Übermittlung der Nachricht
text: Inhalt der Nachricht
parse_mode: optional, Stringinhalt nur "HTML" oder "Markdown", gibt an ob die Nachricht Formatierungen enthält
disable_web_page_preview: optional,
disable_notification: optional, Senden ohne Benachrichtigung
reply_to_message_id: optional, wenn die Nachricht eine Antwort ist, ID der ursprünglichen Nachricht.
reply_markup: optional, Additional interface options
Returns: Result from Telegram
"""
params = {"chat_id": str(chat_id)}
if isinstance(text, str):
params["text"] = text
if isinstance(parse_mode, str):
params["parse_mode"] = parse_mode
if isinstance(disable_web_page_preview, bool):
params["disable_web_page_preview"] = disable_web_page_preview
if isinstance(disable_notification, bool):
params["disable_notification"] = disable_notification
if isinstance(reply_to_message_id, int):
params["reply_to_message_id"] = reply_to_message_id
if reply_markup is not None:
params["reply_markup"] = reply_markup
url = "{}{}/sendMessage".format(API_URL, self.token)
r = requests.get(url, params=params)
result = check_results(r, "send_message")
return result
Alles anzeigen
Bot erstellen bei Botfather (Token erhalten)
Wie erwähnt, benötigt man einen Token, dieser Token ist für jeden Bot einmalig. Um einen Bot erstellen zu können, muss man im Telegram Chat mit den Bot Botfather Kontakt aufnehmen. Bei diesem erstellt man neue Bots und erhält den benötigten Token, hinterlegt Botkommandos, editiert bestehende Bots und kann diese auch löschen. Auch sind dort Sicherheitseinstellungen vorzunehmen.
Im Chat mit Botfather /newbot eingeben und den Fragen folgen. Der Rest sollte selbsterkärend sein.
Teile mit niemanden deinen Bottoken
Ebenso würde ich die Telegramid nicht veröffentlichen.
Die Programmbeispiele
Beispiel 1
Bei Beispiel 1 handelt es sich um einen Bot, der auf einem virtuellen Server den dedizierten Gameserver von dem Spiel Empyrion - Galactic Survival steuert.
Zum aktuellen Funktionsumfang gehört, den Server
- starten
- stoppen
- updaten
- status prüfen
zusätzlich ist noch geplant aber noch nicht integriert, dass die IP Adresse des Servers mitgeteilt wird.
Beispiel 1 - Empyrion Server Bot
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()
Alles anzeigen
Beispiel 2
Dieses Beispiel ist deutlich Forennäher. Vor einiger Zeit wurde der Tutorialbereich mit Hilfe von einigen Usern und diesem Botskript hier "aufgeräumt". In einem anderen Skript wurden alle Threads des Tutorialbereiches erfasst und die URLs in einer Datenbank abgespeichert.
Der Bot kümmerte sich um die Zuteilung der Threads der einzelnen helfenden User. Sodass die Threads nicht doppelt bearbeitet worden sind.
Der Thread wurde dann von dem User geprüft und er gab an dem Bot Rückmeldung ob der Thread ok ist, oder von einem Moderator geprüft und / oder verschoben werden muss. Der Moderator wiederrum sah mit Hilfe des Bots alle gemeldeten Threads und konnte dann den Status auf erledigt setzen.
Außerdem konnte man sich ein Bild zur Statistik über den Fortschritt der Aktion senden lassen
Beispiel 2 - Raspberry Pi Tutorial Cleaning Bot
import telegram_api.telegram_bot_api as api
import db_model_tutorials_cleaning as db
import os
import toml
# import setup_logging
import time
from peewee import SqliteDatabase
import datetime
import matplotlib.pyplot as plt
def config_laden():
configfile = os.path.join(SKRIPTPFAD, "config.toml")
with open(configfile) as file:
return toml.loads(file.read())
SKRIPTPFAD = os.path.abspath(os.path.dirname(__file__))
CONFIG = config_laden()
# LOGGER = setup_logging.create_logger("tutorials_cleaning_telegram_bot", 10)
class User:
def __init__(self, telegram_id, name):
self.telegram_id = telegram_id
self.name = name
self.menue = None
self.umenue = None
class TutorialBot:
def __init__(self, token):
self.bot = api.Bot(token)
# self.logger = logger
self.offset = 0
def get_updates(self, users):
"""
Hole neue Nachrichten vom Telegramserver ab und quittiere diese
:return:
"""
self.bot.get_updates(self.offset)
# self.logger.debug(self.bot.result)
# Wenn das Update erfolgreich ist, das Ergebnis vom Objekt in lokale Variable übertragen und auswerten
if self.bot.result["ok"]:
result = self.bot.result["result"]
# self.logger.debug("Das Update enthält {anzahl} Nachrichten".format(anzahl=len(result)))
update_id = []
for counter, nachricht in enumerate(result):
# self.logger.debug("Inhalt Nachricht {nr}: {inhalt}".format(nr=counter, inhalt=nachricht))
if "message" not in nachricht:
continue
users, key = user_handler(users, nachricht)
nachrichten_handler(nachricht, self.bot, users, key)
update_id.append(nachricht["update_id"])
# Die größte update_id um 1 erhöhen und als Offsetwert für nächste Abfrage speichern
if len(result) != 0:
self.offset = max(update_id) + 1
else:
self.offset = 0
else:
# self.logger.info("Telegram Abruf fehlgeschlagen")
pass
return users
def user_handler(users, nachricht):
key = nachricht["message"]["from"]["id"]
if not nachricht["message"]["from"]["id"] in users.keys():
user = User(nachricht["message"]["from"]["id"], "Test")
users[key] = user
return users, key
def nachrichten_handler(nachricht, bot, users, key):
"""Handling der vorliegenden Nachricht"""
if "message" in nachricht:
# Prüfen ob es sich um ein Botkommando handelt
if "bot_command" in nachricht["message"].get("entities", [{}])[0].get("type", ""):
users = bot_command(nachricht, bot, users, key)
elif users[key].menue is not None:
users[key].menue(nachricht, bot, users, key)
return users
# ---------------------------------------------------------------------------------------------------------------------
# Ab hier kommen die Botkommandos
# ---------------------------------------------------------------------------------------------------------------------
def bot_command(nachricht, bot, users, key):
"""Hier werden alle Verfügbaren Telegramkommdos angelegt"""
kommando = nachricht["message"]["text"]
if kommando == "/start":
return m_start(nachricht, bot, users, key)
user_db = get_user_db(users, key)
if user_db is None:
bot.send_message(users[key].telegram_id, "User unbekannt, bitte /start ausführen")
return users
if kommando == "/help":
m_help(nachricht, bot, users, key)
elif kommando == "/del_acc":
m_del_acc(nachricht, bot, users, key)
elif kommando == "/stop":
users = m_abbrechen(nachricht, bot, users, key)
elif kommando == "/statistik":
m_statistik_anzeigen(nachricht, bot, users, key)
elif kommando == "/thread_bearbeiten":
users = m_thread_bearbeiten(nachricht, bot, users, key)
elif kommando == "/threads_mod_anzeigen":
m_threads_mod_anzeigen(nachricht, bot, users, key)
elif kommando == "/thread_mod_bearbeiten":
users = m_thread_mod_bearbeiten(nachricht, bot, users, key)
else:
bot.send_message(users[key].telegram_id, "Kommando unbekannt")
return users
def m_start(nachricht, bot, users, key):
if users[key].umenue is None:
with open(os.path.join(SKRIPTPFAD, "starttext.txt"), encoding="UTF-8") as file:
starttext = file.read()
bot.send_message(users[key].telegram_id, starttext)
user_db = get_user_db(users, key)
if user_db is not None:
bot.send_message(users[key].telegram_id, "Du hast schon einen Account, "
"welcher mit /del_acc gelöscht werden kann")
else:
users[key].menue = m_start
users[key].umenue = 1
return users
elif users[key].umenue == 1:
if nachricht["message"]["text"] == "ja":
acc_erstellen(users, key, nachricht["message"]["from"]["username"])
bot.send_message(users[key].telegram_id, "Account erstellt")
else:
bot.send_message(users[key].telegram_id, "Erlaubnis nicht erteilt, Bot kann nicht genutzt werden")
users[key].menue = None
users[key].umenue = None
return users
def m_help(_, bot, users, key):
with open(os.path.join(SKRIPTPFAD, "Hilfetext.txt",), encoding="UTF-8") as file:
hilfetext = file.read()
with open(os.path.join(SKRIPTPFAD, "botkommandos.txt"), encoding="UTF-8") as file:
botkommandos = file.read()
ausgabe = f"{hilfetext}\n{botkommandos}"
bot.send_message(users[key].telegram_id, ausgabe)
def m_del_acc(nachricht, bot, users, key):
if users[key].umenue is None:
bot.send_message(users[key].telegram_id, "Möchtest du sicher deinen Account löschen? So antworte mit ja")
users[key].menue = m_del_acc
users[key].umenue = 1
elif users[key].umenue == 1:
if nachricht["message"]["text"] == "ja":
acc_loeschen(users, key)
bot.send_message(users[key].telegram_id, "Account erfolgreich gelöscht")
else:
bot.send_message(users[key].telegram_id, "Falsches Format, Befehl wenn gewünscht bitte wiederholen")
users[key].menue = None
users[key].umenue = None
return users
def m_abbrechen(_, bot, users, key):
users[key].menue = None
users[key].umenue = None
bot.send_message(users[key].telegram_id, "Befehl beendet")
return users
def m_statistik_anzeigen(_, bot, users, key):
kuchendiagramm_erstellen()
ausgabe = ""
threadstatuse = db.Threadstatus.select()
for threadstatus in threadstatuse:
threadanzahl = db.Thread.select().where(db.Thread.status == threadstatus).count()
ausgabe = f"{ausgabe}{threadstatus.text}: {threadanzahl} \n"
bot.send_message(users[key].telegram_id, ausgabe)
bot.send_photo(users[key].telegram_id, CONFIG["pic_pfad"], disable_notification=True)
def m_thread_bearbeiten(nachricht, bot, users, key):
thread_status_zugewiesen = db.Threadstatus.get(db.Threadstatus.status == 1)
thread_status_offen = db.Threadstatus.get(db.Threadstatus.status == 0)
if users[key].umenue is None:
users[key].menue = m_thread_bearbeiten
ausgabe = "Bitte ThreadID und Ergebnis mitteilen. 1 wenn Thread ok ist, 0 wenn Thread von einem Moderator " \
"bearbeitet werden soll\n" \
"Außerdem kann ein Bemerkungstext hinzugefügt werden welchen dem Moderator angezeigt wird\n" \
"Schema: <ID> <(0 oder 1)> <Text>\n" \
"Beispiel: 234 0 Keine Anleitung sondern Frage" \
"Es wird solange ein neuer Thread ausgegeben, bis mit /stop der Modus beendet wird"
bot.send_message(users[key].telegram_id, ausgabe)
users[key].umenue = 1
if users[key].umenue == 1:
user_db = get_user_db(users, key)
thread = db.Thread.get_or_none((db.Thread.bearbeiter == user_db)
& (db.Thread.status == thread_status_zugewiesen))
if thread is None:
thread = db.Thread.get_or_none(db.Thread.status == thread_status_offen)
thread.status = thread_status_zugewiesen
thread.bearbeiter = user_db
thread.letzte_aenderung = datetime.datetime.now()
thread.save()
bot.send_message(users[key].telegram_id, f"ID: {thread.get_id()}\n{thread.link}")
users[key].umenue = 2
elif users[key].umenue == 2:
id_, status, bemerkung = thread_eingabe_auswerten(nachricht["message"]["text"])
if id_ is None or status is None or bemerkung is None:
bot.send_message(users[key].telegram_id, "Formatfehler, bitte wiederholen oder /stop")
return users
thread = db.Thread.get_by_id(id_)
if thread.bearbeiter == get_user_db(users, key):
if status:
status = db.Threadstatus.get(db.Threadstatus.status == 3)
else:
status = db.Threadstatus.get(db.Threadstatus.status == 2)
thread.status = status
thread.letzte_aenderung = datetime.datetime.now()
thread.bemerkung_korrektur = bemerkung
thread.save()
bot.send_message(
users[key].telegram_id,
f"Datensatz aktualisiert für Threadid {thread.get_id()}\n"
f"Wenn fertig /stop verwenden, ansonsten nächstes Ergebnis posten"
)
users[key].umenue = 1
users = m_thread_bearbeiten("", bot, users, key)
else:
bot.send_message(users[key].telegram_id, "Dieser Thread ist dir nicht zugewiesen. Falsche ID?")
return users
def m_threads_mod_anzeigen(_, bot, users, key):
ausgabe = "Offene Threads die zur Bearbeitung für Moderator zugewiesen sind:\n"
for thread in db.Thread.select().where((db.Thread.status == 2)):
ausgabe = f"{ausgabe}ID: {thread.get_id()}\n{thread.link}\n" \
f"Bemerkung von {thread.bearbeiter.username}: {thread.bemerkung_korrektur}\n\n"
bot.send_message(users[key].telegram_id, ausgabe)
def m_thread_mod_bearbeiten(nachricht, bot, users, key):
if users[key].umenue is None:
user_db = get_user_db(users, key)
if user_db.berechtigung.status == 1 or user_db.berechtigung.status == 2:
users[key].menue = m_thread_mod_bearbeiten
ausgabe = "Bitte ThreadID und optional Bemerkungstext angeben \n" \
"Schema: <ID> <Text>\n" \
"Beispiel: 234 Anleitung für mich doch ok"
bot.send_message(users[key].telegram_id, ausgabe)
users[key].umenue = 1
else:
bot.send_message(users[key].telegram_id, "Kommando kann nicht ausgeführt werden - Fehlende Berechtigung")
return users
elif users[key].umenue == 1:
id_, bemerkung = thread_mod_eingabe_auswerten(nachricht["message"]["text"])
if id_ is None or bemerkung is None:
bot.send_message(users[key].telegram_id, "Formatfehler, bitte wiederholen oder /stop")
return users
thread = db.Thread.get_by_id(id_)
status = db.Threadstatus.get(db.Threadstatus.status == 3)
thread.status = status
thread.bemerkung_moderator = bemerkung
thread.letzte_aenderung = datetime.datetime.now()
thread.save()
bot.send_message(
users[key].telegram_id,
f"Datensatz aktualisiert für Thread {thread.link}\n"
f"Wenn fertig /stop verwenden, ansonsten nächstes Ergebnis posten"
)
return users
def thread_eingabe_auswerten(nachricht):
eingabe = nachricht.split()
try:
id_ = int(eingabe[0])
except (ValueError, IndexError):
id_ = None
try:
status = int(eingabe[1])
except (ValueError, IndexError):
status = None
if status == 0:
status = False
elif status == 1:
status = True
else:
status = None
if len(eingabe) > 2:
bemerkung = " ".join(eingabe[2:])
else:
bemerkung = ""
return id_, status, bemerkung
def thread_mod_eingabe_auswerten(nachricht):
eingabe = nachricht.split()
try:
id_ = int(eingabe[0])
except (ValueError, IndexError):
id_ = None
if len(eingabe) > 1:
bemerkung = " ".join(eingabe[1:])
else:
bemerkung = ""
return id_, bemerkung
def check_zugewiesene_threads(users, key):
threadanzahl = db.Thread.select().where(((db.Thread.bearbeiter == get_user_db(users, key))
& (db.Thread.status == 1))).count()
return threadanzahl
def get_user_db(users, key):
return db.User.get_or_none(db.User.telegram_id == users[key].telegram_id)
def acc_erstellen(users, key, username):
berechtigung = db.Userberechtigung.get(db.Userberechtigung == 0)
db.User.create(telegram_id=users[key].telegram_id,
username=username,
berechtigung=berechtigung)
def acc_loeschen(users, key):
user_db = get_user_db(users, key)
db.Thread.update(bearbeiter="unbekannt").where((db.Thread.bearbeiter == user_db)
& (db.Thread.status == 2)
| (db.Thread.status == 3)).execute()
db.Thread.update(bearbeiter=None, status=0).where((db.Thread.bearbeiter == user_db)
& (db.Thread.status == 1)).execute()
db.User.delete().where(db.User.telegram_id == users[key].telegram_id).execute()
def table_user_erstellen():
berechtigung = db.Userberechtigung.get(db.Userberechtigung.status == 2)
db.User.create(telegram_id=CONFIG["telegram"]["administrator_id"], username="admin", berechtigung=berechtigung)
def kuchendiagramm_erstellen():
labels = []
values = []
query = db.Threadstatus.select()
for label in query:
value = db.Thread.select().where(db.Thread.status == label).count()
labels.append(label.text)
values.append(value)
fig1, ax1 = plt.subplots()
ax1.pie(values, labels=labels, startangle=90, autopct='%1.1f%%')
ax1.axis('equal') # Equal aspect ratio ensures that pie is drawn as a circle.
plt.savefig(CONFIG["pic_pfad"])
def threadzuweisung_pruefen(now):
thread_status_zugewiesen = db.Threadstatus.get(db.Threadstatus.status == 1)
thread_status_offen = db.Threadstatus.get(db.Threadstatus.status == 0)
threads = db.Thread.select().where(db.Thread.status == thread_status_zugewiesen)
for thread in threads:
if (now - thread.letzte_aenderung) > datetime.timedelta(days=CONFIG["max_dauer_korrekturlesen"]):
thread.letzte_aenderung = None
thread.bearbeiter = None
thread.status = thread_status_offen
thread.save()
def main():
if db.User.get_or_none() is None:
table_user_erstellen()
users = {}
bot = TutorialBot(CONFIG["telegram"]["token"])
letzte_threadzuweisungspruefung = datetime.datetime(1970, 1, 1)
while True:
users = bot.get_updates(users)
now = datetime.datetime.now()
if (datetime.datetime.now() - letzte_threadzuweisungspruefung).total_seconds() > CONFIG[
"intervall_threadzuweisungspruefung"]:
threadzuweisung_pruefen(now)
letzte_threadzuweisungspruefung = now
time.sleep(0.5)
if __name__ == "__main__":
db.DB.initialize(SqliteDatabase(CONFIG["database"]))
main()
Alles anzeigen
Die beiden Beispiele unterscheiden sich zum einen natürlich in ihrer Funktion, den verwendeten Funktionen der API und Beispiel 1 erbt von der Klasse aus der API, welches in Beispiel 2 nicht der Fall ist.
Erste Schritte
Bevor näher auf die Beispiele eingegangen wird, senden wir uns eine Nachricht mit den neu erstellten Bot.
Wer das Beispiel ausführen will, muss die Telegram API für Python und ebenso toml installiert haben.
Grundsätzlich empfiehlt es sich, Daten von Code zu trennen. Gerade sensible Daten wie der Token hat in dem Code auf keinem Fall etwas verloren. Zu schnell ist der Code geteilt bei einem Problem und man hat es versäumt den Token unkenntlich zu machen.
Aus diesem Grund wird auch das Beispiel aus 2 Teilen bestehen. Die Daten kommen in ein seperates Toml File, der Code in das Skript (auch zu finden im Examples Ordner auf Github).
Die Daten in den <> sind anzupassen.
Die eigene Telegramid lässt sich einfach mit dem Raw Data Bot herausfinden.
Der Bot kann immer nur mit einem User kommunizieren, wenn der User zunächst selbst Kontakt mit dem Bot aufnahm indem er auf starten klickte oder ihm eine eine gesendet hat. Zuvor kann der Bot keine Nachricht an den User senden.
import os
import toml
import telegram_bot_api as api
def config_laden():
configfile = os.path.join(SKRIPTPFAD, "example_cfg.toml")
with open(configfile) as file:
return toml.loads(file.read())
SKRIPTPFAD = os.path.abspath(os.path.dirname(__file__))
CONFIG = config_laden()
def main():
user = CONFIG["telegram"]["user"]
bot = api.Bot(CONFIG["telegram"]["token"])
bot.send_message(user, CONFIG["text"])
if __name__ == "__main__":
main()
Alles anzeigen
Bei der Einführung zur Gestaltung der API hieß es zuletzt, dass die Funktion das Resultat zurückgibt, würde man nun also die send_message Zeile wie folgt abändern, kommt man an die Rückmeldung des Telegram Servers
Die Ausgabe davon:
{'ok': True, 'result': {'message_id': ***, 'from': {'id': ***, 'is_bot': True, 'first_name': '***', 'username': '***'}, 'chat': {'id': ***, 'first_name': 'Hofei', 'username': 'Hofei', 'type': 'private'}, 'date': 1613917215, 'text': 'Hello World'}}
Das Format von erhaltenen Nachrichten sieht ebenso aus.
Statt nun eine Nachricht zu versenden, warten wir auf eine eintreffende Nachricht.
import os
import toml
import telegram_bot_api as api
def config_laden():
configfile = os.path.join(SKRIPTPFAD, "example_cfg.toml")
with open(configfile) as file:
return toml.loads(file.read())
SKRIPTPFAD = os.path.abspath(os.path.dirname(__file__))
CONFIG = config_laden()
def main():
bot = api.Bot(CONFIG["telegram"]["token"])
while True:
messages = bot.get_updates()
for message in messages:
print("Gesamte Rückgabe mit sämlichen Informationen:")
print(message)
print(f"Nur Text: {message['message']['text']}")
if __name__ == "__main__":
main()
Alles anzeigen
Sobald nun eine Nachricht an den Bot gesendet wird, erhalten wir eine Nachricht. Die Struktur der Nachricht kann hier nachgelesen werden.
Zum Abschluss des Einstieges möchte ich noch erwähnen, dass es neben der Polling Methode zum Nachrichten empfangen noch die Möglichkeit zum Einrichten eines Webhooks gibt. Diese Möglichkeit macht es aber notwendig Ports zu öffnen, weswegen ich bisher durchgängig auf Polling setze.
Programmaufbau
Alles folgende bezieht sich auf Beispiel 2.
Beispiel 1 + 2 liegen von der Entstehung zeitlich etwas auseinander, weswegen sich meine herangehensweise etwas geändert hat.
Klassen
Aus telegramtechnischer Sicht arbeite ich im Hauptteil mit 2 Klassen.
Klasse User enthält folgende Attribute
- telegram_id - Enthält die Telegramid des Users um ihm antworten zu können
- name - Name des Users
- menue - Ist ein Menüpunkt aktiv so ist die entsprechende Funktion hier hinterlegt
- umenue - =Untermenü um die Navigation innerhalb eines Menüpunktes zu ermöglichen
Nimmt ein neuer/unbekannter User Kontakt mit dem Bot auf, so wird eine neue Instanz der Klasse User erstellt um die notwendigen Userbezogenen Daten hier speichern zu können.
Klasse TutorialBot enthält folgende Attribute
- bot - Ist wird die Botinstanz abgespeichert (Instanzierung des Bot von der API mit der zuweisung des Tokens)
- offset - Wird benötigt um den Erhalt der Nachrichten quittieren zu können. Offset muss immer 1 höher sein als die letzte größte Zahl der Message ID
Enthaltende Methoden
- get_updates - Bestätigt mit Hilfe des Attributs offset den erhalt der Nachrichten
Die API habe ich seit der Erstellung von Beispiel 2 so überarbeitet, dass get_updates direkt in der API enthalten ist. Deswegen ist im Beispiel 1 kein get_updates enthalten. Dafür werden in Beispiel 1 alle Botkommandos in der Klassse (hier TutorialBot) gespeichert
Aufbau
Nachrichten empfangen
Zunächst fange ich immer an, dass ich Nachrichten mit dem Bot empfangen kann.
Somit müssen wir erstmal unseren Bot mit dem zugehörigen Token instanzieren (Zeile 378). In Zeile 377 wird ein Dictonary erstellt um unsere später ankommenden Userinstanzen speichern zu können.
In der while True Schleife wird alle ~0.5 Sekunden (Zeile 387) geprüft ob neue Nachrichten vorliegen -> Zeile 381 users = bot.get_updates(users)
~0,5 Sekunden deshalb, da es ja nur die Zeit ist, welche an der Stelle pausiert wird, nicht noch die Zeit welche benötigt wird um die aufgerufenen Funktionen abzuarbeiten.
Wie ich seit der Erstellung des Codes durch Advent of Code durch __blackjack__ gelernt habe, ist es außerdem unüblich/nicht richtig das Dictonary zurück übergeben. Das users = müsste also eigentlich Weg. Jedoch möchte ich aktuell den Code nicht ändern, sondern mit dem Stand posten, wo ich wusste dass es im Produktiven Einsatz war.
Nun müssen wir dem Code folgen, wir prüfen also ob eine Nachricht vorhanden ist.
Die geschieht in der Methode get_updates() in Zeile 37 - 64.
Ist seit dem letzten Aufruf also eine Nachricht vorhanden wird in Zeile 49 über das Resultat iteriert.
Die Iteration ist deshalb nötig, da ja seit dem letzten Aufruf schon mehr als eine Nachricht vorliegen könnte. Mit diesem vorgehen arbeiten wir nun alle vorliegenden Nachrichten ab.
Für den Fall das das Resultat etwas anderes enthält als eine Nachricht und somit der Key "message" nicht enthalten ist, soll mit der nächsten Iteration fortgefahren werden (Zeile 51, 52)
Als nächstes wird mit Hilfe des user_handlers geprüft, ob in unserem Dictonary users der User von der vorliegenden Nachricht schon steht. Wenn nein wird der User aufgenommen (Zeile 69 - 72). Da ich mich entschieden hatte den Name nicht mit aufzunehmen, steht in Zeile 70 nur Test.
Als Key der Einträge im Dictonary wird die Telegramid des Users verwendet.
nachrichten_handler
Als nächstes geht die uns vorliegende Nachricht zu Funktion nachrichten_handler() der, wie ich finde, zentrale Punkt des Programmes.
Wir springen als von Zeile 54 nach Zeile 75 - 83.
Hier wird (nochmals wie ich gerade merke) geprüft, ob die Nachricht den Schlüssel "message" enhält.
Wenn dies der Fall ist, wird überprüft, ob es sich im vorliegenden Falle um ein Botkommando handelt. Ist es ein Botkommando wird die Nachricht an die Funktion bot_command weitergereicht. Um ein Botkommando handelt es sich immer dann, wenn der Text mit / beginnt
Ansonsten wird überprüft ob bei dem Antwortenden User schon ein Menüpunkt aktiv ist (also ob schon in einer vorherigen Nachricht ein Menüpunkt einem Botkommando aktiviert worden ist). Ein Menüpunkt ist aktiv wenn bei dem Attribut nicht None hinterlegt ist.
Da wir aktive Menüpunkte direkt an das Attribut menue binden, müssen wir von dem User nur menue aufrufen.
Aufgrund der Tatsache, dass wir diesen Aufruf so durchführen, ist es zwingend erforderlich, dass alle Funktionen die mittels bot_command angesprochen werden, die selbe Struktur aufweisen! In meinem gezeigten Fall also dass jede Funktion die Paramter nachricht, bot, users, key entgegen nimmt.
Da dieser Bot in diesem Beispiel nichts mit Nachrichten anfängt wenn keine Funktion aktiv ist, ist die Prüfung im nachrichten_handler beendet.
Als nächstes wird die Struktuer der Funktion bot_command betrachtet und wie das Attribut menue seinen Status annimmt.
bot_command
In dieser Funktion (Zeile 89 - 114) werden alle verfügbaren Botkommandos aufgeführt.
In einer if elif else Abfrage wird so jeder Eintrag durchgegangen ob er mit dem gesendeten Botkommando übereinstimmt.
Innerhalb des Ifzeiges wird dann die entsprechende zugehörige Funktion zum Botkommando augerufen.
z.B wird an unserem Bot das Kommando /statistik gesendet, so wird in Zeile 104, 105 die Funktion m_statistik_anzeigen() aufgerufen
m = Menü zur Besseren Übersicht im Strukturbaum der IDE.
Erreicht man den Else Teil der Abfrage, so erhält der Benutzer den Hinweis, dass das Kommando unbekannt ist.
Nur /start darf ausgeführt werden, wenn der Benutzer noch nicht in der Datenbank steht, deshalb ist /start in einer seperaten if Abfrage, bei allen anderen Kommandos wird vorausgesetzt, dass der Benutzer bekannt ist. (Zeile 94, 95)
Damit im Messanger die verfügbaren Kommandos angezeigt werden, muss Botfather eine Liste mit Botkommandos und kurzer Beschreibungstext mitgeteilt werden.
Dazu bei Botfather den passenden Bot auswählen
- /mybots
- In der Liste gewünschten Bot wählen
- Edit Bot wählen
- Edit Commands
- Liste nach vorgegebener Formatierung schicken
Nach dem Beispiel würde das wie folgt aussehen, hierzu erstelle ich mir immer eine kleine Textdatei botkommandos.txt
Die hat den Vorteil, dass man z.B mit /help dann alle verfügbaren Kommandos anzeigen lassen kann. (siehe Zeile 141)
/start - Erstmalige Ausführung zum Bestätigung der Datenschutzhinweise
/help - Öffnet das Hilfemenü
/del_acc - Löscht deinen Account
/stop - Bricht alle offenen Befehle ab
/statistik - Zeigt den Stand der Threadstatuse an
/thread_bearbeiten - Zum mitteilen des Ergebnisses
/threads_mod_anzeigen - Zeigt Threads an, die auf Bearbeitung durch Moderator warten
/thread_mod_bearbeiten - Bearbeitung durch Moderator
Ausführen von Kommandos
ohne Zwischenspeicherung
Wurde nun das Kommando /statistik an den Bot gesendet, wird über die Funktion bot_command die Funktion m_statistik_anzeigen() aufgerufen. (Zeile 173 - 181).
In dieser Funktion ist es nicht notwendig, den Menüpunkt in user.menue zu speichern, da alles notwendige in einem Durchlauf abgearbeitet werden kann.
Es wird zunächst das Bild für die Statistik generiert und im eingestellten Pfad abgespeichert.
Anschließend bekommt der User mit der Zeile 180 den Text gesendet und in Zeile 181 das Bild.
Für Fotos wird die Methode send_photo() anstelle send_message() benötigt.
mit Zwischenspeicherung
Nun wird es etwas "komplizierter".
Als Beispiel beziehen wir uns auf die Funktion m_thread_bearbeiten() von Zeile 184 - 234.
Der User hat also das Botkommando /thread_bearbeiten gesendet.
Über den Nachrichten Handler wird die Funktion bot_command aufgerufen und in der If Abfrage wurde Zeile 107 ausgeführt.
Wir wollen jetzt aber mit dem User interagieren können. Also Texte, Anweisungen senden, Angaben entgegen nehmen usw.
Aus diesem Zweck gibt es in der Klasse User die Attribute menue und umenue
In Zeile 187 wird der Zustand von umenue geprüft und festgestellt, dass der Zustand None ist.
Somit wissen wir, die Funktion wird das erste mal betreten.
Als erstes wird die aktive Funktion m_thread_bearbeiten in dem Attribut menue hinterlegt. Wichtig ist, die Funktion hinterlegen, nicht aufrufen. Das heißt die Funktion ohne () an menue binden.
Wir erinnern uns, im Nachrichten Handler wird geprüft ob menue None ist, da dies jetzt nicht mehr der Fall ist, wird die Funktion bei der nächsten Nachricht erneut aufgerufen, obwohl kein Bot Kommando gesendet wird.
Dann wird dem Benutzer der Anweisungstesxt gesendet Zeile 189 - 195 und das Attribute umenue auf 1 gesetzt (Zeile 196). Dadurch, dass nun umenu auf 1 ist und nicht mehr None, wissen wir beim nächsten Aufruf, wenn der User etwas postet, dass die Funktion schon aktiv ist und der Teil von Zeile 187 - 196 wird nicht mehr betreten sondern die If Part == 1 von Zeile 197 - 209.
Da die Prüfung mit is None und == 1 nicht mit elif Verknüpft ist, wird im selben Durchlauf auch noch der Teil von 197 - 208 bearbeitet.
umenue ist also zum Zeitpunkt des ersten Verlassen der Funktion auf 2.
Nun erfolgt die erste Antwort seit die Funktion aktiv ist bei dem User.
Der Nachrichtenhandler ruft m_thread_bearbeiten auf, das menue enthält m_thread_bearbeiten.
Fortgefahren wird in Zeile 209, da umenue die 2 enthält.
Ist die Antwort des Users nicht nach dem geforderten Format, so wird das dem User mitgeteilt und die Funktion in Zeile 213 verlassen. Der Inhalt von menueund umenu bleibt also unverändert, sodass er die Eingabe erneut machen kann und an der selben Stelle der Funktion wieder landet.
Da in dem Beispiel eine Mehrfacheingabe gewünscht war, ohne dass jedesmal /thread_bearbeiten eingegeben werden musste, wird nach korrektem Format nachdem der Datensatz aktualisiert wurde und der User benachricht, umenue und menue NICHT auf None gesetzt, wie es nach jeder überlichen Funktion nun nötig wäre, sondern umenue wird wieder zurück gesetzt auf 1 um auf die Nächste Eingabe zu warten. Ansonsten wäre es der korrekte Weg nun umenue und menue auf None zu setzen.
In dem vorliegenden Fall ist es nur Möglich, dass die beiden Attribute wieder den Zustand None annehmen, wenn der Benutzer /abbrechen als Botkommando sendet und in Zeile 166 - 170 landet.
Damit dies so funktioniert ist es jedoch nötig, die Reihenfolge der Behandlung im Nachrichten Handler wie gezeigt einzuhalten. Es muss also erst auf ein bot_command geprüft werden und erst danach darf geprüft werden ob menue None ist oder nicht.
if "bot_command" in nachricht["message"].get("entities", [{}])[0].get("type", ""):
users = bot_command(nachricht, bot, users, key)
elif users[key].menue is not None:
users[key].menue(nachricht, bot, users, key)
Da der Punkt bisher noch etwas zu kurz kam.
Damit der Bot es unterscheiden kann, von wem die Nachricht kommt und bei wem welcher Menüpunkt aktiv ist wird über das Dictonary users realisiert und dem ansprechen des Dictonarys über dem Key was der Telegram ID des Users entspricht.
Mit diesem Wissen, so hoffe ich, sollte es dir Möglich sein deinen eigenen Bot zu erstellen.
Was dieser Bot für Funktionen erfüllt, hängt natürlich davon ab, was in den einzelnen Botkommandos integriert wird.
Jetzt noch etwas aus Beispiel 1
Klassen
Zum Abschluss sehen wir uns noch einige Punkte aus dem Beispiel 1 an
Auch hier gibt es wieder 2 relevante Klassen Namens User und EmpyrionBot.
User ist schon bekannt von dem vorherigen Beispiel.
EmpyrionBot erbt nun von telegram_bot_api.Bot. Dies ist der erste Unterschied. Hat den Vorteil, dass ich innerhalb von Methoden der Klasse EmpyrionBot direkt auf die Methoden der API zugreifen kann.
z.B.: Zeile 96 self.send_message(key, "Abgebrochen")
Ebenso habe ich die bot_commands (hier nur commands genannt) als Methode zur Klasse zugeordnet und auch alle einzelnen Botkommandos als Methode hinterlegt anstatt einzelner Funktionen.
Zugriffskontrolle
In der Konfigurationsdatei gibt es einen Eintrag mit allowed_ids = {"username" = <telegramid>}
In Zeile 170 wird diesesmal das Dictionary users über eine Dictionary Comprehension erstellt, wo für jeden Eintrag in allowed_ids eine Instanz der Klasse User erstellt und gespeichert wird.
Kommt nun eine neue Nachricht prüft der Nachrichten Handler in Zeile 153 ob die Telegramid im Dictionary users bekannt ist, wenn nicht erhält der User eine Nachricht "Permission denied" in Zeile 154.
Fließende Nachrichtenaktualisierung
Der Bot aus Beispiel 1 soll einen Gameserver steuern und wenn nötig ein Update anstoßen. Die Konsolenausgabe des Updateprozesses wird mit dem Bot auf den Messanger wieder gegeben. (Zeile 122 - 140)
Hierfür gibt es die Funktion edit_message_text in der API.
Die Nachrichten auf Telegram sind auf 4096 Zeichen beschränkt, deshalb wird vor jedem editieren der Nachricht die Länge der Nachricht überprüft. Befindet es sich innerhalb der Zeichengrenze wird die Nachricht editiert, ansonsten wird eine neue Nachricht gesendet und dieses wieder editiert.
Abschluss
Anmerkungen und Kritik zum gezeigten Code bitte in einem seperaten Thread äußern.
Anmerkungen und Kritik zum Tutorial sind gerne hier gesehen.
Bei Problemen zur Umsetzung des hier gezeigten, bitte ebenfalls einen neuen Thread eröffnen.