关于区块链钱包私钥碰撞概率计算分析

from web3 import Web3
from loguru import logger
import binascii, os
import threading
import time
def generate_private_key():
    return binascii.b2a_hex(os.urandom(32)).decode('utf8')
# 初始化 Web3
# 这里用alechemy提供的接口来连接Ethereum 节点
http_provider = Web3.HTTPProvider('https://eth-mainnet.g.alchemy.com/v2/YOUR-PROJECT-ID')
web3 = Web3(http_provider)
# ABI
abi = """
[
    {
        "constant": true,
        "name": "balanceOf",
        "outputs": [{"name": "", "type": "uint256", "value": "2000000000"}],
        "inputs": [{"name": "_owner", "type": "address"}],
        "anonymous": false,
        "type": "function"
    }
]
"""
# 智能合同地址和对应的单位
contracts_info = {
    'USDT': {
        'address': "0xdAC17F958D2ee523a2206206994597C13D831ec7",
        'decimal': 10**6
    },
    'USDC': {
        'address': "0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48",
        'decimal': 10**6
    },
    'Dfyn': {
        'address': "0x9695e0114e12c0d3a3636fab5a18e6b737529023",
        'decimal': 10**18
    },
    'BigTime': {
        'address': "0x64bc2ca1be492be7185faa2c8835d9b824c8a194",
        'decimal': 10**18
    },
    'MANA': {
        'address': "0x0F5D2fB29fb7d3CFeE444a200298f468908cC942",
        'decimal': 10**18
    }
}
# 存有所有的合约实例
contracts = {}
# 创建所有的合约实例
for token, contract_info in contracts_info.items():
    contracts[token] = {
        'contract': web3.eth.contract(address=web3.to_checksum_address(contract_info['address']), abi=abi),
        'decimal': contract_info['decimal']
    }
try:
    latest_block = web3.eth.block_number
    logger.success(f'Web3 成功连接！最新的区块编号是: {latest_block}')
except Exception as e:
    logger.error('Web3连接失败!')
    logger.error(str(e))
# 创建一个线程锁
lock = threading.Lock()
def task():
    while True:
        try:
            myPrivateKey = generate_private_key()
            myAccount = web3.eth.account.from_key(myPrivateKey)
            wallet_address = web3.to_checksum_address(myAccount.address)
            # 请求会继续尝试直到成功
            while True:
                try:
                    # 查询ETH余额
                    balance_wei = web3.eth.get_balance(wallet_address)
                    balance_eth = web3.from_wei(balance_wei, 'ether')
                    break  # 如果请求成功则跳出循环
                except Exception as e:
                    logger.error('Failed to query ETH balance. Retrying...')
                    time.sleep(0.5)  # 暂停0.5秒再重试
            # 查询 Token 余额
            balances = dict()
            non_zero_balance = False
            for token, contract_info in contracts.items():
                contract = contract_info['contract']
                decimal = contract_info['decimal']
                # 请求会继续尝试直到成功
                while True:
                    try:
                        balance = contract.functions.balanceOf(wallet_address).call() / decimal
                        break  # 如果请求成功则跳出循环
                    except Exception as e:
                        logger.error(f'Failed to query {token} balance. Retrying...')
                        time.sleep(0.5)  # 暂停0.5秒再重试
                balances[token] = balance
                # 检查余额是否大于0
                if balance > 0:
                    non_zero_balance = True
            # 打印所有余额
            logger.info(f'PrivateKey:{myPrivateKey},ETH:{balance_eth},' + ','.join([f'{token}:{balance}' for token, balance in balances.items()]))
            # 如果余额大于0，加锁向文件追加数据
            if non_zero_balance:
                with lock:
                    logger.success(f'发现非空钱包,PrivateKey:{myPrivateKey},ETH:{balance_eth},' + ','.join([f'{token}:{balance}' for token, balance in balances.items()]))
                    with open('key.txt', 'a') as f:
                        f.write(f'PrivateKey:{myPrivateKey},ETH:{balance_eth},' + ','.join([f'{token}:{balance}' for token, balance in balances.items()]) + '\n')
        except Exception as e:
            logger.error('Error occurred. Retrying...')
            continue
def main():
    # 线程数量
    logger.info('请输入线程数:')
    num_threads = int(input())
    for _ in range(num_threads):
        t = threading.Thread(target=task)
        t.start()
if __name__ == '__main__':
    main()