scripts/hcaptcha_bot.py

"""hCaptcha Playwright automation wrapper.

Runs on your local machine. Detects hCaptcha on faucet pages,
screenshots tiles, calls the vision model API to classify them,
and clicks matching tiles with human-like delays.

Usage:
    python scripts/hcaptcha_bot.py https://faucet-wave.online/tron-faucet/

Or import and use in your existing claimer scripts:
    from scripts.hcaptcha_bot import solve_hcaptcha
    await solve_hcaptcha(page, "https://faucet-wave.online/tron-faucet/")
"""

from __future__ import annotations

import asyncio
import base64
import io
import os
import random
import sys
import time
from typing import Optional

import httpx

try:
    from playwright.async_api import async_playwright, Page, ElementHandle
except ImportError:
    print("ERROR: playwright not installed. Run: pip install playwright && playwright install chromium")
    sys.exit(1)


# Config
API_URL = os.environ.get("CAPTCHA_SOLVER_URL", "http://localhost:8765")
HF_SPACE_URL = os.environ.get("HF_SPACE_URL", "")  # Set this for HF Spaces deployment
SOLVER_URL = HF_SPACE_URL or API_URL
TILE_DELAY_MIN = 0.8  # Min seconds between tile clicks
TILE_DELAY_MAX = 2.5  # Max seconds between tile clicks
CLICK_DELAY_MIN = 0.3
CLICK_DELAY_MAX = 0.8


async def _call_classify_api(
    image_bytes: bytes,
    instruction: str,
    client: httpx.AsyncClient,
) -> dict:
    """Call POST /classify with a tile image."""
    b64 = base64.b64encode(image_bytes).decode("ascii")
    payload = {
        "image_base64": b64,
        "instruction": instruction,
    }
    try:
        r = await client.post(f"{SOLVER_URL}/classify", json=payload, timeout=30)
        r.raise_for_status()
        return r.json()
    except Exception as exc:
        return {"match": False, "confidence": 0.0, "caption": "", "error": str(exc)}


def _extract_instruction(modal_content: str) -> str:
    """Extract hCaptcha instruction text from the challenge header."""
    # Common hCaptcha patterns
    patterns = [
        r"(?:click|find|select|choose)\s+all\s+[^.]+",
        r"(?:click|find|select|choose)\s+the\s+[^.]+",
        r"(?:which|what)\s+[^?]+\?",
    ]
    text_lower = modal_content.lower()
    for pattern in patterns:
        import re
        match = re.search(pattern, text_lower, re.IGNORECASE)
        if match:
            return match.group(0).strip()
    return modal_content.strip()


async def _human_like_mouse_move(page: Page, x: float, y: float) -> None:
    """Move mouse to (x, y) with human-like curve."""
    await page.mouse.move(x, y, steps=random.randint(10, 25))
    await asyncio.sleep(random.uniform(CLICK_DELAY_MIN, CLICK_DELAY_MAX))


async def _click_with_human_delay(page: Page, selector: str) -> None:
    """Click an element with random human-like delay."""
    await asyncio.sleep(random.uniform(TILE_DELAY_MIN, TILE_DELAY_MAX))
    element = await page.query_selector(selector)
    if element:
        box = await element.bounding_box()
        if box:
            # Click slightly off-center for realism
            x = box["x"] + box["width"] * random.uniform(0.3, 0.7)
            y = box["y"] + box["height"] * random.uniform(0.3, 0.7)
            await _human_like_mouse_move(page, x, y)
            await asyncio.sleep(random.uniform(0.1, 0.3))
            await page.mouse.click(x, y)


async def _screenshot_tile(page: Page, tile_index: int, grid_selector: str = ".task-image") -> Optional[bytes]:
    """Screenshot a specific tile from the hCaptcha grid."""
    tiles = await page.query_selector_all(grid_selector)
    if tile_index >= len(tiles):
        return None
    tile = tiles[tile_index]
    screenshot = await tile.screenshot()
    return screenshot


