import json import time from web3 import Web3 from loguru import logger from typing import Dict, List CONFIG_PATH = "config.json" class InfuraMonitor: def __init__(self, infura_url: str): self.infura_url = infura_url self.last_block = None self.addresses = [] self.web3 = Web3(Web3.HTTPProvider(self.infura_url)) if self.web3.is_connected(): self.last_block = self.web3.eth.block_number logger.info(f"Connected to Infura. Last block: #{self.last_block}") else: raise Exception("Couldn't connect to Infura") def check_txs(self, start_block: int, end_block: int, address_names: Dict[str, str]): for block_num in range(start_block, end_block + 1): try: block = self.web3.eth.get_block(block_num, full_transactions=True) except Exception as e: logger.error(f"Error fetching block #{block_num}: {e}") continue for tx in block['transactions']: from_address = tx['from'] to_address = tx['to'] from_name = address_names.get(from_address, from_address) to_name = address_names.get(to_address, to_address) if from_address in self.addresses or to_address in self.addresses: logger.info(f"Transaction from {from_name} to {to_name} detected in block #{block_num}") logger.info(f"Transaction details: {tx}") def start_monitor(self, addresses: Dict[str, str]): self.addresses = list(addresses.keys()) while True: try: current_block = self.web3.eth.block_number logger.info(f"Current block: #{current_block}. Checking transactions from block #{self.last_block + 1} to #{current_block}") if current_block > self.last_block: self.check_txs(self.last_block + 1, current_block, addresses) self.last_block = current_block else: logger.info(f"No new blocks since last check. Last block: #{self.last_block}") except Exception as e: logger.error(f"Error during monitoring: {e}") time.sleep(15) def load_config(path: str) -> dict: try: with open(path, 'r', encoding='utf-8') as f: config = json.load(f) return config except FileNotFoundError: logger.error(f"Configuration file not found: {path}") raise except json.JSONDecodeError: logger.error(f"Error decoding JSON config: {path}") raise def main(): try: config = load_config(CONFIG_PATH) infura_url = config['infura_url'] addresses = { "0x482702745260Ffd69FC19943f70cFFE2caCd70e9": "$JENNER CONTRACT", "0xb41af5ce8c1b86e0204a0bc6625041376c70ba81": "DEV WALLET", "0xC152A863312F4AB4C5B6a52447abe8bFDD741aa2": "MARKETING WALLET", "0xedc3D54605d6d25cF405f214B56d63b7bCD80d1f": "LIQUIDITY WALLET", } monitor = InfuraMonitor(infura_url) monitor.start_monitor(addresses) except Exception as e: logger.error(f"Error in main execution: {e}") if __name__ == '__main__': main()