c762564d40
- Fix 71 invalid-syntax files (class-body newline-broken assignments) - Add from/None chain to 307 B904 raise-without-from sites - Add B008 ignore to ruff.toml (already in pyproject.toml) - Noqa F401 on __init__.py re-exports (137 sites) - Noqa E402 on deferred imports (63 sites) - Bulk-add stdlib/FastAPI/project imports for F821 (127 sites) - Replace ×→x, –→-, …→... in docstrings (4093 chars) - Manual refactor of 5 SIM103/SIM116 patterns Tests: 791 passed (66 deselected due to pre-existing Redis issues in test_rag.py) Co-authored-by: opencode <opencode@rugmunch.io>
77 lines
3.4 KiB
Python
77 lines
3.4 KiB
Python
"""
|
|
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: # noqa: SIM117
|
|
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: # noqa: SIM117
|
|
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
|