async def solve_hcaptcha(
    page: Page,
    max_retries: int = 3,
    verbose: bool = True,
) -> bool:
    """Solve hCaptcha on the current page.

    Args:
        page: Playwright page with hCaptcha loaded.
        max_retries: Number of attempts before giving up.
        verbose: Print progress info.

    Returns:
        True if captcha was solved successfully, False otherwise.
    """
    for attempt in range(max_retries):
        if verbose:
            print(f"[hCaptcha] Attempt {attempt + 1}/{max_retries}")

        try:
            # Wait for hCaptcha iframe to appear
            hcaptcha_frame = None
            for frame in page.frames:
                if "hcaptcha" in frame.url.lower():
                    hcaptcha_frame = frame
                    break

            if not hcaptcha_frame:
                # Try clicking the checkbox first
                checkbox = await page.query_selector(".h-captcha iframe, iframe[src*='hcaptcha']")
                if checkbox:
                    await checkbox.click()
                    await asyncio.sleep(2)
                    # Re-check frames
                    for frame in page.frames:
                        if "hcaptcha" in frame.url.lower():
                            hcaptcha_frame = frame
                            break

            if not hcaptcha_frame:
                if verbose:
                    print("[hCaptcha] No hCaptcha iframe found")
                return False

            # Wait for challenge to load (if checkbox was already solved)
            await asyncio.sleep(1)

            # Check if already solved (checkbox is green)
            checkbox_frame = None
            for frame in page.frames:
                if "hcaptcha" in frame.url.lower() and "checkbox" in frame.url.lower():
                    checkbox_frame = frame
                    break

            if checkbox_frame:
                is_checked = await checkbox_frame.query_selector(".check")
                if is_checked:
                    if verbose:
                        print("[hCaptcha] Already solved!")
                    return True

            # Look for the challenge iframe (separate from checkbox iframe)
            challenge_frame = None
            for frame in page.frames:
                if "hcaptcha" in frame.url.lower() and "challenge" in frame.url.lower():
                    challenge_frame = frame
                    break

            if not challenge_frame:
                # Challenge might be in the main frame's modal
                challenge_frame = page

            # Extract instruction text
            instruction = ""
            try:
                header = await challenge_frame.query_selector(".challenge-header, .prompt-text, [class*='header']")
                if header:
                    instruction = await header.inner_text()
                else:
                    # Try getting all visible text in the challenge
                    body = await challenge_frame.query_selector("body")
                    if body:
                        all_text = await body.inner_text()
                        instruction = _extract_instruction(all_text)
            except Exception:
                pass

            if not instruction:
                if verbose:
                    print("[hCaptcha] Could not extract instruction")
                continue

            if verbose:
                print(f"[hCaptcha] Instruction: {instruction}")

            # Screenshot all tiles
            tiles = await challenge_frame.query_selector_all(".task-image, [class*='tile'], [class*='image']")
            if not tiles:
                if verbose:
                    print("[hCaptcha] No tiles found in challenge")
                continue

            if verbose:
                print(f"[hCaptcha] Found {len(tiles)} tiles")

            # Classify each tile via API
            matches = []
            async with httpx.AsyncClient() as client:
                for i, tile in enumerate(tiles):
                    try:
                        tile_screenshot = await tile.screenshot()
                        result = await _call_classify_api(tile_screenshot, instruction, client)
                        if result.get("match"):
                            matches.append(i)
                        if verbose:
                            status = "MATCH" if result.get("match") else "skip"
                            caption = result.get("caption", "")
                            conf = result.get("confidence", 0)
                            print(f"  Tile {i + 1}: {status} (conf={conf:.2f}) caption={caption}")
                    except Exception as exc:
                        if verbose:
                            print(f"  Tile {i + 1}: error - {exc}")

            if not matches:
                if verbose:
                    print("[hCaptcha] No matches found, skipping")
                # Click "Get a new challenge" button
                try:
                    refresh_btn = await challenge_frame.query_selector(
                        ".refresh-button, [class*='refresh'], button[aria-label*='refresh']"
                    )
                    if refresh_btn:
                        await refresh_btn.click()
                        await asyncio.sleep(2)
                        continue
                except Exception:
                    pass
                continue

            if verbose:
                print(f"[hCaptcha] Clicking tiles: {[m + 1 for m in matches]}")

            # Click matching tiles with human-like delays
            for idx in matches:
                tile = tiles[idx]
                try:
                    box = await tile.bounding_box()
                    if box:
                        x = box["x"] + box["width"] * random.uniform(0.3, 0.7)
                        y = box["y"] + box["height"] * random.uniform(0.3, 0.7)
                        await _human_like_mouse_move(page, x, y)
                        await asyncio.sleep(random.uniform(0.1, 0.3))
                        await page.mouse.click(x, y)
                        await asyncio.sleep(random.uniform(TILE_DELAY_MIN, TILE_DELAY_MAX))
                except Exception as exc:
                    if verbose:
                        print(f"  Failed to click tile {idx + 1}: {exc}")

            # Click Verify button
            await asyncio.sleep(random.uniform(1.0, 2.0))
            try:
                verify_btn = await challenge_frame.query_selector(
                    ".verify-button, button[type='submit'], [class*='verify']"
                )
                if verify_btn:
                    await verify_btn.click()
                    await asyncio.sleep(3)

                    # Check if solved
                    for frame in page.frames:
                        if "hcaptcha" in frame.url.lower() and "checkbox" in frame.url.lower():
                            is_checked = await frame.query_selector(".check")
                            if is_checked:
                                if verbose:
                                    print("[hCaptcha] Solved successfully!")
                                return True

                    # If we get here, might need another attempt
                    if verbose:
                        print("[hCaptcha] Verification pending, checking...")

            except Exception as exc:
                if verbose:
                    print(f"[hCaptcha] Verify click failed: {exc}")

        except Exception as exc:
            if verbose:
                print(f"[hCaptcha] Attempt {attempt + 1} error: {exc}")

        # Wait before retry
        await asyncio.sleep(random.uniform(2.0, 4.0))

    if verbose:
        print(f"[hCaptcha] Failed after {max_retries} attempts")
    return False


