transfers.py

from web3 import Web3
import json
from threading import Thread
from queue import Queue

# Connect to an Ethereum node
infura_url = "https://mainnet.infura.io/v3/5708c51bf624479984fd7b28fc4c5f7c"
web3 = Web3(Web3.HTTPProvider(infura_url))

# Ensure connection is established
if not web3.is_connected():
    raise Exception("Failed to connect to Ethereum node")

# Minimal ABI for ERC-20 name, symbol, and decimals
erc20_abi = json.loads('[{"constant":true,"inputs":[],"name":"name","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"symbol","outputs":[{"name":"","type":"string"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"decimals","outputs":[{"name":"","type":"uint8"}],"payable":false,"stateMutability":"view","type":"function"}]')

# Minimal ABI for ERC-165's supportsInterface function
erc165_abi = json.loads('[{"constant":true,"inputs":[{"name":"interfaceID","type":"bytes4"}],"name":"supportsInterface","outputs":[{"name":"","type":"bool"}],"payable":false,"stateMutability":"view","type":"function"}]')

# ERC-721 interface ID
erc721_interface_id = "0x80ac58cd"

# The signature hash of the Transfer events (ERC-20, ERC-721)
transfer_hash = web3.keccak(text="Transfer(address,address,uint256)").hex()

# The signature hash of the TransferSingle, TransferBatch events (ERC-1155)
transfer_single_hash = web3.keccak(text="TransferSingle(address,address,address,uint256,uint256)").hex()
transfer_batch_hash  = web3.keccak(text="TransferBatch(address,address,address,uint256[],uint256[])").hex()


def is_721(token_address):
    # Check if the contract supports ERC-721
    try:
        contract = web3.eth.contract(address=token_address, abi=erc165_abi)
        return contract.functions.supportsInterface(erc721_interface_id).call()
    except Exception as e:
        return False


def add_token_info(token_address, tx):
    # Call name and symbol functions
    contract = web3.eth.contract(address=token_address, abi=erc20_abi)
    try:
        tx['token_name']   = contract.functions.name().call()
        tx['token_symbol'] = contract.functions.symbol().call()
    except Exception as e:
        print(f"Failed to get Name/Symbol for {token_address}")

    # No decimals on ERC721 and many non-standard tokens
    try:
        tx['token_decimals'] = contract.functions.decimals().call()
    except Exception as e:
        pass

    return tx


def decode_transfer_batch_data(ids_data, values_data):
    # Convert the hex strings to integers
    ids = [int(ids_data[i:i+64], 16) for i in range(0, len(ids_data), 64)]
    values = [int(values_data[i:i+64], 16) for i in range(0, len(values_data), 64)]

    return ids, values


def process_transaction(transaction):
    tx = {
      "hash":   transaction.hash.hex(),
      "to":     transaction.to,
      "from":   transaction['from'],
      "value":  web3.from_wei(transaction.value, 'ether'),
    }

    # Check for native ETH transfer
    if transaction.value > 0:
        tx['type']  = 'Native ETH Transfer'
        return tx

    receipt = web3.eth.get_transaction_receipt(transaction.hash)

    # Check each log for the Transfer event
    for log in receipt.logs:
        # The token address that emitted the event
        token_address = log.address

        # Convert binary topic to hexadecimal string
        topic_hex = log.topics[0].hex()

        # Convert hex address to checksum address
        def to_checksum_address(topic):
            return web3.to_checksum_address('0x' + topic.hex()[-40:])

        # Convert to integer from hex
        def to_int(topic):
            value_hex = topic.hex()
            if value_hex == '0x':
                return 0
            return int(value_hex, 16)

        # Check if ERC20/721
        if topic_hex == transfer_hash:
            tx['type']       = 'ERC-20 Transfer'
            tx['from']       = to_checksum_address(log.topics[1])
            tx['to_address'] = to_checksum_address(log.topics[2])
            tx['value']      = to_int(log.data)

            # Check if 721
            if is_721(token_address):
                tx['type']     = 'ERC-721 Transfer'
                tx['token_id'] = to_int(log.topics[3])

        # ERC1155
        elif topic_hex == transfer_single_hash:
            tx['operator_address'] = to_checksum_address(log.topics[0])
            tx['from_address']     = to_checksum_address(log.topics[1])
            tx['to_address']       = to_checksum_address(log.topics[2])
            tx['token_id']         = to_int(log.topics[3])
            try:
              tx['value']            = to_int(log.topics[4])
            except:
              print('wut', tx)

        elif topic_hex == transfer_batch_hash:
            tx['operator_address'] = to_checksum_address(log.topics[0])
            tx['from_address']     = to_checksum_address(log.topics[1])
            tx['to_address']       = to_checksum_address(log.topics[2])

            # Extract token IDs from the fourth topic and values from the fifth topic
            token_ids_hex = log.topics[3].hex()
            values_hex = log.topics[4].hex()

            # Decode token IDs and values from hex
            ids, values  = decode_transfer_batch_data(token_ids_hex, values_hex)
            tx['ids']    = ids
            tx['values'] = values

        return add_token_info(token_address, tx)

if __name__ == "__main__":
    num_threads = 100

    # The block number to inspect
    block_number = 18808322  # Replace with the desired block number
    block = web3.eth.get_block(block_number, full_transactions=True)
    print(f"Processing block {block_number} with transactions: {len(block.transactions)}")

    def worker(q, results):
        while True:
            transaction = q.get()
            if transaction is None:
                break
            results.append(process_transaction(transaction))
            q.task_done()

    # Queue for transactions
    transaction_queue = Queue()
    results = []

    # Start thread workers
    threads = [Thread(target=worker, args=(transaction_queue, results)) for _ in range(num_threads)]
    for t in threads:
        t.start()

    # Add transactions to queue
    for transaction in block.transactions:
        transaction_queue.put(transaction)

    # Block until all tasks are done
    transaction_queue.join()

    # Stop workers
    for _ in range(num_threads):
        transaction_queue.put(None)
    for t in threads:
        t.join()

    print(results)