当前位置：首页 > news >正文

Python与区块链：如何用Web3.py与以太坊交互

news 2025/11/2 8:05:07

Python与区块链：如何用Web3.py与以太坊交互
- 1. 引言
- - 1.1 区块链技术概述
  - 1.2 Python在区块链开发中的优势
- 2. Web3.py基础概念
- - 2.1 Web3.py架构解析
  - 2.2 核心组件与功能
- 3. 环境搭建与配置
- - 3.1 安装Web3.py
  - 3.2 配置以太坊节点连接
- 4. 基础交互操作
- - 4.1 区块链信息查询
  - 4.2 交易处理与Gas管理
- 5. 智能合约交互
- - 5.1 智能合约基础
  - 5.2 合约编译与部署
- 6. 完整DApp示例：去中心化投票系统
- - 6.1 投票合约实现
  - 6.2 DApp后端服务
- 7. 完整代码实现与测试
- - 7.1 完整的项目结构
  - 7.2 综合演示脚本
  - 7.3 测试用例
- 8. 安全最佳实践
- - 8.1 私钥安全管理
- 9. 性能优化与高级特性
- - 9.1 批量操作与异步处理
- 10. 总结与展望
- - 10.1 技术总结
  - 10.2 数学基础回顾
  - 10.3 未来发展方向
  - 10.4 实践建议

『宝藏代码胶囊开张啦！』—— 我的 CodeCapsule 来咯！✨
写代码不再头疼！我的新站点 CodeCapsule 主打一个 “白菜价”+“量身定制”！无论是卡脖子的毕设/课设/文献复现，需要灵光一现的算法改进，还是想给项目加个“外挂”，这里都有便宜又好用的代码方案等你发现！低成本，高适配，助你轻松通关！速来围观 👉 CodeCapsule官网

Python与区块链：如何用Web3.py与以太坊交互

1. 引言

1.1 区块链技术概述

区块链技术作为数字时代的重要创新，正在重塑我们对信任、交易和数据管理的认知。从本质上讲，区块链是一个去中心化的分布式账本，它通过密码学技术确保数据的安全性和不可篡改性。以太坊作为第二代区块链技术的代表，不仅支持加密货币交易，更重要的是引入了智能合约的概念，使得在区块链上执行复杂的业务逻辑成为可能。

区块链的核心特性可以用以下数学公式表示：

区块链完整性公式：
$Hn=Hash(Hn−1∣∣Tn∣∣Nonce)H_{n} = \text{Hash}(H_{n-1} || T_{n} || \text{Nonce})$

其中：

$H_{n}$ 是当前区块的哈希值
$H_{n-1}$ 是前一个区块的哈希值
$T_{n}$ 是当前区块的交易集合
$Nonce\text{Nonce}$ 是工作量证明的随机数

1.2 Python在区块链开发中的优势

Python凭借其简洁的语法、丰富的库生态系统和强大的社区支持，已成为区块链开发的重要工具。Web3.py作为Python与以太坊交互的核心库，为开发者提供了：

直观的API设计：简化了与以太坊节点的复杂交互
类型安全：支持类型注解，提高代码可靠性
异步支持：充分利用现代Python的异步特性
完善的文档：降低学习曲线，加速开发进程

2. Web3.py基础概念

2.1 Web3.py架构解析

Web3.py库采用分层架构设计，为开发者提供了从底层JSON-RPC调用到高级抽象接口的完整工具链。

2.2 核心组件与功能

Web3.py的核心功能模块包括：

Web3：主入口点，管理节点连接和基础操作
Eth：以太坊区块链交互模块
Net：网络信息查询
Geth：Geth特定功能的扩展（如挖矿控制）
Parity：Parity客户端的特定功能

3. 环境搭建与配置

3.1 安装Web3.py

首先创建隔离的Python环境并安装必要依赖：

# scripts/setup_environment.py
#!/usr/bin/env python3
"""
区块链开发环境设置脚本
"""
import subprocess
import sys
import os
from pathlib import Pathdef run_command(cmd, check=True):"""运行shell命令并返回结果"""print(f"执行命令: {cmd}")result = subprocess.run(cmd, shell=True, capture_output=True, text=True)if check and result.returncode != 0:print(f"错误: {result.stderr}")sys.exit(result.returncode)return resultdef setup_blockchain_environment():"""设置区块链开发环境"""# 创建虚拟环境if not Path("venv").exists():print("创建虚拟环境...")run_command("python -m venv venv")# 确定激活命令if sys.platform == "win32":pip_cmd = "venv\\Scripts\\pip"python_cmd = "venv\\Scripts\\python"else:pip_cmd = "venv/bin/pip"python_cmd = "venv/bin/python"# 升级pip并安装依赖print("安装依赖包...")run_command(f"{pip_cmd} install --upgrade pip")requirements = ["web3==6.11.0","python-dotenv==1.0.0","cryptography==41.0.7","eth-account==0.9.0","eth-typing==3.5.0","eth-utils==2.3.1","hexbytes==0.3.1","ipfshttpclient==0.8.0a2","requests==2.31.0","websockets==12.0"]for package in requirements:run_command(f"{pip_cmd} install {package}")# 验证安装print("验证安装...")result = run_command(f"{python_cmd} -c \"import web3; print(f'Web3.py版本: {web3.__version__}')\"")print("区块链开发环境设置完成!")print(f"使用以下命令激活环境:")if sys.platform == "win32":print("venv\\Scripts\\activate")else:print("source venv/bin/activate")if __name__ == "__main__":setup_blockchain_environment()

3.2 配置以太坊节点连接

配置与不同以太坊网络的连接：

