Add celery poller for phone index

apps/contract-migration/scripts/balance.py

@@ -0,0 +1,42 @@
+# standard imports
+import logging
+
+# external imports
+from eth_contract_registry import Registry
+from eth_token_index import TokenUniqueSymbolIndex
+from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.nonce import RPCNonceOracle
+
+logg = logging.getLogger().getChild(__name__)
+
+
+class BalanceProcessor:
+
+    def __init__(self, conn, chain_spec, registry_address, signer_address):
+
+        self.chain_spec = chain_spec
+        self.conn = conn
+        self.signer_address = signer_address
+        self.registry_address = registry_address
+
+        self.token_index_address = None
+        self.token_address = None
+
+        gas_oracle = OverrideGasOracle(conn=conn, limit=8000000)
+        nonce_oracle = RPCNonceOracle(signer_address, conn)
+
+
+    def init(self):
+        # Get Token registry address
+        registry = Registry(self.chain_spec)
+        o = registry.address_of(self.registry_address, 'TokenRegistry')
+        r = self.conn.do(o)
+        self.token_index_address = registry.parse_address_of(r)
+        logg.info('found token index address {}'.format(self.token_index_address))
+
+        token_registry = TokenUniqueSymbolIndex(self.chain_spec)
+        o = token_registry.address_of(self.token_index_address, 'SRF')
+        r = self.conn.do(o)
+        self.token_address = token_registry.parse_address_of(r)
+        logg.info('found SRF token address {}'.format(self.token_address))
+

apps/contract-migration/scripts/config/meta.ini

@@ -1,2 +1,5 @@
 [meta]
 url = http://localhost:63380
+host = localhost
+port = 63380
+ssl = 0

apps/contract-migration/scripts/create_import_users.py

@@ -206,7 +206,7 @@ if __name__ == '__main__':
         f.close()
 
         pidx = genPhoneIndex(phone)
-        d = prepareLocalFilePath(os.path.join(user_dir, 'phone'), uid)
+        d = prepareLocalFilePath(os.path.join(user_dir, 'phone'), pidx)
         f = open('{}/{}'.format(d, pidx), 'w')
         f.write(eth)
         f.close()

apps/contract-migration/scripts/import_balance.py

@@ -10,7 +10,7 @@ import hashlib
 import csv
 import json
 
