a7b6e39cdf
- 55 CLI files: handlers/, aitbc_cli/commands/, cli/core/, cli/utils/, top-level scripts
- Click-based files: print() -> click.echo()
- Library modules: print() -> logger.info/error/warning
- Fixed pre-existing indentation bugs in monitor.py dashboard function
- Fixed bare print() -> logger.info('') in chain_manager.py
- 0 remaining print() in production CLI code
- All files compile cleanly
96 lines
3.4 KiB
Python
96 lines
3.4 KiB
Python
"""
|
|
Blockchain utility functions for AITBC CLI
|
|
"""
|
|
|
|
from typing import Optional, Dict
|
|
|
|
from aitbc import AITBCHTTPClient, NetworkError
|
|
import logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
|
|
def get_chain_info(rpc_url: str = "http://localhost:8006") -> Optional[Dict]:
|
|
"""Get blockchain information"""
|
|
try:
|
|
result = {}
|
|
# Get chain metadata from health endpoint
|
|
http_client = AITBCHTTPClient(base_url=rpc_url, timeout=30)
|
|
health = http_client.get("/health")
|
|
chains = health.get('supported_chains', [])
|
|
result['chain_id'] = chains[0] if chains else 'ait-mainnet'
|
|
result['supported_chains'] = ', '.join(chains) if chains else 'ait-mainnet'
|
|
result['proposer_id'] = health.get('proposer_id', '')
|
|
# Get head block for height
|
|
head = http_client.get("/rpc/head")
|
|
result['height'] = head.get('height', 0)
|
|
result['hash'] = head.get('hash', "")
|
|
result['timestamp'] = head.get('timestamp', 'N/A')
|
|
result['tx_count'] = head.get('tx_count', 0)
|
|
return result if result else None
|
|
except NetworkError as e:
|
|
logger.error(f"Error: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
return None
|
|
|
|
|
|
def get_network_status(rpc_url: str = "http://localhost:8006") -> Optional[Dict]:
|
|
"""Get network status and health"""
|
|
try:
|
|
# Get head block
|
|
http_client = AITBCHTTPClient(base_url=rpc_url, timeout=30)
|
|
return http_client.get("/rpc/head")
|
|
except NetworkError as e:
|
|
logger.error(f"Error getting network status: {e}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
return None
|
|
|
|
|
|
def get_blockchain_analytics(analytics_type: str, limit: int = 10, rpc_url: str = "http://localhost:8006") -> Optional[Dict]:
|
|
"""Get blockchain analytics and statistics"""
|
|
try:
|
|
if analytics_type == "blocks":
|
|
# Get recent blocks analytics
|
|
http_client = AITBCHTTPClient(base_url=rpc_url, timeout=30)
|
|
head = http_client.get("/rpc/head")
|
|
return {
|
|
"type": "blocks",
|
|
"current_height": head.get("height", 0),
|
|
"latest_block": head.get("hash", ""),
|
|
"timestamp": head.get("timestamp", ""),
|
|
"tx_count": head.get("tx_count", 0),
|
|
"status": "Active"
|
|
}
|
|
|
|
elif analytics_type == "supply":
|
|
# Get total supply info
|
|
return {
|
|
"type": "supply",
|
|
"total_supply": "1000000000", # From genesis
|
|
"circulating_supply": "999997980", # After transactions
|
|
"genesis_minted": "1000000000",
|
|
"status": "Available"
|
|
}
|
|
|
|
elif analytics_type == "accounts":
|
|
# Account statistics
|
|
return {
|
|
"type": "accounts",
|
|
"total_accounts": 3, # Genesis + treasury + user
|
|
"active_accounts": 2, # Accounts with transactions
|
|
"genesis_accounts": 2, # Genesis and treasury
|
|
"user_accounts": 1,
|
|
"status": "Healthy"
|
|
}
|
|
|
|
else:
|
|
return {"type": analytics_type, "status": "Not implemented yet"}
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error getting analytics: {e}")
|
|
return None
|