# config/blockchain_config.py
"""
区块链配置模块
"""
import os
from typing import Dict, Optional
from dotenv import load_dotenv
from web3 import Web3
from web3.middleware import geth_poa_middleware# 加载环境变量
load_dotenv()class BlockchainConfig:"""区块链配置类"""# 网络配置NETWORKS: Dict[str, Dict[str, str]] = {"mainnet": {"rpc_url": os.getenv("MAINNET_RPC_URL", "https://mainnet.infura.io/v3/your-project-id"),"chain_id": "1","name": "Ethereum Mainnet"},"goerli": {"rpc_url": os.getenv("GOERLI_RPC_URL", "https://goerli.infura.io/v3/your-project-id"),"chain_id": "5","name": "Goerli Testnet"},"sepolia": {"rpc_url": os.getenv("SEPOLIA_RPC_URL", "https://sepolia.infura.io/v3/your-project-id"),"chain_id": "11155111","name": "Sepolia Testnet"},"ganache": {"rpc_url": "http://localhost:8545","chain_id": "1337","name": "Ganache Local"},"polygon": {"rpc_url": os.getenv("POLYGON_RPC_URL", "https://polygon-mainnet.infura.io/v3/your-project-id"),"chain_id": "137","name": "Polygon Mainnet"}}@classmethoddef get_web3_connection(cls, network: str = "goerli") -> Web3:"""获取Web3连接实例Args:network: 网络名称Returns:Web3连接实例"""if network not in cls.NETWORKS:raise ValueError(f"不支持的网络: {network}。可用网络: {list(cls.NETWORKS.keys())}")rpc_url = cls.NETWORKS[network]["rpc_url"]w3 = Web3(Web3.HTTPProvider(rpc_url))# 对于POA网络（如Goerli、Polygon），需要注入POA中间件if network in ["goerli", "polygon"]:w3.middleware_onion.inject(geth_poa_middleware, layer=0)return w3@classmethoddef test_connection(cls, network: str = "goerli") -> bool:"""测试网络连接Args:network: 网络名称Returns:连接是否成功"""try:w3 = cls.get_web3_connection(network)connected = w3.is_connected()if connected:print(f"✅ 成功连接到 {cls.NETWORKS[network]['name']}")print(f"   最新区块: {w3.eth.block_number}")print(f"   链ID: {w3.eth.chain_id}")else:print(f"❌ 无法连接到 {cls.NETWORKS[network]['name']}")return connectedexcept Exception as e:print(f"❌ 连接错误: {str(e)}")return False# 环境变量模板
ENV_TEMPLATE = """
# 区块链开发环境配置
MAINNET_RPC_URL=https://mainnet.infura.io/v3/your-project-id
GOERLI_RPC_URL=https://goerli.infura.io/v3/your-project-id
SEPOLIA_RPC_URL=https://sepolia.infura.io/v3/your-project-id
POLYGON_RPC_URL=https://polygon-mainnet.infura.io/v3/your-project-id# 钱包配置（谨慎处理私钥！）
PRIVATE_KEY=your_private_key_here
WALLET_ADDRESS=your_wallet_address_here# 智能合约地址
CONTRACT_ADDRESS=your_contract_address_here
"""

4. 基础交互操作

4.1 区块链信息查询

实现基础的区块链数据查询功能：

# src/blockchain_basics.py
"""
区块链基础操作模块
"""
from typing import Dict, Any, Optional
from decimal import Decimal
from web3 import Web3
from web3.types import BlockData, Wei, TxDataclass BlockchainBasics:"""区块链基础操作类"""def __init__(self, w3: Web3):"""初始化Args:w3: Web3实例"""self.w3 = w3def get_network_info(self) -> Dict[str, Any]:"""获取网络信息Returns:网络信息字典"""try:block_number = self.w3.eth.block_numberchain_id = self.w3.eth.chain_idgas_price = self.w3.eth.gas_priceclient_version = self.w3.client_versionreturn {"connected": True,"block_number": block_number,"chain_id": chain_id,"gas_price_wei": gas_price,"gas_price_gwei": self.w3.from_wei(gas_price, "gwei"),"client_version": client_version,"sync_status": self.w3.eth.syncing}except Exception as e:return {"connected": False,"error": str(e)}def get_block_info(self, block_identifier: Optional[str] = None) -> BlockData:"""获取区块信息Args:block_identifier: 区块标识（最新区块为"latest"）Returns:区块数据"""if block_identifier is None:block_identifier = "latest"return self.w3.eth.get_block(block_identifier)def get_transaction(self, tx_hash: str) -> TxData:"""获取交易信息Args:tx_hash: 交易哈希Returns:交易数据"""return self.w3.eth.get_transaction(tx_hash)def get_balance(self, address: str, unit: str = "ether") -> Decimal:"""获取地址余额Args:address: 钱包地址unit: 单位（wei, gwei, ether）Returns:余额数值"""balance_wei = self.w3.eth.get_balance(address)return self.w3.from_wei(balance_wei, unit)def get_transaction_count(self, address: str) -> int:"""获取地址的交易数量（nonce）Args:address: 钱包地址Returns:交易数量"""return self.w3.eth.get_transaction_count(address)def wei_converter(self, amount: Wei, from_unit: str = "wei", to_unit: str = "ether") -> Decimal:"""Wei单位转换器Args:amount: 金额from_unit: 原单位to_unit: 目标单位Returns:转换后的金额"""wei_amount = self.w3.to_wei(amount, from_unit)return self.w3.from_wei(wei_amount, to_unit)def demonstrate_basics():"""演示基础功能"""from config.blockchain_config import BlockchainConfig# 连接到测试网络w3 = BlockchainConfig.get_web3_connection("goerli")basics = BlockchainBasics(w3)# 获取网络信息network_info = basics.get_network_info()print("=== 网络信息 ===")for key, value in network_info.items():print(f"{key}: {value}")# 获取最新区块信息latest_block = basics.get_block_info("latest")print(f"\n=== 最新区块 (#{latest_block['number']}) ===")print(f"哈希: {latest_block['hash'].hex()}")print(f"时间戳: {latest_block['timestamp']}")print(f"交易数量: {len(latest_block['transactions'])}")print(f"矿工: {latest_block['miner']}")# 单位转换示例print(f"\n=== 单位转换 ===")one_ether_in_wei = basics.wei_converter(1, "ether", "wei")one_gwei_in_ether = basics.wei_converter(1, "gwei", "ether")print(f"1 ETH = {one_ether_in_wei} Wei")print(f"1 Gwei = {one_gwei_in_ether} ETH")if __name__ == "__main__":demonstrate_basics()

4.2 交易处理与Gas管理

实现交易创建、签名和发送功能：

# src/transaction_manager.py
"""
交易管理模块
"""
import json
from typing import Dict, Any, Optional
from decimal import Decimal
from web3 import Web3
from web3.types import TxParams, Wei, Nonce
from eth_account import Account
from eth_account.signers.local import LocalAccountclass TransactionManager:"""交易管理器"""def __init__(self, w3: Web3):"""初始化Args:w3: Web3实例"""self.w3 = w3def create_transaction(self,from_address: str,to_address: str,value: Wei,gas_limit: Optional[int] = None,gas_price: Optional[Wei] = None,max_fee_per_gas: Optional[Wei] = None,max_priority_fee_per_gas: Optional[Wei] = None,data: bytes = b"",nonce: Optional[Nonce] = None) -> TxParams:"""创建交易参数Args:from_address: 发送地址to_address: 接收地址value: 发送金额（Wei）gas_limit: Gas限制gas_price: Gas价格（传统交易）max_fee_per_gas: 最大Gas费用（EIP-1559）max_priority_fee_per_gas: 最大优先费用（EIP-1559）data: 交易数据nonce: 交易序号Returns:交易参数"""if nonce is None:nonce = self.w3.eth.get_transaction_count(from_address)if gas_limit is None:gas_limit = 21000  # 标准ETH转账的Gas限制# 构建交易参数transaction: TxParams = {"nonce": nonce,"to": to_address,"value": value,"data": data,"gas": gas_limit,"chainId": self.w3.eth.chain_id}# EIP-1559交易if max_fee_per_gas is not None and max_priority_fee_per_gas is not None:transaction["maxFeePerGas"] = max_fee_per_gastransaction["maxPriorityFeePerGas"] = max_priority_fee_per_gastransaction["type"] = 2  # EIP-1559交易类型else:# 传统交易if gas_price is None:gas_price = self.w3.eth.gas_pricetransaction["gasPrice"] = gas_pricereturn transactiondef estimate_gas_cost(self, transaction: TxParams) -> Dict[str, Any]:"""估算Gas成本Args:transaction: 交易参数Returns:Gas成本信息"""try:gas_estimate = self.w3.eth.estimate_gas(transaction)gas_price = transaction.get("gasPrice") or transaction.get("maxFeePerGas", self.w3.eth.gas_price)total_cost_wei = gas_estimate * gas_pricetotal_cost_eth = self.w3.from_wei(total_cost_wei, "ether")return {"gas_estimate": gas_estimate,"gas_price_wei": gas_price,"gas_price_gwei": self.w3.from_wei(gas_price, "gwei"),"total_cost_wei": total_cost_wei,"total_cost_eth": total_cost_eth,"success": True}except Exception as e:return {"success": False,"error": str(e)}def sign_transaction(self, transaction: TxParams, private_key: str) -> bytes:"""签名交易Args:transaction: 交易参数private_key: 私钥Returns:签名后的交易数据"""return self.w3.eth.account.sign_transaction(transaction, private_key).rawTransactiondef send_transaction(self, signed_transaction: bytes) -> str:"""发送已签名的交易Args:signed_transaction: 签名后的交易Returns:交易哈希"""tx_hash = self.w3.eth.send_raw_transaction(signed_transaction)return tx_hash.hex()def wait_for_transaction(self, tx_hash: str, timeout: int = 120) -> Dict[str, Any]:"""等待交易确认Args:tx_hash: 交易哈希timeout: 超时时间（秒）Returns:交易收据"""receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash, timeout=timeout)return dict(receipt)def get_gas_prices(self) -> Dict[str, Any]:"""获取当前Gas价格信息Returns:Gas价格信息"""try:# 传统Gas价格gas_price = self.w3.eth.gas_price# EIP-1559费用fee_history = self.w3.eth.fee_history(1, "latest")base_fee = fee_history["baseFeePerGas"][0]# 估算优先费用max_priority_fee = self.w3.eth.max_priority_feereturn {"gas_price_wei": gas_price,"gas_price_gwei": self.w3.from_wei(gas_price, "gwei"),"base_fee_wei": base_fee,"base_fee_gwei": self.w3.from_wei(base_fee, "gwei"),"max_priority_fee_wei": max_priority_fee,"max_priority_fee_gwei": self.w3.from_wei(max_priority_fee, "gwei"),"recommended_max_fee_wei": base_fee * 2 + max_priority_fee,"success": True}except Exception as e:return {"success": False,"error": str(e)}def demonstrate_transactions():"""演示交易功能"""from config.blockchain_config import BlockchainConfigimport os# 连接到本地Ganache（演示用）w3 = BlockchainConfig.get_web3_connection("ganache")tx_manager = TransactionManager(w3)# 从环境变量获取测试账户（仅用于演示，生产环境要保护私钥！）private_key = os.getenv("TEST_PRIVATE_KEY")if not private_key:print("请设置TEST_PRIVATE_KEY环境变量")returnaccount: LocalAccount = Account.from_key(private_key)from_address = account.addressprint(f"=== 账户信息 ===")print(f"地址: {from_address}")print(f"余额: {w3.from_wei(w3.eth.get_balance(from_address), 'ether')} ETH")# 获取Gas价格gas_info = tx_manager.get_gas_prices()print(f"\n=== Gas信息 ===")for key, value in gas_info.items():print(f"{key}: {value}")# 创建测试交易（发送到零地址）to_address = "0x0000000000000000000000000000000000000000"value = w3.to_wei(0.001, "ether")transaction = tx_manager.create_transaction(from_address=from_address,to_address=to_address,value=value,max_fee_per_gas=w3.to_wei(20, "gwei"),max_priority_fee_per_gas=w3.to_wei(2, "gwei"))# 估算Gas成本gas_estimate = tx_manager.estimate_gas_cost(transaction)print(f"\n=== Gas估算 ===")for key, value in gas_estimate.items():print(f"{key}: {value}")# 在实际环境中，这里会签名并发送交易print(f"\n=== 交易详情 ===")print(f"从: {from_address}")print(f"到: {to_address}")print(f"金额: {w3.from_wei(value, 'ether')} ETH")print(f"Nonce: {transaction['nonce']}")# 注意：在生产环境中发送真实交易# signed_tx = tx_manager.sign_transaction(transaction, private_key)# tx_hash = tx_manager.send_transaction(signed_tx)# print(f"交易已发送: {tx_hash}")if __name__ == "__main__":demonstrate_transactions()

5. 智能合约交互

5.1 智能合约基础

智能合约是在区块链上运行的自动化合约，其执行逻辑可以用以下公式表示：

智能合约状态转换：
$St+1=Contract(St,T,Msg)S_{t+1} = \text{Contract}(S_t, T, \text{Msg})$

其中：

$S_t$ 是时间 $t$ 的合约状态
$T$ 是交易输入
$Msg\text{Msg}$ 是消息调用上下文
$S_{t+1}$ 是执行后的新状态

5.2 合约编译与部署

# src/contract_manager.py
"""
智能合约管理器
"""
import json
import os
from typing import Dict, Any, Optional, List
from pathlib import Path
from web3 import Web3
from web3.contract import Contract
from web3.types import TxReceipt, HexStrclass ContractManager:"""智能合约管理器"""def __init__(self, w3: Web3):"""初始化Args:w3: Web3实例"""self.w3 = w3self.contracts: Dict[str, Contract] = {}def compile_contract(self, solidity_source: str, contract_name: str, optimizer_runs: int = 200) -> Dict[str, Any]:"""编译Solidity合约（需要py-solc-x）Args:solidity_source: Solidity源代码contract_name: 合约名称optimizer_runs: 优化器运行次数Returns:编译结果"""try:from solcx import compile_source, install_solc# 安装最新Solidity编译器install_solc('0.8.19')compiled_sol = compile_source(solidity_source,output_values=['abi', 'bin'],solc_version='0.8.19',optimizer={"enabled": True, "runs": optimizer_runs})contract_interface = compiled_sol[f'<stdin>:{contract_name}']return {"success": True,"abi": contract_interface['abi'],"bytecode": contract_interface['bin']}except ImportError:return {"success": False,"error": "请安装py-solc-x: pip install py-solc-x"}except Exception as e:return {"success": False,"error": str(e)}def load_contract_from_file(self, abi_path: str, bytecode_path: Optional[str] = None) -> Dict[str, Any]:"""从文件加载合约Args:abi_path: ABI文件路径bytecode_path: 字节码文件路径Returns:合约接口"""try:with open(abi_path, 'r') as f:abi = json.load(f)bytecode = Noneif bytecode_path and os.path.exists(bytecode_path):with open(bytecode_path, 'r') as f:bytecode = f.read().strip()return {"success": True,"abi": abi,"bytecode": bytecode}except Exception as e:return {"success": False,"error": str(e)}def deploy_contract(self, abi: List[Dict], bytecode: str, deployer_private_key: str, constructor_args: Optional[list] = None,gas_limit: int = 2000000) -> Dict[str, Any]:"""部署合约Args:abi: 合约ABIbytecode: 合约字节码deployer_private_key: 部署者私钥constructor_args: 构造函数参数gas_limit: Gas限制Returns:部署结果"""try:from eth_account import Accountaccount = Account.from_key(deployer_private_key)deployer_address = account.address# 创建合约对象contract = self.w3.eth.contract(abi=abi, bytecode=bytecode)# 构建部署交易constructor = contract.constructor(*(constructor_args or []))transaction = constructor.build_transaction({'from': deployer_address,'nonce': self.w3.eth.get_transaction_count(deployer_address),'gas': gas_limit,'gasPrice': self.w3.eth.gas_price})# 签名并发送交易signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key=deployer_private_key)tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)# 等待交易确认tx_receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)if tx_receipt.status == 1:contract_address = tx_receipt.contractAddress# 保存合约实例deployed_contract = self.w3.eth.contract(address=contract_address, abi=abi)self.contracts[contract_address] = deployed_contractreturn {"success": True,"contract_address": contract_address,"transaction_hash": tx_hash.hex(),"gas_used": tx_receipt.gasUsed,"block_number": tx_receipt.blockNumber}else:return {"success": False,"error": "合约部署失败","transaction_hash": tx_hash.hex()}except Exception as e:return {"success": False,"error": str(e)}def load_existing_contract(self, address: str, abi: List[Dict]) -> Contract:"""加载已部署的合约Args:address: 合约地址abi: 合约ABIReturns:合约实例"""contract = self.w3.eth.contract(address=address, abi=abi)self.contracts[address] = contractreturn contractdef call_contract_function(self, contract_address: str, function_name: str, args: Optional[list] = None, call_options: Optional[Dict] = None) -> Any:"""调用合约视图函数（只读）Args:contract_address: 合约地址function_name: 函数名称args: 函数参数call_options: 调用选项Returns:函数调用结果"""try:if contract_address not in self.contracts:raise ValueError(f"合约未加载: {contract_address}")contract = self.contracts[contract_address]function = getattr(contract.functions, function_name)if args:result = function(*args).call(**(call_options or {}))else:result = function().call(**(call_options or {}))return {"success": True,"result": result}except Exception as e:return {"success": False,"error": str(e)}def transact_contract_function(self, contract_address: str, function_name: str,private_key: str, args: Optional[list] = None,transaction_options: Optional[Dict] = None) -> Dict[str, Any]:"""执行合约交易函数（修改状态）Args:contract_address: 合约地址function_name: 函数名称private_key: 签名私钥args: 函数参数transaction_options: 交易选项Returns:交易结果"""try:from eth_account import Accountif contract_address not in self.contracts:raise ValueError(f"合约未加载: {contract_address}")account = Account.from_key(private_key)from_address = account.addresscontract = self.contracts[contract_address]function = getattr(contract.functions, function_name)# 构建交易if args:transaction = function(*args).build_transaction({'from': from_address,'nonce': self.w3.eth.get_transaction_count(from_address),'gasPrice': self.w3.eth.gas_price,**(transaction_options or {})})else:transaction = function().build_transaction({'from': from_address,'nonce': self.w3.eth.get_transaction_count(from_address),'gasPrice': self.w3.eth.gas_price,**(transaction_options or {})})# 签名并发送交易signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key=private_key)tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)# 等待交易确认tx_receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)return {"success": tx_receipt.status == 1,"transaction_hash": tx_hash.hex(),"gas_used": tx_receipt.gasUsed,"block_number": tx_receipt.blockNumber,"status": "success" if tx_receipt.status == 1 else "failed"}except Exception as e:return {"success": False,"error": str(e)}# 示例智能合约
SIMPLE_STORAGE_CONTRACT = """
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;contract SimpleStorage {uint256 private storedData;address public owner;event ValueChanged(uint256 newValue, address changedBy);event OwnershipTransferred(address indexed previousOwner, address indexed newOwner);modifier onlyOwner() {require(msg.sender == owner, "Only owner can call this function");_;}constructor(uint256 initialValue) {storedData = initialValue;owner = msg.sender;}function set(uint256 x) public {storedData = x;emit ValueChanged(x, msg.sender);}function get() public view returns (uint256) {return storedData;}function transferOwnership(address newOwner) public onlyOwner {require(newOwner != address(0), "New owner is the zero address");emit OwnershipTransferred(owner, newOwner);owner = newOwner;}function getOwner() public view returns (address) {return owner;}
}
"""

6. 完整DApp示例：去中心化投票系统

6.1 投票合约实现

