ethereum-etl学习2

ethereum-etl学习2

> ethereumetl export_token_transfers --start-block 0 --end-block 500000 \
--provider-uri file://$HOME/Library/Ethereum/geth.ipc --output token_transfers.csv

function export_token_transfers

def export_token_transfers(start_block, end_block, batch_size, output, max_workers, provider_uri, tokens):
    """Exports ERC20/ERC721 transfers."""
    job = ExportTokenTransfersJob(
        start_block=start_block,
        end_block=end_block,
        batch_size=batch_size,
        web3=ThreadLocalProxy(lambda: build_web3(get_provider_from_uri(provider_uri))),
        item_exporter=token_transfers_item_exporter(output),
        max_workers=max_workers,
        tokens=tokens)
    job.run()

功能

​ 获取ERC20/ERC721的交易信息（协议内容可以参照https://github.com/ethereum/EIPs/tree/master/EIPS）

主体

1.构建ExportTokenTransfersJob对象

2.调用run方法

---

class ExportTokenTransfersJob

class ExportTokenTransfersJob(BaseJob):

构造（部分）

web3 获取web3的连接对象

batch_work_executor 构造批处理的执行者

item_exporter 设置挖掘匹配的字段，以及存储文件

receipt_log_mapper 实现交易的log对象和json的转换解析

transaction_mapper 实现transaction对象和json的转换解析

token_transfer_extractor 一个EthTokenTransferExtractor对象，实现最后获取transaction的log解码为交易内容

run()方法调用

def _start(self):
    self.item_exporter.open()

​ 准备存储block和transaction的文件

def _export(self):
    self.batch_work_executor.execute(
        range(self.start_block, self.end_block + 1),
        self._export_batch,
        total_items=self.end_block - self.start_block + 1
    )

​ 调用batch_work_executor的execute方法

​ _export_batch批次获取以太坊的块中与TRANSFER事件匹配的TOPIC–‘0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef’

def _end(self):
    self.batch_work_executor.shutdown()
    self.item_exporter.close()

​ 释放batch_work_executor对象，以及关闭存储block和transaction信息的文件

_export_batch方法

def _export_batch(self, block_number_batch):
    assert len(block_number_batch) > 0
    # https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_getfilterlogs
    filter_params = {
        'fromBlock': block_number_batch[0],
        'toBlock': block_number_batch[-1],
        'topics': [TRANSFER_EVENT_TOPIC]
    }

    if self.tokens is not None and len(self.tokens) > 0:
        filter_params['address'] = self.tokens

    event_filter = self.web3.eth.filter(filter_params)
    events = event_filter.get_all_entries()
    for event in events:
        log = self.receipt_log_mapper.web3_dict_to_receipt_log(event)
        token_transfer = self.token_transfer_extractor.extract_transfer_from_log(log)
        if token_transfer is not None:
          self.item_exporter.export_item(self.token_transfer_mapper.
             token_transfer_to_dict(token_transfer))

    self.web3.eth.uninstallFilter(event_filter.filter_id)

​ 定义筛选的条件，对获取内容进行筛选，让对于log和transfer进行转化，最后完成所有筛选事件之后注销过滤器