64 lines
1.9 KiB
Python
64 lines
1.9 KiB
Python
import json
|
|
import requests
|
|
from loguru import logger
|
|
from typing import Any, Optional, Dict
|
|
from web3.types import ChecksumAddress
|
|
|
|
STANDARD_ERC20_ABI = [
|
|
{
|
|
"constant": True,
|
|
"inputs": [],
|
|
"name": "name",
|
|
"outputs": [{"name": "", "type": "string"}],
|
|
"payable": False,
|
|
"stateMutability": "view",
|
|
"type": "function",
|
|
},
|
|
{
|
|
"constant": True,
|
|
"inputs": [],
|
|
"name": "symbol",
|
|
"outputs": [{"name": "", "type": "string"}],
|
|
"payable": False,
|
|
"stateMutability": "view",
|
|
"type": "function",
|
|
},
|
|
{
|
|
"constant": True,
|
|
"inputs": [],
|
|
"name": "totalSupply",
|
|
"outputs": [{"name": "", "type": "uint256"}],
|
|
"payable": False,
|
|
"stateMutability": "view",
|
|
"type": "function",
|
|
}
|
|
]
|
|
|
|
|
|
def fetch_abi(address: ChecksumAddress, headers: Optional[Dict[str, Any]] = None,
|
|
params: Optional[Dict[str, Any]] = None, key: Optional[str] = None) -> Optional[Dict[str, Any]]:
|
|
url = f'https://api.etherscan.io/api?module=contract&action=getabi&address={address}&apikey={key}'
|
|
response = requests.get(url, headers=headers, params=params)
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
|
|
# Check for error in the response
|
|
if data['status'] == '0':
|
|
logger.error(f"Error fetching ABI: {data.get('result')}")
|
|
logger.warning(f"Using fallback ABI.")
|
|
return STANDARD_ERC20_ABI
|
|
|
|
return json.loads(data['result'])
|
|
|
|
|
|
def load_config(path: str) -> dict:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return json.loads(f.read())
|
|
|
|
|
|
def format_matrix_alert(token0: dict, token1: dict) -> str:
|
|
return f"""
|
|
🚨 New Uniswap Pair Alert 🚨\n
|
|
🌐 **Asset Pair:** [{token0['name']}](https://etherscan.io/token/{token0['address']}) + [{token1['name']}](https://etherscan.io/token/{token1['address']})\n
|
|
🏠 **Network:** ETH
|
|
""" |