Here is transaction https://etherscan.io/tx/0x6465187a7bb43a6db42ee63e5f5cc30fb094393957a7f1ce6c08b5afddf3e0bc
I don't know why my script can not find ABI Transfer() function after decoded on Chainlink: LINK Token
address. The main problem is that script works correct, I tested on many transactions and appended 2 of them for example.
Perhaps problem is in Transfer() input params after decoding? But how to solve it?
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']}")
abi = json.loads(json.loads(requests.get(f"https://api.etherscan.io/api?module=contract&action=getabi&address={log['address']}&apikey={API_ETHERSCAN}").text)['result'])
contract = w3.eth.contract(log["address"], abi=abi)
event_abi = [a for a in contract.abi if a["type"] == "event"]
topic2abi = {event_abi_to_log_topic(_): _ for _ in event_abi}
log_ = {
'address': None, #Web3.toChecksumAddress(address),
'blockHash': None, #HexBytes(blockHash),
'blockNumber': None,
'data': log['data'],
'logIndex': None,
'topics': [HexBytes(_) for _ in log["topics"]],
'transactionHash': None, #HexBytes(transactionHash),
'transactionIndex': None
}
try:
event_abi = topic2abi[log['topics'][0]]
except KeyError as e:
exit('ABI event not found!')
data = get_event_data(w3.codec, event_abi, log_)['args']
print(f" {event_abi['name']} {data['value'] if 'value' in data else ''}")
Output:
tx_hash=0xff8db775b90935b1ade58182054c0be04f613ea23f9e1df73a6114a726e76237
1) log['address']=0xf21661D0D1d76d3ECb8e1B9F1c923DBfffAe4097
Transfer 95352033474036727055914
tx_hash=0x7bdfe6c2a9309773ddeafddc2624edc2b38f13ec257182576d95d8b5f5ea2cd1
1) log['address']=0xe28b3B32B6c345A34Ff64674606124Dd5Aceca30
Transfer 29000000000000000000
tx_hash=0x6465187a7bb43a6db42ee63e5f5cc30fb094393957a7f1ce6c08b5afddf3e0bc
1) log['address']=0x514910771AF9Ca656af840dff83E8264EcF986CA
ABI event not found!
I added this line to check the structure of event_abi
print("event_abi['name']",event_abi)
I was getting those results:
{'anonymous': False, 'inputs': [{'indexed': True, 'internalType': 'address', 'name': 'from', 'type': 'address'}, {'indexed': True, 'internalType': 'address', 'name': 'to', 'type': 'address'}, {'indexed': False, 'internalType': 'uint256', 'name': 'value', 'type': 'uint256'}], 'name': 'Transfer', 'type': 'event'}
{'anonymous': False, 'inputs': [{'indexed': True, 'internalType': 'address', 'name': 'from', 'type': 'address'}, {'indexed': True, 'internalType': 'address', 'name': 'to', 'type': 'address'}, {'indexed': False, 'internalType': 'uint256', 'name': 'value', 'type': 'uint256'}], 'name': 'Transfer', 'type': 'event'}
[{'anonymous': False, 'inputs': [{'indexed': True, 'name': 'from', 'type': 'address'}, {'indexed': True, 'name': 'to', 'type': 'address'}, {'indexed': False, 'name': 'value', 'type': 'uint256'}, {'indexed': False, 'name': 'data', 'type': 'bytes'}], 'name': 'Transfer', 'type': 'event'}, {'anonymous': False, 'inputs': [{'indexed': True, 'name': 'owner', 'type': 'address'}, {'indexed': True, 'name': 'spender', 'type': 'address'}, {'indexed': False, 'name': 'value', 'type': 'uint256'}], 'name': 'Approval', 'type': 'event'}]
first 2 results were dict
but the third one is a list
. I was first getting this error:
---> 69 print(f" {event_abi['name']} {data['value'] if 'value' in data else ''}")
TypeError: list indices must be integers or slices, not str
Because event_abi
can either be a dictionary or a list of dictionaries. To handle this variability, I had to adjust how to access the 'name' key depending on the type of event_abi. so I added this
try:
event_abi_entry = topic2abi[log['topics'][0]]
if isinstance(event_abi_entry, dict):
event_name = event_abi_entry.get('name', 'DefaultNmae')
data = get_event_data(w3.codec, event_abi_entry, log_)['args']
print(f" {event_name} {data.get('value', '')}")
elif isinstance(event_abi_entry, list):
for abi_entry in event_abi_entry:
event_name = abi_entry.get('name', 'DefaultName')
data = get_event_data(w3.codec, abi_entry, log_)['args']
print(f" {event_name} {data.get('value', '')}")
else:
print("Event ABI entry is not in the expected format.")
except KeyError as e:
exit('ABI event not found!')
and it does not throw KeyError. proof of work:
This is the full code:
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 = "YOUR-API-KEY-HERE" # 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)
# print(tx_receipt)
for i, log in enumerate(tx_receipt['logs']):
print(f" {i+1}) log['address']={log['address']}")
abi = json.loads(json.loads(requests.get(f"https://api.etherscan.io/api?module=contract&action=getabi&address={log['address']}&apikey={API_ETHERSCAN}").text)['result'])
contract = w3.eth.contract(log["address"], abi=abi)
event_abi = [a for a in contract.abi if a["type"] == "event"]
topic2abi = {event_abi_to_log_topic(_): _ for _ in event_abi}
log_ = {
'address': None, #Web3.toChecksumAddress(address),
'blockHash': None, #HexBytes(blockHash),
'blockNumber': None,
'data': log['data'],
'logIndex': None,
'topics': [HexBytes(_) for _ in log["topics"]],
'transactionHash': None, #HexBytes(transactionHash),
'transactionIndex': None
}
try:
event_abi_entry = topic2abi[log['topics'][0]]
if isinstance(event_abi_entry, dict):
event_name = event_abi_entry.get('name', 'DefaultNmae')
data = get_event_data(w3.codec, event_abi_entry, log_)['args']
print(f" {event_name} {data.get('value', '')}")
elif isinstance(event_abi_entry, list):
for abi_entry in event_abi_entry:
event_name = abi_entry.get('name', 'DefaultName')
data = get_event_data(w3.codec, abi_entry, log_)['args']
print(f" {event_name} {data.get('value', '')}")
else:
print("Event ABI entry is not in the expected format.")
except KeyError as e:
exit('ABI event not found!')
print()
print("loop completed")
Here are the current versions of the software and libraries in use:
Python 3.11.6
Name: web3
Version: 6.15.1
Name: eth-utils
Version: 4.0.0
Name: hexbytes
Version: 0.3.1
I also test the app on Kaggle and here is the successful result: