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_getfilterlogsfilter_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.tokensevent_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进行转化，最后完成所有筛选事件之后注销过滤器