-# third-party impotts
+# external imports
 import eth_abi
 import confini
 from hexathon import (
@@ -42,7 +42,7 @@ from cic_types.models.person import Person
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
 
-config_dir = '/usr/local/etc/cic-syncer'
+config_dir = './config'
 
 argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
 argparser.add_argument('-p', '--provider', dest='p', type=str, help='chain rpc provider address')

apps/contract-migration/scripts/import_ussd_balance.py

@@ -0,0 +1,174 @@
+# standard imports
+import os
+import sys
+import logging
+import argparse
+import hashlib
+import redis
+import celery
+
+# external imports
+import confini
+from chainlib.eth.connection import EthHTTPConnection
+from chainlib.chain import ChainSpec
+from hexathon import (
+        strip_0x,
+        add_0x,
+        )
+from chainlib.eth.address import to_checksum_address
+from crypto_dev_signer.eth.signer import ReferenceSigner as EIP155Signer
+from crypto_dev_signer.keystore.dict import DictKeystore
+from cic_types.models.person import Person
+
+# local imports
+from balance import BalanceProcessor
+from task import *
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+config_dir = './config'
+
+argparser = argparse.ArgumentParser(description='daemon that monitors transactions in new blocks')
+argparser.add_argument('-p', '--provider', dest='p', type=str, help='chain rpc provider address')
+argparser.add_argument('-y', '--key-file', dest='y', type=str, help='Ethereum keystore file to use for signing')
+argparser.add_argument('-c', type=str, default=config_dir, help='config root to use')
+argparser.add_argument('--old-chain-spec', type=str, dest='old_chain_spec', default='evm:oldchain:1', help='chain spec')
+argparser.add_argument('-i', '--chain-spec', type=str, dest='i', help='chain spec')
+argparser.add_argument('-r', '--registry-address', type=str, dest='r', help='CIC Registry address')
+argparser.add_argument('--meta-host', dest='meta_host', type=str, help='metadata server host')
+argparser.add_argument('--meta-port', dest='meta_port', type=int, help='metadata server host')
+argparser.add_argument('--redis-host', dest='redis_host', type=str, help='redis host to use for task submission')
+argparser.add_argument('--redis-port', dest='redis_port', type=int, help='redis host to use for task submission')
+argparser.add_argument('--redis-db', dest='redis_db', type=int, help='redis db to use for task submission and callback')
+argparser.add_argument('--token-symbol', default='SRF', type=str, dest='token_symbol', help='Token symbol to use for trnsactions')
+argparser.add_argument('--head', action='store_true', help='start at current block height (overrides --offset)')
+argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
+argparser.add_argument('-q', type=str, default='cic-eth', help='celery queue to submit transaction tasks to')
+argparser.add_argument('--offset', type=int, default=0, help='block offset to start syncer from')
+argparser.add_argument('-v', help='be verbose', action='store_true')
+argparser.add_argument('-vv', help='be more verbose', action='store_true')
+argparser.add_argument('user_dir', type=str, help='user export directory')
+args = argparser.parse_args(sys.argv[1:])
+
+if args.v == True:
+    logging.getLogger().setLevel(logging.INFO)
+elif args.vv == True:
+    logging.getLogger().setLevel(logging.DEBUG)
+
+config_dir = os.path.join(args.c)
+os.makedirs(config_dir, 0o777, True)
+config = confini.Config(config_dir, args.env_prefix)
+config.process()
+# override args
+args_override = {
+        'CIC_CHAIN_SPEC': getattr(args, 'i'),
+        'ETH_PROVIDER': getattr(args, 'p'),
+        'CIC_REGISTRY_ADDRESS': getattr(args, 'r'),
+        'REDIS_HOST': getattr(args, 'redis_host'),
+        'REDIS_PORT': getattr(args, 'redis_port'),
+        'REDIS_DB': getattr(args, 'redis_db'),
+        'META_HOST': getattr(args, 'meta_host'),
+        'META_PORT': getattr(args, 'meta_port'),
+        }
+config.dict_override(args_override, 'cli flag')
+config.censor('PASSWORD', 'DATABASE')
+config.censor('PASSWORD', 'SSL')
+logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
+
+redis_host = config.get('REDIS_HOST')
+redis_port = config.get('REDIS_PORT')
+redis_db = config.get('REDIS_DB')
+r = redis.Redis(redis_host, redis_port, redis_db)
+celery_app = celery.Celery(backend=config.get('CELERY_RESULT_URL'),  broker=config.get('CELERY_BROKER_URL'))
+
+signer_address = None
+keystore = DictKeystore()
+if args.y != None:
+    logg.debug('loading keystore file {}'.format(args.y))
+    signer_address = keystore.import_keystore_file(args.y)
+    logg.debug('now have key for signer address {}'.format(signer_address))
+signer = EIP155Signer(keystore)
+
+queue = args.q
+chain_str = config.get('CIC_CHAIN_SPEC')
+block_offset = 0
+if args.head:
+    block_offset = -1
+else:
+    block_offset = args.offset
+
+chain_spec = ChainSpec.from_chain_str(chain_str)
+old_chain_spec_str = args.old_chain_spec
+old_chain_spec = ChainSpec.from_chain_str(old_chain_spec_str)
+
+user_dir = args.user_dir # user_out_dir from import_users.py
+
+token_symbol = args.token_symbol
+
+MetadataTask.meta_host = config.get('META_HOST')
+MetadataTask.meta_port = config.get('META_PORT')
+
+def main():
+    conn = EthHTTPConnection(config.get('ETH_PROVIDER'))
+   
+    processor = BalanceProcessor(conn, chain_spec, config.get('CIC_REGISTRY_ADDRESS'), signer_address)
+    processor.init()
+
+
+#    syncer_backend = MemBackend(chain_str, 0)
+#
+#    if block_offset == -1:
+#        o = block_latest()
+#        r = conn.do(o)
+#        block_offset = int(strip_0x(r), 16) + 1
+##
+##    addresses = {}
+##    f = open('{}/addresses.csv'.format(user_dir, 'r'))
+##    while True:
+##        l = f.readline()
+##        if l == None:
+##            break
+##        r = l.split(',')
+##        try:
+##            k = r[0]
+##            v = r[1].rstrip()
+##            addresses[k] = v
+##            sys.stdout.write('loading address mapping {} -> {}'.format(k, v).ljust(200) + "\r")
+##        except IndexError as e:
+##            break
+##    f.close()
+#
+#    # TODO get decimals from token
+    balances = {}
+    f = open('{}/balances.csv'.format(user_dir, 'r'))
+    remove_zeros = 10**6
+    i = 0
+    while True:
+        l = f.readline()
+        if l == None:
+            break
+        r = l.split(',')
+        try:
+            address = to_checksum_address(r[0])
+            sys.stdout.write('loading balance {} {} {}'.format(i, address, r[1]).ljust(200) + "\r")
+        except ValueError:
+            break
+        balance = int(int(r[1].rstrip()) / remove_zeros)
+        balances[address] = balance
+        i += 1
+
+    f.close()
+
+    argv = ['worker', '-Q', 'cic-import-ussd', '--loglevel=DEBUG']
+    celery_app.worker_main(argv)
+
+#    syncer_backend.set(block_offset, 0)
+#    syncer = HeadSyncer(syncer_backend, progress_callback=progress_callback)
+#    handler = Handler(conn, chain_spec, user_dir, balances, sarafu_token_address, signer, gas_oracle, nonce_oracle)
+#    syncer.add_filter(handler)
+#    syncer.loop(1, conn)
+    
+
+if __name__ == '__main__':
+    main()

apps/contract-migration/scripts/import_ussd_users.py

@@ -145,6 +145,19 @@ if __name__ == '__main__':
 
             new_address = register_ussd(i, u)
 
+            phone_object = phonenumbers.parse(u.tel)
+            phone = phonenumbers.format_number(phone_object, phonenumbers.PhoneNumberFormat.E164)
+
+            s = celery.signature(
+                    'task.resolve_phone',
+                    [
+                        phone,
+                        ],
+                    queue='cic-import-ussd',
+                    )
+            s.apply_async()
+
+
 #            if u.identities.get('evm') == None:
 #                u.identities['evm'] = {}
 #            sub_chain_str = '{}:{}'.format(chain_spec.common_name(), chain_spec.network_id())

apps/contract-migration/scripts/task.py

@@ -0,0 +1,43 @@
+# standard imports
+import logging
+import urllib.parse
+import urllib.error
+import urllib.request
+
+# external imports
+import celery
+from cic_types.processor import generate_metadata_pointer
+
+logg = logging.getLogger().getChild(__name__)
+
+celery_app = celery.current_app
+
+
+class MetadataTask(celery.Task):
+
+    meta_host = None
+    meta_port = None
+    meta_path = ''
+    meta_ssl = False
+    autoretry_for = (
+            urllib.error.HTTPError,
+            )
+    retry_kwargs = {
+        'countdown': 1,
+            }
+
+    @classmethod
+    def meta_url(self):
+        scheme = 'http'
+        if self.meta_ssl:
+            scheme += s
+        url = urllib.parse.urlparse('{}://{}:{}/{}'.format(scheme, self.meta_host, self.meta_port, self.meta_path))
+        return urllib.parse.urlunparse(url)
+
+@celery_app.task(bind=True, base=MetadataTask)
+def resolve_phone(self, phone):
+    identifier = generate_metadata_pointer(phone.encode('utf-8'), 'cic.phone')
+    url = urllib.parse.urljoin(self.meta_url(), identifier)
+    logg.debug('attempt getting phone pointer at {}'.format(url))
+    r = urllib.request.urlopen(url)
+    logg.debug('phone pointer result {}'.format(r))