web3/web3Interactive
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

Browse Source
This commit is contained in:
jack
2025-11-12 16:47:21 +08:00
commit 266ba871a4
43 changed files with 3747 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/load_key_to_wallet.py
+6
View File
@@ -0,0 +1,6 @@
# -*- coding: utf-8 -*-
from web3 import Web3
privateKeyList = [
'0x3991542110242368f4770716be904b0ca6d44a8dbe4501771833b1a3642198d1'
]
test/test_contract.py
+88
View File
@@ -0,0 +1,88 @@
from web3 import Web3
import json
# Monad 测试网 RPC URL
rpc_url = "https://testnet-rpc.monad.xyz"
# 连接到 Monad 测试网
w3 = Web3(Web3.HTTPProvider(rpc_url))
# 检查是否成功连接
if not w3.is_connected():
raise Exception("无法连接到 Monad 测试网")
# 合约地址和 ABI（仅保留 transfer 和 decimals）
contract_address = w3.to_checksum_address("0x760AfE86e5de5fa0Ee542fc7B7B713e1c5425701")
abi = json.loads('''[
{"inputs":[{"internalType":"address","name":"dst","type":"address"},{"internalType":"uint256","name":"wad","type":"uint256"}],"name":"transfer","outputs":[{"internalType":"bool","name":"","type":"bool"}],"stateMutability":"nonpayable","type":"function"},
{"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"}
]''')
# 初始化合约
contract = w3.eth.contract(address=contract_address, abi=abi)
# 你的钱包私钥和地址
private_key = "b898cf63a5ec89105ba755ef3b7533c25ea9130ab50fb0db14779fb6efd4f9c6" # 替换为你的私钥
account = w3.eth.account.from_key(private_key)
wallet_address = account.address
# 辅助函数：签名并发送交易
def send_transaction(tx):
"""
签名并发送交易
参数：
tx: 交易字典
返回：
交易回执
"""
try:
tx['from'] = wallet_address
tx['nonce'] = w3.eth.get_transaction_count(wallet_address)
tx['gas'] = 100000 # Gas 限额，适用于简单的 transfer
tx['gasPrice'] = w3.eth.gas_price
signed_tx = w3.eth.account.sign_transaction(tx, private_key)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
return w3.eth.wait_for_transaction_receipt(tx_hash)
except Exception as e:
raise Exception(f"发送交易失败: {str(e)}")
# 转账代币
def transfer(recipient, amount):
"""
转账代币给指定地址
参数：
recipient: 接收者地址
amount: 转账的代币数量（以代币单位为准，例如 10 表示 10 个代币）
"""
try:
# 验证接收者地址
if not w3.is_address(recipient):
raise ValueError("接收者地址无效")
recipient = w3.to_checksum_address(recipient)
# 获取小数位并转换金额
decimals = contract.functions.decimals().call()
if not isinstance(amount, (int, float)) or amount < 0:
raise ValueError("转账金额必须是非负数")
amount_wei = int(amount * 10**decimals) # 转换为最小单位
# 构建交易
tx = contract.functions.transfer(recipient, amount_wei).build_transaction({
'chainId': w3.eth.chain_id,
'gas': 100000, # 显式指定 Gas 限额
'gasPrice': w3.eth.gas_price,
'nonce': w3.eth.get_transaction_count(wallet_address),
})
# 发送交易
receipt = send_transaction(tx)
print(f"转账成功，交易哈希: {receipt.transactionHash.hex()}")
except Exception as e:
print(f"转账失败: {str(e)}")
# 示例用法
if __name__ == "__main__":
# 转账 10 个代币给指定地址
recipient_address = "0x760AfE86e5de5fa0Ee542fc7B7B713e1c5425701" # 0x904d6CEf48D78448E332B90f66e23a5aAedC1A47
transfer(recipient_address, 0.000000005231854) # 修改此处的 10 为你想要转账的代币数量
test/test_contract1.py
+144
View File
@@ -0,0 +1,144 @@
# -*- coding: utf-8 -*-
from web3 import Web3
import random
import time
import os
# Monad 测试网 RPC URL
provider_url = "https://testnet-rpc.monad.xyz"
w3 = Web3(Web3.HTTPProvider(provider_url))
# 合约地址、合约方法哈希值及描述
contract_data = [
{
"address": "0x64Aff7245EbdAAECAf266852139c67E4D8DBa4de",
"data_hash": "0x7ff36ab5",
"description": "TEST"
}
]
# 读取私钥文件
file_path = os.path.join(os.path.dirname(__file__), 'AccountList.txt')
def get_random_amount():
# 生成随机交易金额 例如 0.00000000****（最后4位随机数）
# 小数位数也是随机, 14 - 18 位
random_tail = str(random.randint(1000, 9999))
random_bit = random.randint(14, 18)
random_balance = f"0.{random_tail.zfill(random_bit)}"
return Web3.to_wei(random_balance, "ether")
def get_random_contract_data():
# 随机获取合约地址、合约方法哈希值及描述
if not contract_data:
print("The contract data array is empty, the contract cannot be selected, and the program exits.")
exit(1)
return random.choice(contract_data)
def check_wallet_balance(balance, minimum_balance):
# 如果钱包余额小于等于最小余额
if balance <= Web3.to_wei(minimum_balance, "ether"):
print("The wallet balance is invalid.")
exit(1)
async def execute_contract(account, minimum_balance):
try:
# 随机选择一个合约地址、合约方法哈希值及描述
contract_info = get_random_contract_data()
address = contract_info["address"]
data_hash = contract_info["data_hash"]
description = contract_info["description"]
print(f"Selected contract: {description}")
print(f"Selected contract address: {address}")
print(f"Selected contract dataHash: {data_hash}")
# 检查原生代币余额
balance = w3.eth.get_balance(account.address)
before_balance = Web3.from_wei(balance, "ether")
print(f"Native balance: {before_balance}")
# 检查钱包余额，不能小于 minimum_balance
check_wallet_balance(balance, minimum_balance)
amount_to_stake = get_random_amount()
print(f"Amount to contract: {Web3.from_wei(amount_to_stake, 'ether')}")
if balance < amount_to_stake:
raise Exception("Insufficient balance for contract execution.")
# 构建交易
nonce = w3.eth.get_transaction_count(account.address)
tx = {
"to": Web3.to_checksum_address(address),
"value": amount_to_stake,
"data": data_hash,
"gas": 100000,
"gasPrice": w3.eth.gas_price,
"nonce": nonce,
"chainId": w3.eth.chain_id
}
# 签名并发送交易
signed_tx = w3.eth.account.sign_transaction(tx, account.key)
tx_hash = w3.eth.send_raw_transaction(signed_tx.raw_transaction)
print(f"Contract Execution hash: {tx_hash.hex()}")
# 等待交易确认
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
execute_status = receipt["status"]
if execute_status == 1:
print("Contract Execution status: seccessful")
elif execute_status == 0:
print("Contract Execution status: failed")
else:
print("Contract Execution status: unknown")
# 检查交易后的余额
balance_staking = w3.eth.get_balance(account.address)
after_balance = Web3.from_wei(balance_staking, "ether")
print(f"Native balance after transaction: {after_balance}")
print(f"The cost of this transaction: {float(before_balance) - float(after_balance)}")
except Exception as e:
print(f"Error executing contract: {e}")
async def run_multiple_staking(private_keys):
last_times = 9999
min_delay = 10
max_delay = 20
minimum_balance = 0.01
for private_key in private_keys:
account = w3.eth.account.from_key(private_key.strip())
print(f"Processing wallet with address: {account.address}")
for i in range(last_times):
if last_times > 1:
print(f"Starting Contract attempt {i + 1}")
await execute_contract(account, minimum_balance)
print(f"Completed Contract attempt {i + 1}")
if i < last_times - 1:
delay = random.uniform(min_delay, max_delay)
print(f"Waiting for {delay} seconds...")
print('------------------------------------------------------------')
time.sleep(random.uniform(min_delay, max_delay))
async def main():
try:
await run_multiple_staking(['3991542110242368f4770716be904b0ca6d44a8dbe4501771833b1a3642198d1'])
except Exception as err:
print(f'读取文件时发生错误: {err}')
exit(1)
if __name__ == "__main__":
import asyncio
asyncio.run(main())