manual/web3/faucet.py

# -*- coding: utf-8 -*-
import re

import httpx
import asyncio
import json
import random
import logging
from typing import List
from fake_useragent import UserAgent  # 导入 fake_useragent 模块

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# 初始化 fake_useragent
ua = UserAgent()


async def send_post_request(wallet: str, timeout: float = 10.0) -> None:
    """
    向目标网站发送 POST 请求以领取水。

    Args:
        wallet: 钱包地址
        timeout: 请求超时时间（秒）
    """
    url = "https://faucet.nerzo.xyz/monad"

    headers = {
        "accept": "text/x-component",
        "accept-encoding": "gzip, deflate, br, zstd",
        "accept-language": "zh-CN,zh;q=0.9",
        "content-type": "text/plain;charset=UTF-8",
        "next-action": "405d6e23437f844564e65cdd65851724c449d7f778",
        "next-router-state-tree": '["",{"children":["(faucet)",{"children":["monad",{"children":["__PAGE__",{},"/monad","refresh"]}]}]},null,null,true]',
        "origin": "https://faucet.nerzo.xyz",
        "priority": "u=1, i",
        "referer": "https://faucet.nerzo.xyz/monad",
        "sec-ch-ua": '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "sec-fetch-dest": "empty",
        "sec-fetch-mode": "cors",
        "sec-fetch-site": "same-origin",
        "user-agent": ua.random,
        "x-deployment-id": "dpl_DWgqqeB8V4ASD1aCKUqWZwcPwnVd"
    }
    payload = json.dumps([wallet])

    async with httpx.AsyncClient(timeout=timeout, proxy={'http://': 'http://127.0.0.1:7890'}) as client:
        retry = 3
        while retry > 0:  # 修复了 retry 的逻辑错误
            try:
                response = await client.post(url, headers=headers, content=payload)
                # logger.info(f"Wallet: {wallet} | Status Code: {response.status_code} | Response: {response.text}")
                logger.info(f"Wallet: {wallet} | Status Code: {response.status_code}")
                if 'message' in response.text:
                    message = re.findall(r'"message":"(.*?)"', response.text)[0]
                    logger.info(f"Wallet: {wallet} | Message: {message}")
                return response
            except httpx.RequestError as e:
                logger.error(f"Wallet: {wallet} | Request failed: {e}")
                retry -= 1

                if retry <= 0:
                    return None


async def main(wallet_list: List[str], min_delay: float = 5.0, max_delay: float = 8.0) -> None:
    """
    批量处理钱包地址，发送 POST 请求，并添加随机延时。

    Args:
        wallet_list: 钱包地址列表
        min_delay: 最小延时（秒）
        max_delay: 最大延时（秒）
    """
    for i, wallet in enumerate(wallet_list, 1):
        logger.info(f"Processing wallet {i}/{len(wallet_list)}: {wallet}")
        await send_post_request(wallet)
        # 在最后一个钱包之前添加随机延时
        if i < len(wallet_list):
            delay = random.uniform(min_delay, max_delay)
            logger.info(f"Waiting for {delay:.2f} seconds before next request")
            await asyncio.sleep(delay)


if __name__ == "__main__":
    wallet_list = [
        '0xe50B77Cd771243b8Ae1d6ce33b4E13ECC5Fa28a6',
        '0x9ea2ECAD4090E32916e03b77d7C75CbF6C8E0A55',
        '0xE8A4b0C04300154DC9B1D0e565Ba70F996614690',
        '0x1b623c5d70c93b437d93c305bf2cfa389095f636',
        '0x06D25c3e0E1F753ac0486a3f8aaD7259149656cB',
        '0x15cFEE34Ca4541CAc9a1c4B6F6aB47A65877E240',
        '0x7aBF0dA8Ac07B6dE7206e467988455E1AD0b60B5',
        '0xF736f45d4663a8D8DfF7EFA55b1Cf6Fe38D026c8',
        '0x83173eECf3a6d9ABB79682568e16c2eAd361620e',
        '0xa401b85B4849Fc7610Bd180cc937859C78528F47',
        '0x10A43E7Fe77E2D84adBeC26cF0bFc6f403841266',
        '0x70D5EE1DfddD3726f0D71F4CD5a8Ef43aC651a75'
    ]
    random.shuffle(wallet_list)
    asyncio.run(main(wallet_list))
    print("done")