124 lines
4.6 KiB
Python
124 lines
4.6 KiB
Python
# standard imports
|
|
import datetime
|
|
import logging
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from chainlib.encode import TxHexNormalizer
|
|
from chainlib.eth.address import is_address, to_checksum_address
|
|
# external imports
|
|
from cic_eth_registry import CICRegistry
|
|
from cic_eth_registry.lookup.tokenindex import TokenIndexLookup
|
|
from cic_types.models.person import Person
|
|
from clicada.error import MetadataNotFoundError
|
|
from clicada.token import FileTokenStore, token_balance
|
|
# local imports
|
|
from clicada.tx import ResolvedTokenTx, TxGetter
|
|
from clicada.user import FileUserStore
|
|
from hexathon import add_0x
|
|
|
|
logg = logging.getLogger(__name__)
|
|
|
|
tx_normalizer = TxHexNormalizer()
|
|
|
|
|
|
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', action='store_true', help='Update records of mutable entries')
|
|
argparser.add_argument('identifier', type=str, help='user identifier (phone_number or address)')
|
|
|
|
|
|
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'), 10)
|
|
|
|
store_path = os.path.join(str(Path.home()), '.clicada')
|
|
user_phone_file_label = 'phone'
|
|
user_phone_store = FileUserStore(ctrl.opener('meta'), ctrl.chain(), user_phone_file_label, store_path, int(ctrl.get('FILESTORE_TTL')), encrypter=ctrl.encrypter)
|
|
|
|
identifier = ctrl.get('_IDENTIFIER')
|
|
ctrl.notify('resolving identifier {} to wallet address'.format(identifier))
|
|
if is_address(identifier):
|
|
user_address = identifier
|
|
else:
|
|
user_address = user_phone_store.by_phone(identifier, update=ctrl.get('_FORCE'))
|
|
if user_address == None:
|
|
ctrl.ouch('unknown identifier: {}\n'.format(identifier))
|
|
sys.exit(1)
|
|
try:
|
|
user_address = to_checksum_address(user_address)
|
|
except ValueError:
|
|
ctrl.ouch('invalid response "{}" for {}\n'.format(user_address, identifier))
|
|
sys.exit(1)
|
|
|
|
logg.debug('loaded user address {} for {}'.format(user_address, identifier))
|
|
|
|
user_address_normal = tx_normalizer.wallet_address(user_address)
|
|
ctrl.notify('retrieving txs for address {}'.format(user_address_normal))
|
|
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.opener('meta'), ctrl.chain(), user_address_file_label, store_path, int(ctrl.get('FILESTORE_TTL')), encrypter=ctrl.encrypter)
|
|
|
|
ctrl.notify('resolving metadata for address {}'.format(user_address_normal))
|
|
r = None
|
|
try:
|
|
r = user_address_store.by_address(user_address_normal, update=ctrl.get('_FORCE'))
|
|
except MetadataNotFoundError as e:
|
|
ctrl.ouch('could not resolve metadata for user: {}'.format(e))
|
|
|
|
ctrl.write(f"""Phone: {ctrl.get('_IDENTIFIER')}
|
|
Network address: {add_0x(user_address)}
|
|
Chain: {ctrl.chain().common_name()}
|
|
Name: {str(r)}
|
|
Registered: {r and datetime.datetime.fromtimestamp(r).ctime()}
|
|
Gender: {r and r.gender}
|
|
Location: {r and r.location['area_name']}
|
|
Products: {r and ','.join(r.products)}
|
|
Tags: {r and ','.join(r.tags)}"""
|
|
)
|
|
|
|
tx_lines = []
|
|
seen_tokens = {}
|
|
for tx_src in txs['data']:
|
|
ctrl.notify('resolve details for tx {}'.format(tx_src['tx_hash']))
|
|
tx = ResolvedTokenTx.from_dict(tx_src)
|
|
tx.resolve(token_store, user_address_store, show_decimals=True, update=ctrl.get('_FORCE'))
|
|
tx_lines.append(tx)
|
|
seen_tokens[tx.source_token_label] = tx.source_token
|
|
seen_tokens[tx.destination_token_label] = tx.destination_token
|
|
|
|
for k in seen_tokens.keys():
|
|
ctrl.notify('resolve token {}'.format(seen_tokens[k]))
|
|
(token_symbol, token_decimals) = token_store.by_address(seen_tokens[k])
|
|
ctrl.notify('get token balance for {} => {}'.format(token_symbol, seen_tokens[k]))
|
|
balance = token_balance(ctrl.chain(), ctrl.conn(), seen_tokens[k], user_address)
|
|
fmt = '{:.' + str(token_decimals) + 'f}'
|
|
decimal_balance = fmt.format(balance / (10 ** token_decimals))
|
|
ctrl.write("Balances:\n {} {}".format(token_symbol, decimal_balance))
|
|
|
|
print()
|
|
for l in tx_lines:
|
|
ctrl.write(l)
|