clicada/clicada/cli/user.py

# 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,
    )

    r = None
    ctrl.write(
        f"""
Phone: {ctrl.get("_IDENTIFIER")}
Network address: {add_0x(user_address)}
Chain: {ctrl.chain().common_name()}"""
    )
    ctrl.notify("resolving metadata for address {}".format(user_address_normal))
    try:
        r = user_address_store.by_address(
            user_address_normal, update=ctrl.get("_FORCE")
        )
        if r:
            ctrl.write(
                f"""
Name: { str(r)}
Registered: {datetime.datetime.fromtimestamp(r.date_registered).ctime()}
Gender: {r.gender}
Location: {r.location["area_name"]}
Products: {",".join(r.products)}
Tags: {",".join(r.tags)}"""
            )
    except MetadataNotFoundError as e:
        ctrl.ouch(f"MetadataNotFoundError: Could not resolve metadata for user {e}\n")

    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)