여러분, 안녕하세요! 몇 년 동안 저는 Python으로 여러 가지 작은 일상적인 작업(무언가 알림, 서비스 가동 시간 확인, 텔레그램 채널 및 채팅에서 흥미로운 콘텐츠 전달 등)을 처리하는 다양한 "보조" 텔레그램 봇을 작성해 왔습니다. 휴대폰이 항상 가까이에 있기 때문에 편리하고, 노트북을 열지 않고도 서버의 문제를 해결할 수 있다는 점이 특별한 즐거움을 선사합니다. 일반적으로 HackerNoon 독자들과 공유하고 싶은 다양한 소규모 프로젝트 템플릿이 많이 축적되어 있습니다. 예제는 "있는 그대로" 애플리케이션이라는 측면에서 틈새 시장에 속할 수 있지만 몇 줄의 코드를 자신만의 코드로 변경하여 대부분의 코드를 재사용할 수 있는 부분을 표시하겠습니다. 귀하의 프로젝트를 위한 개발. 저는 며칠 전에 이 특정 프로젝트를 완료했는데, 이미 많은 이점을 얻었습니다. 저는 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) 함수가 포함된 파일 (클래스로 구현하면 되지만 예제가 짧기 때문에 모듈로 나눌 생각은 없었지만 멀티스레딩 라이브러리는 함수를 별도의 파일로 옮겨야 합니다). 종속성 가져오기: function.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 systemd 단위 파일을 구성한 후. 결과적으로 봇의 작업은 다음과 같습니다. 모든 것이 괜찮다면: 지연이 너무 커지면 다음과 같이 됩니다. 다음 문서에서는 나머지 두 가지 작업을 구현하는 방법을 설명합니다. 요청 시 지난 X시간 동안 발생한 이벤트를 보여주는 그래프를 검색합니다. 현재 어떤 일이 발생하고 있으며 조치가 필요함을 나타내는 경고를 받습니다. 프로젝트의 소스 코드는 GitHub 에서 사용할 수 있습니다. 이 튜토리얼이 도움이 되었다면 GitHub에서 별표를 남겨주시면 감사하겠습니다🙂 저장소
