web3Interactive/project/monad/of_apr_io_daily.py

# -*- coding: utf-8 -*-
"""
Monad 测试网批量签到脚本（双合约版本）
1. signIn()  → 0x9E1DDC2a90bc0bec6325dF48Cbb3125Ce2879F7C  方法 0x9e4cda43
2. checkIn() → 0x703e753E9a2aCa1194DED65833EAec17dcFeAc1b  方法 0x183ff085
"""

import asyncio
import os
import random
import time
from typing import List

from web3 import Web3
from web3.types import TxReceipt

# ------------------------ 用户配置区 ------------------------
RPC_URL = "https://testnet-rpc.monad.xyz"

# 两个合约地址
SIGN_IN_CONTRACT  = "0x9E1DDC2a90bc0bec6325dF48Cbb3125Ce2879F7C"
CHECK_IN_CONTRACT = "0x703e753E9a2aCa1194DED65833EAec17dcFeAc1b"

# 私钥文件
KEY_FILE = "AccountList.txt"

# 余额保护
MIN_KEEP_BALANCE = 0.02

# 延时
MIN_DELAY = 15
MAX_DELAY = 30
# -----------------------------------------------------------

w3 = Web3(Web3.HTTPProvider(RPC_URL))
if not w3.is_connected():
    raise RuntimeError("无法连接 RPC")

# 方法选择器
SIGN_IN_SELECTOR  = "0x9e4cda43"
CHECK_IN_SELECTOR = "0x183ff085"


# ------------------------ 工具函数 ------------------------
def load_private_keys(path: str) -> List[str]:
    if not os.path.isfile(path):
        raise FileNotFoundError(f"找不到私钥文件: {path}")
    with open(path, "r", encoding="utf-8") as f:
        return [line.strip() for line in f if line.strip()]


def random_gas_limit(base: int = 100_000) -> int:
    return base + random.randint(-5_000, 10_000)


async def wait_tx(tx_hash: str) -> TxReceipt:
    print(f"⏳ 等待交易确认: {tx_hash}")
    receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
    if receipt.status == 1:
        print("✅ 成功")
    else:
        print("❌ 失败")
    return receipt


async def send_contract_call(private_key: str,
                             contract_addr: str,
                             selector: str,
                             desc: str):
    account = w3.eth.account.from_key(private_key)
    addr = account.address
    balance = w3.eth.get_balance(addr)
    if w3.from_wei(balance, "ether") < MIN_KEEP_BALANCE:
        print(f"⚠️  {addr} 余额不足，跳过")
        return

    nonce = w3.eth.get_transaction_count(addr)
    tx = {
        "to": Web3.to_checksum_address(contract_addr),
        "value": 0,
        "data": selector,
        "gas": random_gas_limit(),
        "gasPrice": w3.eth.gas_price,
        "nonce": nonce,
        "chainId": w3.eth.chain_id,
    }

    signed = account.sign_transaction(tx)
    tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction).hex()  # <- 修正处
    print(f"{desc} tx_hash: {tx_hash}")
    await wait_tx(tx_hash)


async def process_account(private_key: str):
    account = w3.eth.account.from_key(private_key)
    print(f"\n========== 开始处理 {account.address} ==========")

    try:
        # 1. signIn
        await send_contract_call(
            private_key,
            SIGN_IN_CONTRACT,
            SIGN_IN_SELECTOR,
            "1️⃣ signIn"
        )
        await asyncio.sleep(random.uniform(MIN_DELAY, MAX_DELAY))

        # 2. checkIn
        await send_contract_call(
            private_key,
            CHECK_IN_CONTRACT,
            CHECK_IN_SELECTOR,
            "2️⃣ checkIn"
        )
    except Exception as e:
        print(f"执行出错: {e}")


async def main():
    keys = load_private_keys(KEY_FILE)
    if not keys:
        print("私钥文件为空")
        return

    print(f"共读取到 {len(keys)} 个钱包")
    for pk in keys:
        await process_account(pk)
        await asyncio.sleep(random.uniform(MIN_DELAY, MAX_DELAY))


if __name__ == "__main__":
    asyncio.run(main())