# contracts/voting_contract.py
"""
投票系统智能合约
"""
VOTING_CONTRACT_SOURCE = """
// SPDX-License-Identifier: MIT
pragma solidity ^0.8.0;contract Voting {struct Voter {uint weight; // 投票权重bool voted;  // 是否已投票address delegate; // 委托投票地址uint vote;   // 投票的提案索引}struct Proposal {bytes32 name;   // 提案名称（短名称）uint voteCount; // 得票数}address public chairperson;mapping(address => Voter) public voters;Proposal[] public proposals;uint public votingStart;uint public votingEnd;bool public votingClosed;// 事件event VoteCast(address indexed voter, uint proposal);VoteDelegate(address indexed from, address indexed to);VotingStarted(uint startTime, uint endTime);VotingEnded(uint endTime);modifier onlyChairperson() {require(msg.sender == chairperson, "Only chairperson can call this function");_;}modifier duringVotingPeriod() {require(block.timestamp >= votingStart && block.timestamp <= votingEnd, "Not during voting period");require(!votingClosed, "Voting is closed");_;}constructor(bytes32[] memory proposalNames, uint votingDurationInMinutes) {chairperson = msg.sender;voters[chairperson].weight = 1;votingStart = block.timestamp;votingEnd = block.timestamp + votingDurationInMinutes * 1 minutes;for (uint i = 0; i < proposalNames.length; i++) {proposals.push(Proposal({name: proposalNames[i],voteCount: 0}));}emit VotingStarted(votingStart, votingEnd);}function giveRightToVote(address voter) external onlyChairperson {require(!voters[voter].voted, "The voter already voted.");require(voters[voter].weight == 0, "Voter already has right to vote.");voters[voter].weight = 1;}function delegate(address to) external duringVotingPeriod {Voter storage sender = voters[msg.sender];require(!sender.voted, "You already voted.");require(to != msg.sender, "Self-delegation is disallowed.");while (voters[to].delegate != address(0)) {to = voters[to].delegate;require(to != msg.sender, "Found loop in delegation.");}sender.voted = true;sender.delegate = to;Voter storage delegate_ = voters[to];if (delegate_.voted) {proposals[delegate_.vote].voteCount += sender.weight;} else {delegate_.weight += sender.weight;}emit VoteDelegate(msg.sender, to);}function vote(uint proposal) external duringVotingPeriod {Voter storage sender = voters[msg.sender];require(sender.weight != 0, "Has no right to vote");require(!sender.voted, "Already voted.");sender.voted = true;sender.vote = proposal;proposals[proposal].voteCount += sender.weight;emit VoteCast(msg.sender, proposal);}function winningProposal() public view returns (uint winningProposal_) {uint winningVoteCount = 0;for (uint p = 0; p < proposals.length; p++) {if (proposals[p].voteCount > winningVoteCount) {winningVoteCount = proposals[p].voteCount;winningProposal_ = p;}}}function winnerName() external view returns (bytes32 winnerName_) {winnerName_ = proposals[winningProposal()].name;}function getProposalsCount() external view returns (uint) {return proposals.length;}function getProposal(uint index) external view returns (bytes32 name, uint voteCount) {require(index < proposals.length, "Invalid proposal index");Proposal memory proposal = proposals[index];return (proposal.name, proposal.voteCount);}function closeVoting() external onlyChairperson {require(block.timestamp > votingEnd || msg.sender == chairperson, "Cannot close voting yet");votingClosed = true;emit VotingEnded(block.timestamp);}function getVotingStatus() external view returns (bool isActive, uint timeRemaining) {isActive = !votingClosed && block.timestamp >= votingStart && block.timestamp <= votingEnd;if (block.timestamp < votingEnd) {timeRemaining = votingEnd - block.timestamp;} else {timeRemaining = 0;}}
}
"""

6.2 DApp后端服务

