codekingpro/portable-devtools

1import json2import os3import re4import time5from typing import Iterator, List, Literal, Optional6 7import requests8from langchain_core.documents import Document9 10from langchain_community.document_loaders.base import BaseLoader11 12 13class MintbaseDocumentLoader(BaseLoader):14    """Load elements from a blockchain smart contract.15 16    The supported blockchains are: Near mainnet, Near testnet.17 18    If no BlockchainType is specified, the default is Near mainnet.19 20    The Loader uses the Mintbase API to interact with the blockchain.21    MB_API_KEY environment variable must be set to use this loader.22 23    The API returns 100 NFTs per request and can be paginated using the24    startToken parameter.25 26    If get_all_tokens is set to True, the loader will get all tokens27    on the contract.  Note that for contracts with a large number of tokens,28    this may take a long time (e.g. 10k tokens is 100 requests).29    Default value is false for this reason.30 31    The max_execution_time (sec) can be set to limit the execution time32    of the loader.33 34    Future versions of this loader can:35        - Support additional Mintbase APIs (e.g. getTokens, etc.)36 37    Example:38        .. code-block:: python39 40            contractAddress = "nft.yearofchef.near"  # Year of chef contract address41            blockchainLoader = MintbaseDocumentLoader(42                contract_address=contractAddress, blockchain_type="mainnet",api_key="omni-site"43            )44    """  # noqa: E50145 46    def __init__(47        self,48        contract_address: str,49        *,50        blockchain_type: Literal["mainnet", "testnet"],51        api_key: str = "",52        table: str = "",53        select: str = "",54        fields: Optional[List[str]] = None,55        get_all_tokens: bool = False,56        max_execution_time: Optional[int] = None,57    ):58        """59 60        Args:61            contract_address: The address of the smart contract.62            blockchainType: The blockchain type.63            api_key: The Mintbase API key.64            table: name of the table to query65            select: Conditions for querying66            fields: Information to display after query67            get_all_tokens: Whether to get all tokens on the contract.68            max_execution_time: The maximum execution time (sec).69        """70        self.contract_address = contract_address71        self.blockchainType = blockchain_type72        self.api_key = os.environ.get("MB_API_KEY") or api_key73        self.table = "mb_views_nft_tokens" or table74        self.select = 'where: {nft_contract_id: {_eq: "contract_address"}}' or select75        self.fields = fields or [76            "base_uri",77            "burned_receipt_id",78            "burned_timestamp",79            "copies",80            "description",81            "expires_at",82            "extra",83            "issued_at",84            "last_transfer_receipt_id",85            "last_transfer_timestamp",86            "media",87            "media_hash",88            "metadata_content_flag",89            "metadata_id",90            "mint_memo",91            "minted_receipt_id",92            "minted_timestamp",93            "minter",94            "nft_contract_content_flag",95            "nft_contract_created_at",96            "nft_contract_icon",97            "nft_contract_id",98            "nft_contract_is_mintbase",99            "nft_contract_name",100            "nft_contract_owner_id",101            "nft_contract_reference",102            "nft_contract_spec",103            "nft_contract_symbol",104            "owner",105            "reference",106            "reference_blob",107            "reference_hash",108            "royalties",109            "royalties_percent",110            "splits",111            "starts_at",112            "title",113            "token_id",114            "updated_at",115        ]116 117        self.get_all_tokens = get_all_tokens118        self.max_execution_time = max_execution_time119 120        if not self.api_key:121            raise ValueError("Mintbase API key not provided.")122 123        if not re.match(124            r"^(([a-z\d]+[\-_])*[a-z\d]+\.)*([a-z\d]+[\-_])*[a-z\d]+$",125            self.contract_address,126        ):127            raise ValueError(f"Invalid contract address {self.contract_address}")128 129    def load(self) -> List[Document]:130        result = []131 132        start_time = time.time()133 134        while True:135            # Define the GraphQL query as a multi-line string136            operations_doc = """137            query MyQuery {138            table(select) {139                fields140            }141            }142            """143 144            # Replace the placeholder with the actual contract address145            operations_doc = operations_doc.replace("select", self.select)146            operations_doc = operations_doc.replace(147                "contract_address", self.contract_address148            )149            operations_doc = operations_doc.replace("table", self.table)150            operations_doc = operations_doc.replace("fields", "\n".join(self.fields))151 152            # Define the headers153            headers = {"mb-api-key": self.api_key, "Content-Type": "application/json"}154            # Define the POST data155            data = {156                "query": operations_doc,157                "variables": {},158                "operationName": "MyQuery",159            }160 161            url = f"https://graph.mintbase.xyz/{self.blockchainType}"162 163            response = requests.post(url, headers=headers, data=json.dumps(data))164 165            if response.status_code != 200:166                raise ValueError(167                    f"Request failed with status code {response.status_code}"168                )169 170            items = response.json()["data"]["mb_views_nft_tokens"]171 172            if not items:173                break174 175            for item in items:176                content = str(item)177                token_id = item["token_id"]178                metadata = {179                    "source": self.contract_address,180                    "blockchain": self.blockchainType,181                    "tokenId": token_id,182                }183                result.append(Document(page_content=content, metadata=metadata))184 185            # exit after the first API call if get_all_tokens is False186            if not self.get_all_tokens:187                break188 189            if (190                self.max_execution_time is not None191                and (time.time() - start_time) > self.max_execution_time192            ):193                raise RuntimeError("Execution time exceeded the allowed time limit.")194 195        if not result:196            raise ValueError(197                f"No NFTs found for contract address {self.contract_address}"198            )199 200        return result201 202    def lazy_load(self) -> Iterator[Document]:203        start_time = time.time()204 205        while True:206            # Define the GraphQL query as a multi-line string207            operations_doc = """208            query MyQuery {209            table(select) {210                fields211            }212            }213            """214 215            # Replace the placeholder with the actual contract address216            operations_doc = operations_doc.replace("select", self.select)217            operations_doc = operations_doc.replace(218                "contract_address", self.contract_address219            )220            operations_doc = operations_doc.replace("table", self.table)221            operations_doc = operations_doc.replace("fields", "\n".join(self.fields))222 223            # Define the headers224            headers = {"mb-api-key": self.api_key, "Content-Type": "application/json"}225            # Define the POST data226            data = {227                "query": operations_doc,228                "variables": {},229                "operationName": "MyQuery",230            }231 232            url = f"https://graph.mintbase.xyz/{self.blockchainType}"233 234            response = requests.post(url, headers=headers, data=json.dumps(data))235 236            if response.status_code != 200:237                raise ValueError(238                    f"Request failed with status code {response.status_code}"239                )240 241            items = response.json()["data"]["mb_views_nft_tokens"]242 243            if not items:244                break245 246            for item in items:247                content = str(item)248                tokenId = item["token_id"]249                metadata = {250                    "source": self.contract_address,251                    "blockchain": self.blockchainType,252                    "tokenId": tokenId,253                }254                yield Document(page_content=content, metadata=metadata)255 256            # exit after the first API call if get_all_tokens is False257            if not self.get_all_tokens:258                break259 260            if (261                self.max_execution_time is not None262                and (time.time() - start_time) > self.max_execution_time263            ):264                raise RuntimeError("Execution time exceeded the allowed time limit.")265