- pip install web3.py
/ I$ p& u* ?: o# f6 J4 K" D
2、注册 Infura 获得节点服务
使用邮箱注册 Infura 账户后，创建一个项目，即可获得以太坊节点服务，进入设置即可看到链接的URL8 R( G: j3 c: Y7 Q/ S
可以选择主网测试网等，会有两个链接，一个是使用HTTPS的一个是使用WebSocket，按你的需求选择一个就行了，注意：Infura 个人免费请求次数，是每天有十万个请求。5 Q4 k( c8 j7 n/ l7 J
3、代码示例
- from web3 import Web38 K: B9 k% j/ j ]2 F+ Z" i3 r) V
- import json
- import time
- import os! `4 t" `& {1 W, J" C [9 V5 s3 R( J4 ~
- import logging
- from django.conf import settings9 M0 [" u) e" f- R) N, k* O
- from decimal import Decimal
- class PayEthOrToken(object):
- # v! a* ]0 h6 { G1 j
- def __init__(self):
- # 设置web3' n' p+ N' g/ \
- self.web3 = Web3(Web3.HTTPProvider('your infura http url'))1 H# U4 R% Y2 n O
- # token合约地址+ A6 }! S! ?3 Y+ [6 t, h. P% X
- self.contract_address = 'your contract address'
- # 主钱包地址" ~; X1 @" e' ~) |7 Y( b+ B
- self.wallet = 'your wallet address'- B- t% n5 T( Z9 v- Q% N: a
- # 钱包的私钥, a9 a- K# J/ ~
- self.wallet_key = 'your wallet key'0 ]. Y- q* H" f' i) D: T" Q
- # 合约的abi test.json 是eth的abi json文件，可以在eth区块链浏览器上获得
- with open('test.json', 'r') as f:! }1 \% l o P# O5 z7 H' @5 d
- self.abi = json.loads(f.read())
- # 生成合约1 X+ B2 X6 ?9 D4 t+ \- x4 I
- self.contract = self.web3.eth.contract(address=self.contract_address, abi=self.abi). Z, `8 ]/ x; {' ^: r$ Q
- # 代币简写; M/ B6 v9 g" A+ ?; i; I: w. @% |3 V
- self.token_name = 'USDT'
-
- def transfer_usdt(self, to, value):4 r2 E) x* h$ W4 }
- '''进行代币转账' x7 Q1 ~: T* ]$ E w. a
- args：" F; J: Y6 e" I% U4 o
- to str：接收代币的地址 Z. b6 V/ F( ^
- value str/int：代币数量，以ether为单位，可以是字符串和int类型
- returns：: S6 E* ^& u" h( M1 b2 U
- (str, str)：返回交易哈希，以及异常信息
- '''! |" v1 j2 ?" g& ?! k4 J
- try:6 \1 ] r; f/ n& B# f
- token_balance = self.web3.fromWei(self.contract.functions.balanceOf(self.wallet).call(), 'ether')
- # 如果代币不足返回异常. }8 F e& ~' E) g
- if Decimal(token_balance) < Decimal(value):
- return None, 'Platform USDT token is insufficient, please try again later'
- # 进行转账代币
- nonce = self.web3.eth.get_transaction_count(self.wallet)
- tx = {
- 'from': self.wallet,
- 'nonce': nonce,# j' d; c' g1 s4 g d9 ]
- 'gas': 100000,
- 'gasPrice': self.web3.toWei('50', 'gwei'),
- 'chainId': 1
- }
- to = Web3.toChecksumAddress(to)
- txn = self.contract.functions.transfer(to, self.web3.toWei(value, 'ether')).buildTransaction(tx)
- signed_txn = self.web3.eth.account.sign_transaction(txn, private_key=self.wallet_key)* P/ w1 ?4 n+ a5 o2 y! w4 ~
- tx_hash = self.web3.eth.send_raw_transaction(signed_txn.rawTransaction)
- return self.web3.toHex(tx_hash), 'pay success'6 O( n5 G5 U& E4 N5 @& n
- except Exception as e:( \: `/ _7 N# D- A
- logging.error(f'转账{self.token_name}代币时发生异常：{e}')
- logging.exception(e): c2 y/ l/ z6 V2 m0 h- |1 o v* y; o
- return None, str(e)( W f o* d P( \7 w9 c3 U
- 9 K- B3 R. t W
- def transfer_eth(self, to, value):6 Q5 L" X& w' x
- '''进行eth转账
- args：) J$ k' S& ?- X
- to str：接收以太坊的地址
- value str/int：数量，以ether为单位，可以是字符串和int类型
- returns：
- str：返回交易哈希
- '''" c# |. T( O0 _! U/ K
- try:
- token_balance = self.web3.fromWei(self.web3.eth.get_balance(self.wallet), 'ether')7 |' @& D! U- S% L4 g6 _
- # 如果代币不足返回异常
- if Decimal(token_balance) < Decimal(value):
- return None, 'Platform ETH token is insufficient, please try again later'7 p. a5 Z1 {% e a3 W3 f5 a' k- z% {
- # 获取 nonce，这个是交易计数' ?3 h7 N3 n, ~$ c& ]. I
- to = Web3.toChecksumAddress(to)
- nonce = self.web3.eth.get_transaction_count(self.wallet)
- tx = {8 F& _! h4 p0 T& o
- 'nonce': nonce,4 y- Z' U* L* T3 X
- 'to': to,
- 'gas': 100000,& J% B' f2 x c, J1 W; H
- 'gasPrice': self.web3.toWei('50', 'gwei'),
- 'value': self.web3.toWei(value, 'ether'),) G$ f/ s i" q) K. G
- 'chainId': 1, g+ j& _! o6 B; \3 Q
- }1 q: p6 X6 K% G+ }; k' I. e
- # 签名交易5 ~( i% w; {0 K! ~+ m) l
- signed_tx = self.web3.eth.account.sign_transaction(tx, self.wallet_key)
- tx_hash = self.web3.eth.send_raw_transaction(signed_tx.rawTransaction)
- return self.web3.toHex(tx_hash), 'pay success'
- except Exception as e:: @# Q. e+ G+ [8 H
- logging.error(f'转账eth时发生异常：{e}'), B$ \7 Q1 a" g; ]; U
- logging.exception(e)
- return None, str(e)
