codekingpro/portable-devtools

1import os2import re3from typing import Iterator, List4 5import requests6from langchain_core.documents import Document7 8from langchain_community.document_loaders.base import BaseLoader9 10 11class EtherscanLoader(BaseLoader):12    """Load transactions from `Ethereum` mainnet.13 14    The Loader use Etherscan API to interact with Ethereum mainnet.15 16    ETHERSCAN_API_KEY environment variable must be set use this loader.17    """18 19    def __init__(20        self,21        account_address: str,22        api_key: str = "docs-demo",23        filter: str = "normal_transaction",24        page: int = 1,25        offset: int = 10,26        start_block: int = 0,27        end_block: int = 99999999,28        sort: str = "desc",29    ):30        self.account_address = account_address31        self.api_key = os.environ.get("ETHERSCAN_API_KEY") or api_key32        self.filter = filter33        self.page = page34        self.offset = offset35        self.start_block = start_block36        self.end_block = end_block37        self.sort = sort38 39        if not self.api_key:40            raise ValueError("Etherscan API key not provided")41 42        if not re.match(r"^0x[a-fA-F0-9]{40}$", self.account_address):43            raise ValueError(f"Invalid contract address {self.account_address}")44        if filter not in [45            "normal_transaction",46            "internal_transaction",47            "erc20_transaction",48            "eth_balance",49            "erc721_transaction",50            "erc1155_transaction",51        ]:52            raise ValueError(f"Invalid filter {filter}")53 54    def lazy_load(self) -> Iterator[Document]:55        """Lazy load Documents from table."""56        result = []57        if self.filter == "normal_transaction":58            result = self.getNormTx()59        elif self.filter == "internal_transaction":60            result = self.getInternalTx()61        elif self.filter == "erc20_transaction":62            result = self.getERC20Tx()63        elif self.filter == "eth_balance":64            result = self.getEthBalance()65        elif self.filter == "erc721_transaction":66            result = self.getERC721Tx()67        elif self.filter == "erc1155_transaction":68            result = self.getERC1155Tx()69        else:70            raise ValueError(f"Invalid filter {filter}")71        for doc in result:72            yield doc73 74    def getNormTx(self) -> List[Document]:75        url = (76            f"https://api.etherscan.io/api?module=account&action=txlist&address={self.account_address}"77            f"&startblock={self.start_block}&endblock={self.end_block}&page={self.page}"78            f"&offset={self.offset}&sort={self.sort}&apikey={self.api_key}"79        )80        try:81            response = requests.get(url)82            response.raise_for_status()83        except requests.exceptions.RequestException as e:84            print("Error occurred while making the request:", e)  # noqa: T20185        items = response.json()["result"]86        result = []87        if len(items) == 0:88            return [Document(page_content="")]89        for item in items:90            content = str(item)91            metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}92            result.append(Document(page_content=content, metadata=metadata))93        print(len(result))  # noqa: T20194        return result95 96    def getEthBalance(self) -> List[Document]:97        url = (98            f"https://api.etherscan.io/api?module=account&action=balance"99            f"&address={self.account_address}&tag=latest&apikey={self.api_key}"100        )101 102        try:103            response = requests.get(url)104            response.raise_for_status()105        except requests.exceptions.RequestException as e:106            print("Error occurred while making the request:", e)  # noqa: T201107        return [Document(page_content=response.json()["result"])]108 109    def getInternalTx(self) -> List[Document]:110        url = (111            f"https://api.etherscan.io/api?module=account&action=txlistinternal"112            f"&address={self.account_address}&startblock={self.start_block}"113            f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"114            f"&sort={self.sort}&apikey={self.api_key}"115        )116 117        try:118            response = requests.get(url)119            response.raise_for_status()120        except requests.exceptions.RequestException as e:121            print("Error occurred while making the request:", e)  # noqa: T201122        items = response.json()["result"]123        result = []124        if len(items) == 0:125            return [Document(page_content="")]126        for item in items:127            content = str(item)128            metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}129            result.append(Document(page_content=content, metadata=metadata))130        return result131 132    def getERC20Tx(self) -> List[Document]:133        url = (134            f"https://api.etherscan.io/api?module=account&action=tokentx"135            f"&address={self.account_address}&startblock={self.start_block}"136            f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"137            f"&sort={self.sort}&apikey={self.api_key}"138        )139 140        try:141            response = requests.get(url)142            response.raise_for_status()143        except requests.exceptions.RequestException as e:144            print("Error occurred while making the request:", e)  # noqa: T201145        items = response.json()["result"]146        result = []147        if len(items) == 0:148            return [Document(page_content="")]149        for item in items:150            content = str(item)151            metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}152            result.append(Document(page_content=content, metadata=metadata))153        return result154 155    def getERC721Tx(self) -> List[Document]:156        url = (157            f"https://api.etherscan.io/api?module=account&action=tokennfttx"158            f"&address={self.account_address}&startblock={self.start_block}"159            f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"160            f"&sort={self.sort}&apikey={self.api_key}"161        )162 163        try:164            response = requests.get(url)165            response.raise_for_status()166        except requests.exceptions.RequestException as e:167            print("Error occurred while making the request:", e)  # noqa: T201168        items = response.json()["result"]169        result = []170        if len(items) == 0:171            return [Document(page_content="")]172        for item in items:173            content = str(item)174            metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}175            result.append(Document(page_content=content, metadata=metadata))176        return result177 178    def getERC1155Tx(self) -> List[Document]:179        url = (180            f"https://api.etherscan.io/api?module=account&action=token1155tx"181            f"&address={self.account_address}&startblock={self.start_block}"182            f"&endblock={self.end_block}&page={self.page}&offset={self.offset}"183            f"&sort={self.sort}&apikey={self.api_key}"184        )185 186        try:187            response = requests.get(url)188            response.raise_for_status()189        except requests.exceptions.RequestException as e:190            print("Error occurred while making the request:", e)  # noqa: T201191        items = response.json()["result"]192        result = []193        if len(items) == 0:194            return [Document(page_content="")]195        for item in items:196            content = str(item)197            metadata = {"from": item["from"], "tx_hash": item["hash"], "to": item["to"]}198            result.append(Document(page_content=content, metadata=metadata))199        return result200