[ASSET-27] Add Routescan puller (#33)

* ASSET-24 Fix bugs Signed-off-by: jormal <[email protected]> * ASSET-24 Add routescan puller Signed-off-by: jormal <[email protected]> * ASSET-24 Update version Signed-off-by: jormal <[email protected]> * ASSET-28 Fix dependency Signed-off-by: jormal <[email protected]> --------- Signed-off-by: jormal <[email protected]>
bifrost-platform · Jun 10, 2024 · ad09895 · ad09895
1 parent b65367e
commit ad09895
Show file tree

Hide file tree

Showing 8 changed files with 146 additions and 12 deletions.
diff --git a/VERSION b/VERSION
@@ -1 +1 @@
-2.0.1
+2.0.2
diff --git a/libraries/puller/runner.py b/libraries/puller/runner.py
@@ -16,12 +16,14 @@
 from libraries.puller.token_pullers.token_puller_klaytnscope import (
     TokenPullerKlaytnscope,
 )
+from libraries.puller.token_pullers.token_puller_routescan import TokenPullerRoutescan
 
 TOKEN_PULLER_CLASS_MAP: dict[Id, Type[TokenPullerAbstracted]] = {
     Id("blockscout"): TokenPullerBlockscout,
     Id("dexguru"): TokenPullerDexguru,
     Id("etherscan"): TokenPullerEtherscan,
     Id("klaytnscope"): TokenPullerKlaytnscope,
+    Id("routescan"): TokenPullerRoutescan,
 }
 
 

diff --git a/libraries/puller/token_pullers/token_puller_abstracted.py b/libraries/puller/token_pullers/token_puller_abstracted.py
@@ -127,9 +127,8 @@ def __run_body(self, address: Address) -> None:
         address, name, symbol, decimals = self.__get_contract_info(address)
         printf(HTML(f"<b>  Name: {name}</b>"))
         printf(HTML(f"<b>  Symbol: {symbol}</b>"))
-        printf(
-            HTML(f"<b>  Source: <skyblue>{self._get_token_url(address)}</skyblue></b>")
-        )
+        printed_url = str(self._get_token_url(address)).replace("&", "&amp;")
+        printf(HTML(f"<b>  Source: <skyblue>{printed_url}</skyblue></b>"))
         info = self.network_assets.get(address, None)
         if info is not None and not confirm("Would you like to renew the information?"):
             return None
@@ -215,7 +214,8 @@ def __check_node_url(network: Network, url: URL) -> None:
                 not web3.is_connected()
                 or str(web3.eth.chain_id) != str(network.id).split("-")[-1]
             ):
-                printf(HTML(f"<red>Invalid node URL: {url}</red>"))
+                printed_url = str(url).replace("&", "&amp;")
+                printf(HTML(f"<red>Invalid node URL: {printed_url}</red>"))
                 raise ValueError("Invalid node URL")
 
     @staticmethod
