|
|
@@ -0,0 +1,683 @@
|
|
|
+import os
|
|
|
+from web3 import Web3
|
|
|
+from web3.middleware import construct_sign_and_send_tx_middleware
|
|
|
+from web3.exceptions import TransactionNotFound
|
|
|
+from eth_account import Account
|
|
|
+from eth_account.signers.local import LocalAccount
|
|
|
+
|
|
|
+# 定义一个简单的异常类
|
|
|
+class EthTransactionError(Exception):
|
|
|
+ pass
|
|
|
+
|
|
|
+class ERC20Error(Exception):
|
|
|
+ pass
|
|
|
+
|
|
|
+class EthClient:
|
|
|
+ def __init__(self, rpc_url: str):
|
|
|
+ """
|
|
|
+ 初始化 EthClient 客户端。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ rpc_url: 以太坊 RPC 节点的 URL (例如: "https://mainnet.infura.io/v3/你的项目ID" 或 "http://localhost:8545")
|
|
|
+ """
|
|
|
+ self.w3 = Web3(Web3.HTTPProvider(rpc_url))
|
|
|
+
|
|
|
+ # 检查 RPC 连接
|
|
|
+ if not self.w3.is_connected():
|
|
|
+ raise ConnectionError(f"无法连接到以太坊 RPC 节点: {rpc_url}")
|
|
|
+ print(f"成功连接到以太坊 RPC 节点: {rpc_url}")
|
|
|
+
|
|
|
+ # 从环境变量加载私钥
|
|
|
+ private_key = os.environ.get("PRIVATE_KEY")
|
|
|
+ if not private_key:
|
|
|
+ raise EnvironmentError("环境变量 'PRIVATE_KEY' 未设置")
|
|
|
+
|
|
|
+ try:
|
|
|
+ self.account: LocalAccount = Account.from_key(private_key)
|
|
|
+ print(f"成功加载私钥，账户地址: {self.account.address}")
|
|
|
+ except Exception as e:
|
|
|
+ raise ValueError(f"加载私钥失败: {e}")
|
|
|
+
|
|
|
+ # 添加签名中间件，使得交易可以被自动签名
|
|
|
+ self.w3.middleware_onion.add(construct_sign_and_send_tx_middleware(self.account))
|
|
|
+
|
|
|
+ # 获取链 ID
|
|
|
+ self.chain_id = self.w3.eth.chain_id
|
|
|
+ print(f"当前连接的链 ID: {self.chain_id}")
|
|
|
+
|
|
|
+ def get_balance(self, address: str) -> int:
|
|
|
+ """
|
|
|
+ 获取指定地址的 ETH 余额。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ address: 查询余额的以太坊地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 地址的 ETH 余额 (以最小单位 wei 为单位)。
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ balance = self.w3.eth.get_balance(address)
|
|
|
+ return balance
|
|
|
+ except Exception as e:
|
|
|
+ raise EthTransactionError(f"获取 {address} 的 ETH 余额失败: {e}")
|
|
|
+
|
|
|
+ def get_nonce(self, address: str) -> int:
|
|
|
+ """
|
|
|
+ 获取指定地址的下一个可用 nonce。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ address: 查询 nonce 的以太坊地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 地址的下一个可用 nonce。
|
|
|
+ """
|
|
|
+ # 为了提高并发安全性，如果需要频繁发送交易，建议在类的实例中管理 nonce
|
|
|
+ # 或在发送交易前实时获取并处理潜在的 nonce 冲突
|
|
|
+ try:
|
|
|
+ nonce = self.w3.eth.get_transaction_count(address)
|
|
|
+ return nonce
|
|
|
+ except Exception as e:
|
|
|
+ raise EthTransactionError(f"获取 {address} 的 nonce 失败: {e}")
|
|
|
+
|
|
|
+ def estimate_gas(self, transaction: dict) -> int:
|
|
|
+ """
|
|
|
+ 估算交易所需的 Gas。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ transaction: 未签名的交易字典。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 估算的 Gas 量。
|
|
|
+ """
|
|
|
+ # 移除 from 字段，因为估算 Gas 時不需要签名者的 from
|
|
|
+ tx_copy = transaction.copy()
|
|
|
+ tx_copy.pop('from', None)
|
|
|
+ try:
|
|
|
+ gas_estimate = self.w3.eth.estimate_gas(tx_copy)
|
|
|
+ return gas_estimate
|
|
|
+ except Exception as e:
|
|
|
+ # 尝试解析错误信息，提供更具体的提示
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower():
|
|
|
+ raise EthTransactionError(f"估算 Gas 失败，交易可能会 Revert: {error_message}")
|
|
|
+ raise EthTransactionError(f"估算 Gas 失败: {e}")
|
|
|
+
|
|
|
+ def build_eth_transaction(self, to_address: str, amount_ether: float, gas_limit: int = 21000, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None) -> dict:
|
|
|
+ # ... (与之前相同，省略代码以保持简洁)
|
|
|
+ """
|
|
|
+ 构建一个发送 ETH 的交易字典。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ to_address: 接收 ETH 的地址。
|
|
|
+ amount_ether: 发送的 ETH 数量 (以 Ether 为单位)。
|
|
|
+ gas_limit: 交易的 Gas Limit (发送 ETH 通常为 21000)。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。如果为 None，则由节点自动估算。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 未签名的交易字典。
|
|
|
+ """
|
|
|
+ value_wei = self.w3.to_wei(amount_ether, 'ether')
|
|
|
+ base_fee = self.w3.eth.get_block('latest').baseFeePerGas
|
|
|
+
|
|
|
+ # EIP-1559 Gas Fee Calculation
|
|
|
+ if max_fee_per_gas_gwei is None:
|
|
|
+ max_fee_per_gas_wei = base_fee * 2 + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ else:
|
|
|
+ max_fee_per_gas_wei = self.w3.to_wei(max_fee_per_gas_gwei, 'gwei')
|
|
|
+
|
|
|
+ required_min_max_fee = base_fee + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ if max_fee_per_gas_wei < required_min_max_fee:
|
|
|
+ max_fee_per_gas_wei = required_min_max_fee
|
|
|
+
|
|
|
+ transaction = {
|
|
|
+ 'to': to_address,
|
|
|
+ 'value': value_wei,
|
|
|
+ 'gas': gas_limit,
|
|
|
+ 'maxPriorityFeePerGas': self.w3.to_wei(max_priority_fee_gwei, 'gwei'),
|
|
|
+ 'maxFeePerGas': max_fee_per_gas_wei,
|
|
|
+ 'nonce': self.get_nonce(self.account.address),
|
|
|
+ 'chainId': self.chain_id,
|
|
|
+ 'type': 2
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ estimated_gas = self.estimate_gas(transaction)
|
|
|
+ print(f"估算 ETH 交易 Gas 量: {estimated_gas}")
|
|
|
+ except EthTransactionError as e:
|
|
|
+ print(f"警告: ETH 交易 Gas 估算失败，使用默认值 {gas_limit}: {e}")
|
|
|
+
|
|
|
+ return transaction
|
|
|
+
|
|
|
+ def send_transaction(self, transaction: dict, wait_for_confirm: bool = True, timeout: int = 180):
|
|
|
+ # ... (与之前相同，省略代码以保持简洁)
|
|
|
+ """
|
|
|
+ 发送交易并等待确认（可选）。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ transaction: 未签名的交易字典。
|
|
|
+ wait_for_confirm: 是否等待交易被打包确认。
|
|
|
+ timeout: 等待交易确认的超时时间（秒）。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 交易哈希字符串。
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ tx_hash = self.w3.eth.send_transaction(transaction)
|
|
|
+ print(f"交易已发送，哈希: {tx_hash.hex()}")
|
|
|
+
|
|
|
+ if wait_for_confirm:
|
|
|
+ print("等待交易确认...")
|
|
|
+ try:
|
|
|
+ tx_receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)
|
|
|
+ print(f"交易确认成功，区块号: {tx_receipt.blockNumber}")
|
|
|
+ if tx_receipt.status == 0:
|
|
|
+ print("警告: 交易失败 (status=0)")
|
|
|
+ raise EthTransactionError(f"交易失败 (status=0)，哈希: {tx_hash.hex()}")
|
|
|
+ return tx_hash.hex()
|
|
|
+ except TimeoutError:
|
|
|
+ raise EthTransactionError(f"等待交易确认超时 ({timeout}秒)，哈希: {tx_hash.hex()}")
|
|
|
+ except TransactionNotFound:
|
|
|
+ raise EthTransactionError(f"发送的交易哈希未找到，可能未广播成功或节点同步问题，哈希: {tx_hash.hex()}")
|
|
|
+ except Exception as e:
|
|
|
+ # 捕获可能的 ContractLogicError 等
|
|
|
+ raise EthTransactionError(f"等待或处理交易回执时发生错误: {e}，哈希: {tx_hash.hex()}")
|
|
|
+ else:
|
|
|
+ return tx_hash.hex()
|
|
|
+
|
|
|
+ except ValueError as e:
|
|
|
+ raise EthTransactionError(f"发送交易时出现值错误: {e}")
|
|
|
+ except Exception as e:
|
|
|
+ # 捕获可能的 RPC 错误等
|
|
|
+ raise EthTransactionError(f"发送交易时发生未知错误: {e}")
|
|
|
+
|
|
|
+ def send_eth(self, to_address: str, amount_ether: float, gas_limit: int = 21000, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None, wait_for_confirm: bool = True, timeout: int = 180):
|
|
|
+ # ... (与之前相同，省略代码以保持简洁)
|
|
|
+ """
|
|
|
+ 发送原生 ETH。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ to_address: 接收 ETH 的地址。
|
|
|
+ amount_ether: 发送的 ETH 数量 (以 Ether 为单位)。
|
|
|
+ gas_limit: 交易的 Gas Limit。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。
|
|
|
+ wait_for_confirm: 是否等待交易被打包确认。
|
|
|
+ timeout: 等待交易确认的超时时间（秒）。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 交易哈希字符串。
|
|
|
+ """
|
|
|
+ print(f"准备发送 {amount_ether} ETH 到 {to_address}")
|
|
|
+ transaction = self.build_eth_transaction(to_address, amount_ether, gas_limit, max_priority_fee_gwei, max_fee_per_gas_gwei)
|
|
|
+ return self.send_transaction(transaction, wait_for_confirm=wait_for_confirm, timeout=timeout)
|
|
|
+
|
|
|
+ # --- 扩展的 ERC-20 代币功能和 ABI ---
|
|
|
+
|
|
|
+ # 扩展 ERC-20 标准 ABI
|
|
|
+ ERC20_ABI = [
|
|
|
+ # transfer function
|
|
|
+ {
|
|
|
+ "constant": False,
|
|
|
+ "inputs": [
|
|
|
+ {"name": "_to", "type": "address"},
|
|
|
+ {"name": "_value", "type": "uint256"}
|
|
|
+ ],
|
|
|
+ "name": "transfer",
|
|
|
+ "outputs": [{"name": "", "type": "bool"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "nonpayable",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # balanceOf function (read-only)
|
|
|
+ {
|
|
|
+ "constant": True,
|
|
|
+ "inputs": [{"name": "_owner", "type": "address"}],
|
|
|
+ "name": "balanceOf",
|
|
|
+ "outputs": [{"name": "balance", "type": "uint256"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "view",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # allowance function (read-only)
|
|
|
+ {
|
|
|
+ "constant": True,
|
|
|
+ "inputs": [
|
|
|
+ {"name": "_owner", "type": "address"},
|
|
|
+ {"name": "_spender", "type": "address"}
|
|
|
+ ],
|
|
|
+ "name": "allowance",
|
|
|
+ "outputs": [{"name": "", "type": "uint256"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "view",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # approve function (write)
|
|
|
+ {
|
|
|
+ "constant": False,
|
|
|
+ "inputs": [
|
|
|
+ {"name": "_spender", "type": "address"},
|
|
|
+ {"name": "_value", "type": "uint256"}
|
|
|
+ ],
|
|
|
+ "name": "approve",
|
|
|
+ "outputs": [{"name": "", "type": "bool"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "nonpayable",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # decimals function (read-only)
|
|
|
+ {
|
|
|
+ "constant": True,
|
|
|
+ "inputs": [],
|
|
|
+ "name": "decimals",
|
|
|
+ "outputs": [{"name": "", "type": "uint8"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "view",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # symbol function (read-only)
|
|
|
+ {
|
|
|
+ "constant": True,
|
|
|
+ "inputs": [],
|
|
|
+ "name": "symbol",
|
|
|
+ "outputs": [{"name": "", "type": "string"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "view",
|
|
|
+ "type": "function"
|
|
|
+ },
|
|
|
+ # name function (read-only)
|
|
|
+ {
|
|
|
+ "constant": True,
|
|
|
+ "inputs": [],
|
|
|
+ "name": "name",
|
|
|
+ "outputs": [{"name": "", "type": "string"}],
|
|
|
+ "payable": False,
|
|
|
+ "stateMutability": "view",
|
|
|
+ "type": "function"
|
|
|
+ }
|
|
|
+ ]
|
|
|
+
|
|
|
+ def get_erc20_contract(self, token_contract_address: str):
|
|
|
+ """获取 ERC-20 代币合约实例"""
|
|
|
+ if not self.w3.is_address(token_contract_address):
|
|
|
+ raise ERC20Error(f"无效的代币合约地址: {token_contract_address}")
|
|
|
+ try:
|
|
|
+ return self.w3.eth.contract(address=token_contract_address, abi=self.ERC20_ABI)
|
|
|
+ except Exception as e:
|
|
|
+ raise ERC20Error(f"获取代币合约实例失败: {e}")
|
|
|
+
|
|
|
+ def get_erc20_balance(self, token_contract_address: str, account_address: str) -> int:
|
|
|
+ """
|
|
|
+ 获取指定账户在指定 ERC-20 代币上的余额。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ account_address: 查询余额的以太坊地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 账户的代币余额 (以代币的最小单位为单位)。
|
|
|
+ """
|
|
|
+ if not self.w3.is_address(account_address):
|
|
|
+ raise ValueError(f"无效的账户地址: {account_address}")
|
|
|
+
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+ try:
|
|
|
+ balance = token_contract.functions.balanceOf(account_address).call()
|
|
|
+ return balance
|
|
|
+ except Exception as e:
|
|
|
+ # 尝试解析 Revert 错误
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower():
|
|
|
+ raise ERC20Error(f"查询 ERC-20 余额失败，可能是合约地址错误或账户不存在: {error_message}")
|
|
|
+ raise ERC20Error(f"查询 ERC-20 余额失败: {e}")
|
|
|
+
|
|
|
+ def get_erc20_allowance(self, token_contract_address: str, owner_address: str, spender_address: str) -> int:
|
|
|
+ """
|
|
|
+ 获取 owner 授权给 spender 的代币花费额度。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ owner_address: 授权方的地址。
|
|
|
+ spender_address: 被授权方的地址 (通常是另一个智能合约)。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 授权的代币数量 (以代币的最小单位为单位)。
|
|
|
+ """
|
|
|
+ if not self.w3.is_address(owner_address):
|
|
|
+ raise ValueError(f"无效的 owner 地址: {owner_address}")
|
|
|
+ if not self.w3.is_address(spender_address):
|
|
|
+ raise ValueError(f"无效的 spender 地址: {spender_address}")
|
|
|
+
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+ try:
|
|
|
+ allowance = token_contract.functions.allowance(owner_address, spender_address).call()
|
|
|
+ return allowance
|
|
|
+ except Exception as e:
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower():
|
|
|
+ raise ERC20Error(f"查询 ERC-20 allowance 失败，可能是合约地址错误或参数错误: {error_message}")
|
|
|
+ raise ERC20Error(f"查询 ERC-20 allowance 失败: {e}")
|
|
|
+
|
|
|
+ def get_erc20_decimals(self, token_contract_address: str) -> int:
|
|
|
+ """
|
|
|
+ 获取 ERC-20 代币的 decimals。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 代币的 decimal 数量 (通常 18 或 6)。
|
|
|
+ """
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+ try:
|
|
|
+ decimals = token_contract.functions.decimals().call()
|
|
|
+ return decimals
|
|
|
+ except Exception as e:
|
|
|
+ # 有些不完全符合标准的 ERC-20 可能没有 decimals() 方法
|
|
|
+ # 对于这种情况，可能需要硬编码或使用其他方法获取
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower() or "MethodNotFound" in error_message:
|
|
|
+ raise ERC20Error(f"获取 ERC-20 decimals 失败，可能是合约未实现该方法或地址错误: {error_message}")
|
|
|
+ raise ERC20Error(f"获取 ERC-20 decimals 失败: {e}")
|
|
|
+
|
|
|
+ def get_erc20_symbol(self, token_contract_address: str) -> str:
|
|
|
+ """
|
|
|
+ 获取 ERC-20 代币的 symbol。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 代币的 symbol 字符串。
|
|
|
+ """
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+ try:
|
|
|
+ symbol = token_contract.functions.symbol().call()
|
|
|
+ # Web3.py 返回的 bytes 需要解码
|
|
|
+ if isinstance(symbol, bytes):
|
|
|
+ symbol = symbol.decode('utf-8').replace('\x00', '').strip() # 移除可能的空字节
|
|
|
+ return symbol
|
|
|
+ except Exception as e:
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower() or "MethodNotFound" in error_message:
|
|
|
+ raise ERC20Error(f"获取 ERC-20 symbol 失败，可能是合约未实现该方法或地址错误: {error_message}")
|
|
|
+ raise ERC20Error(f"获取 ERC-20 symbol 失败: {e}")
|
|
|
+
|
|
|
+ def get_erc20_name(self, token_contract_address: str) -> str:
|
|
|
+ """
|
|
|
+ 获取 ERC-20 代币的 name。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 代币的 name 字符串。
|
|
|
+ """
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+ try:
|
|
|
+ name = token_contract.functions.name().call()
|
|
|
+ if isinstance(name, bytes):
|
|
|
+ name = name.decode('utf-8').replace('\x00', '').strip() # 移除可能的空字节
|
|
|
+ return name
|
|
|
+ except Exception as e:
|
|
|
+ error_message = str(e)
|
|
|
+ if "execution reverted" in error_message or "revert" in error_message.lower() or "MethodNotFound" in error_message:
|
|
|
+ raise ERC20Error(f"获取 ERC-20 name 失败，可能是合约未实现该方法或地址错误: {error_message}")
|
|
|
+ raise ERC20Error(f"获取 ERC-20 name 失败: {e}")
|
|
|
+
|
|
|
+ def build_erc20_transfer_transaction(self, token_contract_address: str, to_address: str, amount: int, gas_limit: int = None, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None) -> dict:
|
|
|
+ # ... (与之前相同，但内部调用 get_erc20_contract 获取合约实例)
|
|
|
+ """
|
|
|
+ 构建一个发送 ERC-20 代币的交易字典。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ to_address: 接收代币的地址。
|
|
|
+ amount: 发送的代币数量 (以代币的最小单位为单位)。
|
|
|
+ 注意: 这不是 Ether 单位，取决于代币的 decimal。
|
|
|
+ 例如，如果代币 decimal 是 18，发送 1 个代币需要传入 10^18。
|
|
|
+ gas_limit: 交易的 Gas Limit。如果为 None，将尝试估算。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 未签名的交易字典。
|
|
|
+ """
|
|
|
+ if not self.w3.is_address(to_address):
|
|
|
+ raise ValueError(f"无效的接收地址: {to_address}")
|
|
|
+
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+
|
|
|
+ transaction = token_contract.functions.transfer(
|
|
|
+ to_address,
|
|
|
+ amount
|
|
|
+ ).build_transaction({
|
|
|
+ 'chainId': self.chain_id,
|
|
|
+ 'from': self.account.address,
|
|
|
+ 'nonce': self.get_nonce(self.account.address),
|
|
|
+ 'type': 2
|
|
|
+ })
|
|
|
+
|
|
|
+ base_fee = self.w3.eth.get_block('latest').baseFeePerGas
|
|
|
+
|
|
|
+ if max_fee_per_gas_gwei is None:
|
|
|
+ max_fee_per_gas_wei = base_fee * 2 + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ else:
|
|
|
+ max_fee_per_gas_wei = self.w3.to_wei(max_fee_per_gas_gwei, 'gwei')
|
|
|
+
|
|
|
+ required_min_max_fee = base_fee + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ if max_fee_per_gas_wei < required_min_max_fee:
|
|
|
+ max_fee_per_gas_wei = required_min_max_fee
|
|
|
+
|
|
|
+ transaction['maxPriorityFeePerGas'] = self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ transaction['maxFeePerGas'] = max_fee_per_gas_wei
|
|
|
+
|
|
|
+ if gas_limit is None:
|
|
|
+ try:
|
|
|
+ tx_to_estimate = transaction.copy()
|
|
|
+ tx_to_estimate['from'] = self.account.address # 估算需要 from 字段
|
|
|
+ estimated_gas = self.estimate_gas(tx_to_estimate)
|
|
|
+ transaction['gas'] = estimated_gas
|
|
|
+ print(f"估算 ERC-20 transfer 交易 Gas 量: {estimated_gas}")
|
|
|
+ except EthTransactionError as e:
|
|
|
+ print(f"警告: ERC-20 transfer 交易 Gas 估算失败，请手动指定 gas_limit 或检查参数: {e}")
|
|
|
+ else:
|
|
|
+ transaction['gas'] = gas_limit
|
|
|
+
|
|
|
+ return transaction
|
|
|
+
|
|
|
+ def send_erc20_transfer(self, token_contract_address: str, to_address: str, amount: int, gas_limit: int = None, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None, wait_for_confirm: bool = True, timeout: int = 180):
|
|
|
+ # ... (与之前相同)
|
|
|
+ """
|
|
|
+ 发送 ERC-20 代币。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ to_address: 接收代币的地址。
|
|
|
+ amount: 发送的代币数量 (以代币的最小单位为单位)。
|
|
|
+ 注意: 这不是 Ether 单位，取决于代币的 decimal。
|
|
|
+ 例如，如果代币 decimal 是 18，发送 1 个代币需要传入 10^18。
|
|
|
+ gas_limit: 交易的 Gas Limit。如果为 None，将尝试估算。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。
|
|
|
+ wait_for_confirm: 是否等待交易被打包确认。
|
|
|
+ timeout: 等待交易确认的超时时间（秒）。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 交易哈希字符串。
|
|
|
+ """
|
|
|
+ print(f"准备发送 {amount} (以最小单位计) ERC-20 代币从 {token_contract_address} 到 {to_address}")
|
|
|
+ transaction = self.build_erc20_transfer_transaction(
|
|
|
+ token_contract_address,
|
|
|
+ to_address,
|
|
|
+ amount,
|
|
|
+ gas_limit,
|
|
|
+ max_priority_fee_gwei,
|
|
|
+ max_fee_per_gas_gwei
|
|
|
+ )
|
|
|
+ return self.send_transaction(transaction, wait_for_confirm=wait_for_confirm, timeout=timeout)
|
|
|
+
|
|
|
+ def build_erc20_approve_transaction(self, token_contract_address: str, spender_address: str, amount: int, gas_limit: int = None, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None) -> dict:
|
|
|
+ """
|
|
|
+ 构建一个授权 spender 花费自己代币的交易字典。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ spender_address: 被授权 spender 的地址 (通常是另一个智能合约，例如去中心化交易所)。
|
|
|
+ amount: 授权 spender 可以花费的代币数量 (以代币的最小单位为单位)。
|
|
|
+ gas_limit: 交易的 Gas Limit。如果为 None，将尝试估算。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 未签名的交易字典。
|
|
|
+ """
|
|
|
+ if not self.w3.is_address(spender_address):
|
|
|
+ raise ValueError(f"无效的 spender 地址: {spender_address}")
|
|
|
+
|
|
|
+ token_contract = self.get_erc20_contract(token_contract_address)
|
|
|
+
|
|
|
+ transaction = token_contract.functions.approve(
|
|
|
+ spender_address,
|
|
|
+ amount
|
|
|
+ ).build_transaction({
|
|
|
+ 'chainId': self.chain_id,
|
|
|
+ 'from': self.account.address,
|
|
|
+ 'nonce': self.get_nonce(self.account.address),
|
|
|
+ 'type': 2
|
|
|
+ })
|
|
|
+
|
|
|
+ base_fee = self.w3.eth.get_block('latest').baseFeePerGas
|
|
|
+
|
|
|
+ if max_fee_per_gas_gwei is None:
|
|
|
+ max_fee_per_gas_wei = base_fee * 2 + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ else:
|
|
|
+ max_fee_per_gas_wei = self.w3.to_wei(max_fee_per_gas_gwei, 'gwei')
|
|
|
+
|
|
|
+ required_min_max_fee = base_fee + self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ if max_fee_per_gas_wei < required_min_max_fee:
|
|
|
+ max_fee_per_gas_wei = required_min_max_fee
|
|
|
+
|
|
|
+ transaction['maxPriorityFeePerGas'] = self.w3.to_wei(max_priority_fee_gwei, 'gwei')
|
|
|
+ transaction['maxFeePerGas'] = max_fee_per_gas_wei
|
|
|
+
|
|
|
+ if gas_limit is None:
|
|
|
+ try:
|
|
|
+ tx_to_estimate = transaction.copy()
|
|
|
+ tx_to_estimate['from'] = self.account.address # 估算需要 from 字段
|
|
|
+ estimated_gas = self.estimate_gas(tx_to_estimate)
|
|
|
+ transaction['gas'] = estimated_gas
|
|
|
+ print(f"估算 ERC-20 approve 交易 Gas 量: {estimated_gas}")
|
|
|
+ except EthTransactionError as e:
|
|
|
+ print(f"警告: ERC-20 approve 交易 Gas 估算失败，请手动指定 gas_limit 或检查参数: {e}")
|
|
|
+ else:
|
|
|
+ transaction['gas'] = gas_limit
|
|
|
+
|
|
|
+ return transaction
|
|
|
+
|
|
|
+ def send_erc20_approve(self, token_contract_address: str, spender_address: str, amount: int, gas_limit: int = None, max_priority_fee_gwei: int = 2, max_fee_per_gas_gwei: int = None, wait_for_confirm: bool = True, timeout: int = 180):
|
|
|
+ """
|
|
|
+ 授权 spender 花费自己代币。
|
|
|
+
|
|
|
+ Args:
|
|
|
+ token_contract_address: ERC-20 代币合约地址。
|
|
|
+ spender_address: 被授权 spender 的地址 (通常是另一个智能合约，例如去中心化交易所)。
|
|
|
+ amount: 授权 spender 可以花费的代币数量 (以代币的最小单位为单位)。
|
|
|
+ gas_limit: 交易的 Gas Limit。如果为 None，将尝试估算。
|
|
|
+ max_priority_fee_gwei: EIP-1559 的 Max Priority Fee (以 Gwei 为单位)。
|
|
|
+ max_fee_per_gas_gwei: EIP-1559 的 Max Fee Per Gas (以 Gwei 为单位)。
|
|
|
+ wait_for_confirm: 是否等待交易被打包确认。
|
|
|
+ timeout: 等待交易确认的超时时间（秒）。
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 交易哈希字符串。
|
|
|
+ """
|
|
|
+ print(f"准备授权 {spender_address} 花费 {amount} (以最小单位计) ERC-20 代币从 {token_contract_address}")
|
|
|
+ transaction = self.build_erc20_approve_transaction(
|
|
|
+ token_contract_address,
|
|
|
+ spender_address,
|
|
|
+ amount,
|
|
|
+ gas_limit,
|
|
|
+ max_priority_fee_gwei,
|
|
|
+ max_fee_per_gas_gwei
|
|
|
+ )
|
|
|
+ return self.send_transaction(transaction, wait_for_confirm=wait_for_confirm, timeout=timeout)
|
|
|
+
|
|
|
+# --- 使用示例 ---
|
|
|
+
|
|
|
+if __name__ == "__main__":
|
|
|
+ # 请将 YOUR_RPC_URL_HERE 替换为您的实际 RPC 节点 URL
|
|
|
+ rpc_url = "YOUR_RPC_URL_HERE" # 请替换为你的 RPC 节点 URL
|
|
|
+
|
|
|
+ # Sepolia 测试网上的 WETH (Wrapped Ether) 合约地址作为示例
|
|
|
+ # 在真实环境中，你需要获取你想要交互的代币的合约地址
|
|
|
+ sepolia_weth_address = "0x3e57f1cffDD0c89Aec32056fEf5bB60F7c348345" # 示例 WETH 合约地址在 Sepolia
|
|
|
+
|
|
|
+ try:
|
|
|
+ client = EthClient(rpc_url)
|
|
|
+ my_address = client.account.address
|
|
|
+ print(f"我的账户地址: {my_address}")
|
|
|
+
|
|
|
+ # ---- 示例 1: 获取 ERC-20 代币信息 ----
|
|
|
+ try:
|
|
|
+ token_name = client.get_erc20_name(sepolia_weth_address)
|
|
|
+ token_symbol = client.get_erc20_symbol(sepolia_weth_address)
|
|
|
+ token_decimals = client.get_erc20_decimals(sepolia_weth_address)
|
|
|
+ print(f"\n代币名称: {token_name}")
|
|
|
+ print(f"代币符号: {token_symbol}")
|
|
|
+ print(f"代币 decimals: {token_decimals}")
|
|
|
+ except ERC20Error as e:
|
|
|
+ print(f"获取代币信息失败: {e}")
|
|
|
+
|
|
|
+ # ---- 示例 2: 获取 ERC-20 代币余额 ----
|
|
|
+ # 请替换为你要查询余额的地址
|
|
|
+ # address_to_check = "某个地址" # 例如你的地址或另一个地址
|
|
|
+ # try:
|
|
|
+ # erc20_balance_wei = client.get_erc20_balance(sepolia_weth_address, address_to_check)
|
|
|
+ # # 将最小单位转换为用户友好的单位
|
|
|
+ # # 如果知道 decimals，可以计算
|
|
|
+ # # amount_in_token_units = erc20_balance_wei / (10**token_decimals)
|
|
|
+ # print(f"\n地址 {address_to_check} 的 {token_symbol} 余额 (最小单位): {erc20_balance_wei}")
|
|
|
+ # # print(f"地址 {address_to_check} 的 {token_symbol} 余额 ({token_symbol} 单位): {amount_in_token_units}") # 如果 decimals 已知
|
|
|
+ # except ERC20Error as e:
|
|
|
+ # print(f"获取 ERC-20 余额失败: {e}")
|
|
|
+
|
|
|
+ # ---- 示例 3: 获取 ERC-20 allowance ----
|
|
|
+ # 请替换为 owner 和 spender 地址
|
|
|
+ # owner_address = my_address # 例如你的地址
|
|
|
+ # spender_address = "某个需要花费你代币的合约地址" # 例如 Uniswap router 地址
|
|
|
+ # try:
|
|
|
+ # erc20_allowance = client.get_erc20_allowance(sepolia_weth_address, owner_address, spender_address)
|
|
|
+ # print(f"\n地址 {owner_address} 授权给 {spender_address} 的 {token_symbol} 额度 (最小单位): {erc20_allowance}")
|
|
|
+ # except ERC20Error as e:
|
|
|
+ # print(f"获取 ERC-20 allowance 失败: {e}")
|
|
|
+
|
|
|
+ # ---- 示例 4: 发送 ERC-20 代币交易 ----
|
|
|
+ # 请替换为你要发送的 ERC-20 代币合约地址、目标地址和数量
|
|
|
+ # target_token_address = "目标以太坊地址" # 接收代币的地址
|
|
|
+ # # 发送 0.01 个 WETH (假设 decimals 是 18)
|
|
|
+ # amount_to_send_token = 0.01 * (10**18)
|
|
|
+
|
|
|
+ # print(f"\n尝试发送 ERC-20 代币...")
|
|
|
+ # try:
|
|
|
+ # tx_hash_erc20_transfer = client.send_erc20_transfer(sepolia_weth_address, target_token_address, int(amount_to_send_token))
|
|
|
+ # print(f"ERC-20 transfer 交易成功，哈希: {tx_hash_erc20_transfer}")
|
|
|
+ # except EthTransactionError as e:
|
|
|
+ # print(f"发送 ERC-20 transfer 交易失败: {e}")
|
|
|
+
|
|
|
+ # ---- 示例 5: 发送 ERC-20 approve 交易 ----
|
|
|
+ # 请替换为你要授权的 spender 地址和数量
|
|
|
+ # spender_contract_address = "某个需要你授权代币的合约地址"
|
|
|
+ # # 授权 spender 可以花费 100 个 WETH (假设 decimals 是 18)
|
|
|
+ # amount_to_approve_token = 100 * (10**18)
|
|
|
+ # amount_to_approve_token_max = 2**256 - 1 # 或者授权一个非常大的数 (无限授权)
|
|
|
+
|
|
|
+ # print(f"\n尝试授权 ERC-20 代币花费权限...")
|
|
|
+ # try:
|
|
|
+ # tx_hash_erc20_approve = client.send_erc20_approve(sepolia_weth_address, spender_contract_address, int(amount_to_approve_token)) # or amount_to_approve_token_max
|
|
|
+ # print(f"ERC-20 approve 交易成功，哈希: {tx_hash_erc20_approve}")
|
|
|
+ # except EthTransactionError as e:
|
|
|
+ # print(f"发送 ERC-20 approve 交易失败: {e}")
|
|
|
+
|
|
|
+ except (ConnectionError, EnvironmentError, ValueError, EthTransactionError, ERC20Error) as e:
|
|
|
+ print(f"客户端初始化或操作失败: {e}")
|
