Cách tạo Bot Telegram để theo dõi thời gian hoạt động của dịch vụ của bạn bằng Python (Phần 1: Số liệu tức thì)

Xin chào tất cả mọi người! Trong vài năm nay, tôi đã viết nhiều chương trình điện tín "trợ lý" khác nhau cho chính mình bằng Python để xử lý các tác vụ nhỏ thường ngày khác nhau cho tôi - thông báo cho tôi về điều gì đó, kiểm tra thời gian hoạt động của dịch vụ, chuyển tiếp nội dung thú vị từ các kênh điện tín và trò chuyện, v.v.

Điều này rất tiện lợi vì điện thoại luôn ở trong tầm tay và việc có thể sửa một số thứ trên máy chủ mà không cần mở máy tính xách tay mang lại cho tôi niềm vui đặc biệt.

Nói chung, tôi đã tích lũy được rất nhiều mẫu dự án nhỏ khác nhau mà tôi muốn chia sẻ với độc giả của HackerNoon.

Tôi sẽ nói ngay rằng các ví dụ có thể thích hợp về mặt ứng dụng của chúng "nguyên trạng", nhưng tôi sẽ đánh dấu những nơi mà bằng cách thay đổi một vài dòng mã thành mã của riêng bạn, bạn sẽ có thể sử dụng lại hầu hết các ví dụ. phát triển cho các dự án của bạn.

Tôi đã hoàn thành dự án cụ thể này vài ngày trước và nó đã mang lại cho tôi rất nhiều lợi ích. Tôi làm việc tại nhà cung cấp cơ sở hạ tầng Web3 chainstack.com, xử lý dịch vụ lập chỉ mục dữ liệu từ hợp đồng thông minh trên chuỗi khối EVM.

Và chất lượng của dịch vụ đang được phát triển chủ yếu phụ thuộc vào mức độ "tốt" của các nút mà dịch vụ truy xuất dữ liệu trực tuyến đang hoạt động.

Tôi đã dành nhiều giờ để cố gắng sử dụng các công cụ có sẵn mà bộ phận cơ sở hạ tầng của chúng tôi sử dụng, chẳng hạn như Grafana, BetterUptime và các công cụ khác, nhưng vì tôi ít quan tâm đến nội bộ của hệ thống nên trọng tâm chính của tôi là các chỉ số ở lối vào và lối ra, tôi quyết định viết bot của riêng mình, nó sẽ làm như sau:

• Theo yêu cầu của tôi, nó sẽ chuyển đến dịch vụ, kiểm tra các số liệu và gửi cho tôi một báo cáo ngắn gọn về tình hình hiện tại.

• Theo yêu cầu khác của tôi, nó sẽ gửi cho tôi các biểu đồ về những gì đã xảy ra trong X giờ qua.

• Trong trường hợp đặc biệt, nó sẽ gửi cho tôi một thông báo rằng có điều gì đó đang xảy ra vào thời điểm đó.

Trong bài viết này, tôi sẽ tập trung vào phần đầu tiên, đó là nhận số liệu theo yêu cầu.

Chúng tôi sẽ cần một môi trường ảo mới cho công việc.

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

Cài đặt phụ thuộc:

 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)

Tệp có các hàm functions.py (bạn có thể triển khai nó với các lớp, nhưng vì ví dụ này ngắn nên tôi không định chia nó thành các mô-đun, nhưng một thư viện đa luồng yêu cầu các hàm được chuyển sang một tệp riêng). Phụ thuộc nhập khẩu:

 import numpy as np import multiprocessing from web3 import Web3 # add those libraries needed for your task

Mô tả chức năng kiểm tra trạng thái. Trong trường hợp của tôi, nó liên quan đến việc lặp qua các nút công khai được chọn trước, truy xuất khối cuối cùng của chúng, lấy giá trị trung bình để lọc ra bất kỳ sai lệch nào và sau đó, kiểm tra nút của chính chúng ta dựa trên giá trị trung bình này.

Chức năng kiểm tra trạng thái dịch vụ (bạn có thể thay thế bằng chức năng của riêng mình):

 # 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

Tệp quan trọng tiếp theo của bot là uptime_bot.py . Chúng tôi nhập các thư viện và hàm từ tệp ở trên và đặt các hằng số cần thiết:

 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

Tiếp theo, hãy mô tả một chức năng sẽ được gọi khi lệnh được đưa ra từ giao diện người dùng của 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")

Bây giờ, tất cả những gì còn lại là thêm phần khởi tạo bot và chức năng xử lý được kết nối:

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

Cuối cùng, bạn có thể chạy mã trên máy chủ VPS giá rẻ bằng cách sử dụng:

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

Sau khi định cấu hình tệp đơn vị systemd.

Kết quả là công việc của bot sẽ như thế này.

1. Nếu mọi thứ đều ổn:

   
   

2. Và nếu độ trễ trở nên quá lớn, thì như sau:

   
   

Trong các bài viết sau, tôi sẽ mô tả cách thực hiện hai tác vụ còn lại:

1. Truy xuất biểu đồ theo yêu cầu hiển thị các sự kiện đã xảy ra trong X giờ qua.

   
   

2. Nhận thông báo cho biết có điều gì đó hiện đang xảy ra và yêu cầu hành động.

Mã nguồn của dự án có sẵn trong kho lưu trữ GitHub. Nếu bạn thấy hướng dẫn này hữu ích, vui lòng cho nó một ngôi sao trên GitHub, tôi sẽ đánh giá cao nó🙂