async def main():
    """CLI entry point: solve hCaptcha on a given URL."""
    import argparse
    parser = argparse.ArgumentParser(description="Solve hCaptcha on a faucet page")
    parser.add_argument("url", help="URL of the page with hCaptcha")
    parser.add_argument("--address", help="FaucetPay address to fill in")
    parser.add_argument("--api", default=SOLVER_URL, help="Captcha solver API URL")
    parser.add_argument("--headless", action="store_true", help="Run headless (no browser window)")
    args = parser.parse_args()

    global SOLVER_URL
    SOLVER_URL = args.api

    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=args.headless,
            args=[
                "--disable-blink-features=AutomationControlled",
                "--no-sandbox",
            ],
        )
        context = await browser.new_context(
            viewport={"width": 1280, "height": 800},
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
        )
        page = await context.new_page()

        print(f"[bot] Navigating to {args.url}")
        await page.goto(args.url, wait_until="networkidle")
        await asyncio.sleep(2)

        # Fill in address if provided
        if args.address:
            addr_input = await page.query_selector("#fp-address, input[name='address']")
            if addr_input:
                await addr_input.fill(args.address)
                print(f"[bot] Address filled: {args.address}")

        # Click Claim button to open modal
        claim_btn = await page.query_selector("#faucet-claim-btn, .faucet-btn")
        if claim_btn:
            await claim_btn.click()
            print("[bot] Claim button clicked, modal opened")
            await asyncio.sleep(2)

        # Solve hCaptcha
        solved = await solve_hcaptcha(page, max_retries=3, verbose=True)

        if solved:
            # Click Submit
            submit_btn = await page.query_selector("#faucet-submit-btn, button.faucet-btn")
            if submit_btn:
                await submit_btn.click()
                print("[bot] Submit clicked!")
                await asyncio.sleep(5)

                # Check for success message
                success = await page.query_selector(".faucet-success, [class*='success']")
                if success:
                    msg = await success.inner_text()
                    print(f"[bot] SUCCESS: {msg}")
                else:
                    print("[bot] Claim submitted, check results")
        else:
            print("[bot] Failed to solve hCaptcha")

        await browser.close()


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