Merge remote-tracking branch 'origin/master' into lash/demurrage-task
This commit is contained in:
nolash
commit
185458e320
@ -4,3 +4,4 @@ pytest-mock==3.3.1
|
||||
pytest-cov==2.10.1
|
||||
eth-tester==0.5.0b3
|
||||
py-evm==0.3.0a20
|
||||
eth-erc20==0.0.10a2
|
||||
|
||||
@ -127,6 +127,8 @@ def main():
|
||||
argv.append('--loglevel=INFO')
|
||||
argv.append('-Q')
|
||||
argv.append(args.q)
|
||||
argv.append('-n')
|
||||
argv.append(args.q)
|
||||
|
||||
current_app.worker_main(argv)
|
||||
|
||||
|
||||
@ -10,7 +10,7 @@ import redis
|
||||
import celery
|
||||
from cic_eth_registry.registry import CICRegistry
|
||||
from chainsyncer.backend.memory import MemBackend
|
||||
from chainsyncer.driver import HeadSyncer
|
||||
from chainsyncer.driver.head import HeadSyncer
|
||||
from chainlib.eth.connection import EthHTTPConnection
|
||||
from chainlib.chain import ChainSpec
|
||||
from chainlib.eth.gas import RPCGasOracle
|
||||
@ -24,6 +24,7 @@ from cic_base import (
|
||||
rpc,
|
||||
signer as signer_funcs,
|
||||
)
|
||||
from cic_base.eth.syncer import chain_interface
|
||||
|
||||
# local imports
|
||||
#import common
|
||||
@ -120,7 +121,7 @@ def main():
|
||||
'api_queue': config.get('_CELERY_QUEUE'),
|
||||
}
|
||||
|
||||
syncer = HeadSyncer(syncer_backend, block_callback=handler.refresh)
|
||||
syncer = HeadSyncer(syncer_backend, chain_interface, block_callback=handler.refresh)
|
||||
syncer.add_filter(handler)
|
||||
syncer.loop(1, conn)
|
||||
|
||||
|
||||
@ -1,28 +1,22 @@
|
||||
# standard imports
|
||||
import os
|
||||
import sys
|
||||
import logging
|
||||
import argparse
|
||||
import hashlib
|
||||
import redis
|
||||
import celery
|
||||
import logging
|
||||
import sys
|
||||
import os
|
||||
|
||||
# external imports
|
||||
import celery
|
||||
import confini
|
||||
from chainlib.eth.connection import EthHTTPConnection
|
||||
import redis
|
||||
from chainlib.chain import ChainSpec
|
||||
from hexathon import (
|
||||
strip_0x,
|
||||
add_0x,
|
||||
)
|
||||
from chainlib.eth.address import to_checksum_address
|
||||
from chainlib.eth.connection import EthHTTPConnection
|
||||
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 import_util import BalanceProcessor
|
||||
from import_task import *
|
||||
from import_task import ImportTask, MetadataTask
|
||||
from import_util import BalanceProcessor, get_celery_worker_status
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
logg = logging.getLogger()
|
||||
@ -41,37 +35,41 @@ argparser.add_argument('--meta-port', dest='meta_port', type=int, help='metadata
|
||||
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('--token-symbol', default='GFT', type=str, dest='token_symbol',
|
||||
help='Token symbol to use for transactions')
|
||||
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('--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-import-ussd', 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', default='out', type=str, help='user export directory')
|
||||
args = argparser.parse_args(sys.argv[1:])
|
||||
|
||||
if args.v == True:
|
||||
if args.v:
|
||||
logging.getLogger().setLevel(logging.INFO)
|
||||
elif args.vv == True:
|
||||
|
||||
elif args.vv:
|
||||
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'),
|
||||
'KEYSTORE_FILE_PATH': getattr(args, 'y')
|
||||
}
|
||||
'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'),
|
||||
'KEYSTORE_FILE_PATH': getattr(args, 'y')
|
||||
}
|
||||
config.dict_override(args_override, 'cli flag')
|
||||
config.censor('PASSWORD', 'DATABASE')
|
||||
config.censor('PASSWORD', 'SSL')
|
||||
@ -81,14 +79,19 @@ 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'))
|
||||
|
||||
# create celery apps
|
||||
celery_app = celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY_BROKER_URL'))
|
||||
status = get_celery_worker_status(celery_app=celery_app)
|
||||
|
||||
signer_address = None
|
||||
keystore = DictKeystore()
|
||||
if args.y != None:
|
||||
if args.y is not 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))
|
||||
|
||||
# define signer
|
||||
signer = EIP155Signer(keystore)
|
||||
|
||||
queue = args.q
|
||||
@ -103,7 +106,7 @@ 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
|
||||
user_dir = args.user_dir # user_out_dir from import_users.py
|
||||
|
||||
token_symbol = args.token_symbol
|
||||
|
||||
@ -111,20 +114,22 @@ MetadataTask.meta_host = config.get('META_HOST')
|
||||
MetadataTask.meta_port = config.get('META_PORT')
|
||||
ImportTask.chain_spec = chain_spec
|
||||
|
||||
|
||||
def main():
|
||||
conn = EthHTTPConnection(config.get('ETH_PROVIDER'))
|
||||
|
||||
ImportTask.balance_processor = BalanceProcessor(conn, chain_spec, config.get('CIC_REGISTRY_ADDRESS'), signer_address, signer)
|
||||
|
||||
ImportTask.balance_processor = BalanceProcessor(conn, chain_spec, config.get('CIC_REGISTRY_ADDRESS'),
|
||||
signer_address, signer)
|
||||
ImportTask.balance_processor.init(token_symbol)
|
||||
|
||||
# TODO get decimals from token
|
||||
balances = {}
|
||||
f = open('{}/balances.csv'.format(user_dir, 'r'))
|
||||
remove_zeros = 10**6
|
||||
remove_zeros = 10 ** 6
|
||||
i = 0
|
||||
while True:
|
||||
l = f.readline()
|
||||
if l == None:
|
||||
if l is None:
|
||||
break
|
||||
r = l.split(',')
|
||||
try:
|
||||
@ -143,15 +148,23 @@ def main():
|
||||
ImportTask.import_dir = user_dir
|
||||
|
||||
s = celery.signature(
|
||||
'import_task.send_txs',
|
||||
[
|
||||
MetadataTask.balance_processor.nonce_offset,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
'import_task.send_txs',
|
||||
[
|
||||
MetadataTask.balance_processor.nonce_offset,
|
||||
],
|
||||
queue=queue,
|
||||
)
|
||||
s.apply_async()
|
||||
|
||||
argv = ['worker', '-Q', 'cic-import-ussd', '--loglevel=DEBUG']
|
||||
argv = ['worker']
|
||||
if args.vv:
|
||||
argv.append('--loglevel=DEBUG')
|
||||
elif args.v:
|
||||
argv.append('--loglevel=INFO')
|
||||
argv.append('-Q')
|
||||
argv.append(args.q)
|
||||
argv.append('-n')
|
||||
argv.append(args.q)
|
||||
celery_app.worker_main(argv)
|
||||
|
||||
|
||||
|
||||
@ -9,7 +9,7 @@ import celery
|
||||
import confini
|
||||
|
||||
# local imports
|
||||
from import_task import *
|
||||
from import_util import get_celery_worker_status
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
logg = logging.getLogger()
|
||||
@ -39,9 +39,12 @@ elif args.vv:
|
||||
config_dir = args.c
|
||||
config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
|
||||
config.process()
|
||||
config.censor('PASSWORD', 'DATABASE')
|
||||
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
|
||||
|
||||
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
|
||||
status = get_celery_worker_status(celery_app=celery_app)
|
||||
|
||||
|
||||
db_configs = {
|
||||
'database': config.get('DATABASE_NAME'),
|
||||
@ -61,7 +64,8 @@ def main():
|
||||
(db_configs, phone_to_pins),
|
||||
queue=args.q
|
||||
)
|
||||
s_import_pins.apply_async()
|
||||
result = s_import_pins.apply_async()
|
||||
logg.debug(f'TASK: {result.id}, STATUS: {result.status}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
@ -1,36 +1,33 @@
|
||||
# standard imports
|
||||
import os
|
||||
import logging
|
||||
import random
|
||||
import urllib.parse
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import urllib.error
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
# external imports
|
||||
import celery
|
||||
import psycopg2
|
||||
from psycopg2 import extras
|
||||
from hexathon import (
|
||||
strip_0x,
|
||||
add_0x,
|
||||
)
|
||||
from chainlib.eth.address import to_checksum_address
|
||||
from chainlib.eth.tx import (
|
||||
unpack,
|
||||
raw,
|
||||
)
|
||||
from cic_types.processor import generate_metadata_pointer
|
||||
unpack,
|
||||
raw,
|
||||
)
|
||||
from cic_types.models.person import Person
|
||||
from cic_types.processor import generate_metadata_pointer
|
||||
from hexathon import (
|
||||
strip_0x,
|
||||
add_0x,
|
||||
)
|
||||
|
||||
#logg = logging.getLogger().getChild(__name__)
|
||||
logg = logging.getLogger()
|
||||
|
||||
celery_app = celery.current_app
|
||||
|
||||
|
||||
class ImportTask(celery.Task):
|
||||
|
||||
balances = None
|
||||
import_dir = 'out'
|
||||
count = 0
|
||||
@ -38,16 +35,16 @@ class ImportTask(celery.Task):
|
||||
balance_processor = None
|
||||
max_retries = None
|
||||
|
||||
class MetadataTask(ImportTask):
|
||||
|
||||
class MetadataTask(ImportTask):
|
||||
meta_host = None
|
||||
meta_port = None
|
||||
meta_path = ''
|
||||
meta_ssl = False
|
||||
autoretry_for = (
|
||||
urllib.error.HTTPError,
|
||||
OSError,
|
||||
)
|
||||
urllib.error.HTTPError,
|
||||
OSError,
|
||||
)
|
||||
retry_jitter = True
|
||||
retry_backoff = True
|
||||
retry_backoff_max = 60
|
||||
@ -64,12 +61,12 @@ class MetadataTask(ImportTask):
|
||||
def old_address_from_phone(base_path, phone):
|
||||
pidx = generate_metadata_pointer(phone.encode('utf-8'), ':cic.phone')
|
||||
phone_idx_path = os.path.join('{}/phone/{}/{}/{}'.format(
|
||||
base_path,
|
||||
pidx[:2],
|
||||
pidx[2:4],
|
||||
pidx,
|
||||
)
|
||||
)
|
||||
base_path,
|
||||
pidx[:2],
|
||||
pidx[2:4],
|
||||
pidx,
|
||||
)
|
||||
)
|
||||
f = open(phone_idx_path, 'r')
|
||||
old_address = f.read()
|
||||
f.close()
|
||||
@ -97,11 +94,11 @@ def generate_metadata(self, address, phone):
|
||||
logg.debug('address {}'.format(address))
|
||||
old_address_upper = strip_0x(old_address).upper()
|
||||
metadata_path = '{}/old/{}/{}/{}.json'.format(
|
||||
self.import_dir,
|
||||
old_address_upper[:2],
|
||||
old_address_upper[2:4],
|
||||
old_address_upper,
|
||||
)
|
||||
self.import_dir,
|
||||
old_address_upper[:2],
|
||||
old_address_upper[2:4],
|
||||
old_address_upper,
|
||||
)
|
||||
|
||||
f = open(metadata_path, 'r')
|
||||
o = json.load(f)
|
||||
@ -116,12 +113,12 @@ def generate_metadata(self, address, phone):
|
||||
|
||||
new_address_clean = strip_0x(address)
|
||||
filepath = os.path.join(
|
||||
self.import_dir,
|
||||
'new',
|
||||
new_address_clean[:2].upper(),
|
||||
new_address_clean[2:4].upper(),
|
||||
new_address_clean.upper() + '.json',
|
||||
)
|
||||
self.import_dir,
|
||||
'new',
|
||||
new_address_clean[:2].upper(),
|
||||
new_address_clean[2:4].upper(),
|
||||
new_address_clean.upper() + '.json',
|
||||
)
|
||||
os.makedirs(os.path.dirname(filepath), exist_ok=True)
|
||||
|
||||
o = u.serialize()
|
||||
@ -131,10 +128,10 @@ def generate_metadata(self, address, phone):
|
||||
|
||||
meta_key = generate_metadata_pointer(bytes.fromhex(new_address_clean), ':cic.person')
|
||||
meta_filepath = os.path.join(
|
||||
self.import_dir,
|
||||
'meta',
|
||||
'{}.json'.format(new_address_clean.upper()),
|
||||
)
|
||||
self.import_dir,
|
||||
'meta',
|
||||
'{}.json'.format(new_address_clean.upper()),
|
||||
)
|
||||
os.symlink(os.path.realpath(filepath), meta_filepath)
|
||||
|
||||
# write ussd data
|
||||
@ -180,10 +177,8 @@ def generate_metadata(self, address, phone):
|
||||
|
||||
@celery_app.task(bind=True, base=MetadataTask)
|
||||
def opening_balance_tx(self, address, phone, serial):
|
||||
|
||||
|
||||
old_address = old_address_from_phone(self.import_dir, phone)
|
||||
|
||||
|
||||
k = to_checksum_address(strip_0x(old_address))
|
||||
balance = self.balances[k]
|
||||
logg.debug('found balance {} for address {} phone {}'.format(balance, old_address, phone))
|
||||
@ -196,39 +191,39 @@ def opening_balance_tx(self, address, phone, serial):
|
||||
logg.debug('generated tx token value {} to {} tx hash {}'.format(decimal_balance, address, tx_hash_hex))
|
||||
|
||||
tx_path = os.path.join(
|
||||
self.import_dir,
|
||||
'txs',
|
||||
strip_0x(tx_hash_hex),
|
||||
)
|
||||
|
||||
self.import_dir,
|
||||
'txs',
|
||||
strip_0x(tx_hash_hex),
|
||||
)
|
||||
|
||||
f = open(tx_path, 'w')
|
||||
f.write(strip_0x(o))
|
||||
f.close()
|
||||
|
||||
tx_nonce_path = os.path.join(
|
||||
self.import_dir,
|
||||
'txs',
|
||||
'.' + str(tx['nonce']),
|
||||
)
|
||||
self.import_dir,
|
||||
'txs',
|
||||
'.' + str(tx['nonce']),
|
||||
)
|
||||
os.symlink(os.path.realpath(tx_path), tx_nonce_path)
|
||||
|
||||
return tx['hash']
|
||||
|
||||
|
||||
@celery_app.task(bind=True, base=ImportTask, autoretry_for=(FileNotFoundError,), max_retries=None, default_retry_delay=0.1)
|
||||
@celery_app.task(bind=True, base=ImportTask, autoretry_for=(FileNotFoundError,), max_retries=None,
|
||||
default_retry_delay=0.1)
|
||||
def send_txs(self, nonce):
|
||||
|
||||
if nonce == self.count + self.balance_processor.nonce_offset:
|
||||
logg.info('reached nonce {} (offset {} + count {}) exiting'.format(nonce, self.balance_processor.nonce_offset, self.count))
|
||||
logg.info('reached nonce {} (offset {} + count {}) exiting'.format(nonce, self.balance_processor.nonce_offset,
|
||||
self.count))
|
||||
return
|
||||
|
||||
|
||||
logg.debug('attempt to open symlink for nonce {}'.format(nonce))
|
||||
tx_nonce_path = os.path.join(
|
||||
self.import_dir,
|
||||
'txs',
|
||||
'.' + str(nonce),
|
||||
)
|
||||
self.import_dir,
|
||||
'txs',
|
||||
'.' + str(nonce),
|
||||
)
|
||||
f = open(tx_nonce_path, 'r')
|
||||
tx_signed_raw_hex = f.read()
|
||||
f.close()
|
||||
@ -238,21 +233,20 @@ def send_txs(self, nonce):
|
||||
o = raw(add_0x(tx_signed_raw_hex))
|
||||
tx_hash_hex = self.balance_processor.conn.do(o)
|
||||
|
||||
logg.info('sent nonce {} tx hash {}'.format(nonce, tx_hash_hex)) #tx_signed_raw_hex))
|
||||
logg.info('sent nonce {} tx hash {}'.format(nonce, tx_hash_hex)) # tx_signed_raw_hex))
|
||||
|
||||
nonce += 1
|
||||
|
||||
queue = self.request.delivery_info.get('routing_key')
|
||||
s = celery.signature(
|
||||
'import_task.send_txs',
|
||||
[
|
||||
nonce,
|
||||
],
|
||||
queue=queue,
|
||||
)
|
||||
'import_task.send_txs',
|
||||
[
|
||||
nonce,
|
||||
],
|
||||
queue=queue,
|
||||
)
|
||||
s.apply_async()
|
||||
|
||||
|
||||
return nonce
|
||||
|
||||
|
||||
@ -310,4 +304,3 @@ def set_ussd_data(config: dict, ussd_data: dict):
|
||||
# close connections
|
||||
db_cursor.close()
|
||||
db_conn.close()
|
||||
|
||||
|
||||
@ -17,6 +17,9 @@ import redis
|
||||
from chainlib.chain import ChainSpec
|
||||
from cic_types.models.person import Person
|
||||
|
||||
# local imports
|
||||
from import_util import get_celery_worker_status
|
||||
|
||||
logging.basicConfig(level=logging.WARNING)
|
||||
logg = logging.getLogger()
|
||||
|
||||
@ -28,11 +31,14 @@ argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='Chain spe
|
||||
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('--batch-size', dest='batch_size', default=100, type=int, help='burst size of sending transactions to node') # batch size should be slightly below cumulative gas limit worth, eg 80000 gas txs with 8000000 limit is a bit less than 100 batch size
|
||||
argparser.add_argument('--batch-size', dest='batch_size', default=100, type=int,
|
||||
help='burst size of sending transactions to node') # batch size should be slightly below cumulative gas limit worth, eg 80000 gas txs with 8000000 limit is a bit less than 100 batch size
|
||||
argparser.add_argument('--batch-delay', dest='batch_delay', default=3, type=int, help='seconds delay between batches')
|
||||
argparser.add_argument('--timeout', default=60.0, type=float, help='Callback timeout')
|
||||
argparser.add_argument('--ussd-host', dest='ussd_host', type=str, help="host to ussd app responsible for processing ussd requests.")
|
||||
argparser.add_argument('--ussd-port', dest='ussd_port', type=str, help="port to ussd app responsible for processing ussd requests.")
|
||||
argparser.add_argument('--ussd-host', dest='ussd_host', type=str,
|
||||
help="host to ussd app responsible for processing ussd requests.")
|
||||
argparser.add_argument('--ussd-port', dest='ussd_port', type=str,
|
||||
help="port to ussd app responsible for processing ussd requests.")
|
||||
argparser.add_argument('--ussd-no-ssl', dest='ussd_no_ssl', help='do not use ssl (careful)', action='store_true')
|
||||
argparser.add_argument('-q', type=str, default='cic-eth', help='Task queue')
|
||||
argparser.add_argument('-v', action='store_true', help='Be verbose')
|
||||
@ -49,13 +55,16 @@ config_dir = args.c
|
||||
config = confini.Config(config_dir, os.environ.get('CONFINI_ENV_PREFIX'))
|
||||
config.process()
|
||||
args_override = {
|
||||
'CIC_CHAIN_SPEC': getattr(args, 'i'),
|
||||
'REDIS_HOST': getattr(args, 'redis_host'),
|
||||
'REDIS_PORT': getattr(args, 'redis_port'),
|
||||
'REDIS_DB': getattr(args, 'redis_db'),
|
||||
}
|
||||
'CIC_CHAIN_SPEC': getattr(args, 'i'),
|
||||
'REDIS_HOST': getattr(args, 'redis_host'),
|
||||
'REDIS_PORT': getattr(args, 'redis_port'),
|
||||
'REDIS_DB': getattr(args, 'redis_db'),
|
||||
}
|
||||
config.dict_override(args_override, 'cli')
|
||||
logg.debug('config loaded from {}:\n{}'.format(args.c, config))
|
||||
|
||||
celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
|
||||
get_celery_worker_status(celery_app=celery_app)
|
||||
|
||||
redis_host = config.get('REDIS_HOST')
|
||||
redis_port = config.get('REDIS_PORT')
|
||||
@ -65,22 +74,22 @@ r = redis.Redis(redis_host, redis_port, redis_db)
|
||||
ps = r.pubsub()
|
||||
|
||||
user_new_dir = os.path.join(args.user_dir, 'new')
|
||||
os.makedirs(user_new_dir)
|
||||
os.makedirs(user_new_dir, exist_ok=True)
|
||||
|
||||
ussd_data_dir = os.path.join(args.user_dir, 'ussd')
|
||||
os.makedirs(ussd_data_dir)
|
||||
os.makedirs(ussd_data_dir, exist_ok=True)
|
||||
|
||||
preferences_dir = os.path.join(args.user_dir, 'preferences')
|
||||
os.makedirs(os.path.join(preferences_dir, 'meta'))
|
||||
os.makedirs(os.path.join(preferences_dir, 'meta'), exist_ok=True)
|
||||
|
||||
meta_dir = os.path.join(args.user_dir, 'meta')
|
||||
os.makedirs(meta_dir)
|
||||
os.makedirs(meta_dir, exist_ok=True)
|
||||
|
||||
user_old_dir = os.path.join(args.user_dir, 'old')
|
||||
os.stat(user_old_dir)
|
||||
|
||||
txs_dir = os.path.join(args.user_dir, 'txs')
|
||||
os.makedirs(txs_dir)
|
||||
os.makedirs(txs_dir, exist_ok=True)
|
||||
|
||||
chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
|
||||
chain_str = str(chain_spec)
|
||||
@ -95,6 +104,7 @@ if ussd_no_ssl is True:
|
||||
else:
|
||||
ussd_ssl = True
|
||||
|
||||
|
||||
def build_ussd_request(phone, host, port, service_code, username, password, ssl=False):
|
||||
url = 'http'
|
||||
if ssl:
|
||||
@ -109,13 +119,13 @@ def build_ussd_request(phone, host, port, service_code, username, password, ssl=
|
||||
|
||||
session = uuid.uuid4().hex
|
||||
data = {
|
||||
'sessionId': session,
|
||||
'serviceCode': service_code,
|
||||
'phoneNumber': phone,
|
||||
'text': service_code,
|
||||
}
|
||||
'sessionId': session,
|
||||
'serviceCode': service_code,
|
||||
'phoneNumber': phone,
|
||||
'text': service_code,
|
||||
}
|
||||
req = urllib.request.Request(url)
|
||||
req.method=('POST')
|
||||
req.method = 'POST'
|
||||
data_str = urlencode(data)
|
||||
data_bytes = data_str.encode('utf-8')
|
||||
req.add_header('Content-Type', 'application/x-www-form-urlencoded')
|
||||
@ -150,7 +160,7 @@ if __name__ == '__main__':
|
||||
j = 0
|
||||
for x in os.walk(user_old_dir):
|
||||
for y in x[2]:
|
||||
if y[len(y)-5:] != '.json':
|
||||
if y[len(y) - 5:] != '.json':
|
||||
continue
|
||||
# handle json containing person object
|
||||
filepath = os.path.join(x[0], y)
|
||||
@ -164,35 +174,35 @@ if __name__ == '__main__':
|
||||
f.close()
|
||||
u = Person.deserialize(o)
|
||||
|
||||
new_address = register_ussd(i, u)
|
||||
register_ussd(i, u)
|
||||
|
||||
phone_object = phonenumbers.parse(u.tel)
|
||||
phone = phonenumbers.format_number(phone_object, phonenumbers.PhoneNumberFormat.E164)
|
||||
|
||||
s_phone = celery.signature(
|
||||
'import_task.resolve_phone',
|
||||
[
|
||||
phone,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
'import_task.resolve_phone',
|
||||
[
|
||||
phone,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
|
||||
s_meta = celery.signature(
|
||||
'import_task.generate_metadata',
|
||||
[
|
||||
phone,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
'import_task.generate_metadata',
|
||||
[
|
||||
phone,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
|
||||
s_balance = celery.signature(
|
||||
'import_task.opening_balance_tx',
|
||||
[
|
||||
phone,
|
||||
i,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
'import_task.opening_balance_tx',
|
||||
[
|
||||
phone,
|
||||
i,
|
||||
],
|
||||
queue='cic-import-ussd',
|
||||
)
|
||||
|
||||
s_meta.link(s_balance)
|
||||
s_phone.link(s_meta)
|
||||
@ -206,4 +216,3 @@ if __name__ == '__main__':
|
||||
if j == batch_size:
|
||||
time.sleep(batch_delay)
|
||||
j = 0
|
||||
|
||||
|
||||
@ -2,15 +2,16 @@
|
||||
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 OverrideNonceOracle
|
||||
from eth_erc20 import ERC20
|
||||
from chainlib.eth.tx import (
|
||||
count,
|
||||
TxFormat,
|
||||
)
|
||||
count,
|
||||
TxFormat,
|
||||
)
|
||||
from celery import Celery
|
||||
from eth_contract_registry import Registry
|
||||
from eth_erc20 import ERC20
|
||||
from eth_token_index import TokenUniqueSymbolIndex
|
||||
|
||||
logg = logging.getLogger().getChild(__name__)
|
||||
|
||||
@ -18,10 +19,9 @@ logg = logging.getLogger().getChild(__name__)
|
||||
class BalanceProcessor:
|
||||
|
||||
def __init__(self, conn, chain_spec, registry_address, signer_address, signer):
|
||||
|
||||
self.chain_spec = chain_spec
|
||||
self.conn = conn
|
||||
#self.signer_address = signer_address
|
||||
# self.signer_address = signer_address
|
||||
self.registry_address = registry_address
|
||||
|
||||
self.token_index_address = None
|
||||
@ -35,7 +35,6 @@ class BalanceProcessor:
|
||||
self.gas_oracle = OverrideGasOracle(conn=conn, limit=8000000)
|
||||
|
||||
self.value_multiplier = 1
|
||||
|
||||
|
||||
def init(self, token_symbol):
|
||||
# Get Token registry address
|
||||
@ -57,16 +56,25 @@ class BalanceProcessor:
|
||||
n = tx_factory.parse_decimals(r)
|
||||
self.value_multiplier = 10 ** n
|
||||
|
||||
|
||||
def get_rpc_tx(self, recipient, value, i):
|
||||
logg.debug('initiating nonce offset {} for recipient {}'.format(self.nonce_offset + i, recipient))
|
||||
nonce_oracle = OverrideNonceOracle(self.signer_address, self.nonce_offset + i)
|
||||
tx_factory = ERC20(self.chain_spec, signer=self.signer, nonce_oracle=nonce_oracle, gas_oracle=self.gas_oracle)
|
||||
return tx_factory.transfer(self.token_address, self.signer_address, recipient, value, tx_format=TxFormat.RLP_SIGNED)
|
||||
#(tx_hash_hex, o) = tx_factory.transfer(self.token_address, self.signer_address, recipient, value)
|
||||
#self.conn.do(o)
|
||||
#return tx_hash_hex
|
||||
|
||||
return tx_factory.transfer(self.token_address, self.signer_address, recipient, value,
|
||||
tx_format=TxFormat.RLP_SIGNED)
|
||||
# (tx_hash_hex, o) = tx_factory.transfer(self.token_address, self.signer_address, recipient, value)
|
||||
# self.conn.do(o)
|
||||
# return tx_hash_hex
|
||||
|
||||
def get_decimal_amount(self, value):
|
||||
return value * self.value_multiplier
|
||||
|
||||
|
||||
def get_celery_worker_status(celery_app: Celery):
|
||||
inspector = celery_app.control.inspect()
|
||||
availability = inspector.ping()
|
||||
status = {
|
||||
'availability': availability,
|
||||
}
|
||||
logg.debug(f'RUNNING WITH STATUS: {status}')
|
||||
return status
|
||||
|
||||
Loading…
Reference in New Issue
Block a user