cic-cli/cic/contract/components/meta.py

from __future__ import annotations

# standard imports
import os
import json
import logging
import base64
from typing import TYPE_CHECKING

# external imports
from cic_types import MetadataPointer
from cic_types.processor import generate_metadata_pointer
from hexathon import strip_0x

# local imports
from cic.contract.base import Data, data_dir
from cic.writers import OutputWriter
from cic_types.ext.metadata import MetadataRequestsHandler
from cic.utils import object_to_str

logg = logging.getLogger(__name__)


class Meta(Data):
    """Serialize and publish metadata for token.

    The token metadata is any mutable data that is not part of the initial token proof, but published simultaneously as the token nonetheless.

    :param path: Path to settings directory
    :type path: str
    :param writer: Writer interface receiving the output of the processor
    :type writer: cic.writers.OutputWriter
    """

    def __init__(
        self, path=".", writer=None, name="", location="", country_code="KE", contact={}, interactive=False
    ):
        super(Meta, self).__init__()
        self.name = name
        self.contact = contact
        self.country_code = country_code
        self.location = location
        self.path = path
        self.writer = writer
        self.meta_path = os.path.join(self.path, "meta.json")

        if interactive:
            self.name = input(f"Enter Metadata Name ({self.name}): ") or self.name
            self.country_code = input(f"Enter Metadata Country Code ({self.country_code}): ") or self.country_code
            self.location = input(f"Enter Metadata Location ({self.location}): ") or self.location
            
            adding_contact_info = True
            contact = {}
            while adding_contact_info:
                value = input("Enter Metadata 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
            self.contact = contact

    def load(self):
        """Load metadata from settings."""
        super(Meta, self).load()

        f = open(self.meta_path, "r", encoding="utf-8")
        o = json.load(f)
        f.close()

        self.name = o["name"]
        self.contact = o["contact"]
        self.country_code = o["country_code"]
        self.location = o["location"]
        self.inited = True

    def start(self):
        """Initialize metadata settings from template."""
        super(Meta, self).start()

        meta_template_file_path = os.path.join(
            data_dir, f"meta_template_v{self.version()}.json"
        )

        f = open(meta_template_file_path, encoding="utf-8")
        o = json.load(f)
        f.close()

        o["name"] = self.name
        o["contact"] = self.contact
        o["country_code"] = self.country_code
        o["location"] = self.location

        f = open(self.meta_path, "w", encoding="utf-8")
        json.dump(o, f, sort_keys=True, indent="\t")
        f.close()

    def reference(self, token_address):
        """Calculate the mutable reference for the token metadata."""
        token_address_bytes = bytes.fromhex(strip_0x(token_address))
        return generate_metadata_pointer(
            token_address_bytes, MetadataPointer.TOKEN_META
        )

    def asdict(self):
        """Output proof state to dict."""
        return {
            "name": self.name,
            "country_code": self.country_code,
            "location": self.location,
            "contact": self.contact,
        }

    def process(self, token_address=None, token_symbol=None, writer=None):
        """Serialize and publish metadata.

        See cic.processor.Processor.process
        """
        if writer is None:
            writer = self.writer

        v = json.dumps(self.asdict(), separators=(",", ":"))

        token_address_bytes = bytes.fromhex(strip_0x(token_address))
        k = generate_metadata_pointer(token_address_bytes, MetadataPointer.TOKEN_META)
        writer.write(k, v.encode("utf-8"))

        token_symbol_bytes = token_symbol.encode("utf-8")
        k = generate_metadata_pointer(
            token_symbol_bytes, MetadataPointer.TOKEN_META_SYMBOL
        )
        writer.write(k, v.encode("utf-8"))

        return (k, v)

    def __str__(self):
        return object_to_str(self, ["name", "contact", "country_code", "location"])


class MetadataWriter(OutputWriter):
    """Custom writer for publishing data under immutable content-addressed pointers in the cic-meta storage backend.

    Data that is not utf-8 will be converted to base64 before publishing.

    Implements cic.writers.OutputWriter
    """

    def write(self, k, v):
        rq = MetadataRequestsHandler(MetadataPointer.NONE, bytes.fromhex(k))
        try:
            v = v.decode("utf-8")
            v = json.loads(v)
            logg.debug(f"metadatawriter bindecode {k} {v}")
        except UnicodeDecodeError:
            v = base64.b64encode(v).decode("utf-8")
            v = json.loads(json.dumps(v, separators=(",", ":")))
            logg.debug(f"metadatawriter b64encode {k} {v}")
        r = rq.create(v)
        logg.info(f"metadata submitted at {k}")
        return r