web3Interactive/project/monad/signin-lumiterra.py

# -*- coding: utf-8 -*-
"""
Monad 测试网 talentum 批量签到脚本
网址: https://monad.talentum.id
合约: 0x9E1DDC2a90bc0bec6325dF48Cbb3125Ce2879F7C  方法: checkIn() → 0x9e4cda43
"""

import asyncio
import os
import random
from typing import List
import traceback

from web3 import Web3
from web3.types import TxReceipt
from eth_account import Account

# ------------------------ 用户配置区 ------------------------
RPC_URL = "https://testnet-rpc.monad.xyz"
CHECK_IN_CONTRACT = "0x9E1DDC2a90bc0bec6325dF48Cbb3125Ce2879F7C"
KEY_FILE = "AccountList.txt"
MIN_KEEP_BALANCE = 0.02
MIN_DELAY = 15
MAX_DELAY = 30

CHECK_IN_ABI = [{
    "inputs": [],
    "name": "checkIn",
    "outputs": [],
    "stateMutability": "nonpayable",
    "type": "function"
}]
# -----------------------------------------------------------

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

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:
    limit = base + random.randint(-5_000, 10_000)
    return max(21_000, limit)  # 保证不会小于最低gas

async def wait_tx(tx_hash: str, tx_params: dict):
    print(f"⏳ 等待交易确认: {tx_hash}")
    try:
        receipt = w3.eth.wait_for_transaction_receipt(tx_hash, timeout=120)
        if receipt.status == 1:
            print("✅ 成功")
        else:
            print("❌ 失败（链上执行返回 status=0），尝试获取失败原因...")
            # 用 eth_call 复现错误，获取revert reason
            try:
                w3.eth.call(tx_params, block_identifier="pending")
                print("但 eth_call 没报错，说明没有 revert reason（或已被修复）")
            except Exception as call_err:
                # eth_call 抛出的异常中通常包含 revert reason
                err_msg = str(call_err)
                print(f"失败原因（revert reason）: {err_msg}")
            print("-" * 40)
        return receipt
    except Exception as e:
        print(f"❌ 等待交易时出错: {e}")
        import traceback
        traceback.print_exc()
        return None

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

    nonce = w3.eth.get_transaction_count(addr)
    contract = w3.eth.contract(address=Web3.to_checksum_address(contract_addr), abi=contract_abi)
    try:
        data = contract.encodeABI(fn_name='checkIn')
        tx = {
            "to": Web3.to_checksum_address(contract_addr),
            "value": 0,
            "data": data,
            "gas": random_gas_limit(),
            "gasPrice": w3.eth.gas_price,
            "nonce": nonce,
            "chainId": w3.eth.chain_id,
        }
        signed = Account.sign_transaction(tx, private_key)
        try:
            tx_hash = w3.eth.send_raw_transaction(signed.rawTransaction).hex()
            print(f"{desc} {addr} 提交签到交易: {tx_hash}")
            # 传入 tx_params 方便 wait_tx 用 eth_call 复现
            await wait_tx(tx_hash, {
                "from": addr,
                "to": tx["to"],
                "data": tx["data"],
                "value": tx["value"],
                "gas": tx["gas"]
            })
        except Exception as send_e:
            print(f"❌ {addr} 广播交易时遇到异常：{send_e}")
            import traceback
            traceback.print_exc()
    except Exception as e:
        print(f"❌ {addr} 构造或签名交易时出错: {e}")
        import traceback
        traceback.print_exc()

async def process_account(private_key: str):
    acct = Account.from_key(private_key)
    print(f"\n========== 开始处理 {acct.address} ==========")
    await send_contract_call(
        private_key,
        CHECK_IN_CONTRACT,
        CHECK_IN_ABI,
        "SignIn"
    )

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

    print(f"共读取到 {len(keys)} 个钱包")
    for idx, pk in enumerate(keys):
        await process_account(pk)
        if idx != len(keys) - 1:
            wait_sec = random.uniform(MIN_DELAY, MAX_DELAY)
            print(f"⏳ 随机等待 {wait_sec:.2f} 秒，防批量风控")
            await asyncio.sleep(wait_sec)

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