¡Hola a todos! Durante varios años, he estado escribiendo varios bots de telegramas "asistentes" para mí en Python que manejan varias pequeñas tareas de rutina para mí: notificarme sobre algo, verificar el tiempo de actividad del servicio, reenviar contenido interesante de los canales y chats de Telegram, y así sucesivamente. Esto es conveniente porque el teléfono está siempre a mano, y poder arreglar algo en el servidor sin siquiera abrir mi computadora portátil me brinda un placer especial. En general, he acumulado muchas plantillas de proyectos pequeños diferentes que quiero compartir con los lectores de HackerNoon. Diré de inmediato que los ejemplos pueden ser de nicho en términos de su aplicación "tal como están", pero marcaré aquellos lugares donde, al cambiar unas pocas líneas de código por las tuyas, podrás reutilizar la mayoría de los desarrollos para sus proyectos. Completé este proyecto específico hace unos días, y ya me ha traído muchos beneficios. Trabajo en un proveedor de infraestructura Web3, chainstack.com, y me ocupo de un servicio para indexar datos de contratos inteligentes en cadenas de bloques EVM. Y la calidad del servicio que se está desarrollando depende críticamente de qué tan "bien" estén funcionando los nodos desde los cuales el servicio recupera datos en línea. Pasé muchas horas tratando de usar herramientas listas para usar que usa nuestra división de infraestructura, como Grafana, BetterUptime y otras, pero como tengo poco interés en las partes internas del sistema, el enfoque principal para mí son las métricas en la entrada y la salida, decidí escribir mi propio bot, que haría lo siguiente: A mi pedido, iría al servicio, verificaría las métricas y me enviaría un breve informe sobre la situación actual. En mi otro pedido, me enviaría gráficos de lo que ha estado sucediendo durante las últimas X horas. En caso de una situación especial, me enviaría una notificación de que algo está pasando en ese momento. En este artículo me centraré en la primera parte, es decir, recibir métricas a pedido. Necesitaremos un nuevo entorno virtual para trabajar. cd ~ virtualenv -p python3.8 up_env # crete a virtualenv source ~/up_env/bin/activate # activate the virtualenl Instalar dependencias: 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) Archivo con funciones (puede implementarlo con clases, pero dado que el ejemplo es corto, no planeé dividirlo en módulos, pero una biblioteca de subprocesos múltiples requiere que las funciones se muevan a un archivo separado). Dependencias de importación: functions.py import numpy as np import multiprocessing from web3 import Web3 # add those libraries needed for your task Describir una función para comprobar el estado. En mi caso, implicó recorrer nodos públicos preseleccionados, recuperar su último bloque, tomar el valor de la mediana para filtrar cualquier desviación y luego comparar nuestro propio nodo con esta mediana. Función de verificación del estado del servicio (puede reemplazarla por la suya): # 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 El siguiente archivo importante del bot es . Importamos bibliotecas y funciones del archivo anterior y establecemos las constantes necesarias: 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 A continuación, describamos una función que se llamará cuando se emita el comando desde la interfaz de usuario del bot. 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") Ahora, todo lo que queda es agregar la parte donde se inicializa el bot y se conecta la función del controlador: 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() Finalmente, puede ejecutar el código en un servidor VPS económico usando: source ~/up_env/bin/activate python uptime_bot.py Después de configurar el archivo de unidad systemd. Como resultado, el trabajo del bot se verá así. Si todo está bien: Y si el retraso se vuelve demasiado grande, entonces de la siguiente manera: En los siguientes artículos, describiré cómo implementar las dos tareas restantes: Recupere gráficos a pedido que muestren los eventos que ocurrieron durante las últimas X horas. Reciba una alerta que indica que algo está sucediendo actualmente y requiere acción. El código fuente del proyecto está disponible en el de GitHub. Si este tutorial le resultó útil, no dude en darle una estrella en GitHub, se lo agradecería 🙂 repositorio