# src/voting_dapp.py
"""
去中心化投票DApp后端服务
"""
import asyncio
import json
from typing import Dict, Any, List, Optional
from datetime import datetime, timedelta
from web3 import Web3
from web3.contract import Contract
from web3.middleware import geth_poa_middlewareclass VotingDApp:"""去中心化投票DApp"""def __init__(self, w3: Web3, contract_address: str, contract_abi: List[Dict]):"""初始化投票DAppArgs:w3: Web3实例contract_address: 合约地址contract_abi: 合约ABI"""self.w3 = w3self.contract_address = contract_addressself.contract = w3.eth.contract(address=contract_address, abi=contract_abi)# 事件过滤器self._setup_event_filters()def _setup_event_filters(self):"""设置事件过滤器"""try:self.vote_cast_filter = self.contract.events.VoteCast.create_filter(fromBlock='latest')self.vote_delegate_filter = self.contract.events.VoteDelegate.create_filter(fromBlock='latest')except:# 如果创建过滤器失败，使用轮询方式self.vote_cast_filter = Noneself.vote_delegate_filter = Noneasync def watch_events(self, callback: callable):"""监听合约事件Args:callback: 事件回调函数"""while True:try:# 检查新事件if self.vote_cast_filter:vote_events = self.vote_cast_filter.get_new_entries()for event in vote_events:await callback("VoteCast", dict(event))if self.vote_delegate_filter:delegate_events = self.vote_delegate_filter.get_new_entries()for event in delegate_events:await callback("VoteDelegate", dict(event))await asyncio.sleep(2)  # 每2秒检查一次except Exception as e:print(f"事件监听错误: {e}")await asyncio.sleep(5)def get_voting_info(self) -> Dict[str, Any]:"""获取投票信息Returns:投票信息字典"""try:chairperson = self.contract.functions.chairperson().call()proposal_count = self.contract.functions.getProposalsCount().call()voting_status = self.contract.functions.getVotingStatus().call()# 获取所有提案proposals = []for i in range(proposal_count):name, vote_count = self.contract.functions.getProposal(i).call()proposals.append({"index": i,"name": name.decode('utf-8').rstrip('\x00'),  # 清理bytes32"vote_count": vote_count})# 获取获胜提案try:winner_index = self.contract.functions.winningProposal().call()winner_name = self.contract.functions.winnerName().call()winner_name = winner_name.decode('utf-8').rstrip('\x00')except:winner_index = Nonewinner_name = Nonereturn {"success": True,"chairperson": chairperson,"proposal_count": proposal_count,"proposals": proposals,"winner_index": winner_index,"winner_name": winner_name,"voting_active": voting_status[0],"time_remaining": voting_status[1],"voting_closed": self.contract.functions.votingClosed().call()}except Exception as e:return {"success": False,"error": str(e)}def get_voter_info(self, address: str) -> Dict[str, Any]:"""获取投票者信息Args:address: 投票者地址Returns:投票者信息"""try:voter_data = self.contract.functions.voters(address).call()return {"success": True,"address": address,"weight": voter_data[0],"voted": voter_data[1],"delegate": voter_data[2],"vote": voter_data[3]}except Exception as e:return {"success": False,"error": str(e)}def give_voting_right(self, chairperson_private_key: str, voter_address: str) -> Dict[str, Any]:"""授予投票权（仅主席可调用）Args:chairperson_private_key: 主席私钥voter_address: 投票者地址Returns:交易结果"""try:from eth_account import Accountchairperson = Account.from_key(chairperson_private_key)# 构建交易transaction = self.contract.functions.giveRightToVote(voter_address).build_transaction({'from': chairperson.address,'nonce': self.w3.eth.get_transaction_count(chairperson.address),'gasPrice': self.w3.eth.gas_price,'gas': 100000})# 签名并发送signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key=chairperson_private_key)tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)# 等待确认receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)return {"success": receipt.status == 1,"transaction_hash": tx_hash.hex(),"voter_address": voter_address}except Exception as e:return {"success": False,"error": str(e)}def cast_vote(self, voter_private_key: str, proposal_index: int) -> Dict[str, Any]:"""投票Args:voter_private_key: 投票者私钥proposal_index: 提案索引Returns:投票结果"""try:from eth_account import Accountvoter = Account.from_key(voter_private_key)# 构建交易transaction = self.contract.functions.vote(proposal_index).build_transaction({'from': voter.address,'nonce': self.w3.eth.get_transaction_count(voter.address),'gasPrice': self.w3.eth.gas_price,'gas': 200000})# 签名并发送signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key=voter_private_key)tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)# 等待确认receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)return {"success": receipt.status == 1,"transaction_hash": tx_hash.hex(),"voter": voter.address,"proposal_index": proposal_index}except Exception as e:return {"success": False,"error": str(e)}def delegate_vote(self, voter_private_key: str, delegate_address: str) -> Dict[str, Any]:"""委托投票Args:voter_private_key: 投票者私钥delegate_address: 委托地址Returns:委托结果"""try:from eth_account import Accountvoter = Account.from_key(voter_private_key)# 构建交易transaction = self.contract.functions.delegate(delegate_address).build_transaction({'from': voter.address,'nonce': self.w3.eth.get_transaction_count(voter.address),'gasPrice': self.w3.eth.gas_price,'gas': 200000})# 签名并发送signed_txn = self.w3.eth.account.sign_transaction(transaction, private_key=voter_private_key)tx_hash = self.w3.eth.send_raw_transaction(signed_txn.rawTransaction)# 等待确认receipt = self.w3.eth.wait_for_transaction_receipt(tx_hash)return {"success": receipt.status == 1,"transaction_hash": tx_hash.hex(),"from": voter.address,"to": delegate_address}except Exception as e:return {"success": False,"error": str(e)}async def demo_voting_dapp():"""演示投票DApp"""from config.blockchain_config import BlockchainConfig# 连接到本地测试网络w3 = BlockchainConfig.get_web3_connection("ganache")# 这里应该先部署合约，然后使用合约地址# 为了演示，我们假设合约已经部署contract_address = "0x..."  # 实际部署后的合约地址# 加载合约ABI（实际应从编译结果加载）with open("contracts/voting_abi.json", "r") as f:contract_abi = json.load(f)# 创建DApp实例dapp = VotingDApp(w3, contract_address, contract_abi)# 获取投票信息voting_info = dapp.get_voting_info()print("=== 投票系统信息 ===")print(json.dumps(voting_info, indent=2, default=str))# 事件处理回调async def handle_event(event_type: str, event_data: Dict):print(f"新事件: {event_type}")print(f"数据: {event_data}")# 开始监听事件（在实际应用中）# asyncio.create_task(dapp.watch_events(handle_event))if __name__ == "__main__":asyncio.run(demo_voting_dapp())

7. 完整代码实现与测试

7.1 完整的项目结构

python-blockchain/
├── config/
│   ├── __init__.py
│   └── blockchain_config.py
├── contracts/
│   ├── voting_contract.py
│   └── compiled/
├── src/
│   ├── __init__.py
│   ├── blockchain_basics.py
│   ├── transaction_manager.py
│   ├── contract_manager.py
│   └── voting_dapp.py
├── scripts/
│   ├── setup_environment.py
│   ├── deploy_contracts.py
│   └── run_demo.py
├── tests/
│   ├── test_basics.py
│   ├── test_transactions.py
│   └── test_contracts.py
├── requirements.txt
├── .env.example
└── README.md

7.2 综合演示脚本

