""" Blockchain explorer: look up a transaction by hash on one of five networks and return a normalized result (source/destination address, asset, amount, date, status, network, block-explorer link). This is a direct Python port of the reference n8n workflow ("Blockchain Explorer"): same data sources, same per-network parsing rules, including their quirks (e.g. the TRC20 branch's date field falling back to a hex-parsed timestamp, and its unused "ETH on TRC-20" fallback) — kept as-is rather than "corrected", since the reference flow is the spec here. Providers (identical to the reference flow): BEP20 -> Moralis (header auth: X-API-Key) ERC20 -> Etherscan v2 API (query auth: apikey) BTC -> mempool.space (public, no key) SOL -> Solana public JSON-RPC (public, no key) TRC20 -> Tronscan (public, no key) """ import time from datetime import datetime, timedelta, timezone import requests from .. import config from .price_service import _gregorian_to_jalali _TEHRAN_OFFSET_SECONDS = 3600 * 3.5 NETWORKS = ("BEP20", "TRC20", "ERC20", "BTC", "SOL") class TransactionNotFound(Exception): """Raised when the network's API confirms the hash does not exist.""" def _jalali_dash_format(epoch_seconds) -> str: """'YYYY-MM-DD HH:MM:SS', Tehran time, from a UTC unix timestamp (seconds).""" tehran = datetime.fromtimestamp(epoch_seconds, tz=timezone.utc) + timedelta( seconds=_TEHRAN_OFFSET_SECONDS ) jy, jm, jd = _gregorian_to_jalali(tehran.year, tehran.month, tehran.day) return f"{jy:04d}-{jm:02d}-{jd:02d} {tehran.hour:02d}:{tehran.minute:02d}:{tehran.second:02d}" def _iso_to_epoch_seconds(iso_text: str): if not iso_text: return None try: return datetime.fromisoformat(iso_text.replace("Z", "+00:00")).timestamp() except ValueError: return None def _request_with_retry(method: str, url: str, tries: int, **kwargs) -> requests.Response: last_error = None for attempt in range(1, max(tries, 1) + 1): try: resp = requests.request(method, url, timeout=config.EXPLORER_HTTP_TIMEOUT, **kwargs) resp.raise_for_status() return resp except Exception as exc: # network/HTTP error last_error = exc if attempt < tries: time.sleep(config.EXPLORER_HTTP_RETRY_WAIT_SECONDS) raise last_error # --- BEP20 (Moralis) -------------------------------------------------------------- _BEP20_TOKENS = { "0x55d398326f99059ff775485246999027b3197955": {"symbol": "USDT", "decimals": 18}, "0xe9e7cea3dedca5984780bafc599bd69add087d56": {"symbol": "BUSD", "decimals": 18}, "0x8ac76a51cc950d9822d68b83fe1ad97b32cd580d": {"symbol": "USDC", "decimals": 18}, "0x2170ed0880ac9a755fd29b2688956bd959f933f8": {"symbol": "ETH", "decimals": 18}, "0x7130d2a12b9bcbfae4f2634d864a1ee1ce3ead9c": {"symbol": "BTCB", "decimals": 18}, } _ERC20_TRANSFER_SIG = "0xa9059cbb" _TRANSFER_EVENT_TOPIC0 = "0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef" def _fetch_bep20(tx_hash: str): url = config.MORALIS_BEP20_TX_URL.format(tx_hash=tx_hash) headers = {"X-API-Key": config.MORALIS_API_KEY} if config.MORALIS_API_KEY else {} resp = _request_with_retry("GET", url, tries=1, headers=headers) data = resp.json() tx = data[0] if isinstance(data, list) else data if not tx or not isinstance(tx, dict) or not tx.get("hash"): raise TransactionNotFound() source_address = tx.get("from_address") or "" destination_address = tx.get("to_address") or "" amount = "0" asset = "BNB" input_data = tx.get("input") or "" to_address = (tx.get("to_address") or "").lower() if input_data.startswith(_ERC20_TRANSFER_SIG) and len(input_data) == 138: destination_address = "0x" + input_data[34:74] token = _BEP20_TOKENS.get(to_address) decimals = token["decimals"] if token else 18 asset = token["symbol"] if token else "BEP20 Token" amount_raw = int(input_data[74:138], 16) amount = f"{amount_raw / (10 ** decimals):.6f}" elif tx.get("value") and int(tx["value"]) > 0: amount = f"{int(tx['value']) / 1e18:.6f}" asset = "BNB" else: transfer_log = next( ( log for log in (tx.get("logs") or []) if log.get("topic0") == _TRANSFER_EVENT_TOPIC0 and log.get("data") not in (None, "0x") ), None, ) if transfer_log is not None: token_address = (transfer_log.get("address") or "").lower() token = _BEP20_TOKENS.get(token_address) decimals = token["decimals"] if token else 18 asset = token["symbol"] if token else "BEP20 Token" amount_raw = int(transfer_log["data"], 16) amount = f"{amount_raw / (10 ** decimals):.6f}" topic2 = transfer_log.get("topic2") if topic2 and topic2 != "0x": destination_address = "0x" + topic2[26:] status = "تایید شده" if tx.get("receipt_status") in ("1", 1) else "ناموفق" epoch = _iso_to_epoch_seconds(tx.get("block_timestamp")) return { "source_address": source_address, "destination_address": destination_address, "asset": asset, "amount": amount, "date": _jalali_dash_format(epoch) if epoch is not None else "", "status": status, "network": "BEP-20", "source": f"https://bscscan.com/tx/{tx['hash']}", } # --- ERC20 (Etherscan v2) ----------------------------------------------------------- _ERC20_TOKENS = { "0xdac17f958d2ee523a2206206994597c13d831ec7": {"symbol": "USDT", "decimals": 6}, "0xa0b86991c6218b36c1d19d4a2e9eb0ce3606eb48": {"symbol": "USDC", "decimals": 6}, "0x6b175474e89094c44da98b954eedeac495271d0f": {"symbol": "DAI", "decimals": 18}, "0x2260fac5e5542a773aa44fbcfedf7c193bc2c599": {"symbol": "WBTC", "decimals": 8}, } def _fetch_erc20(tx_hash: str): params = { "chainid": "1", "module": "proxy", "action": "eth_getTransactionByHash", "txhash": tx_hash, } if config.ETHERSCAN_API_KEY: params["apikey"] = config.ETHERSCAN_API_KEY resp = _request_with_retry("POST", config.ETHERSCAN_API_URL, tries=5, params=params) data = resp.json() tx = data.get("result") if isinstance(data, dict) else None if not tx or not isinstance(tx, dict): raise TransactionNotFound() source_address = tx.get("from") or "" destination_address = tx.get("to") or "" amount = "0" asset = "ETH" value_wei = int(tx.get("value") or "0x0", 16) input_data = tx.get("input") or "0x" to_address = (tx.get("to") or "").lower() if input_data.startswith(_ERC20_TRANSFER_SIG) and len(input_data) == 138: destination_address = "0x" + input_data[34:74] token = _ERC20_TOKENS.get(to_address) decimals = token["decimals"] if token else 18 asset = token["symbol"] if token else "ERC20 Token" amount_raw = int(input_data[74:138], 16) amount = str(amount_raw / (10 ** decimals)) elif value_wei > 0: amount = str(value_wei / 1e18) asset = "ETH" block_timestamp = tx.get("blockTimestamp") date_str = "" if block_timestamp: date_str = _jalali_dash_format(int(block_timestamp, 16)) return { "source_address": source_address, "destination_address": destination_address, "asset": asset, "amount": amount, "date": date_str, "status": "تایید شده", "network": "ERC-20", "source": f"https://etherscan.io/tx/{tx['hash']}" if tx.get("hash") else "", } # --- BTC (mempool.space) ------------------------------------------------------------ def _fetch_btc(tx_hash: str): url = config.MEMPOOL_SPACE_TX_URL.format(tx_hash=tx_hash) resp = _request_with_retry("GET", url, tries=5) tx = resp.json() if not tx or not isinstance(tx, dict) or not tx.get("txid"): raise TransactionNotFound() status = tx.get("status") or {} status_text = "تایید شده" if status.get("confirmed") is True else "در انتظار تایید شبکه" raw_epoch = status.get("block_time") if raw_epoch is None: raw_epoch = time.time() date_str = _jalali_dash_format(raw_epoch) vin = tx.get("vin") or [] vout = tx.get("vout") or [] source_address = ((vin[0] or {}).get("prevout") or {}).get("scriptpubkey_address", "") if vin else "" destination_address = (vout[0] or {}).get("scriptpubkey_address", "") if vout else "" amount_sats = (vout[0] or {}).get("value", 0) if vout else 0 amount_btc = amount_sats / 100_000_000 return { "source_address": source_address, "destination_address": destination_address, "asset": "BTC", "amount": f"{amount_btc:,.8f}", "date": date_str, "status": status_text, "network": "BTC", "source": f"https://mempool.space/tx/{tx['txid']}", } # --- SOL (Solana public RPC) -------------------------------------------------------- def _fetch_sol(tx_hash: str): payload = { "jsonrpc": "2.0", "id": 1, "method": "getTransaction", "params": [tx_hash, {"commitment": "confirmed", "maxSupportedTransactionVersion": 0, "encoding": "jsonParsed"}], } resp = _request_with_retry( "POST", config.SOLANA_RPC_URL, tries=1, json=payload, headers={"Content-Type": "application/json"}, ) data = resp.json() if isinstance(data, list): data = data[0] if data else {} tx = data.get("result") if isinstance(data, dict) else None if not tx: raise TransactionNotFound() block_time = tx.get("blockTime") date_str = _jalali_dash_format(block_time + _TEHRAN_OFFSET_SECONDS) if block_time else "" message = ((tx.get("transaction") or {}).get("message")) or {} signatures = (tx.get("transaction") or {}).get("signatures") or [] instructions = message.get("instructions") or [] account_keys = message.get("accountKeys") or [] transfer_ix = next( ( ix for ix in instructions if ix.get("program") == "system" and (ix.get("parsed") or {}).get("type") == "transfer" and (ix.get("parsed") or {}).get("info") ), None, ) if transfer_ix is not None: info = transfer_ix["parsed"]["info"] source_address = info.get("source", "") destination_address = info.get("destination", "") amount = f"{int(info.get('lamports', 0)) / 1e9:.9f}" else: signer = next((k.get("pubkey", "") for k in account_keys if k.get("signer")), "") source_address = signer destination_address = "" amount = "0" status_text = "تایید شده" if (tx.get("meta") or {}).get("err") is None else "ناموفق" tx_signature = signatures[0] if signatures else "" return { "source_address": source_address, "destination_address": destination_address, "asset": "SOL", "amount": amount, "date": date_str, "status": status_text, "network": "SOL", "source": f"https://solscan.io/tx/{tx_signature}" if tx_signature else "", } # --- TRC20 (Tronscan) ---------------------------------------------------------------- _TRC20_TOKENS = { "tr7nhqjekqxgtci8q8zy4pl8otszgjlj6t": {"symbol": "USDT", "decimals": 6}, "tekxitehnzsmse2xqrbj4w32run966rdz8": {"symbol": "USDC", "decimals": 6}, } def _fetch_trc20(tx_hash: str): url = config.TRONSCAN_TX_URL.format(tx_hash=tx_hash) resp = _request_with_retry("GET", url, tries=5) data = resp.json() tx = data.get("result") or data if isinstance(data, dict) else {} transfers = [] if tx.get("transfersAllList"): transfers = tx["transfersAllList"] elif tx.get("trc20TransferInfo"): transfers = tx["trc20TransferInfo"] elif tx.get("tokenTransferInfo"): transfers = [tx["tokenTransferInfo"]] if not tx.get("hash") and not transfers and not tx.get("contractData"): raise TransactionNotFound() default_source = tx.get("from") or tx.get("ownerAddress") or (tx.get("contractData") or {}).get("owner_address", "") default_dest = tx.get("to") or tx.get("toAddress") or tx.get("Destination address", "") timestamp = tx.get("timestamp") or tx.get("blockTimestamp") date_str = "" if timestamp: epoch_ms = int(timestamp, 16) * 1000 if isinstance(timestamp, str) else timestamp date_str = _jalali_dash_format(epoch_ms / 1000) status_text = ( "تایید شده" if tx.get("Status") == "SUCCESS" or tx.get("contractRet") == "SUCCESS" else "در انتظار تایید شبکه" ) source_link = f"https://tronscan.org/#/transaction/{tx['hash']}" if tx.get("hash") else "" results = [] if transfers: for t in transfers: contract_addr = (t.get("contract_address") or t.get("contractaddr") or "").lower() token = _TRC20_TOKENS.get(contract_addr) decimals = int(t.get("decimals") or (token["decimals"] if token else 6)) asset = t.get("symbol") or (token["symbol"] if token else "TRC20 Token") amount = f"{float(t.get('amount_str', 0)) / (10 ** decimals):.6f}" results.append({ "source_address": t.get("from_address") or default_source, "destination_address": t.get("to_address") or default_dest, "asset": asset, "amount": amount, "date": date_str, "status": status_text, "network": "TRC-20", "source": source_link, }) else: amount = "0" asset = "TRX" network = "TRC-20" contract_data = tx.get("contractData") or {} if tx.get("contractType") == 1 and contract_data.get("amount") is not None: amount = f"{contract_data['amount'] / 1_000_000:.6f}" asset = "TRX" elif contract_data.get("balance") is not None: amount = f"{contract_data['balance'] / 1_000_000:.6f}" asset = "TRX" elif tx.get("value") and int(tx["value"]) > 0: amount = f"{int(tx['value']) / 1e18:.6f}" asset = "ETH" network = "ERC-20" results.append({ "source_address": default_source, "destination_address": default_dest, "asset": asset, "amount": amount, "date": date_str, "status": status_text, "network": network, "source": source_link, }) return results _FETCHERS = { "BEP20": _fetch_bep20, "ERC20": _fetch_erc20, "BTC": _fetch_btc, "SOL": _fetch_sol, "TRC20": _fetch_trc20, } def fetch_transaction(network: str, tx_hash: str): """ Look up `tx_hash` on `network` (one of NETWORKS). Returns a single result dict, or a list of dicts for TRC20 when the transaction contains more than one token transfer. Raises TransactionNotFound when the API confirms the hash doesn't exist, or lets any other exception (network/HTTP error) propagate to the caller. """ fetcher = _FETCHERS.get(network) if fetcher is None: raise ValueError(f"unknown network: {network!r}") return fetcher(tx_hash)