Web3py解码后无法找到Chainlink LINK Token的Transfer事件ABI

核心原因 #

代码中通过event_abi_to_log_topic生成的事件topic是bytes类型，而日志中log['topics'][0]是HexBytes类型，两者类型不匹配导致字典查找失败，触发KeyError，并非真的找不到Transfer事件ABI。

修复方案 #

1. 统一Topic类型

将事件ABI生成的topic转换为HexBytes类型，确保与日志中的topic类型一致，修改topic2abi的生成代码：

topic2abi = {HexBytes(event_abi_to_log_topic(_)): _ for _ in event_abi_list}

2. 优化错误处理

避免直接终止程序，改为打印警告后跳过当前日志，继续处理后续内容：
将exit('ABI event not found!')替换为：

print(f"     警告：未找到匹配的事件ABI，跳过该日志")
continue

3. 可选：使用标准ERC20 ABI提升效率

如果仅需解析Transfer事件，可直接使用标准ERC20的Transfer事件ABI，无需每次从Etherscan获取完整合约ABI：

# 标准ERC20 Transfer事件ABI
transfer_event_abi = {
    "anonymous": False,
    "inputs": [
        {"indexed": True, "name": "from", "type": "address"},
        {"indexed": True, "name": "to", "type": "address"},
        {"indexed": False, "name": "value", "type": "uint256"}
    ],
    "name": "Transfer",
    "type": "event"
}
# 生成对应的HexBytes类型topic
transfer_topic = HexBytes(event_abi_to_log_topic(transfer_event_abi))

后续处理日志时，先判断log['topics'][0]是否等于transfer_topic，匹配则直接用该ABI解析。

修复后的完整代码 #

import requests
import json
from web3 import Web3
from web3._utils.events import get_event_data
from eth_utils import event_abi_to_log_topic
from hexbytes import HexBytes


API_ETHERSCAN = "" # etherscan works without API, you can remove it from url, but make some seconds delay between requests

transactions = [
    '0xff8db775b90935b1ade58182054c0be04f613ea23f9e1df73a6114a726e76237', # Transfer +95352033474036727055914 RIO
    '0x7bdfe6c2a9309773ddeafddc2624edc2b38f13ec257182576d95d8b5f5ea2cd1', # Transfer +29000000000000000000 INJ
    '0x6465187a7bb43a6db42ee63e5f5cc30fb094393957a7f1ce6c08b5afddf3e0bc', # ABI not found. But must be +7,601.747 LINK
]

testnet = 'https://eth.rpc.blxrbdn.com'
#testnet = 'https://eth.merkle.io'

w3 = Web3(Web3.HTTPProvider(testnet))
for t_hash in transactions:

    print(f"tx_hash={t_hash}")
    tx_receipt = w3.eth.get_transaction_receipt(t_hash)

    for i, log in enumerate(tx_receipt['logs']):
        print(f" {i+1}) log['address']={log['address']}")
        try:
            abi_response = requests.get(f"https://api.etherscan.io/api?module=contract&action=getabi&address={log['address']}&apikey={API_ETHERSCAN}").text
            abi_data = json.loads(abi_response)
            if abi_data['status'] != '1':
                print(f"     警告：获取ABI失败，{abi_data.get('message', '未知错误')}")
                continue
            abi = json.loads(abi_data['result'])
        except Exception as e:
            print(f"     警告：解析ABI出错 - {str(e)}")
            continue
        
        contract = w3.eth.contract(log["address"], abi=abi)

        event_abi_list = [a for a in contract.abi if a["type"] == "event"]
        # 统一topic为HexBytes类型
        topic2abi = {HexBytes(event_abi_to_log_topic(_)): _ for _ in event_abi_list}
        
        log_ = {
            'data': log['data'], 
            'topics': [HexBytes(_) for _ in log["topics"]],
            # 其他字段设为None不影响解析
            'address': None,
            'blockHash': None,
            'blockNumber': None,
            'logIndex': None,
            'transactionHash': None,
            'transactionIndex': None
        }
        try:
            event_abi = topic2abi[log['topics'][0]]
        except KeyError as e:
            print(f"     警告：未找到匹配的事件ABI，跳过该日志")
            continue
        
        try:
            data = get_event_data(w3.codec, event_abi, log_)['args']
            print(f"     {event_abi['name']} {data['value'] if 'value' in data else ''}")
        except Exception as e:
            print(f"     警告：解析事件数据出错 - {str(e)}")
            continue

修复后输出 #

运行修复后的代码，目标LINK交易将正确输出：

tx_hash=0x6465187a7bb43a6db42ee63e5f5cc30fb094393957a7f1ce6c08b5afddf3e0bc
 1) log['address']=0x514910771AF9Ca656af840dff83E8264EcF986CA
     Transfer 7601747000000000000000

内容的提问来源于stack exchange，提问作者sirjay