# scripts/run_demo.py
#!/usr/bin/env python3
"""
区块链交互综合演示脚本
"""
import asyncio
import json
import os
from decimal import Decimal
from config.blockchain_config import BlockchainConfig
from src.blockchain_basics import BlockchainBasics
from src.transaction_manager import TransactionManager
from src.contract_manager import ContractManagerclass BlockchainDemo:"""区块链交互演示类"""def __init__(self, network: str = "goerli"):"""初始化演示环境Args:network: 区块链网络"""self.network = networkself.w3 = BlockchainConfig.get_web3_connection(network)self.basics = BlockchainBasics(self.w3)self.tx_manager = TransactionManager(self.w3)self.contract_manager = ContractManager(self.w3)def demo_network_info(self):"""演示网络信息查询"""print("=" * 60)print("1. 网络信息查询演示")print("=" * 60)network_info = self.basics.get_network_info()if network_info["connected"]:print("✅ 网络连接成功")print(f"   网络: {self.network}")print(f"   最新区块: {network_info['block_number']}")print(f"   Chain ID: {network_info['chain_id']}")print(f"   Gas价格: {network_info['gas_price_gwei']} Gwei")else:print("❌ 网络连接失败")print(f"   错误: {network_info['error']}")def demo_block_exploration(self):"""演示区块探索"""print("\n" + "=" * 60)print("2. 区块探索演示")print("=" * 60)# 获取最新区块latest_block = self.basics.get_block_info("latest")print(f"最新区块 #{latest_block['number']}:")print(f"  哈希: {latest_block['hash'].hex()}")print(f"  时间: {latest_block['timestamp']}")print(f"  交易数: {len(latest_block['transactions'])}")print(f"  矿工: {latest_block['miner']}")# 如果有交易，显示第一笔交易信息if latest_block['transactions']:first_tx_hash = latest_block['transactions'][0]if isinstance(first_tx_hash, str):tx_data = self.basics.get_transaction(first_tx_hash)print(f"  示例交易: {tx_data['hash'].hex()}")print(f"    从: {tx_data['from']}")print(f"    到: {tx_data['to']}")print(f"    价值: {self.w3.from_wei(tx_data['value'], 'ether')} ETH")def demo_gas_estimation(self):"""演示Gas估算"""print("\n" + "=" * 60)print("3. Gas费用估算演示")print("=" * 60)gas_info = self.tx_manager.get_gas_prices()if gas_info["success"]:print("当前Gas价格信息:")print(f"  基础费用: {gas_info['base_fee_gwei']} Gwei")print(f"  优先费用: {gas_info['max_priority_fee_gwei']} Gwei")print(f"  推荐最大费用: {self.w3.from_wei(gas_info['recommended_max_fee_wei'], 'gwei')} Gwei")# 估算简单转账的Gas成本sample_tx = self.tx_manager.create_transaction(from_address="0x" + "0" * 40,  # 示例地址to_address="0x" + "1" * 40,value=self.w3.to_wei(0.01, "ether"),max_fee_per_gas=gas_info["recommended_max_fee_wei"],max_priority_fee_per_gas=self.w3.to_wei(2, "gwei"))gas_estimate = self.tx_manager.estimate_gas_cost(sample_tx)if gas_estimate["success"]:print(f"简单转账估算:")print(f"  Gas用量: {gas_estimate['gas_estimate']}")print(f"  总成本: {gas_estimate['total_cost_eth']} ETH")print(f"  总成本: {gas_estimate['total_cost_wei']} Wei")def demo_contract_interaction(self):"""演示合约交互"""print("\n" + "=" * 60)print("4. 智能合约交互演示")print("=" * 60)# 编译简单存储合约from src.contract_manager import SIMPLE_STORAGE_CONTRACTcompile_result = self.contract_manager.compile_contract(SIMPLE_STORAGE_CONTRACT, "SimpleStorage")if compile_result["success"]:print("✅ 合约编译成功")print(f"  ABI长度: {len(compile_result['abi'])}")print(f"  字节码长度: {len(compile_result['bytecode'])}")# 注意：实际部署需要私钥和测试ETH# 这里只演示流程print("\n⚠️  部署合约需要测试ETH和私钥")print("  请在测试网络上获取测试ETH后取消注释部署代码")else:print("❌ 合约编译失败")print(f"  错误: {compile_result['error']}")def demo_unit_conversion(self):"""演示单位转换"""print("\n" + "=" * 60)print("5. 以太坊单位转换演示")print("=" * 60)units = [(1, "ether", "wei"),(1, "gwei", "ether"), (1, "ether", "gwei"),(0.5, "ether", "wei"),(1000000, "gwei", "ether")]print("单位转换示例:")for amount, from_unit, to_unit in units:result = self.basics.wei_converter(amount, from_unit, to_unit)print(f"  {amount} {from_unit} = {result} {to_unit}")async def run_all_demos(self):"""运行所有演示"""print("🚀 开始区块链交互演示")print(f"目标网络: {self.network}")print()self.demo_network_info()self.demo_block_exploration() self.demo_gas_estimation()self.demo_contract_interaction()self.demo_unit_conversion()print("\n" + "=" * 60)print("🎉 所有演示完成!")print("=" * 60)def main():"""主函数"""import argparseparser = argparse.ArgumentParser(description="区块链交互演示")parser.add_argument("--network", default="goerli", choices=["mainnet", "goerli", "sepolia", "ganache", "polygon"],help="区块链网络选择")args = parser.parse_args()# 创建演示实例demo = BlockchainDemo(args.network)# 运行演示asyncio.run(demo.run_all_demos())if __name__ == "__main__":main()

7.3 测试用例

# tests/test_basics.py
"""
基础功能测试
"""
import pytest
from web3 import Web3
from config.blockchain_config import BlockchainConfig
from src.blockchain_basics import BlockchainBasicsclass TestBlockchainBasics:"""区块链基础功能测试类"""@pytest.fixturedef w3(self):"""Web3实例fixture"""return BlockchainConfig.get_web3_connection("goerli")@pytest.fixturedef basics(self, w3):"""BlockchainBasics实例fixture"""return BlockchainBasics(w3)def test_network_connection(self, basics):"""测试网络连接"""network_info = basics.get_network_info()assert network_info["connected"] == Trueassert "block_number" in network_infoassert network_info["block_number"] > 0def test_block_info(self, basics):"""测试区块信息获取"""block_info = basics.get_block_info("latest")assert "number" in block_infoassert "hash" in block_infoassert "timestamp" in block_infodef test_wei_conversion(self, basics):"""测试Wei单位转换"""# 1 ETH = 10^18 Weione_eth_in_wei = basics.wei_converter(1, "ether", "wei")assert one_eth_in_wei == 10**18# 1 Gwei = 10^9 Weione_gwei_in_wei = basics.wei_converter(1, "gwei", "wei")assert one_gwei_in_wei == 10**9# 1 ETH = 10^9 Gweione_eth_in_gwei = basics.wei_converter(1, "ether", "gwei")assert one_eth_in_gwei == 10**9if __name__ == "__main__":pytest.main([__file__, "-v"])

8. 安全最佳实践

8.1 私钥安全管理

