diff --git a/VERSION b/VERSION index 38f77a6..e9307ca 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -2.0.1 +2.0.2 diff --git a/libraries/puller/runner.py b/libraries/puller/runner.py index d6038b3..a24878d 100644 --- 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 index f4cfc90..7e64b32 100644 --- 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" Name: {name}")) printf(HTML(f" Symbol: {symbol}")) - printf( - HTML(f" Source: {self._get_token_url(address)}") - ) + printed_url = str(self._get_token_url(address)).replace("&", "&") + printf(HTML(f" Source: {printed_url}")) 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"Invalid node URL: {url}")) + printed_url = str(url).replace("&", "&") + printf(HTML(f"Invalid node URL: {printed_url}")) 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: {token_image_url}")) + printed_url = str(token_image_url).replace("&", "&") + printf(HTML(f"Image URL found: {printed_url}")) 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 index 945ffce..1400740 100644 --- 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"⎡ Available images for {address}:")) for size, url in available_images.items(): - printf(HTML(f"⎢ ∙ {size}: {url}")) + printed_url = str(url).replace("&", "&") + printf(HTML(f"⎢ ∙ {size}: {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 index 3a86f92..d55f5e2 100644 --- 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"⎡ Available images for {address}:")) for size, url in available_images.items(): - printf(HTML(f"⎢ ∙ {size}: {url}")) + printed_url = str(url).replace("&", "&") + printf(HTML(f"⎢ ∙ {size}: {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 new file mode 100644 index 0000000..0004a83 --- /dev/null +++ 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"⎡ Available images for {address}:")) + for size, url in available_images.items(): + printed_url = str(url).replace("&", "&") + printf(HTML(f"⎢ ∙ {size}: {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 index 6498969..017c931 100644 --- 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 deleted file mode 100644 index 83b76bd..0000000 --- a/requirements.txt +++ /dev/null @@ -1 +0,0 @@ --r requirements/all.txt