225 lines
6.8 KiB
Python
225 lines
6.8 KiB
Python
# Standard
|
|
import importlib
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import TYPE_CHECKING, List
|
|
|
|
import requests
|
|
from chainlib.chain import ChainSpec
|
|
from chainlib.cli.config import Config
|
|
from cic.contract.components.attachment import Attachment
|
|
from cic.contract.components.meta import Meta
|
|
from cic.contract.components.proof import Proof
|
|
from cic.contract.components.token import Token
|
|
from cic.contract.helpers import init_writers_from_config
|
|
from cic.contract.network import Network
|
|
|
|
# Local Modules
|
|
from cic.contract.processor import ContractProcessor
|
|
from cic.writers import HTTPWriter, KeyedWriterFactory, MetadataWriter, OutputWriter
|
|
|
|
# external imports
|
|
from cic_types.ext.metadata import MetadataRequestsHandler
|
|
from cic_types.ext.metadata.signer import Signer as MetadataSigner
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
class Contract:
|
|
""" """
|
|
|
|
def __init__(
|
|
self,
|
|
token: Token,
|
|
proof: Proof,
|
|
meta: Meta,
|
|
attachment: Attachment,
|
|
network: Network,
|
|
):
|
|
self.token = token
|
|
self.proof = proof
|
|
self.meta = meta
|
|
self.attachment = attachment
|
|
self.network = network
|
|
|
|
def __str__(self):
|
|
s = ""
|
|
s += f"\n[cic.header]\nversion = {self.proof.version()}\n\n"
|
|
s += f"[cic.token]\n{self.token}\n"
|
|
s += f"[cic.proof]\n{self.proof}\n"
|
|
s += f"[cic.meta]\n{self.meta}\n"
|
|
s += f"[cic.attachment]\n{self.attachment}\n"
|
|
s += f"[cic.network]\n{self.network}\n"
|
|
return s
|
|
|
|
|
|
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 generate_contract(
|
|
directory: str, targets: List[str], config, interactive=True
|
|
) -> Contract:
|
|
if os.path.exists(directory):
|
|
contine = input(
|
|
f"Directory {directory} already exists, Would you like to delete it? (y/n): "
|
|
)
|
|
if contine.lower() != "y":
|
|
log.debug("Trying to load existing contract")
|
|
return load_contract(directory)
|
|
else:
|
|
print(f"Deleted {directory}")
|
|
os.system(f"rm -rf {directory}")
|
|
os.makedirs(directory)
|
|
log.debug("Generating token")
|
|
token = Token(directory, interactive=interactive)
|
|
token.start()
|
|
|
|
log.debug("Generating proof")
|
|
proof = Proof(directory, interactive=interactive)
|
|
proof.start()
|
|
|
|
log.debug("Generating meta")
|
|
meta = Meta(directory, interactive=interactive)
|
|
meta.start()
|
|
|
|
log.debug("Generating attachment")
|
|
attachment = Attachment(directory, interactive=interactive)
|
|
|
|
log.debug("Generating network")
|
|
network = Network(directory, targets=targets)
|
|
network.start()
|
|
|
|
log.debug(
|
|
f"""Populating infomation from network:
|
|
CIC_REGISTRY_ADDRESS: {config.get("CIC_REGISTRY_ADDRESS")}
|
|
CHAIN_SPEC: {config.get("CHAIN_SPEC")}
|
|
RPC_PROVIDER: {config.get("RPC_PROVIDER")}
|
|
"""
|
|
)
|
|
for target in targets:
|
|
# TODO Clean this up
|
|
modname = f"cic.ext.{target}"
|
|
cmd_mod = importlib.import_module(modname)
|
|
signer_hint = config.get("WALLET_KEY_FILE")
|
|
keys = cmd_mod.list_keys(config, signer_hint)
|
|
if len(keys) > 1:
|
|
print(f"More than one key found, please select one:")
|
|
for idx, key in enumerate(keys):
|
|
print(f"{idx} - {key} ")
|
|
selecting_key = True
|
|
while selecting_key:
|
|
idx = int(input("Select key: "))
|
|
if keys[idx] is not None:
|
|
key_account_address = keys[idx]
|
|
selecting_key = False
|
|
else:
|
|
print("Invalid key, try again")
|
|
else:
|
|
key_account_address = keys[0]
|
|
|
|
m = importlib.import_module(f"cic.ext.{target}.start")
|
|
m.extension_start(
|
|
network,
|
|
registry_address=config.get("CIC_REGISTRY_ADDRESS"),
|
|
chain_spec=ChainSpec.from_chain_str(config.get("CHAIN_SPEC")),
|
|
rpc_provider=config.get("RPC_PROVIDER"),
|
|
key_account_address=key_account_address,
|
|
)
|
|
network.load()
|
|
|
|
return Contract(
|
|
token=token, proof=proof, meta=meta, attachment=attachment, network=network
|
|
)
|
|
|
|
|
|
def deploy_contract(
|
|
config: Config,
|
|
target: str,
|
|
contract_directory: str,
|
|
):
|
|
|
|
modname = f"cic.ext.{target}"
|
|
cmd_mod = importlib.import_module(modname)
|
|
|
|
writers = init_writers_from_config(config)
|
|
output_directory = os.path.join(contract_directory, "out")
|
|
output_writer_path_meta = output_directory
|
|
|
|
metadata_endpoint = config.get("META_URL")
|
|
|
|
if metadata_endpoint is not None:
|
|
MetadataRequestsHandler.base_url = metadata_endpoint
|
|
MetadataSigner.gpg_path = "/tmp"
|
|
MetadataSigner.key_file_path = config.get("AUTH_KEYFILE_PATH")
|
|
MetadataSigner.gpg_passphrase = config.get("AUTH_PASSPHRASE")
|
|
writers["proof"] = KeyedWriterFactory(MetadataWriter, HTTPWriter).new
|
|
writers["attachment"] = KeyedWriterFactory(None, HTTPWriter).new
|
|
writers["meta"] = MetadataWriter
|
|
output_writer_path_meta = metadata_endpoint
|
|
|
|
ct = Token(path=contract_directory)
|
|
cm = Meta(
|
|
path=contract_directory, writer=writers["meta"](path=output_writer_path_meta)
|
|
)
|
|
ca = Attachment(
|
|
path=contract_directory,
|
|
writer=writers["attachment"](path=output_writer_path_meta),
|
|
)
|
|
cp = Proof(
|
|
path=contract_directory,
|
|
attachments=ca,
|
|
writer=writers["proof"](path=output_writer_path_meta),
|
|
)
|
|
cn = Network(path=contract_directory)
|
|
|
|
ca.load()
|
|
ct.load()
|
|
cp.load()
|
|
cm.load()
|
|
cn.load()
|
|
|
|
chain_spec = None
|
|
try:
|
|
chain_spec = config.get("CHAIN_SPEC")
|
|
log.debug(f"using CHAIN_SPEC from config: {chain_spec}")
|
|
except KeyError:
|
|
chain_spec = cn.chain_spec
|
|
config.add(chain_spec, "CHAIN_SPEC", exists_ok=True)
|
|
log.debug(f"using CHAIN_SPEC: {str(chain_spec)} from network")
|
|
|
|
signer_hint = config.get("WALLET_KEY_FILE")
|
|
(rpc, signer) = cmd_mod.parse_adapter(config, signer_hint)
|
|
|
|
target_network_reference = cn.resource(target)
|
|
chain_spec = cn.chain_spec(target)
|
|
log.debug(
|
|
f'found reference {target_network_reference["contents"]} chain spec {chain_spec} for target {target}'
|
|
)
|
|
c = getattr(cmd_mod, "new")(
|
|
chain_spec,
|
|
target_network_reference["contents"],
|
|
cp,
|
|
signer_hint=signer,
|
|
rpc=rpc,
|
|
outputs_writer=writers["ext"](path=output_directory),
|
|
)
|
|
c.apply_token(ct)
|
|
|
|
p = ContractProcessor(proof=cp, attachment=ca, metadata=cm, extensions=[c])
|
|
p.process()
|