পাইথনে আপনার পরিষেবা আপটাইম নিরীক্ষণ করার জন্য কীভাবে একটি টেলিগ্রাম বট তৈরি করবেন (পর্ব 1: তাত্ক্ষণিক মেট্রিক্স)

সবাইকে অভিবাদন! এখন বেশ কয়েক বছর ধরে, আমি পাইথনে নিজের জন্য বিভিন্ন "সহকারী" টেলিগ্রাম বট লিখছি যা আমার জন্য বিভিন্ন ছোট ছোট রুটিন কাজ পরিচালনা করে - আমাকে কিছু সম্পর্কে অবহিত করা, পরিষেবা আপটাইম চেক করা, টেলিগ্রাম চ্যানেল এবং চ্যাট থেকে আকর্ষণীয় বিষয়বস্তু ফরওয়ার্ড করা ইত্যাদি।

এটি সুবিধাজনক কারণ ফোনটি সর্বদা হাতের কাছে থাকে এবং আমার ল্যাপটপ না খুলেও সার্ভারে কিছু ঠিক করতে সক্ষম হওয়া আমাকে বিশেষ আনন্দ দেয়।

সাধারণভাবে, আমি অনেকগুলি বিভিন্ন ছোট প্রকল্পের টেমপ্লেট সংগ্রহ করেছি যা আমি হ্যাকারনুন পাঠকদের সাথে ভাগ করতে চাই৷

আমি অবিলম্বে বলব যে উদাহরণগুলি তাদের প্রয়োগের ক্ষেত্রে "যেমন আছে" এর দিক থেকে কুলুঙ্গি হতে পারে, তবে আমি সেই জায়গাগুলিকে চিহ্নিত করব যেখানে, কোডের কয়েকটি লাইন আপনার নিজের মতো পরিবর্তন করে, আপনি বেশিরভাগই পুনরায় ব্যবহার করতে সক্ষম হবেন আপনার প্রকল্পের জন্য উন্নয়ন.

আমি এই নির্দিষ্ট প্রকল্পটি কয়েক দিন আগে সম্পন্ন করেছি, এবং এটি ইতিমধ্যেই আমাকে অনেক সুবিধা এনে দিয়েছে। আমি একটি Web3 পরিকাঠামো প্রদানকারী chainstack.com-এ কাজ করি, EVM ব্লকচেইনে স্মার্ট চুক্তি থেকে ডেটা ইন্ডেক্স করার জন্য একটি পরিষেবা নিয়ে কাজ করি।

এবং পরিষেবার গুণমানটি সমালোচনামূলকভাবে বিকশিত হচ্ছে তার উপর নির্ভর করে যে নোডগুলি থেকে পরিষেবাটি অনলাইনে ডেটা পুনরুদ্ধার করে সেগুলি কতটা "ভাল" কাজ করছে৷

আমাদের অবকাঠামো বিভাগ যেমন Grafana, BetterUptime এবং অন্যান্য ব্যবহার করে এমন রেডিমেড সরঞ্জামগুলি ব্যবহার করার চেষ্টা করার জন্য আমি অনেক ঘন্টা কাটিয়েছি, কিন্তু সিস্টেমের অভ্যন্তরীণ বিষয়ে আমার খুব কম আগ্রহ নেই, আমার জন্য প্রধান ফোকাস হল প্রবেশদ্বারের মেট্রিক্স এবং প্রস্থান, আমি আমার নিজের বট লিখতে সিদ্ধান্ত নিয়েছে, যা নিম্নলিখিত কাজ করবে:

• আমার অনুরোধে, এটি পরিষেবাতে যাবে, মেট্রিক্স পরীক্ষা করবে এবং বর্তমান পরিস্থিতি সম্পর্কে আমাকে একটি সংক্ষিপ্ত প্রতিবেদন পাঠাবে।

• আমার অন্য অনুরোধে, এটি আমাকে গত 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.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

বটের পরবর্তী গুরুত্বপূর্ণ ফাইল হল 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

সিস্টেমড ইউনিট ফাইল কনফিগার করার পরে।

ফলে বটের কাজ দেখতে এরকম হবে।

1. সবকিছু ঠিক থাকলে:

   
   

2. এবং যদি ল্যাগটি খুব বড় হয়ে যায় তবে নিম্নরূপ:

   
   

নিম্নলিখিত নিবন্ধগুলিতে, আমি কীভাবে অবশিষ্ট দুটি কাজ বাস্তবায়ন করতে পারি তা বর্ণনা করব:

1. অনুরোধে গ্রাফ পুনরুদ্ধার করুন যাতে গত X ঘন্টার মধ্যে ঘটে যাওয়া ঘটনাগুলি দেখানো হয়।

   
   

2. একটি সতর্কতা পান যে ইঙ্গিত করে যে কিছু বর্তমানে ঘটছে এবং পদক্ষেপের প্রয়োজন।

প্রকল্পের সোর্স কোড GitHub সংগ্রহস্থলে উপলব্ধ। আপনি যদি এই টিউটোরিয়ালটিকে সহায়ক বলে মনে করেন তবে নির্দ্বিধায় এটিকে GitHub-এ একটি তারকা দিন, আমি এটির প্রশংসা করব🙂