cic-cli/cic/cmd/easy.py

from __future__ import annotations

# standard import
import importlib
import json
import logging
import os
from typing import TYPE_CHECKING

import requests

# external imports
from chainlib.chain import ChainSpec

# local imports
from cic import Proof
from cic.actions.deploy import deploy
from cic.attachment import Attachment
from cic.meta import Meta
from cic.network import Network
from cic.token import Token

if TYPE_CHECKING:
    from cic.cmd.arg import CmdCtrl
    from cic.actions.types import Options, Contract
log = logging.getLogger(__name__)


def process_args(argparser):
    argparser.add_argument(
        "--skip-gen", action="store_true", default=False, help="Skip Generation"
    )
    argparser.add_argument(
        "--skip-deploy",
        action="store_true",
        help="Skip Deployment",
    )
    argparser.add_argument(
        "--target",
        default="eth",
        help="Contract Tech Target (eth)",
    )
    argparser.add_argument(
        "path",
        type=str,
        help="Path to generate/use contract deployment info",
    )
    argparser.add_argument(
        "-p",
        type=str,
        help="RPC Provider (http://localhost:8545)",
    )


def extra_args():
    return {
        "path": "_TOKEN_PATH",
        "skip_gen": "_TOKEN_SKIP_GEN",
        "skip_deploy": "_TOKEN_SKIP_DEPLOY",
        "target": "_TOKEN_TARGET",
        "p": "RPC_PROVIDER",
    }


def validate_args(_args):
    pass


CONTRACTS = [
    {
        "url": "https://gitlab.com/cicnet/eth-erc20/-/raw/master/python/giftable_erc20_token/data/GiftableToken",
        "name": "Giftable Token",
    },
    {
        "url": "https://gitlab.com/cicnet/erc20-demurrage-token/-/raw/master/python/erc20_demurrage_token/data/DemurrageTokenSingleNocap",
        "name": "Demurrage Token Single No Cap",
    },
]

# Download File from Url
def download_file(url: str, directory: str, filename=None) -> (str, bytes):
    os.makedirs(directory, exist_ok=True)
    filename = filename if filename else url.split("/")[-1]
    path = os.path.join(directory, filename)
    if not os.path.exists(path):
        log.debug(f"Downloading {filename}")
        r = requests.get(url, allow_redirects=True)
        open(path, "wb").write(r.content)
        return path
    return path


def get_contract_args(data: list):
    for item in data:
        if item["type"] == "constructor":
            return item["inputs"]
    raise Exception("No constructor found in contract")


def print_contract_args(json_path: str):
    json_data = json.load(open(json_path, encoding="utf-8"))
    print("Contract Args:")
    for contract_arg in get_contract_args(json_data):
        print(
            f"\t{contract_arg.get('name', '<no name>')} - {contract_arg.get('type', '<no type>')}"
        )


def select_contract():
    print("Contracts:")
    print("\t C - Custom (path/url to contract)")
    for idx, contract in enumerate(CONTRACTS):
        print(f"\t {idx} - {contract['name']}")

    val = input("Select contract (C,0,1..): ")
    if val.isdigit() and int(val) < len(CONTRACTS):
        contract = CONTRACTS[int(val)]
        directory = f"./contracts/{contract['name']}"
        bin_path = os.path.abspath(download_file(contract["url"] + ".bin", directory))
        json_path = download_file(contract["url"] + ".json", directory)
    elif val == "C":
        possible_bin_location = input("Enter path/url to contract: ")
        # possible_bin_location is path
        if possible_bin_location[0] == "." or possible_bin_location[0] == "/":
            if os.path.exists(possible_bin_location):
                bin_path = os.path.abspath(possible_bin_location)
            else:
                raise Exception(f"File {possible_bin_location} does not exist")

            possible_json_path = val.replace(".bin", ".json")
            if os.path.exists(possible_json_path):
                json_path = possible_json_path
        # possible_bin_location is url
        else:
            bin_path = download_file(possible_bin_location, directory)
    else:
        print("Invalid selection")
        exit(1)
    contract_extra_args = []
    contract_extra_args_types = []

    if os.path.exists(json_path):
        json_data = json.load(open(json_path, encoding="utf-8"))
        for contract_arg in get_contract_args(json_data):
            arg_name = contract_arg.get("name")
            arg_type = contract_arg.get("type")
            if arg_name not in ["_decimals", "_name", "_symbol"]:
                val = input(f"Enter value for {arg_name} ({arg_type}): ")
                contract_extra_args.append(val)
                if arg_type == "uint128":
                    contract_extra_args_types.append("uint256")
                else:
                    contract_extra_args_types.append(arg_type)


    return {
        "bin_path": bin_path,
        "json_path": json_path,
        "extra_args": contract_extra_args,
        "extra_args_types": contract_extra_args_types,
    }