@@ -395,7 +395,8 @@ def __save_image(
         # Get the image URL
         if (token_image_url := self._get_token_image_url(address)) is None:
             return None
-        printf(HTML(f"Image URL found: <skyblue>{token_image_url}</skyblue>"))
+        printed_url = str(token_image_url).replace("&", "&amp;")
+        printf(HTML(f"Image URL found: <skyblue>{printed_url}</skyblue>"))
         if not confirm("Would you like to download the image?"):
             return None
         # Download the image
@@ -404,8 +405,13 @@ def __save_image(
         # Save the image
         image_path = Path(mkdtemp(prefix=str(info.id), dir=self.tmp_dir))
         downloaded_type = list()
-        if ".png" in token_image_url.suffix or token_image_url.path.startswith(
-            "image/png"
+        if (
+            # Normal case
+            ".png" in token_image_url.suffix
+            # KlaytnScope case
+            or token_image_url.path.startswith("image/png")
+            # Routescan case
+            or ".png" in URL(token_image_url.query.get("url", "")).suffix
         ):
             downloaded_type.extend(self.__save_png_image(image_path, image))
         elif ".svg" in token_image_url.suffix:

diff --git a/libraries/puller/token_pullers/token_puller_blockscout.py b/libraries/puller/token_pullers/token_puller_blockscout.py
@@ -64,7 +64,8 @@ def _get_token_image_url(self, address: Address) -> URL | None:
                 available_images.update({Id("original"): base})
             printf(HTML(f"⎡ <b>Available images for {address}:</b>"))
             for size, url in available_images.items():
-                printf(HTML(f"⎢ <b>∙ {size}</b>: {url}"))
+                printed_url = str(url).replace("&", "&amp;")
+                printf(HTML(f"⎢ <b>∙ {size}</b>: {printed_url}"))
             selected_type = get_id(
                 "⎣ Select the image type",
                 None,

diff --git a/libraries/puller/token_pullers/token_puller_etherscan.py b/libraries/puller/token_pullers/token_puller_etherscan.py
@@ -82,7 +82,7 @@ def _get_token_image_url(self, address: Address) -> URL | None:
                 url = sub(r".png", f"_{size.size}.png", base_url)
             response = get(url, headers=HEADER)
             if response.status_code == 200 and response.url == url:
-                available_images.update({Id(size.lower()): url})
+                available_images.update({Id(str(size).lower()): url})
         if base_url not in available_images.values():
             available_images.update({Id("original"): base_url})
         # Select the image type.
@@ -91,7 +91,8 @@ def _get_token_image_url(self, address: Address) -> URL | None:
         else:
             printf(HTML(f"⎡ <b>Available images for {address}:</b>"))
             for size, url in available_images.items():
-                printf(HTML(f"⎢ <b>∙ {size}</b>: {url}"))
+                printed_url = str(url).replace("&", "&amp;")
+                printf(HTML(f"⎢ <b>∙ {size}</b>: {printed_url}"))
             selected_type = get_id(
                 "⎣ Select the image type",
                 None,

diff --git a/libraries/puller/token_pullers/token_puller_routescan.py b/libraries/puller/token_pullers/token_puller_routescan.py
@@ -0,0 +1,124 @@
+from bs4 import BeautifulSoup
+from prompt_toolkit import print_formatted_text as printf, HTML
+from requests import get
+from web3 import Web3
+from yarl import URL
+
+from libraries.models.network import Network
+from libraries.models.terminals.address import Address
+from libraries.models.terminals.id import Id
+from libraries.preprocess.image import PNG_TYPES
+from libraries.puller.getters.id_getter import get_id
+from libraries.puller.getters.token_count_getter import TOKEN_COUNT_PER_PAGE
+from libraries.puller.token_pullers.token_puller_abstracted import TokenPullerAbstracted
+
+ROUTESCAN_API_URL: URL = URL("https://api.routescan.io/")
+TOKEN_IMAGE_SELECTOR: str = "#token > div > div > div > div > div > span > img"
+
+
+class TokenPullerRoutescan(TokenPullerAbstracted):
+    """Token puller using Routescan explorer.
+
+    Attributes:
+        routescan_url: The URL of the Routescan explorer.
+    """
+
+    routescan_url: URL
+
+    def __init__(self, network: Network) -> None:
+        """Initialize the token puller routescan class.
+
+        Args:
+            network: The network information.
+        """
+        super().__init__(network)
+        self.routescan_url = URL(
+            str(next(filter(lambda x: x.id == "routescan", self.network.explorers)).url)
+        )
+
+    def _get_top_token_list(self) -> set[tuple[int, Address]]:
+        addresses = []
+        path = (
+            URL("/v2/network/mainnet/")
+            / str(self.network.id).replace("-", "/")
+            / "erc20"
+        )
+        while len(addresses) < self.token_count:
+            token_list, path = self.__get_token_list(path)
+            addresses.extend(
+                [
+                    (idx, address)
+                    for idx, address in enumerate(token_list, len(addresses))
+                ]
+            )
+        return addresses
+
+    def _get_token_url(self, address: Address) -> URL:
+        return self.routescan_url / "token" / str(address)
+
+    def _get_token_image_url(self, address: Address) -> URL | None:
+        # Get the token page for getting the token image URL.
+        token_page = get(str(self._get_token_url(address)))
+        if token_page.status_code != 200:
+            return None
+        soup = BeautifulSoup(token_page.content, "html.parser")
+        if (
+            image_src := soup.select_one(TOKEN_IMAGE_SELECTOR).get("src", None)
+        ) is None:
+            return None
+        base_url = URL(image_src)
+        available_images: dict[Id, str] = dict()
+        for size in PNG_TYPES:
+            url = base_url.update_query([("w", size.size)])
+            response = get(url)
+            if response.status_code == 200 and response.url == str(url):
+                available_images.update({Id(str(size).lower()): url})
+        if base_url not in available_images.values():
+            available_images.update({Id("original"): base_url})
+        # Select the image type.
+        if len(available_images) == 1:
+            return next(iter(available_images.values()))
+        else:
+            printf(HTML(f"⎡ <b>Available images for {address}:</b>"))
+            for size, url in available_images.items():
+                printed_url = str(url).replace("&", "&amp;")
+                printf(HTML(f"⎢ <b>∙ {size}</b>: {printed_url}"))
+            selected_type = get_id(
+                "⎣ Select the image type",
+                None,
+                set(available_images.keys()),
+            )
+            return URL(available_images[selected_type])
+
+    def _download_token_image(self, token_image_url: URL) -> bytes | None:
+        response = get(str(token_image_url))
+        if response.status_code == 200:
+            return response.content
+        return None
+
+    @staticmethod
+    def __get_token_list(sub_path: URL) -> tuple[list[Address], URL | None]:
+        """Get the list of tokens from the Routescan explorer.
+
+        Args:
+            sub_path: The sub-path of the Routescan API URL.
+
+        Returns:
+            A tuple containing the list of token addresses and the next sub-path.
+        """
+        response = get(
+            str(ROUTESCAN_API_URL.join(sub_path)),
+            params={"sort": "marketCap,desc", "limit": TOKEN_COUNT_PER_PAGE},
+        )
+        if response.status_code != 200:
+            return []
+        else:
+            body: dict = response.json()
+            next_sub_path = body.get("link", {}).get("next", None)
+            return (
+                [
+                    Address(Web3.to_checksum_address(item["address"]))
+                    for item in body.get("items", [])
+                ],
+                URL(next_sub_path) if next_sub_path is not None else None,
+            )
diff --git a/pyproject.toml b/pyproject.toml
@@ -28,6 +28,7 @@ dependencies = { file = ["requirements/essential.txt"] }
 [tool.setuptools.dynamic.optional-dependencies]
 all = { file = ["requirements/essential.txt", "requirements/dev.txt", "requirements/test.txt"] }
 dev = { file = ["requirements/essential.txt", "requirements/dev.txt"] }
+essential = { file = ["requirements/essential.txt"] }
 test = { file = ["requirements/essential.txt", "requirements/dev.txt", "requirements/test.txt"] }
 test_simple = { file = ["requirements/essential.txt", "requirements/test.txt"] }
 

diff --git a/requirements.txt b/requirements.txt