Hallo zusammen! Seit einigen Jahren schreibe ich für mich selbst verschiedene „Assistenten“-Telegram-Bots in Python, die verschiedene kleine Routineaufgaben für mich erledigen – mich über etwas benachrichtigen, die Dienstverfügbarkeit überprüfen, interessante Inhalte von Telegram-Kanälen und Chats weiterleiten und so weiter. Das ist praktisch, weil das Telefon immer zur Hand ist und es mir besonders Freude bereitet, etwas auf dem Server reparieren zu können, ohne meinen Laptop öffnen zu müssen. Im Allgemeinen habe ich viele verschiedene Vorlagen für kleine Projekte gesammelt, die ich mit HackerNoon-Lesern teilen möchte. Ich sage gleich, dass die Beispiele in Bezug auf ihre Anwendung „so wie sie sind“ möglicherweise Nischenbeispiele sind, aber ich werde die Stellen markieren, an denen Sie die meisten davon wiederverwenden können, indem Sie ein paar Zeilen Code in Ihren eigenen Code ändern Entwicklungen für Ihre Projekte. Ich habe dieses konkrete Projekt vor ein paar Tagen abgeschlossen und es hat mir bereits viele Vorteile gebracht. Ich arbeite bei Chainstack.com, einem Web3-Infrastrukturanbieter, und beschäftige mich mit einem Dienst zur Indizierung von Daten aus Smart Contracts auf EVM-Blockchains. Und die Qualität des zu entwickelnden Dienstes hängt entscheidend davon ab, wie „gut“ die Knoten funktionieren, von denen der Dienst online Daten abruft. Ich habe viele Stunden damit verbracht, vorgefertigte Tools zu verwenden, die unsere Infrastrukturabteilung verwendet, wie Grafana, BetterUptime und andere, aber da ich wenig Interesse an den Interna des Systems habe, liegt mein Hauptaugenmerk auf den Metriken am Eingang und Nach dem Exit beschloss ich, meinen eigenen Bot zu schreiben, der Folgendes tun würde: Auf meine Bitte hin ging es zum Dienst, überprüfte die Kennzahlen und schickte mir einen kurzen Bericht über die aktuelle Situation. Auf meine andere Anfrage hin würde es mir Diagramme darüber senden, was in den letzten X Stunden passiert ist. Im Falle einer besonderen Situation würde es mir eine Benachrichtigung senden, dass gerade etwas passiert. In diesem Artikel werde ich mich auf den ersten Teil konzentrieren, nämlich den Erhalt von Metriken auf Anfrage. Wir werden eine neue virtuelle Umgebung für die Arbeit brauchen. cd ~ virtualenv -p python3.8 up_env # crete a virtualenv source ~/up_env/bin/activate # activate the virtualenl Abhängigkeiten installieren: pip install python-telegram-bot pip install "python-telegram-bot[job-queue]" --pre pip install --upgrade python-telegram-bot==13.6.0 # the code was written before version 20, so here the version is explicitly specified pip install numpy # needed for the median value function pip install web3 # needed for requests to nodes (replace with what you need) Datei mit Funktionen (Sie können sie mit Klassen implementieren, aber da das Beispiel kurz ist, hatte ich nicht vor, es in Module aufzuteilen, aber eine Multithreading-Bibliothek erfordert, dass Funktionen in eine separate Datei verschoben werden). Abhängigkeiten importieren: „functions.py“ import numpy as np import multiprocessing from web3 import Web3 # add those libraries needed for your task Beschreiben einer Funktion zur Überprüfung des Zustands. In meinem Fall ging es darum, vorher ausgewählte öffentliche Knoten zu durchlaufen, deren letzten Block abzurufen, den Medianwert zu verwenden, um etwaige Abweichungen herauszufiltern, und dann unseren eigenen Knoten mit diesem Median zu vergleichen. Funktion zur Überprüfung des Dienststatus (Sie können sie durch Ihre eigene ersetzen): # Helper function that checks a single node def get_last_block_once(rpc): try: w3 = Web3(Web3.HTTPProvider(rpc)) block_number = w3.eth.block_number if isinstance(block_number, int): return block_number else: return None except Exception as e: print(f'{rpc} - {repr(e)}') return None # Main function to check the status of the service that will be called def check_service(): # pre-prepared list of reference nodes # for any network, it can be found on the website https://chainlist.org/ list_of_public_nodes = [ 'https://polygon.llamarpc.com', 'https://polygon.rpc.blxrbdn.com', 'https://polygon.blockpi.network/v1/rpc/public', 'https://polygon-mainnet.public.blastapi.io', 'https://rpc-mainnet.matic.quiknode.pro', 'https://polygon-bor.publicnode.com', 'https://poly-rpc.gateway.pokt.network', 'https://rpc.ankr.com/polygon', 'https://polygon-rpc.com' ] # parallel processing of requests to all nodes with multiprocessing.Pool(processes=len(list_of_public_nodes)) as pool: results = pool.map(get_last_block_once, list_of_public_nodes) last_blocks = [b for b in results if b is not None and isinstance(b, int)] # define the maximum and median value of the current block med_val = int(np.median(last_blocks)) max_val = int(np.max(last_blocks)) # determine the number of nodes with the maximum and median value med_support = np.sum([1 for x in last_blocks if x == med_val]) max_support = np.sum([1 for x in last_blocks if x == max_val]) return max_val, max_support, med_val, med_support Die nächste wichtige Datei des Bots ist . Wir importieren Bibliotheken und Funktionen aus der Datei oben und legen die notwendigen Konstanten fest: uptime_bot.py import telegram from telegram.ext import Updater, CommandHandler, Filters from functions import get_last_block_once, check_service # Here one can to set a limited circle of bot users, # listing the usernames of the users ALLOWED_USERS = ['your_telegram_account', 'someone_else'] # The address of the node that I am monitoring (also a public node in this case) OBJECT_OF_CHECKING = 'https://polygon-mainnet.chainstacklabs.com' # Threshold for highlighting critical lag THRESHOLD = 5 Als Nächstes beschreiben wir eine Funktion, die aufgerufen wird, wenn der Befehl über die Benutzeroberfläche des Bots ausgegeben wird. def start(update, context): """Send a message when the command /start is issued.""" try: # Get the user user = update.effective_user # Filter out bots if user.is_bot: return # Check if the user is allowed username = str(user.username) if username not in ALLOWED_USERS: return except Exception as e: print(f'{repr(e)}') return # Call the main function to check the network status max_val, max_support, med_val, med_support = check_service() # Call the function to check the status of the specified node last_block = get_last_block_once(OBJECT_OF_CHECKING) # Create the message to send to Telegram message = "" # Information about the state of the nodes in the public network (median, maximum, and number of nodes) message += f"Public median block number {med_val} (on {med_support}) RPCs\n" message += f"Public maximum block number +{max_val - med_val} (on {max_support}) PRCs\n" # Compare with the threshold if last_block is not None: out_text = str(last_block - med_val) if last_block - med_val < 0 else '+' + str(last_block - med_val) if abs(last_block - med_val) > THRESHOLD: message += f"The node block number shift ⚠️<b>{out_text}</b>⚠️" else: message += f"The node block number shift {out_text}" else: # Exception processing if a node has not responded message += f"The node has ⚠️<b>not responded</b>⚠️" # Send the message to the user context.bot.send_message(chat_id=user.id, text=message, parse_mode="HTML") Jetzt müssen Sie nur noch den Teil hinzufügen, in dem der Bot initialisiert wird, und die Handler-Funktion verbinden: token = "xxx" # Bot token obtained from BotFather # set up the bot bot = telegram.Bot(token=token) updater = Updater(token=token, use_context=True) dispatcher = updater.dispatcher # bind the handler function dispatcher.add_handler(CommandHandler("start", start, filters=Filters.chat_type.private)) # run the bot updater.start_polling() Schließlich können Sie den Code auf einem günstigen VPS-Server ausführen, indem Sie Folgendes verwenden: source ~/up_env/bin/activate python uptime_bot.py Nach der Konfiguration der systemd-Unit-Datei. Infolgedessen wird die Arbeit des Bots so aussehen. Wenn alles in Ordnung ist: Und wenn die Verzögerung zu groß wird, dann wie folgt: In den folgenden Artikeln beschreibe ich, wie man die beiden verbleibenden Aufgaben umsetzt: Rufen Sie auf Anfrage Diagramme ab, die die Ereignisse zeigen, die in den letzten X Stunden aufgetreten sind. Erhalten Sie eine Benachrichtigung, die darauf hinweist, dass gerade etwas passiert und Maßnahmen erforderlich sind. Der Quellcode des Projekts ist im GitHub- verfügbar. Wenn Sie dieses Tutorial hilfreich fanden, können Sie ihm gerne einen Stern auf GitHub geben, ich würde mich darüber freuen 🙂 Repository
