Web3Tools/monad/faucet_official.py

# -*- coding: utf-8 -*-
import asyncio
from loguru import logger
import random
import hashlib
from eth_account import Account
from pynocaptcha import CloudFlareCracker, TlsV1Cracker
from curl_cffi.requests import AsyncSession

SETTINGS = {
    "ATTEMPTS": 3,
    "RANDOM_PAUSE_BETWEEN_ACTIONS": [1, 5],
}

FAUCET = {
    "USE_CAPSOLVER_FOR_CLOUDFLARE": False,
    "CAPSOLVER_API_KEY": "your_capsolver_api_key",
    "NOCAPTCHA_API_KEY": "your_nocaptcha_api_key",
}


async def faucet(
        session: AsyncSession,
        account_index: int,
        wallet: Account,
        proxy: str,
) -> bool:
    for retry in range(SETTINGS["ATTEMPTS"]):
        try:
            logger.info(f"[{account_index}] | Starting faucet for account {wallet.address}...")
            user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36"
            href = "https://testnet.monad.xyz/"

            # Solve Cloudflare challenge
            if FAUCET["USE_CAPSOLVER_FOR_CLOUDFLARE"]:
                logger.info(f"[{account_index}] | Solving Cloudflare challenge with Capsolver...")
                # 这里假设Capsolver类已经定义并且可用
                capsolver = Capsolver(api_key=FAUCET["CAPSOLVER_API_KEY"], proxy=proxy, session=session)
                cf_result = await capsolver.solve_turnstile("0x4AAAAAAA-3X4Nd7hf3mNGx", href)
            else:
                logger.info(f"[{account_index}] | Solving Cloudflare challenge with Nocaptcha...")
                # 这里假设CloudFlareCracker类已经定义并且可用
                cracker = CloudFlareCracker(
                    internal_host=True,
                    user_token=FAUCET["NOCAPTCHA_API_KEY"],
                    href=href,
                    sitekey="0x4AAAAAAA-3X4Nd7hf3mNGx",
                    proxy=proxy,
                    debug=False,
                    show_ad=False,
                    timeout=60,
                )
                cf_result = cracker.crack()
                cf_result = cf_result["token"]

            if not cf_result:
                raise Exception("Failed to solve Cloudflare challenge")

            logger.success(f"[{account_index}] | Cloudflare challenge solved")

            # Generate visitor ID
            visitor_id = hashlib.md5(str(random.random()).encode()).hexdigest()

            json_data = {
                "address": wallet.address,
                "visitorId": visitor_id,
                "cloudFlareResponseToken": cf_result,
            }

            headers = {
                "sec-ch-ua-platform": '"Windows"',
                "user-agent": user_agent,
                "sec-ch-ua": '"Not(A:Brand";v="99", "Google Chrome";v="133", "Chromium";v="133"',
                "content-type": "application/json",
                "sec-ch-ua-mobile": "?0",
                "accept": "*/*",
                "origin": href,
                "sec-fetch-site": "same-origin",
                "sec-fetch-mode": "cors",
                "sec-fetch-dest": "empty",
                "referer": href,
                "accept-language": "en-GB,en;q=0.9",
                "priority": "u=1, i",
            }

            response = await session.post(
                f"{href}api/faucet/claim", headers=headers, json=json_data
            )
            response_text = response.text

            if not response_text:
                raise Exception("Failed to send claim request")

            if '"Success"' in response_text or '"message":"Success"' in response_text:
                logger.success(f"[{account_index}] | Successfully got tokens from faucet")
                return True

            if "Claimed already" in response_text:
                logger.success(f"[{account_index}] | Already claimed tokens from faucet")
                return True

            error_messages = {
                "FUNCTION_INVOCATION_TIMEOUT": "Server is not responding, wait...",
                "Vercel Security Checkpoint": "Failed to solve Vercel challenge, trying again...",
                "Server error on QuickNode API": "FAUCET DOES NOT WORK, QUICKNODE IS DOWN",
                "Over Enterprise free quota": "MONAD IS SHIT, FAUCET DOES NOT WORK, TRY LATER",
                "invalid-keys": "PLEASE UPDATE THE BOT USING GITHUB",
            }

            for key, message in error_messages.items():
                if key in response_text:
                    logger.error(f"[{account_index}] | {message}")
                    if key == "Over Enterprise free quota" or key == "invalid-keys":
                        return False
                    break
            else:
                logger.error(f"[{account_index}] | Failed to get tokens from faucet: {response_text}")

            await asyncio.sleep(3)

        except Exception as e:
            random_pause = random.randint(*SETTINGS["RANDOM_PAUSE_BETWEEN_ACTIONS"])
            if "operation timed out" in str(e):
                logger.error(
                    f"[{account_index}] | Error faucet to monad.xyz ({retry + 1}/{SETTINGS['ATTEMPTS']}): Connection timed out. Next faucet in {random_pause} seconds")
            else:
                logger.error(
                    f"[{account_index}] | Error faucet to monad.xyz ({retry + 1}/{SETTINGS['ATTEMPTS']}): {e}. Next faucet in {random_pause} seconds")
            await asyncio.sleep(random_pause)
            continue

    return False