File size: 3,448 Bytes
6993919 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 | """
BSC + Polygon DataBus Providers — Free public APIs, no key needed.
BscScan/PolygonScan free tier: balance, transactions, token transfers.
Rate limited to 1 req/5 sec per IP. No API key required for basic use.
"""
import logging
logger = logging.getLogger("databus.evm_extra")
async def _fetch_bsc_data(address: str = "", action: str = "balance", **kwargs) -> dict | None:
"""BSC intelligence via BscScan free public API."""
import aiohttp
apis = {
"balance": f"https://api.bscscan.com/api?module=account&action=balance&address={address}&tag=latest",
"txlist": f"https://api.bscscan.com/api?module=account&action=txlist&address={address}&startblock=0&endblock=99999999&page=1&offset=10&sort=desc",
"tokentx": f"https://api.bscscan.com/api?module=account&action=tokentx&address={address}&startblock=0&endblock=99999999&page=1&offset=10&sort=desc",
}
url = apis.get(action, apis["balance"])
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status == 200:
data = await resp.json()
if data.get("status") == "1":
return {
"chain": "bsc",
"address": address,
"data": data.get("result"),
"source": "BscScan (free, no API key)",
}
return {
"chain": "bsc",
"address": address,
"error": data.get("message", "No data"),
"source": "BscScan",
}
except Exception as e:
logger.warning(f"BSC fetch failed: {e}")
return None
async def _fetch_polygon_data(address: str = "", action: str = "balance", **kwargs) -> dict | None:
"""Polygon intelligence via PolygonScan free public API."""
import aiohttp
apis = {
"balance": f"https://api.polygonscan.com/api?module=account&action=balance&address={address}&tag=latest",
"txlist": f"https://api.polygonscan.com/api?module=account&action=txlist&address={address}&startblock=0&endblock=99999999&page=1&offset=10&sort=desc",
"tokentx": f"https://api.polygonscan.com/api?module=account&action=tokentx&address={address}&startblock=0&endblock=99999999&page=1&offset=10&sort=desc",
}
url = apis.get(action, apis["balance"])
try:
async with aiohttp.ClientSession() as session:
async with session.get(url, timeout=aiohttp.ClientTimeout(total=10)) as resp:
if resp.status == 200:
data = await resp.json()
if data.get("status") == "1":
return {
"chain": "polygon",
"address": address,
"data": data.get("result"),
"source": "PolygonScan (free, no API key)",
}
return {
"chain": "polygon",
"address": address,
"error": data.get("message", "No data"),
"source": "PolygonScan",
}
except Exception as e:
logger.warning(f"Polygon fetch failed: {e}")
return None
|