# src/security_manager.py
"""
安全管理器
"""
import os
import json
import keyring
from typing import Optional
from eth_account import Account
from cryptography.fernet import Fernet
from getpass import getpassclass SecurityManager:"""安全管理器"""def __init__(self, service_name: str = "python-blockchain-app"):"""初始化安全管理器Args:service_name: 密钥存储服务名称"""self.service_name = service_nameself._fernet = Nonedef _get_encryption_key(self) -> bytes:"""获取加密密钥"""# 从系统密钥环获取加密密钥key = keyring.get_password("system", f"{self.service_name}_encryption_key")if not key:# 生成新密钥key = Fernet.generate_key().decode()keyring.set_password("system", f"{self.service_name}_encryption_key", key)return key.encode()def _get_fernet(self) -> Fernet:"""获取Fernet实例"""if self._fernet is None:key = self._get_encryption_key()self._fernet = Fernet(key)return self._fernetdef encrypt_private_key(self, private_key: str, password: Optional[str] = None) -> str:"""加密私钥Args:private_key: 私钥password: 加密密码Returns:加密后的私钥"""fernet = self._get_fernet()# 使用密码进行额外加密（可选）if password:private_key = private_key + f"|{password}"encrypted_key = fernet.encrypt(private_key.encode())return encrypted_key.decode()def decrypt_private_key(self, encrypted_key: str, password: Optional[str] = None) -> str:"""解密私钥Args:encrypted_key: 加密的私钥password: 解密密码Returns:解密后的私钥"""fernet = self._get_fernet()decrypted_key = fernet.decrypt(encrypted_key.encode()).decode()if password and "|" in decrypted_key:key_part, stored_password = decrypted_key.split("|", 1)if stored_password == password:return key_partelse:raise ValueError("密码错误")return decrypted_keydef store_private_key(self, private_key: str, identifier: str, password: Optional[str] = None):"""安全存储私钥Args:private_key: 私钥identifier: 标识符password: 加密密码"""encrypted_key = self.encrypt_private_key(private_key, password)keyring.set_password(self.service_name, identifier, encrypted_key)def load_private_key(self, identifier: str, password: Optional[str] = None) -> str:"""加载私钥Args:identifier: 标识符password: 解密密码Returns:私钥"""encrypted_key = keyring.get_password(self.service_name, identifier)if not encrypted_key:raise ValueError(f"未找到标识符为 {identifier} 的私钥")return self.decrypt_private_key(encrypted_key, password)def generate_new_account(self, identifier: str, password: Optional[str] = None) -> str:"""生成新账户并安全存储Args:identifier: 标识符password: 加密密码Returns:账户地址"""account = Account.create()self.store_private_key(account.key.hex(), identifier, password)return account.addressdef list_stored_accounts(self) -> list:"""列出所有存储的账户"""# 注意：keyring.get_password在某些后端可能不支持列出所有账户# 这里需要根据具体后端实现return []# 使用示例
def security_demo():"""安全演示"""security = SecurityManager()# 生成新账户address = security.generate_new_account("my_wallet", "my_password")print(f"生成新账户: {address}")# 加载私钥try:private_key = security.load_private_key("my_wallet", "my_password")print(f"成功加载私钥: {private_key[:10]}...")# 验证账户account = Account.from_key(private_key)assert account.address == addressprint("✅ 私钥验证成功")except Exception as e:print(f"❌ 加载失败: {e}")if __name__ == "__main__":security_demo()

9. 性能优化与高级特性

9.1 批量操作与异步处理

# src/advanced_features.py
"""
高级特性与性能优化
"""
import asyncio
import aiohttp
from typing import List, Dict, Any
from concurrent.futures import ThreadPoolExecutor
from web3 import Web3
from web3.types import BlockDataclass AdvancedBlockchainFeatures:"""高级区块链特性"""def __init__(self, w3: Web3, max_workers: int = 10):"""初始化Args:w3: Web3实例max_workers: 最大工作线程数"""self.w3 = w3self.executor = ThreadPoolExecutor(max_workers=max_workers)async def get_multiple_blocks_async(self, block_numbers: List[int]) -> List[BlockData]:"""异步获取多个区块Args:block_numbers: 区块号列表Returns:区块数据列表"""loop = asyncio.get_event_loop()# 将同步调用转换为异步tasks = [loop.run_in_executor(self.executor, self.w3.eth.get_block, block_number)for block_number in block_numbers]return await asyncio.gather(*tasks)def get_blocks_in_range(self, start_block: int, end_block: int, batch_size: int = 10) -> List[BlockData]:"""获取区块范围内的所有区块（批量处理）Args:start_block: 起始区块end_block: 结束区块batch_size: 批次大小Returns:区块数据列表"""all_blocks = []for batch_start in range(start_block, end_block + 1, batch_size):batch_end = min(batch_start + batch_size, end_block + 1)batch_numbers = list(range(batch_start, batch_end))# 同步批量获取（实际应用中建议使用异步）batch_blocks = []for block_number in batch_numbers:try:block = self.w3.eth.get_block(block_number)batch_blocks.append(block)except Exception as e:print(f"获取区块 {block_number} 失败: {e}")all_blocks.extend(batch_blocks)print(f"已获取区块 {batch_start}-{batch_end-1}")return all_blocksdef analyze_gas_trends(self, block_count: int = 100) -> Dict[str, Any]:"""分析Gas价格趋势Args:block_count: 分析的区块数量Returns:Gas趋势分析结果"""latest_block = self.w3.eth.block_numberstart_block = max(0, latest_block - block_count)gas_prices = []base_fees = []for block_number in range(start_block, latest_block + 1):try:block = self.w3.eth.get_block(block_number)# 获取区块中的交易Gas价格for tx_hash in block['transactions']:tx = self.w3.eth.get_transaction(tx_hash)if 'gasPrice' in tx:gas_prices.append(tx['gasPrice'])elif 'maxFeePerGas' in tx:gas_prices.append(tx['maxFeePerGas'])# 获取基础费用（EIP-1559）if hasattr(block, 'baseFeePerGas') and block.baseFeePerGas:base_fees.append(block.baseFeePerGas)except Exception as e:continueif gas_prices:avg_gas_price = sum(gas_prices) / len(gas_prices)max_gas_price = max(gas_prices)min_gas_price = min(gas_prices)else:avg_gas_price = max_gas_price = min_gas_price = 0if base_fees:avg_base_fee = sum(base_fees) / len(base_fees)else:avg_base_fee = 0return {"blocks_analyzed": len(range(start_block, latest_block + 1)),"transactions_analyzed": len(gas_prices),"average_gas_price_gwei": self.w3.from_wei(avg_gas_price, "gwei"),"max_gas_price_gwei": self.w3.from_wei(max_gas_price, "gwei"),"min_gas_price_gwei": self.w3.from_wei(min_gas_price, "gwei"),"average_base_fee_gwei": self.w3.from_wei(avg_base_fee, "gwei") if avg_base_fee else 0}async def demo_advanced_features():"""演示高级特性"""from config.blockchain_config import BlockchainConfigw3 = BlockchainConfig.get_web3_connection("goerli")advanced = AdvancedBlockchainFeatures(w3)# 分析Gas趋势print("分析Gas价格趋势...")gas_analysis = advanced.analyze_gas_trends(50)print("Gas趋势分析结果:")for key, value in gas_analysis.items():print(f"  {key}: {value}")# 异步获取区块（示例）latest_block = w3.eth.block_numberblock_numbers = [latest_block - i for i in range(5)]print(f"\n异步获取最近5个区块...")blocks = await advanced.get_multiple_blocks_async(block_numbers)for i, block in enumerate(blocks):print(f"区块 #{block['number']}: {len(block['transactions'])} 笔交易")if __name__ == "__main__":asyncio.run(demo_advanced_features())