|
|
"""
|
|
|
TronScan Provider - TRON blockchain transaction data
|
|
|
|
|
|
Provides:
|
|
|
- TRON address transaction history
|
|
|
- TRC-20 token transfers
|
|
|
- Account information
|
|
|
- Contract data
|
|
|
|
|
|
API Documentation: https://docs.tronscan.org/
|
|
|
"""
|
|
|
|
|
|
from __future__ import annotations
|
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
|
|
from .base import BaseProvider, create_success_response, create_error_response
|
|
|
|
|
|
|
|
|
class TronscanProvider(BaseProvider):
|
|
|
"""TronScan REST API provider for TRON blockchain data"""
|
|
|
|
|
|
|
|
|
API_KEY = "7ae72726-bffe-4e74-9c33-97b761eeea21"
|
|
|
|
|
|
def __init__(self, api_key: Optional[str] = None):
|
|
|
super().__init__(
|
|
|
name="tronscan",
|
|
|
base_url="https://apilist.tronscanapi.com/api",
|
|
|
api_key=api_key or self.API_KEY,
|
|
|
timeout=10.0,
|
|
|
cache_ttl=30.0
|
|
|
)
|
|
|
|
|
|
def _get_default_headers(self) -> Dict[str, str]:
|
|
|
"""Get headers with TronScan API key"""
|
|
|
return {
|
|
|
"Accept": "application/json",
|
|
|
"User-Agent": "HF-Crypto-Data-Engine/1.0",
|
|
|
"TRON-PRO-API-KEY": self.api_key
|
|
|
}
|
|
|
|
|
|
async def get_transactions(
|
|
|
self,
|
|
|
address: str,
|
|
|
start: int = 0,
|
|
|
limit: int = 50,
|
|
|
sort: str = "-timestamp"
|
|
|
) -> Dict[str, Any]:
|
|
|
"""
|
|
|
Get list of transactions for a TRON address.
|
|
|
|
|
|
Args:
|
|
|
address: TRON address (starts with 'T')
|
|
|
start: Starting index for pagination
|
|
|
limit: Number of transactions to fetch
|
|
|
sort: Sort order ('-timestamp' for descending)
|
|
|
|
|
|
Returns:
|
|
|
Standardized response with transaction list
|
|
|
"""
|
|
|
if not address:
|
|
|
return create_error_response(
|
|
|
self.name,
|
|
|
"Invalid TRON address",
|
|
|
"Address is required"
|
|
|
)
|
|
|
|
|
|
|
|
|
if not address.startswith("T"):
|
|
|
return create_error_response(
|
|
|
self.name,
|
|
|
"Invalid TRON address format",
|
|
|
"TRON address should start with 'T'"
|
|
|
)
|
|
|
|
|
|
params = {
|
|
|
"address": address,
|
|
|
"start": start,
|
|
|
"limit": min(limit, 50),
|
|
|
"sort": sort
|
|
|
}
|
|
|
|
|
|
response = await self.get("transaction", params=params)
|
|
|
|
|
|
if not response.get("success"):
|
|
|
return response
|
|
|
|
|
|
data = response.get("data", {})
|
|
|
|
|
|
|
|
|
if isinstance(data, dict):
|
|
|
transactions = data.get("data", [])
|
|
|
total = data.get("total", 0)
|
|
|
else:
|
|
|
transactions = data if isinstance(data, list) else []
|
|
|
total = len(transactions)
|
|
|
|
|
|
return create_success_response(
|
|
|
self.name,
|
|
|
{
|
|
|
"address": address,
|
|
|
"chain": "tron",
|
|
|
"transactions": self._format_transactions(transactions),
|
|
|
"count": len(transactions),
|
|
|
"total": total
|
|
|
}
|
|
|
)
|
|
|
|
|
|
def _format_transactions(self, transactions: List[Dict]) -> List[Dict]:
|
|
|
"""Format TRON transaction data for clean output"""
|
|
|
formatted = []
|
|
|
for tx in transactions:
|
|
|
|
|
|
raw_amount = tx.get("amount", 0)
|
|
|
try:
|
|
|
amount = int(raw_amount) if raw_amount else 0
|
|
|
except (ValueError, TypeError):
|
|
|
amount = 0
|
|
|
|
|
|
formatted.append({
|
|
|
"hash": tx.get("hash") or tx.get("txID"),
|
|
|
"block": tx.get("block"),
|
|
|
"timestamp": tx.get("timestamp"),
|
|
|
"ownerAddress": tx.get("ownerAddress"),
|
|
|
"toAddress": tx.get("toAddress"),
|
|
|
"contractType": tx.get("contractType"),
|
|
|
"confirmed": tx.get("confirmed", False),
|
|
|
"result": tx.get("result"),
|
|
|
"amount": amount,
|
|
|
"amountTrx": amount / 1e6 if amount else 0,
|
|
|
"fee": tx.get("fee", 0),
|
|
|
"contractData": tx.get("contractData")
|
|
|
})
|
|
|
return formatted
|
|
|
|
|
|
async def get_trc20_transfers(
|
|
|
self,
|
|
|
address: str,
|
|
|
start: int = 0,
|
|
|
limit: int = 50,
|
|
|
contract_address: Optional[str] = None
|
|
|
) -> Dict[str, Any]:
|
|
|
"""
|
|
|
Get TRC-20 token transfer events for a TRON address.
|
|
|
|
|
|
Args:
|
|
|
address: TRON address
|
|
|
start: Starting index
|
|
|
limit: Number of results
|
|
|
contract_address: Optional filter by token contract
|
|
|
"""
|
|
|
if not address or not address.startswith("T"):
|
|
|
return create_error_response(
|
|
|
self.name,
|
|
|
"Invalid TRON address",
|
|
|
"Address must start with 'T'"
|
|
|
)
|
|
|
|
|
|
params = {
|
|
|
"address": address,
|
|
|
"start": start,
|
|
|
"limit": min(limit, 50),
|
|
|
"sort": "-timestamp"
|
|
|
}
|
|
|
|
|
|
if contract_address:
|
|
|
params["contract_address"] = contract_address
|
|
|
|
|
|
response = await self.get("token_trc20/transfers", params=params)
|
|
|
|
|
|
if not response.get("success"):
|
|
|
return response
|
|
|
|
|
|
data = response.get("data", {})
|
|
|
|
|
|
if isinstance(data, dict):
|
|
|
transfers = data.get("token_transfers", [])
|
|
|
total = data.get("total", 0)
|
|
|
else:
|
|
|
transfers = data if isinstance(data, list) else []
|
|
|
total = len(transfers)
|
|
|
|
|
|
return create_success_response(
|
|
|
self.name,
|
|
|
{
|
|
|
"address": address,
|
|
|
"chain": "tron",
|
|
|
"transfers": self._format_token_transfers(transfers),
|
|
|
"count": len(transfers),
|
|
|
"total": total
|
|
|
}
|
|
|
)
|
|
|
|
|
|
def _format_token_transfers(self, transfers: List[Dict]) -> List[Dict]:
|
|
|
"""Format TRC-20 token transfer data"""
|
|
|
formatted = []
|
|
|
for tx in transfers:
|
|
|
decimals = int(tx.get("decimals", 6))
|
|
|
quant = int(tx.get("quant", 0) or 0)
|
|
|
formatted.append({
|
|
|
"hash": tx.get("transaction_id"),
|
|
|
"block": tx.get("block"),
|
|
|
"timestamp": tx.get("block_ts"),
|
|
|
"from": tx.get("from_address"),
|
|
|
"to": tx.get("to_address"),
|
|
|
"quant": str(quant),
|
|
|
"tokenValue": quant / (10 ** decimals) if decimals else quant,
|
|
|
"tokenName": tx.get("tokenInfo", {}).get("tokenName"),
|
|
|
"tokenSymbol": tx.get("tokenInfo", {}).get("tokenAbbr"),
|
|
|
"tokenDecimal": decimals,
|
|
|
"contractAddress": tx.get("contract_address"),
|
|
|
"confirmed": tx.get("confirmed", False)
|
|
|
})
|
|
|
return formatted
|
|
|
|
|
|
async def get_account_info(self, address: str) -> Dict[str, Any]:
|
|
|
"""Get account information and balance for a TRON address"""
|
|
|
if not address or not address.startswith("T"):
|
|
|
return create_error_response(
|
|
|
self.name,
|
|
|
"Invalid TRON address",
|
|
|
"Address must start with 'T'"
|
|
|
)
|
|
|
|
|
|
params = {"address": address}
|
|
|
|
|
|
response = await self.get("accountv2", params=params)
|
|
|
|
|
|
if not response.get("success"):
|
|
|
return response
|
|
|
|
|
|
data = response.get("data", {})
|
|
|
|
|
|
if not data:
|
|
|
return create_error_response(
|
|
|
self.name,
|
|
|
"Account not found",
|
|
|
f"No data found for address {address}"
|
|
|
)
|
|
|
|
|
|
balance = data.get("balance", 0)
|
|
|
return create_success_response(
|
|
|
self.name,
|
|
|
{
|
|
|
"address": address,
|
|
|
"chain": "tron",
|
|
|
"balance": balance,
|
|
|
"balance_trx": balance / 1e6,
|
|
|
"bandwidth": data.get("bandwidth", {}),
|
|
|
"energy": data.get("energy", {}),
|
|
|
"totalFrozen": data.get("totalFrozen", 0),
|
|
|
"totalFrozenV2": data.get("totalFrozenV2", 0),
|
|
|
"tokens": data.get("withPriceTokens", [])[:10],
|
|
|
"transactions": data.get("transactions", 0)
|
|
|
}
|
|
|
)
|
|
|
|
|
|
async def get_token_list(
|
|
|
self,
|
|
|
start: int = 0,
|
|
|
limit: int = 20,
|
|
|
order_by: str = "-volume24hInTrx"
|
|
|
) -> Dict[str, Any]:
|
|
|
"""Get list of TRC-20 tokens sorted by volume"""
|
|
|
params = {
|
|
|
"start": start,
|
|
|
"limit": min(limit, 50),
|
|
|
"order": order_by,
|
|
|
"filter": "trc20"
|
|
|
}
|
|
|
|
|
|
response = await self.get("tokens/overview", params=params)
|
|
|
|
|
|
if not response.get("success"):
|
|
|
return response
|
|
|
|
|
|
data = response.get("data", {})
|
|
|
tokens = data.get("tokens", []) if isinstance(data, dict) else data
|
|
|
|
|
|
formatted_tokens = []
|
|
|
for token in tokens[:limit]:
|
|
|
formatted_tokens.append({
|
|
|
"name": token.get("name"),
|
|
|
"symbol": token.get("abbr"),
|
|
|
"contractAddress": token.get("contractAddress"),
|
|
|
"price": token.get("priceInTrx"),
|
|
|
"priceUsd": token.get("priceInUsd"),
|
|
|
"volume24h": token.get("volume24hInTrx"),
|
|
|
"holders": token.get("holders"),
|
|
|
"marketCap": token.get("marketcap")
|
|
|
})
|
|
|
|
|
|
return create_success_response(
|
|
|
self.name,
|
|
|
{
|
|
|
"chain": "tron",
|
|
|
"tokens": formatted_tokens,
|
|
|
"count": len(formatted_tokens)
|
|
|
}
|
|
|
)
|
|
|
|
