added kafka exporter

.gitignore

@@ -48,3 +48,8 @@ coverage.xml
 .venv
 venv/
 ENV/
+
+.idea/
+
+*.json
+*.txt

Dockerfile

@@ -1,11 +1,17 @@
 FROM python:3.6-alpine
+
 MAINTAINER Omidiora Samuel <[email protected]>
+
 ENV PROJECT_DIR=bitcoin-etl
 
 RUN mkdir /$PROJECT_DIR
+
 WORKDIR /$PROJECT_DIR
+
 COPY . .
+
 RUN apk add --no-cache gcc musl-dev  #for C libraries: <limits.h> <stdio.h>
-RUN pip install --upgrade pip && pip install -e /$PROJECT_DIR/
+
+RUN pip install --upgrade pip && pip install -e /$PROJECT_DIR/ && pip install dist/bitcoin-etl-1.5.2.tar.gz && pip install kafka-python==2.0.2
 
 ENTRYPOINT ["python", "bitcoinetl"]

Dockerfile_with_streaming

@@ -1,15 +1,22 @@
 FROM python:3.6
+
 MAINTAINER Evgeny Medvedev <[email protected]>
+
 ENV PROJECT_DIR=bitcoin-etl
 
 RUN mkdir /$PROJECT_DIR
+
 WORKDIR /$PROJECT_DIR
+
 COPY . .
-RUN pip install --upgrade pip && pip install -e /$PROJECT_DIR/[streaming]
+
+RUN pip install --upgrade pip && pip install dist/bitcoin-etl-1.5.2.tar.gz && pip install -e /$PROJECT_DIR/[streaming]
 
 # Add Tini
 ENV TINI_VERSION v0.18.0
+
 ADD https://github.com/krallin/tini/releases/download/${TINI_VERSION}/tini /tini
+
 RUN chmod +x /tini
 
 ENTRYPOINT ["/tini", "--", "python", "bitcoinetl"]

README_CONTAINER_REGISTRY.md

@@ -0,0 +1,5 @@
+BITCOINETL_STREAMING_VERSION=1.4-streaming
+  docker build -t merklescience/bitcoin-etl:${BITCOINETL_STREAMING_VERSION} -f Dockerfile_with_streaming .
+  docker tag merklescience/bitcoin-etl:${BITCOINETL_STREAMING_VERSION} us.gcr.io/staging-btc-etl/merklescience/bitcoin-etl:${BITCOINETL_STREAMING_VERSION}
+  docker push us.gcr.io/staging-btc-etl/merklescience/bitcoin-etl:${BITCOINETL_STREAMING_VERSION}
+

bitcoinetl/cli/export_all.py

@@ -25,7 +25,7 @@
 import re
 
 from datetime import datetime, timedelta
-from bitcoinetl.enumeration.chain import Chain
+from bitcoinetl.enumeration.chain import Chain, CoinPriceType
 from bitcoinetl.jobs.export_all import export_all as do_export_all
 from bitcoinetl.service.btc_block_range_service import BtcBlockRangeService
 from bitcoinetl.rpc.bitcoin_rpc import BitcoinRpc
@@ -99,4 +99,4 @@ def get_partitions(start, end, partition_batch_size, provider_uri):
 def export_all(start, end, partition_batch_size, provider_uri, output_dir, max_workers, export_batch_size, chain, enrich):
     """Exports all data for a range of blocks."""
     do_export_all(chain, get_partitions(start, end, partition_batch_size, provider_uri),
-                  output_dir, provider_uri, max_workers, export_batch_size, enrich)
+                output_dir, provider_uri, max_workers, export_batch_size, enrich)

bitcoinetl/cli/export_blocks_and_transactions.py

@@ -23,7 +23,7 @@
 
 import click
 
-from bitcoinetl.enumeration.chain import Chain
+from bitcoinetl.enumeration.chain import Chain, CoinPriceType
 from bitcoinetl.jobs.export_blocks_job import ExportBlocksJob
 from bitcoinetl.jobs.exporters.blocks_and_transactions_item_exporter import blocks_and_transactions_item_exporter
 from bitcoinetl.rpc.bitcoin_rpc import BitcoinRpc
