100 lines
3.0 KiB
Python
100 lines
3.0 KiB
Python
# standard imports
|
|
import sys
|
|
import logging
|
|
import datetime
|
|
|
|
# external imports
|
|
from cic_eth_registry import CICRegistry
|
|
from cic_eth_registry.lookup.tokenindex import TokenIndexLookup
|
|
from cic_types.ext.metadata import MetadataRequestsHandler
|
|
from cic_types.models.person import Person
|
|
from chainlib.eth.address import to_checksum_address
|
|
|
|
# local imports
|
|
from clicada.tx import TxGetter
|
|
from clicada.user import FileUserStore
|
|
from clicada.token import FileTokenStore
|
|
from clicada.tx import ResolvedTokenTx
|
|
|
|
logg = logging.getLogger(__name__)
|
|
|
|
|
|
def process_args(argparser):
|
|
argparser.add_argument('-m', '--method', type=str, help='lookup method')
|
|
argparser.add_argument('--meta-url', dest='meta_url', type=str, help='Url to retrieve metadata from')
|
|
argparser.add_argument('-f', '--force-update', dest='force_update', type=str, help='Update records of mutable entries')
|
|
argparser.add_argument('identifier', type=str, help='user identifier')
|
|
|
|
|
|
def extra_args():
|
|
return {
|
|
'force_update': '_FORCE',
|
|
'method': 'META_LOOKUP_METHOD',
|
|
'meta_url': 'META_URL',
|
|
'identifier': '_IDENTIFIER',
|
|
}
|
|
|
|
|
|
def apply_args(config, args):
|
|
if config.get('META_LOOKUP_METHOD'):
|
|
raise NotImplementedError('Sorry, currently only "phone" lookup method is implemented')
|
|
|
|
|
|
def validate(config, args):
|
|
pass
|
|
|
|
|
|
def execute(ctrl):
|
|
tx_getter = TxGetter(ctrl.get('TX_CACHE_URL'))
|
|
|
|
MetadataRequestsHandler.base_url = ctrl.get('META_URL')
|
|
|
|
store_path = '.clicada'
|
|
user_phone_file_label = 'phone'
|
|
user_phone_store = FileUserStore(ctrl.chain(), user_phone_file_label, store_path, int(ctrl.get('FILESTORE_TTL')))
|
|
|
|
user_address = user_phone_store.by_phone(ctrl.get('_IDENTIFIER'), update=ctrl.get('_FORCE'))
|
|
if user_address == None:
|
|
sys.stderr.write('unknown identifier: {}\n'.format(ctrl.get('_IDENTIFIER')))
|
|
sys.exit(1)
|
|
try:
|
|
user_address = to_checksum_address(user_address)
|
|
except ValueError:
|
|
sys.stderr.write('invalid response "{}" for {}\n'.format(user_address, ctrl.get('_IDENTIFIER')))
|
|
sys.exit(1)
|
|
|
|
logg.debug('loaded user address {} for {}'.format(user_address, ctrl.get('_IDENTIFIER')))
|
|
|
|
txs = tx_getter.get(user_address)
|
|
|
|
token_store = FileTokenStore(ctrl.chain(), ctrl.conn(), 'token', store_path)
|
|
|
|
user_address_file_label = 'address'
|
|
user_address_store = FileUserStore(ctrl.chain(), user_address_file_label, store_path, int(ctrl.get('FILESTORE_TTL')))
|
|
|
|
r = user_address_store.by_address(user_address)
|
|
|
|
print("""Phone: {}
|
|
EVM address: {}
|
|
Name: {}
|
|
Registered: {}
|
|
Gender: {}
|
|
Location: {}
|
|
Products: {}
|
|
""".format(
|
|
ctrl.get('_IDENTIFIER'),
|
|
user_address,
|
|
str(r),
|
|
datetime.datetime.fromtimestamp(r.date_registered).ctime(),
|
|
r.gender,
|
|
r.location['area_name'],
|
|
','.join(r.products),
|
|
)
|
|
)
|
|
|
|
|
|
for tx_src in txs['data']:
|
|
tx = ResolvedTokenTx.from_dict(tx_src)
|
|
tx.resolve(token_store, user_address_store, update=ctrl.get('_FORCE'))
|
|
print(tx)
|