Herkese merhaba! Birkaç yıldır Python'da benim için çeşitli küçük rutin görevleri yerine getiren çeşitli "asistan" telgraf botları yazıyorum - beni bir şey hakkında bilgilendirmek, hizmetin çalışma süresini kontrol etmek, telgraf kanallarından ve sohbetlerden ilginç içerikleri iletmek vb. Bu kullanışlı çünkü telefon her zaman elinizin altında ve dizüstü bilgisayarımı açmadan sunucudaki bir şeyi tamir edebilmek bana ayrı bir keyif veriyor. Genel olarak HackerNoon okuyucularıyla paylaşmak istediğim birçok farklı küçük proje şablonunu biriktirdim. Örneklerin "olduğu gibi" uygulanmaları açısından niş olabileceğini hemen söyleyeceğim, ancak birkaç kod satırını kendinize göre değiştirerek çoğu şeyi yeniden kullanabileceğiniz yerleri işaretleyeceğim. Projeleriniz için gelişmeler. Bu özel projeyi birkaç gün önce tamamladım ve bu bana şimdiden pek çok fayda sağladı. Bir Web3 altyapı sağlayıcısı Chainstack.com'da çalışıyorum ve EVM blok zincirlerindeki akıllı sözleşmelerden gelen verileri endekslemeye yönelik bir hizmetle ilgileniyorum. Ve geliştirilmekte olan hizmetin kalitesi, hizmetin çevrimiçi olarak veri aldığı düğümlerin ne kadar "iyi" çalıştığına bağlıdır. Grafana, BetterUptime ve diğerleri gibi altyapı bölümümüzün kullandığı hazır araçları kullanmak için saatler harcadım, ancak sistemin iç kısımlarına pek ilgim olmadığından benim için asıl odak noktası girişteki ölçümler ve çıkışta aşağıdakileri yapacak kendi botumu yazmaya karar verdim: İsteğim üzerine servise gidiyor, metrikleri kontrol ediyor ve bana mevcut durum hakkında kısa bir rapor gönderiyordu. Diğer isteğim üzerine bana son X saat içinde olup bitenlerin grafiklerini gönderecekti. Özel bir durum söz konusu olduğunda bana o anda bir şeyler olduğuna dair bildirim gönderiyordu. Bu yazımda ilk kısma yani istek üzerine metrik alma konusuna odaklanacağım. İş için yeni bir sanal ortama ihtiyacımız olacak. cd ~ virtualenv -p python3.8 up_env # crete a virtualenv source ~/up_env/bin/activate # activate the virtualenl Bağımlılıkları yükleyin: 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) işlevlerini içeren dosya (bunu sınıflarla uygulayabilirsiniz, ancak örnek kısa olduğu için onu modüllere bölmeyi planlamadım, ancak çok iş parçacıklı bir kitaplık, işlevlerin ayrı bir dosyaya taşınmasını gerektirir). İçe aktarma bağımlılıkları: Functions.py import numpy as np import multiprocessing from web3 import Web3 # add those libraries needed for your task Durumu kontrol etmek için bir fonksiyonun tanımlanması. Benim durumumda, önceden seçilmiş genel düğümler arasında döngü yapmayı, son bloklarını almayı, herhangi bir sapmayı filtrelemek için medyan değerini almayı ve ardından kendi düğümümüzü bu medyana göre kontrol etmeyi içeriyordu. Servis durumu kontrol fonksiyonu (kendinizinkiyle değiştirebilirsiniz): # 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 Botun bir sonraki önemli dosyası . Yukarıdaki dosyadan kitaplıkları ve işlevleri içe aktarıyoruz ve gerekli sabitleri ayarlıyoruz: uptime_bot.py'dir 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 Daha sonra botun arayüzünden komut verildiğinde çağrılacak bir fonksiyonu anlatalım. 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") Şimdi geriye kalan tek şey botun başlatıldığı ve işleyici fonksiyonunun bağlandığı kısmı eklemek: 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() Son olarak, aşağıdakileri kullanarak kodu ucuz bir VPS sunucusunda çalıştırabilirsiniz: source ~/up_env/bin/activate python uptime_bot.py Systemd birim dosyasını yapılandırdıktan sonra. Sonuç olarak botun çalışması şu şekilde görünecek. Her şey yolundaysa: Ve eğer gecikme çok büyük olursa, o zaman aşağıdaki gibi: Aşağıdaki makalelerde kalan iki görevin nasıl uygulanacağını anlatacağım: Talep üzerine son X saat içinde meydana gelen olayları gösteren grafikleri alın. Şu anda bir şeyin gerçekleştiğini ve işlem yapılması gerektiğini belirten bir uyarı alın. Projenin kaynak kodu GitHub mevcuttur. Bu eğitimi faydalı bulduysanız GitHub'da yıldız vermekten çekinmeyin, çok memnun olurum🙂 deposunda