@@ -49,7 +49,7 @@
 @click.option('-c', '--chain', default=Chain.BITCOIN, type=click.Choice(Chain.ALL),
               help='The type of chain')
 def export_blocks_and_transactions(start_block, end_block, batch_size, provider_uri,
-                                   max_workers, blocks_output, transactions_output, chain):
+                                   max_workers, blocks_output, transactions_output, chain,):
     """Export blocks and transactions."""
     if blocks_output is None and transactions_output is None:
         raise ValueError('Either --blocks-output or --transactions-output options must be provided')
@@ -63,5 +63,6 @@ def export_blocks_and_transactions(start_block, end_block, batch_size, provider_
         item_exporter=blocks_and_transactions_item_exporter(blocks_output, transactions_output),
         chain=chain,
         export_blocks=blocks_output is not None,
-        export_transactions=transactions_output is not None)
+        export_transactions=transactions_output is not None,
+    )
     job.run()

bitcoinetl/cli/stream.py

@@ -21,7 +21,7 @@
 # SOFTWARE.
 
 import click
-
+import json
 from bitcoinetl.enumeration.chain import Chain
 from bitcoinetl.rpc.bitcoin_rpc import BitcoinRpc
 
@@ -41,16 +41,17 @@
 @click.option('-o', '--output', type=str,
               help='Google PubSub topic path e.g. projects/your-project/topics/bitcoin_blockchain. '
                    'If not specified will print to console.')
+@click.option('--topic-mapping', default=None, type=str, help="Topic Mapping should be json like {\"block\": \"producer-litcoin-blocks-hot\",\"transaction\": \"producer-litcoin-transactions-hot\"}")
 @click.option('-s', '--start-block', default=None, type=int, help='Start block.')
 @click.option('-c', '--chain', default=Chain.BITCOIN, type=click.Choice(Chain.ALL), help='The type of chain.')
-@click.option('--period-seconds', default=10, type=int, help='How many seconds to sleep between syncs.')
-@click.option('-b', '--batch-size', default=2, type=int, help='How many blocks to batch in single request.')
+@click.option('--period-seconds', default=1, type=int, help='How many seconds to sleep between syncs.')
+@click.option('-b', '--batch-size', default=1, type=int, help='How many blocks to batch in single request.')
 @click.option('-B', '--block-batch-size', default=10, type=int, help='How many blocks to batch in single sync round.')
 @click.option('-w', '--max-workers', default=5, type=int, help='The number of workers.')
 @click.option('--log-file', default=None, type=str, help='Log file.')
 @click.option('--pid-file', default=None, type=str, help='pid file.')
 @click.option('--enrich', default=True, type=bool, help='Enable filling in transactions inputs fields.')
