#!/usr/bin/env python3 """ AITBC Blockchain Explorer A simple web interface to explore the blockchain """ import asyncio import httpx import json from datetime import datetime from typing import Dict, List, Optional, Any from fastapi import FastAPI, Request, HTTPException from fastapi.responses import HTMLResponse from fastapi.staticfiles import StaticFiles import uvicorn app = FastAPI(title="AITBC Blockchain Explorer", version="1.0.0") # Configuration BLOCKCHAIN_RPC_URL = "http://localhost:8082" # Local blockchain node EXTERNAL_RPC_URL = "http://aitbc.keisanki.net:8082" # External access # HTML Template HTML_TEMPLATE = r""" AITBC Blockchain Explorer

AITBC Blockchain Explorer

Network: ait-devnet

Current Height

-

Latest Block

-

Node Status

-

Latest Blocks

Height Hash Timestamp Transactions Actions
Loading blocks...
""" async def get_transaction(tx_hash: str) -> Dict[str, Any]: """Get transaction by hash""" try: async with httpx.AsyncClient() as client: response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/tx/{tx_hash}") if response.status_code == 200: return response.json() except Exception as e: print(f"Error getting transaction: {e}") return {} async def get_block(height: int) -> Dict[str, Any]: """Get a specific block by height""" try: async with httpx.AsyncClient() as client: response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/blocks/{height}") if response.status_code == 200: return response.json() except Exception as e: print(f"Error getting block {height}: {e}") return {} @app.get("/", response_class=HTMLResponse) async def root(): """Serve the explorer UI""" return HTML_TEMPLATE.replace("{node_url}", BLOCKCHAIN_RPC_URL) @app.get("/api/chain/head") async def api_chain_head(): """API endpoint for chain head""" return await get_chain_head() @app.get("/api/blocks/{height}") async def api_block(height: int): """API endpoint for block data""" return await get_block(height) @app.get("/api/transactions/{tx_hash}") async def api_transaction(tx_hash: str): """API endpoint for transaction data, normalized for frontend""" async with httpx.AsyncClient() as client: try: response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/tx/{tx_hash}") if response.status_code == 200: tx = response.json() # Normalize for frontend expectations payload = tx.get("payload", {}) return { "hash": tx.get("tx_hash"), "block_height": tx.get("block_height"), "from": tx.get("sender"), "to": tx.get("recipient"), "type": payload.get("type", "transfer"), "amount": payload.get("amount", 0), "fee": payload.get("fee", 0), "timestamp": tx.get("created_at") } elif response.status_code == 404: raise HTTPException(status_code=404, detail="Transaction not found") except httpx.RequestError as e: print(f"Error fetching transaction: {e}") raise HTTPException(status_code=500, detail="Internal server error") @app.get("/health") async def health(): """Health check endpoint""" try: # Test blockchain node connectivity async with httpx.AsyncClient() as client: response = await client.get(f"{BLOCKCHAIN_RPC_URL}/rpc/head", timeout=5.0) node_status = "ok" if response.status_code == 200 else "error" except Exception: node_status = "error" return { "status": "ok" if node_status == "ok" else "degraded", "node_status": node_status, "node_url": BLOCKCHAIN_RPC_URL, "endpoints": { "transactions": "/api/transactions/{tx_hash}", "chain_head": "/api/chain/head", "blocks": "/api/blocks/{height}" } } if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=3001)