Your Name
feat: UI improvements and error suppression - Enhanced dashboard and market pages with improved header buttons, logo, and currency symbol display - Stopped animated ticker - Removed pie chart legends - Added error suppressor for external service errors (SSE, Permissions-Policy warnings) - Improved header button prominence and icon appearance - Enhanced logo with glow effects and better design - Fixed currency symbol visibility in market tables
8b7b267
| """ | |
| Etherscan Provider - Ethereum blockchain transaction data | |
| Provides: | |
| - Address transaction history | |
| - Token transfers | |
| - Contract information | |
| - Account balances | |
| API Documentation: https://docs.etherscan.io/ | |
| """ | |
| from __future__ import annotations | |
| from typing import Any, Dict, List, Optional | |
| from .base import BaseProvider, create_success_response, create_error_response | |
| class EtherscanProvider(BaseProvider): | |
| """Etherscan REST API provider for Ethereum blockchain data""" | |
| # API Keys (temporary hardcoded - will be secured later) | |
| API_KEY_PRIMARY = "SZHYFZK2RR8H9TIMJBVW54V4H81K2Z2KR2" | |
| API_KEY_SECONDARY = "T6IR8VJHX2NE6ZJW2S3FDVN1TYG4PYYI45" | |
| def __init__(self, api_key: Optional[str] = None): | |
| super().__init__( | |
| name="etherscan", | |
| base_url="https://api.etherscan.io/api", | |
| api_key=api_key or self.API_KEY_PRIMARY, | |
| timeout=10.0, | |
| cache_ttl=30.0 | |
| ) | |
| def _build_params(self, **kwargs) -> Dict[str, Any]: | |
| """Build request parameters with API key""" | |
| params = {"apikey": self.api_key} | |
| params.update({k: v for k, v in kwargs.items() if v is not None}) | |
| return params | |
| async def get_transactions( | |
| self, | |
| address: str, | |
| start_block: int = 0, | |
| end_block: int = 99999999, | |
| page: int = 1, | |
| offset: int = 50, | |
| sort: str = "desc" | |
| ) -> Dict[str, Any]: | |
| """ | |
| Get list of transactions for an address. | |
| Args: | |
| address: Ethereum address | |
| start_block: Starting block number | |
| end_block: Ending block number | |
| page: Page number for pagination | |
| offset: Number of transactions per page (max 10000) | |
| sort: Sort order ('asc' or 'desc') | |
| Returns: | |
| Standardized response with transaction list | |
| """ | |
| if not address or not address.startswith("0x"): | |
| return create_error_response( | |
| self.name, | |
| "Invalid Ethereum address", | |
| "Address must start with '0x'" | |
| ) | |
| params = self._build_params( | |
| module="account", | |
| action="txlist", | |
| address=address, | |
| startblock=start_block, | |
| endblock=end_block, | |
| page=page, | |
| offset=min(offset, 100), # Limit for performance | |
| sort=sort | |
| ) | |
| response = await self.get("", params=params) | |
| if not response.get("success"): | |
| return response | |
| # Parse Etherscan response format | |
| data = response.get("data", {}) | |
| status = data.get("status") | |
| message = data.get("message", "") | |
| # Status "1" means success, "0" can mean no data or error | |
| if status == "1" or (status == "0" and "No transactions found" in message): | |
| transactions = data.get("result", []) if status == "1" else [] | |
| if isinstance(transactions, str): | |
| # API returned an error string instead of list | |
| return create_error_response(self.name, message, transactions) | |
| return create_success_response( | |
| self.name, | |
| { | |
| "address": address, | |
| "transactions": self._format_transactions(transactions), | |
| "count": len(transactions) | |
| } | |
| ) | |
| else: | |
| error_msg = message or "Unknown error" | |
| result_msg = data.get("result", "") | |
| if isinstance(result_msg, str) and result_msg: | |
| return create_error_response(self.name, error_msg, result_msg) | |
| return create_error_response(self.name, error_msg) | |
| def _format_transactions(self, transactions: List[Dict]) -> List[Dict]: | |
| """Format transaction data for clean output""" | |
| formatted = [] | |
| for tx in transactions: | |
| formatted.append({ | |
| "hash": tx.get("hash"), | |
| "blockNumber": int(tx.get("blockNumber", 0)), | |
| "timestamp": int(tx.get("timeStamp", 0)), | |
| "from": tx.get("from"), | |
| "to": tx.get("to"), | |
| "value": tx.get("value"), | |
| "valueEth": float(tx.get("value", 0)) / 1e18, | |
| "gas": int(tx.get("gas", 0)), | |
| "gasPrice": tx.get("gasPrice"), | |
| "gasUsed": int(tx.get("gasUsed", 0)), | |
| "isError": tx.get("isError") == "1", | |
| "txreceipt_status": tx.get("txreceipt_status"), | |
| "contractAddress": tx.get("contractAddress") or None, | |
| "functionName": tx.get("functionName") or None | |
| }) | |
| return formatted | |
| async def get_token_transfers( | |
| self, | |
| address: str, | |
| contract_address: Optional[str] = None, | |
| page: int = 1, | |
| offset: int = 50 | |
| ) -> Dict[str, Any]: | |
| """ | |
| Get ERC-20 token transfer events for an address. | |
| Args: | |
| address: Ethereum address | |
| contract_address: Optional token contract address filter | |
| page: Page number | |
| offset: Results per page | |
| """ | |
| if not address or not address.startswith("0x"): | |
| return create_error_response( | |
| self.name, | |
| "Invalid Ethereum address", | |
| "Address must start with '0x'" | |
| ) | |
| params = self._build_params( | |
| module="account", | |
| action="tokentx", | |
| address=address, | |
| page=page, | |
| offset=min(offset, 100), | |
| sort="desc" | |
| ) | |
| if contract_address: | |
| params["contractaddress"] = contract_address | |
| response = await self.get("", params=params) | |
| if not response.get("success"): | |
| return response | |
| data = response.get("data", {}) | |
| if data.get("status") == "1": | |
| transfers = data.get("result", []) | |
| return create_success_response( | |
| self.name, | |
| { | |
| "address": address, | |
| "transfers": self._format_token_transfers(transfers), | |
| "count": len(transfers) | |
| } | |
| ) | |
| else: | |
| error_msg = data.get("message", "Unknown error") | |
| if error_msg == "No transactions found": | |
| return create_success_response( | |
| self.name, | |
| {"address": address, "transfers": [], "count": 0} | |
| ) | |
| return create_error_response(self.name, error_msg) | |
| def _format_token_transfers(self, transfers: List[Dict]) -> List[Dict]: | |
| """Format token transfer data""" | |
| formatted = [] | |
| for tx in transfers: | |
| decimals = int(tx.get("tokenDecimal", 18)) | |
| value = int(tx.get("value", 0)) | |
| formatted.append({ | |
| "hash": tx.get("hash"), | |
| "blockNumber": int(tx.get("blockNumber", 0)), | |
| "timestamp": int(tx.get("timeStamp", 0)), | |
| "from": tx.get("from"), | |
| "to": tx.get("to"), | |
| "value": str(value), | |
| "tokenValue": value / (10 ** decimals) if decimals else value, | |
| "tokenName": tx.get("tokenName"), | |
| "tokenSymbol": tx.get("tokenSymbol"), | |
| "tokenDecimal": decimals, | |
| "contractAddress": tx.get("contractAddress") | |
| }) | |
| return formatted | |
| async def get_balance(self, address: str) -> Dict[str, Any]: | |
| """Get ETH balance for an address""" | |
| if not address or not address.startswith("0x"): | |
| return create_error_response( | |
| self.name, | |
| "Invalid Ethereum address", | |
| "Address must start with '0x'" | |
| ) | |
| params = self._build_params( | |
| module="account", | |
| action="balance", | |
| address=address, | |
| tag="latest" | |
| ) | |
| response = await self.get("", params=params) | |
| if not response.get("success"): | |
| return response | |
| data = response.get("data", {}) | |
| if data.get("status") == "1": | |
| balance_wei = int(data.get("result", 0)) | |
| return create_success_response( | |
| self.name, | |
| { | |
| "address": address, | |
| "balance_wei": str(balance_wei), | |
| "balance_eth": balance_wei / 1e18 | |
| } | |
| ) | |
| else: | |
| return create_error_response(self.name, data.get("message", "Unknown error")) | |
| async def get_gas_price(self) -> Dict[str, Any]: | |
| """Get current gas price""" | |
| params = self._build_params( | |
| module="gastracker", | |
| action="gasoracle" | |
| ) | |
| response = await self.get("", params=params) | |
| if not response.get("success"): | |
| return response | |
| data = response.get("data", {}) | |
| if data.get("status") == "1": | |
| result = data.get("result", {}) | |
| return create_success_response( | |
| self.name, | |
| { | |
| "safeGasPrice": result.get("SafeGasPrice"), | |
| "proposeGasPrice": result.get("ProposeGasPrice"), | |
| "fastGasPrice": result.get("FastGasPrice"), | |
| "suggestBaseFee": result.get("suggestBaseFee"), | |
| "gasUsedRatio": result.get("gasUsedRatio") | |
| } | |
| ) | |
| else: | |
| return create_error_response(self.name, data.get("message", "Unknown error")) | |