def init_token(directory: str, code=""):
    contract = select_contract()
    code = contract["bin_path"]
    contract_extra_args = contract["extra_args"]
    contract_extra_args_types = contract["extra_args_types"]

    name = input("Enter Token Name (Foo Token): ") or "Foo Token"
    symbol = input("Enter Token Symbol (FOO): ") or "FOO"
    precision = input("Enter Token Precision (6): ") or 6
    supply = input("Enter Token Supply (0): ") or 0

    contract_token = Token(
        directory,
        name=name,
        symbol=symbol,
        precision=precision,
        extra_args=contract_extra_args,
        extra_args_types=contract_extra_args_types,
        supply=supply,
        code=code,
    )
    contract_token.start()
    return contract_token


def init_proof(directory):
    description = input("Enter Proof Description (None): ") or None
    namespace = input("Enter Proof Namespace (ge): ") or "ge"
    issuer = input("Enter Proof Issuer (None): ") or None
    contract_proof = Proof(directory, description, namespace, issuer)
    contract_proof.start()
    return contract_proof


def init_meta(directory):
    name = input("Enter Name (None): ") or ""
    country_code = input("Enter Country Code (KE): ") or "KE"
    location = input("Enter Location (None): ") or ""
    adding_contact_info = True
    contact = {}
    while adding_contact_info:
        value = input("Enter contact info (e.g 'phone: +254723522718'): ") or None
        if value:
            data = value.split(":")
            if len(data) != 2:
                print("Invalid contact info, you must enter in the format 'key: value'")
                continue
            contact[data[0].strip()] = data[1].strip()
        else:
            adding_contact_info = False
    contract_meta = Meta(
        directory,
        name=name,
        country_code=country_code,
        location=location,
        contact=contact,
    )
    contract_meta.start()
    return contract_meta


def init_attachment(directory):
    contract_attchment = Attachment(directory)
    contract_attchment.start()
    input(
        f"Please add attachment files to '{os.path.abspath(os.path.join(directory,'attachments'))}' and then press ENTER to continue"
    )
    contract_attchment.load()
    return contract_attchment


def load_contract(directory) -> Contract:
    token = Token(path=directory)
    proof = Proof(path=directory)
    meta = Meta(path=directory)
    attachment = Attachment(path=directory)
    network = Network(directory)

    token.load()
    proof.load()
    meta.load()
    attachment.load()
    network.load()
    return Contract(
        token=token, proof=proof, meta=meta, attachment=attachment, network=network
    )





def init_network(
    directory,
    options: Options,
    targets=["eth"],
):
    contract_network = Network(directory, targets=targets)
    contract_network.start()

    for target in targets:
        m = importlib.import_module(f"cic.ext.{target}.start")
        m.extension_start(
            contract_network,
            registry_address=options.contract_registry,
            chain_spec=options.chain_spec,
            rpc_provider=options.rpc_provider,
            key_account_address=options.key_account,
        )
    contract_network.load()
    return contract_network





def generate(directory: str, target: str, options: Options) -> Contract:
    if os.path.exists(directory):
        contine = input(
            "Directory already exists, Would you like to delete it? (y/n): "
        )
        if contine.lower() != "y":
            print("Exiting")
            exit(1)
        else:
            print(f"Deleted {directory}")
            os.system(f"rm -rf {directory}")
    os.makedirs(directory)

    token = init_token(directory)
    proof = init_proof(directory)
    meta = init_meta(directory)
    attachment = init_attachment(directory)
    network = init_network(
        directory,
        options,
        targets=[target],
    )
    return Contract(
        token=token, proof=proof, meta=meta, attachment=attachment, network=network
    )


