使用多线程和代理查询BNB余额并自动保存进度

from web3 import Web3, HTTPProvider
import requests
import socks
import socket
import time
import threading
import datetime

# 设置Socks5代理
socks.set_default_proxy(socks.SOCKS5, '127.0.0.1', 10808)
socket.socket = socks.socksocket

# 初始化Web3实例使用HTTPProvider连接到Binance Smart Chain节点，并设置代理
w3 = Web3(HTTPProvider('https://bsc-dataseed.binance.org/', request_kwargs={'proxies': {'http': 'socks5://127.0.0.1:10808', 'https': 'socks5://127.0.0.1:10808'}}))

# 添加备用API密钥和备用节点
api_keys = ['B27BYKSACHD69Z2KCD14TJWVWZQWNNKVIB', '5FM54MS2DWM3FTFD8AMPXJN16U1HNWVWJN', 'G1VQAQAJD2J6G1PCFSY6U6E7KGEEDFEW74', 'W9RBNCDTSIJH1KZ1759I6AMQ5UBW6XMQYD']
alt_providers = [
    'https://bsc-dataseed1.binance.org/',
    'https://bsc-dataseed2.binance.org/',
    'https://bsc-dataseed3.binance.org/'
]

# 读取钱包.txt文件
with open('钱包.txt') as file:
    lines = file.readlines()

# 创建已查询.txt文件并定义查询函数
output_lock = threading.Lock()
def query(address):
    for key in api_keys:
        try:
            url = f'https://api.bscscan.com/api?module=account&action=balance&address={address}&tag=latest&apikey={key}'
            response = requests.get(url)
            if 'Max rate limit' in response.json()['result']:
                print(f'达到API速率限制，等待5秒后使用备用API密钥查询：{address}')
                time.sleep(5)
            else:
                balance = int(response.json()['result']) / 10**18

                # 只保存余额大于0的地址
                if balance > 0:
                    with output_lock:
                        with open('已查询.txt', 'a') as output_file:
                            output_file.write(f'{balance},{address}
')

                break
        except Exception as e:
            print(f'查询地址 {address} 时发生了错误：{e}')
            continue

    # 限制查询速率以避免节点被封禁
    time.sleep(0.5)

# 使用多线程查询BNB余额
results = []
threads = []
batch_size = 10 # 每批处理的地址数量
num_threads = 5 # 并发查询线程数量

# 将地址列表分成多批处理
batches = [lines[i:i + batch_size] for i in range(0, len(lines), batch_size)]

# 增加每5分钟保存一次最新一条查询地址为存档.txt
def save_checkpoint(address):
    now = datetime.datetime.now()
    if now.minute % 5 == 0:
        with open('存档.txt', 'w') as checkpoint_file:
            checkpoint_file.write(address)

# 如果程序错误退出，从存档地址开始扫描
start_address = ''
try:
    with open('存档.txt', 'r') as checkpoint_file:
        start_address = checkpoint_file.read().strip()
except:
    pass

# 处理每批地址
for batch in batches:
    threads.clear() # 清空查询线程列表
    # 如果有存档地址，跳过该地址之前的所有地址
    if start_address:
        batch = [line for line in batch if line.split(',')[0] >= start_address]
    # 启动新的查询线程
    for line in batch:
        address = line.split(',')[0]
        print(f'启动查询线程：{address}')
        thread = threading.Thread(target=query, args=(address,))
        threads.append(thread)
        thread.start()
        if len(threads) >= num_threads:
            break
    # 等待本批查询完成
    for thread in threads:
        thread.join()
        # 保存最新查询地址为存档
        save_checkpoint(address)
    threads.clear() # 清空查询线程列表