Etherscan failover ABI fetcher

CHANGELOG.md

@@ -10,6 +10,7 @@ Releases prior to 7.0 has been removed from this file to declutter search result
 
 ### Added
 
+- abi.etherscan: Try to extract ABI from webpage when API call fails.
 - database: support database migrations using [`aerich`](https://github.com/tortoise/aerich)
 - cli: Added `schema` subcommands to manage database migrations: `migrate`, `upgrade`, `downgrade`, `heads` and `history`
 - hasura: Added `ignore` and `ignore_internal` config options to hide specific tables/views.

docs/9.release-notes/_8.0_changelog.md

@@ -3,6 +3,7 @@
 
 ### Added
 
+- abi.etherscan: Try to extract ABI from webpage when API call fails.
 - cli: Added `--pre` flag to `self` group commands to install pre-release versions.
 - cli: Added `--raw` option to `config export` command to dump config preserving the original structure.
 - cli: Added `-C` option, a shorthand for `-c . -c configs/dipdup.<name>.yaml`.

src/dipdup/datasources/__init__.py

@@ -61,13 +61,13 @@ def __init__(self, config: DatasourceConfigT) -> None:
         )
         self._logger = FormattedLogger(__name__, config.name + ': {}')
 
-    @abstractmethod
-    async def run(self) -> None: ...
-
     @property
     def name(self) -> str:
         return self._config.name
 
+    async def run(self) -> None:
+        pass
+
 
 class AbiDatasource(Datasource[DatasourceConfigT], Generic[DatasourceConfigT]):
     @abstractmethod
@@ -86,6 +86,9 @@ def __init__(
         self._on_disconnected_callbacks: set[EmptyCallback] = set()
         self._on_rollback_callbacks: set[RollbackCallback] = set()
 
+    @abstractmethod
+    async def run(self) -> None: ...
+
     @abstractmethod
     async def subscribe(self) -> None: ...
 

src/dipdup/datasources/abi_etherscan.py

@@ -1,4 +1,6 @@
 import asyncio
+import re
+from copy import copy
 from typing import Any
 from typing import cast
 
@@ -7,6 +9,7 @@
 from dipdup.config import HttpConfig
 from dipdup.config.abi_etherscan import AbiEtherscanDatasourceConfig
 from dipdup.datasources import AbiDatasource
+from dipdup.datasources import Datasource
 from dipdup.datasources import EvmAbiProvider
 from dipdup.exceptions import DatasourceError
 
@@ -41,14 +44,40 @@ async def get_abi(self, address: str) -> dict[str, Any]:
                 self._logger.info(message)
 
             if result := response.get('result'):
-                if isinstance(result, str) and 'rate limit reached' in result:
-                    self._logger.warning('Ratelimited; sleeping %s seconds', self._http_config.ratelimit_sleep)
-                    await asyncio.sleep(self._http_config.retry_sleep)
-                    continue
-                try:
+                if isinstance(result, str):
+                    if 'rate limit reached' in result:
+                        self._logger.warning('Ratelimited; sleeping %s seconds', self._http_config.ratelimit_sleep)
+                        await asyncio.sleep(self._http_config.retry_sleep)
+                        continue
+                    if 'API Key' in result:
+                        self._logger.warning('%s, trying workaround', result)
+                        try:
+                            return await self.get_abi_failover(address)
+                        except Exception as e:
+                            self._logger.warning('Failed to get ABI: %s', e)
 
+                try:
                     return cast(dict[str, Any], orjson.loads(result))
                 except orjson.JSONDecodeError as e:
                     raise DatasourceError(result, self.name) from e
 
         raise DatasourceError(message, self.name)
+
+    async def get_abi_failover(self, address: str) -> dict[str, Any]:
+        config = copy(self._config)
+        config.url = f'{self._config.url}/token/{address}'.replace('api.', '').replace('/api', '')
+        html_etherscan = Datasource(config)
+        async with html_etherscan:
+            html = await (
+                await html_etherscan._http._request(
+                    method='get',
+                    url='',
+                    weight=1,
+                    raw=True,
+                )
+            ).text()
+
+        regex = r'id="js-copytextarea2(.*)>(\[.*?)\<\/pre'
+        if (match := re.search(regex, html)) and (abi := match.group(2)):
+            return cast(dict[str, Any], orjson.loads(abi))
+        raise DatasourceError('Failed to get ABI', self.name)