-def stream(last_synced_block_file, lag, provider_uri, output, start_block, chain=Chain.BITCOIN,
+def stream(last_synced_block_file, lag, provider_uri, output, topic_mapping, start_block, chain=Chain.BITCOIN,
            period_seconds=10, batch_size=2, block_batch_size=10, max_workers=5, log_file=None, pid_file=None,
            enrich=True):
     """Streams all data types to console or Google Pub/Sub."""
@@ -61,13 +62,16 @@ def stream(last_synced_block_file, lag, provider_uri, output, start_block, chain
     from bitcoinetl.streaming.btc_streamer_adapter import BtcStreamerAdapter
     from blockchainetl.streaming.streamer import Streamer
 
+    if (topic_mapping is not None):
+        topic_mapping = json.loads(topic_mapping)
+
     streamer_adapter = BtcStreamerAdapter(
         bitcoin_rpc=ThreadLocalProxy(lambda: BitcoinRpc(provider_uri)),
-        item_exporter=get_item_exporter(output),
+        item_exporter=get_item_exporter(output,topic_mapping,chain),
         chain=chain,
         batch_size=batch_size,
         enable_enrich=enrich,
-        max_workers=max_workers
+        max_workers=max_workers,
     )
     streamer = Streamer(
         blockchain_streamer_adapter=streamer_adapter,
@@ -77,5 +81,6 @@ def stream(last_synced_block_file, lag, provider_uri, output, start_block, chain
         period_seconds=period_seconds,
         block_batch_size=block_batch_size,
         pid_file=pid_file,
+        retry_errors=retry_errors
     )
     streamer.stream()

bitcoinetl/domain/block.py

@@ -24,6 +24,7 @@
 
 class BtcBlock(object):
 
+
     def __init__(self):
         self.hash = None
         self.size = None
@@ -36,8 +37,25 @@ def __init__(self):
         self.nonce = None
         self.bits = None
         self.coinbase_param = None
+        self.transaction_count = None
 
         self.transactions = []
 
+        # New fields added
+        self.transaction_ids = []
+
+        self.version_hex = None
+        self.median_timestamp = None
+        self.difficulty = None
+        self.chain_work = None
+        self.previous_block_hash = None
+        self.next_block_hash = None
+        self.input_value = None
+
+        self.block_reward = None
+        self.transaction_fees = None
+        self.coinbase_txid = None
+        self.coinbase_param_decoded = None
+
     def has_full_transactions(self):
         return len(self.transactions) > 0 and isinstance(self.transactions[0], BtcTransaction)

bitcoinetl/domain/transaction.py

@@ -43,6 +43,14 @@ def __init__(self):
         self.join_splits = []
         self.value_balance = 0
 
+        # New fields
+        self.transaction_id = None
+        self.weight = None
+        self.input_count = None
+        self.input_value = None
+        self.output_count = None
+        self.output_value = None
+
     def add_input(self, input):
         if len(self.inputs) > 0:
             input.index = self.inputs[len(self.inputs) - 1].index + 1

bitcoinetl/domain/transaction_input.py

@@ -23,9 +23,12 @@
 
 class BtcTransactionInput(object):
     def __init__(self):
+        self.create_transaction_id = None
+        self.create_output_index = None
+
+        self.spending_transaction_id = None
         self.index = None
-        self.spent_transaction_hash = None
-        self.spent_output_index = None
+
         self.script_asm = None
         self.script_hex = None
         self.coinbase_param = None
@@ -37,4 +40,4 @@ def __init__(self):
         self.value = None
 
     def is_coinbase(self):
-        return self.coinbase_param is not None or self.spent_transaction_hash is None
+        return self.coinbase_param is not None or self.create_transaction_id is None

bitcoinetl/domain/transaction_output.py

@@ -31,3 +31,7 @@ def __init__(self):
 
         self.addresses = []
         self.value = None
+        self.witness = []
+
+        self.create_transaction_id = None
+        self.spending_transaction_id = None

bitcoinetl/enumeration/chain.py

@@ -1,13 +1,35 @@
 class Chain:
     BITCOIN = 'bitcoin'
     BITCOIN_CASH = 'bitcoin_cash'
+    BITCOIN_CASH_SV = 'bitcoin_cash_sv'
     BITCOIN_GOLD = 'bitcoin_gold'
     DOGECOIN = 'dogecoin'
     LITECOIN = 'litecoin'
     DASH = 'dash'
     ZCASH = 'zcash'
     MONACOIN = 'monacoin'
 
-    ALL = [BITCOIN, BITCOIN_CASH, BITCOIN_GOLD, DOGECOIN, LITECOIN, DASH, ZCASH, MONACOIN]
+    ALL = [BITCOIN, BITCOIN_CASH, BITCOIN_CASH_SV, BITCOIN_GOLD, DOGECOIN, LITECOIN, DASH, ZCASH, MONACOIN]
     # Old API doesn't support verbosity for getblock which doesn't allow querying all transactions in a block in 1 go.
     HAVE_OLD_API = [BITCOIN_CASH, DOGECOIN, DASH, MONACOIN]
+
+    @classmethod
+    def ticker_symbol(cls, chain):
+        symbols = {
+            'bitcoin': 'BTC',
+            'bitcoin_cash': 'BCH',
+            'bitcoin_cash_sv': 'BSV',
+            'dogecoin': 'DOGE',
+            'litecoin': 'LTC',
+            'dash': 'DASH',
+            'zcash': 'ZEC',
+            'monacoin': 'MONA',
+        }
+        return symbols.get(chain, None)
+
+
+class CoinPriceType:
+
+    empty = 0
+    daily = 1
+    hourly = 2

bitcoinetl/jobs/enrich_transactions.py

@@ -63,7 +63,7 @@ def _enrich_transactions(self, transactions):
             input_transactions_map = self._get_input_transactions_as_map(transaction_input_batch)
             for input in transaction_input_batch:
                 output = self._get_output_for_input(input, input_transactions_map) \
-                    if input.spent_transaction_hash is not None else None
+                    if input.create_transaction_id is not None else None
                 if output is not None:
                     input.required_signatures = output.required_signatures
                     input.type = output.type
@@ -74,29 +74,29 @@ def _enrich_transactions(self, transactions):
             self.item_exporter.export_item(self.transaction_mapper.transaction_to_dict(transaction))
 
     def _get_input_transactions_as_map(self, transaction_inputs):
-        transaction_hashes = [input.spent_transaction_hash for input in transaction_inputs
-                              if input.spent_transaction_hash is not None]
+        transaction_hashes = [input.create_transaction_id for input in transaction_inputs
+                              if input.create_transaction_id is not None]
 
         transaction_hashes = set(transaction_hashes)
         if len(transaction_hashes) > 0:
             transactions = self.btc_service.get_transactions_by_hashes(transaction_hashes)
-            return {transaction.hash: transaction for transaction in transactions}
+            return {transaction.transaction_id: transaction for transaction in transactions}
         else:
             return {}
 
     def _get_output_for_input(self, transaction_input, input_transactions_map):
-        spent_transaction_hash = transaction_input.spent_transaction_hash
-        input_transaction = input_transactions_map.get(spent_transaction_hash)
+        create_transaction_id = transaction_input.create_transaction_id
+        input_transaction = input_transactions_map.get(create_transaction_id)
         if input_transaction is None:
-            raise ValueError('Input transaction with hash {} not found'.format(spent_transaction_hash))
+            raise ValueError('Input transaction with hash {} not found'.format(create_transaction_id))
 
-        spent_output_index = transaction_input.spent_output_index
-        if input_transaction.outputs is None or len(input_transaction.outputs) < (spent_output_index + 1):
+        create_output_index = transaction_input.create_output_index
+        if input_transaction.outputs is None or len(input_transaction.outputs) < (create_output_index + 1):
             raise ValueError(
                 'There is no output with index {} in transaction with hash {}'.format(
-                    spent_output_index, spent_transaction_hash))
+                    create_output_index, create_transaction_id))
 
-        output = input_transaction.outputs[spent_output_index]
+        output = input_transaction.outputs[create_output_index]
         return output
 
     def _end(self):

bitcoinetl/jobs/export_all.py

@@ -40,7 +40,9 @@
 logger = logging.getLogger('export_all')
 
 
-def export_all(chain, partitions, output_dir, provider_uri, max_workers, batch_size, enrich):
+def export_all(
+        chain, partitions, output_dir, provider_uri, max_workers, batch_size, enrich,
+    ):
     for batch_start_block, batch_end_block, partition_dir, *args in partitions:
         # # # start # # #
 
@@ -101,7 +103,8 @@ def export_all(chain, partitions, output_dir, provider_uri, max_workers, batch_s
             max_workers=max_workers,
             item_exporter=blocks_and_transactions_item_exporter(blocks_file, transactions_file),
             export_blocks=blocks_file is not None,
-            export_transactions=transactions_file is not None)
+            export_transactions=transactions_file is not None,
+            )
         job.run()
 
         if enrich == True:

bitcoinetl/jobs/export_blocks_job.py

@@ -27,6 +27,7 @@
 from blockchainetl.executors.batch_work_executor import BatchWorkExecutor
 from blockchainetl.jobs.base_job import BaseJob
 from blockchainetl.utils import validate_range
+from bitcoinetl.enumeration.chain import CoinPriceType
 
 
 # Exports blocks and transactions
@@ -43,6 +44,7 @@ def __init__(
             export_blocks=True,
             export_transactions=True):
         validate_range(start_block, end_block)
+
         self.start_block = start_block
         self.end_block = end_block