सभी को नमस्कार! अब कई वर्षों से, मैं पायथन में अपने लिए विभिन्न "सहायक" टेलीग्राम बॉट लिख रहा हूं जो मेरे लिए विभिन्न छोटे नियमित कार्यों को संभालते हैं - मुझे किसी चीज़ के बारे में सूचित करना, सेवा अपटाइम की जांच करना, टेलीग्राम चैनलों और चैट से दिलचस्प सामग्री को अग्रेषित करना, इत्यादि। यह सुविधाजनक है क्योंकि फोन हमेशा हाथ में रहता है, और अपना लैपटॉप खोले बिना भी सर्वर पर कुछ ठीक करने में सक्षम होने से मुझे विशेष खुशी मिलती है। सामान्य तौर पर, मैंने बहुत सारे अलग-अलग छोटे प्रोजेक्ट टेम्पलेट जमा कर लिए हैं जिन्हें मैं हैकरनून पाठकों के साथ साझा करना चाहता हूं। मैं तुरंत कहूंगा कि उदाहरण "जैसा है" उनके अनुप्रयोग के संदर्भ में विशिष्ट हो सकते हैं, लेकिन मैं उन स्थानों को चिह्नित करूंगा, जहां, कोड की कुछ पंक्तियों को अपने हिसाब से बदलकर, आप अधिकांश का पुन: उपयोग करने में सक्षम होंगे आपकी परियोजनाओं के लिए विकास। मैंने यह विशिष्ट परियोजना कुछ दिन पहले ही पूरी की है, और इससे मुझे पहले ही बहुत लाभ हुआ है। मैं एक वेब3 इन्फ्रास्ट्रक्चर प्रदाता चेनस्टैक.कॉम पर काम करता हूं, जो ईवीएम ब्लॉकचेन पर स्मार्ट कॉन्ट्रैक्ट से डेटा को अनुक्रमित करने की सेवा से संबंधित है। और विकसित की जा रही सेवा की गुणवत्ता इस बात पर निर्भर करती है कि जिन नोड्स से सेवा ऑनलाइन डेटा पुनर्प्राप्त करती है वे कितने "अच्छी तरह से" काम कर रहे हैं। मैंने तैयार किए गए टूल का उपयोग करने में कई घंटे बिताए, जो हमारे बुनियादी ढांचे प्रभाग का उपयोग करते हैं, जैसे कि ग्राफाना, बेटरअपटाइम और अन्य, लेकिन चूंकि मुझे सिस्टम के आंतरिक में बहुत कम रुचि है, मेरे लिए मुख्य फोकस प्रवेश द्वार पर मेट्रिक्स है और बाहर निकलने पर, मैंने अपना स्वयं का बॉट लिखने का निर्णय लिया, जो निम्नलिखित कार्य करेगा: मेरे अनुरोध पर, यह सेवा में जाएगा, मेट्रिक्स की जांच करेगा, और मुझे वर्तमान स्थिति पर एक संक्षिप्त रिपोर्ट भेजेगा। मेरे दूसरे अनुरोध पर, यह मुझे पिछले 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) फ़ंक्शन के साथ फ़ाइल (आप इसे कक्षाओं के साथ कार्यान्वित कर सकते हैं, लेकिन चूंकि उदाहरण छोटा है, मैंने इसे मॉड्यूल में विभाजित करने की योजना नहीं बनाई थी, लेकिन मल्टी-थ्रेडिंग लाइब्रेरी के लिए फ़ंक्शंस को एक अलग फ़ाइल में ले जाने की आवश्यकता होती है)। आयात निर्भरताएँ: फ़ंक्शन्स.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 आगे, आइए एक फ़ंक्शन का वर्णन करें जिसे बॉट के यूआई से कमांड जारी होने पर कॉल किया जाएगा। 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 सिस्टमडी यूनिट फ़ाइल को कॉन्फ़िगर करने के बाद। परिणामस्वरूप, बॉट का कार्य इस प्रकार दिखेगा. यदि सब कुछ ठीक है: और यदि अंतराल बहुत बड़ा हो जाता है, तो इस प्रकार है: निम्नलिखित लेखों में, मैं वर्णन करूंगा कि शेष दो कार्यों को कैसे कार्यान्वित किया जाए: पिछले X घंटों में हुई घटनाओं को दर्शाने वाले अनुरोध पर ग्राफ़ पुनः प्राप्त करें। एक चेतावनी प्राप्त करें जो दर्शाती है कि वर्तमान में कुछ हो रहा है और कार्रवाई की आवश्यकता है। प्रोजेक्ट का सोर्स कोड GitHub में उपलब्ध है। यदि आपको यह ट्यूटोरियल मददगार लगा, तो बेझिझक इसे GitHub पर एक स्टार दें, मैं इसकी सराहना करूंगा🙂 रिपॉजिटरी
