forked from ymmmmmmmm/BeraChainTools
-
Notifications
You must be signed in to change notification settings - Fork 1
/
honey_swap.py
130 lines (117 loc) · 5.58 KB
/
honey_swap.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
# -*- coding: utf-8 -*-
# Time :2024/1/21 11:36
# Author :ym
# File :honey_swap.py
import concurrent.futures
import os
import random
import time
from typing import Union
from dotenv import load_dotenv
from eth_account import Account
from eth_typing import Address, ChecksumAddress, HexStr
from loguru import logger
from web3 import Web3
from config.address_config import honey_swap_address, usdc_address, honey_address
from config.contract_config import usdc_contract, honey_swap_contract, honey_contract
load_dotenv()
max_workers = int(os.getenv("MaxWorkers"))
rpc_url = os.getenv("RPC_URL")
w3 = Web3(Web3.HTTPProvider(rpc_url))
@logger.catch
def honey_mint(address: Union[Address, ChecksumAddress], private_key: Union[bytes, HexStr, int]) -> str:
usdc_balance = usdc_contract.functions.balanceOf(address).call()
assert usdc_balance != 0
# 支付 usdc 占比
value = int(usdc_balance * 0.8)
allowance_balance = usdc_contract.functions.allowance(address, honey_swap_address).call()
nonce = w3.eth.get_transaction_count(address)
if allowance_balance < value:
# 需要授权
txn = w3.eth.account.sign_transaction(dict(
nonce=nonce,
chainId=80085,
gasPrice=int(w3.eth.gas_price * 1.15),
gas=50000 + random.randint(1, 10000),
to=usdc_address,
data='0x095ea7b300000000000000000000000009ec711b81cd27a6466ec40960f2f8d85bb129d97fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff',
), private_key)
order_hash = w3.eth.send_raw_transaction(txn.rawTransaction)
logger.debug(f'{address}:{order_hash.hex()}')
order_result = w3.eth.wait_for_transaction_receipt(order_hash, timeout=120)
if order_result.status == 1:
logger.success(f'{address}:{order_hash.hex()}')
else:
logger.critical(f'{address}:{order_hash.hex()}')
raise ValueError(f'{address}:{order_hash.hex()}')
nonce += 1
txn = honey_swap_contract.functions.mint(to=address, collateral=usdc_address, amount=value, ).build_transaction(
{'gas': 300000 + random.randint(1, 10000), 'gasPrice': int(w3.eth.gas_price * 1.15), 'nonce': nonce})
signed_txn = w3.eth.account.sign_transaction(txn, private_key=private_key)
order_hash = w3.eth.send_raw_transaction(signed_txn.rawTransaction)
logger.debug(f'{address}:{order_hash.hex()}')
order_result = w3.eth.wait_for_transaction_receipt(order_hash, timeout=120)
if order_result.status == 1:
logger.success(f'{address}:{order_hash.hex()}')
else:
logger.critical(f'{address}:{order_hash.hex()}')
raise ValueError(f'{address}:{order_hash.hex()}')
return order_hash.hex()
@logger.catch
def honey_redeem(address: Union[Address, ChecksumAddress], private_key: Union[bytes, HexStr, int]) -> str:
honey_balance = honey_contract.functions.balanceOf(address).call()
assert honey_balance != 0
# 支付 honey 占比
value = int(honey_balance * 0.8)
allowance_balance = honey_contract.functions.allowance(address, honey_swap_address).call()
logger.debug(allowance_balance)
nonce = w3.eth.get_transaction_count(address)
if allowance_balance < value:
# 需要授权
txn = w3.eth.account.sign_transaction(dict(
nonce=nonce,
chainId=80085,
gasPrice=int(w3.eth.gas_price * 1.15),
gas=50000 + random.randint(1, 10000),
to=honey_address,
data='0x095ea7b300000000000000000000000009ec711b81cd27a6466ec40960f2f8d85bb129d97fffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff',
), private_key)
order_hash = w3.eth.send_raw_transaction(txn.rawTransaction)
logger.debug(f'{address}:{order_hash.hex()}')
order_result = w3.eth.wait_for_transaction_receipt(order_hash, timeout=120)
if order_result.status == 1:
logger.success(f'{address}:{order_hash.hex()}')
else:
logger.critical(f'{address}:{order_hash.hex()}')
raise ValueError(f'{address}:{order_hash.hex()}')
nonce += 1
txn = honey_swap_contract.functions.redeem(to=address, amount=value, collateral=usdc_address).build_transaction(
{'gas': 300000 + random.randint(1, 10000), 'gasPrice': int(w3.eth.gas_price * 1.15), 'nonce': nonce})
signed_txn = w3.eth.account.sign_transaction(txn, private_key=private_key)
order_hash = w3.eth.send_raw_transaction(signed_txn.rawTransaction)
logger.debug(f'{address}:{order_hash.hex()}')
order_result = w3.eth.wait_for_transaction_receipt(order_hash, timeout=120)
if order_result.status == 1:
logger.success(f'{address}:{order_hash.hex()}')
else:
logger.critical(f'{address}:{order_hash.hex()}')
raise ValueError(f'{address}:{order_hash.hex()}')
return order_hash.hex()
def honey_run(key):
account = Account.from_key(key)
honey_mint(account.address, account.key)
time.sleep(random.randint(5, 20))
honey_redeem(account.address, account.key)
def ym_test_run():
account = Account.create()
# mint 操作 usdc换honey
honey_mint(account.address, account.key)
# redeem 操作 honey换usdc
honey_redeem(account.address, account.key)
if __name__ == '__main__':
# 读取当前文件夹下面的bera_claim_success(领取成功文本)
with open('./bera_claim_success.txt', 'r') as f:
wallet_list = f.readlines()
random.shuffle(wallet_list)
with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = [executor.submit(honey_run, i.split('----')[1].replace('\n', '')) for i in wallet_list]