def get_options(ctrl: CmdCtrl) -> Options:
    # Defaults
    default_contract_registry = ctrl.config.get(
        "CIC_REGISTRY_ADDRESS",
        "0xcf60ebc445b636a5ab787f9e8bc465a2a3ef8299",  # Comes from /home/will/grassroots/cic-staff-installer/var/cic-staff-client/CIC_REGISTRY_ADDRESS
    )
    default_key_account = ctrl.config.get(
        "AUTH_KEY",
        "eb3907ecad74a0013c259d5874ae7f22dcbcc95c",  # comes from wallet  `eth-keyfile -z -d $WALLET_KEY_FILE`
    )
    # https://meta.grassrootseconomics.net
    # https://auth.grassrootseconomics.net  Authenticated Meta

    default_metadata_endpoint = ctrl.config.get("META_URL", "https://auth.grassecon.net")
    # Keyring folder needs to be dumped out as a private key file from $HOME/.config/cic/staff-client/.gnupg
    default_wallet_keyfile = ctrl.config.get(
        "WALLET_KEY_FILE",
        "/home/will/grassroots/cic-internal-integration/apps/cic-ussd/tests/data/pgp/privatekeys_meta.asc",
    )  # Show possible wallet keys

    # Should be an input???
    default_wallet_passphrase = ctrl.config.get("WALLET_PASSPHRASE", "merman")

    default_chain_spec = ctrl.config.get("CHAIN_SPEC", "evm:byzantium:8996:bloxberg")
    default_rpc_provider = ctrl.config.get(
        "RPC_PROVIDER", "https://rpc.grassecon.net"
    )
    contract_registry = (
        input(f"Enter Contract Registry ({default_contract_registry}): ")
        or default_contract_registry
    )
    rpc_provider = (
        input(f"Enter RPC Provider ({default_rpc_provider}): ") or default_rpc_provider
    )
    chain_spec = ChainSpec.from_chain_str(
        (input(f"Enter ChainSpec ({default_chain_spec}): ") or default_chain_spec)
    )
    key_account = (
        input(f"Enter KeyAccount ({default_key_account}): ") or default_key_account
    )
    metadata_endpoint = (
        input(f"Enter Metadata Endpoint ({default_metadata_endpoint}): ")
        or default_metadata_endpoint
    )
    auth_passphrase = ctrl.config.get(
        "AUTH_PASSPHRASE"
    )
    auth_keyfile_path = ctrl.config.get(
        "AUTH_KEYFILE_PATH"
    )
    auth_db_path = ctrl.config.get("AUTH_DB_PATH")
    return Options(
        auth_db_path,
        auth_keyfile_path,
        auth_passphrase,
        contract_registry,
        key_account,
        chain_spec,
        rpc_provider,
        metadata_endpoint,
        default_wallet_keyfile,
        default_wallet_passphrase,
    )


def print_contract(contract: Contract):
    print(f"[cic.header]\nversion = {contract.proof.version()}\n")
    print(f"[cic.token]\n{contract.token}")
    print(f"[cic.proof]\n{contract.proof}")
    print(f"[cic.meta]\n{contract.meta}")
    print(f"[cic.attachment]\n{contract.attachment}")
    print(f"[cic.network]\n{contract.network}")


def execute(ctrl: CmdCtrl):
    directory = ctrl.config.get("_TOKEN_PATH")
    target = ctrl.config.get("_TOKEN_TARGET")
    skip_gen = ctrl.config.get("_TOKEN_SKIP_GEN")
    skip_deploy = ctrl.config.get("_TOKEN_SKIP_DEPLOY")

    options = get_options(ctrl)

    if not skip_gen:
        contract = generate(directory, target, options)
    else:
        contract = load_contract(directory)

    print_contract(contract)

    if not skip_deploy:
        ready_to_deploy = input("Ready to deploy? (y/n): ")
        if ready_to_deploy == "y":
            deploy(
                ctrl=ctrl,
                contract_directory=directory,
                options=options,
                keystore_directory="/home/will/grassroots/cic-internal-integration/apps/contract-migration/keystore",  # Meta Signer meta.ge.net but not auth.ge.net(usumbufu determins if you can even interact with the server) and this ensures data integrity
                target=target,
            )
            print("Deployed")
        else:
            print("Not deploying")


if __name__ == "__main__":
    # execute()
    print("Not Implemented")