如何创建电报机器人以监视Python中的正常运行时间（第1部分：即时指标）-DEV365 开发者社区

大家好！几年来，我一直在Python为我自己写各种“助手”电报机器人，这些电报机器人为我处理各种小型日常任务 - 通知我有关某些内容，检查服务正常运行时间，从电报频道和聊天中转发有趣的内容等等。

这很方便，因为手机总是在手头，并且能够在服务器上修复某些东西，甚至不打开我的笔记本电脑会给我特别的乐趣。

通常，我积累了许多我想与dev.to读者分享的不同小型项目模板。

我会马上说，这些示例可能是其应用程序“原样”的利基市场，但是我将标记那些地方，通过将几行代码更改为自己，您将能够重新使用项目的大多数开发。

我几天前完成了这个特定的项目，这已经给我带来了很多好处。我在Web3基础架构提供商Chainstack.com上工作，处理用于索引来自EVM区块链智能合约的数据的服务。

和开发的服务质量在很大程度上取决于如何“井”在线检索数据的节点。

我花了很多时间尝试使用基础设施部门使用的现成工具，例如Grafana，Better Pustime等，但是由于我对系统的内部构件不感兴趣，因此我的主要重点是入口和退出的指标，我决定写下我自己的机器人，这将有以下内容：

>>>>>>>>>>/p>>

>：

应我的要求，它将访问服务，检查指标，并向我发送有关当前情况的简短报告。
根据我的其他要求，它会给我发送有关最近X小时发生的事情的图表。
在特殊情况下，它会给我发送一条通知，说当时正在发生某些事情。

在本文中，我将重点介绍第一部分，即根据要求收到指标。

我们将需要一个新的虚拟环境进行工作。

cd ~ 
virtualenv -p python3.8 up_env  # crete a virtualenv
source ~/up_env/bin/activate  # activate the virtualenl

安装依赖项：

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)

带有functions functions.py的文件（您可以使用类实现它，但是由于示例很短，因此我不打算将其分为模块，但是多线程库需要将函数移至单独的文件）。导入依赖性：

import numpy as np
import multiprocessing

from web3 import Web3 #  add those libraries needed for your task

描述检查状态的函数。就我而言，它涉及通过预先选择的公共节点进行循环，检索其最后一个块，取出中位数以滤除任何偏差，然后检查我们自己的节点针对此中位数。

服务状态检查功能（您可以用自己的替换）：

# 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

bot的下一个重要文件是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

接下来，让我们描述一个从机器人UI发出命令时将调用的函数。

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")

现在，剩下的就是添加初始化机器人的部分，并连接处理程序功能：

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()

最后，您可以使用：
在便宜的VPS服务器上运行代码

source ~/up_env/bin/activate
python uptime_bot.py

配置SystemD单元文件后。

结果，机器人的工作看起来像这样。

如果一切都很好：

如果滞后太大，则如下：

在以下文章中，我将描述如何实施其余两个任务：

在请求上检索图形，显示最后X小时发生的事件。
接收一个警报，指示当前正在发生某些事情并需要采取行动。

项目的源代码可在github repository中找到。如果您发现本教程有帮助，请随时在github上给它一颗星星，我很感激