You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

以太坊私钥转地址查余额脚本条件判断问题排查求助

你的以太坊余额扫描脚本:条件判断及逻辑错误排查

让我一步步帮你找出脚本里的条件判断和相关逻辑错误——这些问题直接导致你的余额检测完全失效:

1. 函数参数与私钥转换的核心错误

你的scanether函数定义的参数是balance,但它实际应该接收单个私钥字符串。更严重的是,你在转换私钥时用了全局变量INPUTFILE(整个输入文件的路径),而不是传入的私钥,这根本不可能生成正确的地址。

修复示例:

def scanether(privkey):
    privkey = privkey.strip()  # 去掉换行符和多余空格
    try:
        # 正确转换私钥:先转成bytes,再生成地址
        priv_bytes = binascii.unhexlify(privkey)
        eth_address = ethereum.utils.privtoaddr(priv_bytes)
        eth_address_hex = binascii.hexlify(eth_address).decode("utf-8")
        # 后续API请求逻辑...

2. 条件判断的逻辑完全混乱

原代码里的if ('result' != 0) in r.eth_balance:有两个致命问题:

  • 变量r根本没定义!你之前的API请求结果存在eth_balance变量里,不是r
  • 逻辑完全错误:API返回的result是字符串类型的wei数值,你需要先转成整数再判断是否大于0,而不是用奇怪的in语法

修复示例:

# 修正API URL:把&改成普通&(HTML转义字符会导致API请求失败)
response = requests.get(f"https://api.etherscan.io/api?module=account&action=balance&address=0x{eth_address_hex}&tag=latest&apikey=APIKEYHERE")
data = response.json()

# 先判断API请求是否成功
if data["status"] != "1":
    print(f"API Error for address 0x{eth_address_hex}: {data['message']}")
    time.sleep(2)
    return

eth_balance_str = data["result"]
eth_balance = int(eth_balance_str)

# 正确的余额判断:wei数大于0
if eth_balance > 0:
    print(f"[*] Address with balance found: 0x{eth_address_hex} | Privkey: {privkey}")
    # 用with语句安全写入,避免资源泄漏
    with open(OUTPUTFILE, 'a') as fHandle:
        fHandle.write(f"0x{eth_address_hex},{privkey},{eth_balance_str}\n")
else:
    # 转换为ETH单位显示(1 ETH = 10^18 wei)
    print(f"balance: {eth_balance / 10**18} ETH | address: 0x{eth_address_hex} | privkey: {privkey}")
    time.sleep(1)

3. 未定义变量问题

原代码里的priv、privkey变量没有定义,你需要直接使用传入的私钥参数(已经用strip()清理过的privkey)。

4. 多进程输入的预处理问题

用readlines()读取的每行私钥会自带换行符,转换私钥时会出错,必须先清理:

# 读取私钥时过滤空行并清理换行符
addresses = [line.strip() for line in open(INPUTFILE, "r").readlines() if line.strip()]

其他潜在优化点

  • 多进程直接追加文件可能出现写入竞争(内容错乱),建议每个进程用独立临时文件,扫描结束后合并,或者使用文件锁
  • 增加API请求的超时设置,避免长时间等待:requests.get(..., timeout=10)
  • Etherscan有请求频次限制,多线程设为200可能会被限流,建议降低到10-20,或者添加动态休眠

修正后的完整脚本参考

#!/usr/bin/python
import sys, os, argparse, requests, ethereum, binascii, time
from multiprocessing import Pool

def scanether(privkey):
    privkey = privkey.strip()
    if not privkey:
        return
    try:
        priv_bytes = binascii.unhexlify(privkey)
        eth_address = ethereum.utils.privtoaddr(priv_bytes)
        eth_address_hex = binascii.hexlify(eth_address).decode("utf-8")
        
        # 修正API URL和错误处理
        api_url = f"https://api.etherscan.io/api?module=account&action=balance&address=0x{eth_address_hex}&tag=latest&apikey=APIKEYHERE"
        response = requests.get(api_url, timeout=10)
        data = response.json()

        if data["status"] != "1":
            print(f"[!] API Error: {data['message']} for address 0x{eth_address_hex}")
            time.sleep(2)
            return

        eth_balance_str = data["result"]
        eth_balance = int(eth_balance_str)

        if eth_balance > 0:
            print(f"[*] Found balance! Address: 0x{eth_address_hex} | Privkey: {privkey} | Balance: {eth_balance / 10**18} ETH")
            with open(OUTPUTFILE, 'a') as fHandle:
                fHandle.write(f"0x{eth_address_hex},{privkey},{eth_balance_str}\n")
        else:
            print(f"[+] No balance. Address: 0x{eth_address_hex} | Privkey: {privkey}")
            time.sleep(1)
    except Exception as e:
        print(f"[!] Error processing privkey {privkey}: {str(e)}")
        return

if __name__ == '__main__':
    print("""# Ethereum Balance Scanner""")
    parser = argparse.ArgumentParser()
    parser.add_argument('-i', '--inputfile', default='input.txt', help='input file with private keys')
    parser.add_argument('-o', '--outputfile', default='output.txt', help='output file for results')
    parser.add_argument('-t', '--threads', default=10, help='number of threads (reduce if rate limited)')
    args = parser.parse_args()

    INPUTFILE = args.inputfile
    OUTPUTFILE = args.outputfile
    MAXPROCESSES = int(args.threads)

    try:
        addresses = [line.strip() for line in open(INPUTFILE, "r").readlines() if line.strip()]
    except FileNotFoundError as e:
        print(f"Error: {e}")
        sys.exit(e.errno)

    print(f"Starting scan with {MAXPROCESSES} threads...")
    pool = Pool(processes=MAXPROCESSES)
    pool.map(scanether, addresses)
    print("Scan finished. Check output file for results.")

内容的提问来源于stack exchange,提问作者Awesim

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.13 08:13:55