如何使用TinyFish AgentQL解决CAPTCHA – 使用CapSolver的分步指南

pip install agentql
playwright install chromium

npm install agentql
npx playwright install chromium

mkdir -p ~/capsolver-extension
unzip CapSolver.Browser.Extension-chrome-v*.zip -d ~/capsolver-extension/

ls ~/capsolver-extension/manifest.json

export const defaultConfig = {
  apiKey: 'CAP-XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', // ← 你的密钥在这里
  useCapsolver: true,
  // ...其余配置
};

import agentql
from playwright.sync_api import sync_playwright
import time
import os

# CapSolver 扩展路径
CAPSOLVER_EXTENSION_PATH = os.path.expanduser("~/capsolver-extension")

def main():
    with sync_playwright() as p:
        # 使用持久化上下文和 CapSolver 扩展启动 Chromium
        context = p.chromium.launch_persistent_context(
            user_data_dir="./browser-data",
            headless=False,  # 扩展需要非无头模式
            args=[
                f"--disable-extensions-except={CAPSOLVER_EXTENSION_PATH}",
                f"--load-extension={CAPSOLVER_EXTENSION_PATH}",
            ],
        )

        # 使用 AgentQL 包装页面以进行 AI 驱动的查询
        page = agentql.wrap(context.pages[0])

        # 导航到目标页面
        page.goto("https://example.com/protected-page")

        # 等待 CapSolver 检测并解决任何验证码
        time.sleep(30)

        # 使用 AgentQL 的自然语言查询查找并点击提交按钮
        response = page.query_elements("""
        {
            submit_button
        }
        """)

        # 点击提交按钮 —— 验证码已解决！
        response.submit_button.click()

        # 提交后提取数据
        result = page.query_data("""
        {
            confirmation_message
        }
        """)

        print(f"结果: {result['confirmation_message']}")

        context.close()

if __name__ == "__main__":
    main()

const { chromium } = require('playwright');
const agentql = require('agentql');
const path = require('path');
const os = require('os');

const CAPSOLVER_EXTENSION_PATH = path.join(os.homedir(), 'capsolver-extension');

(async () => {
  // 使用持久化上下文和 CapSolver 扩展启动 Chromium
  const context = await chromium.launchPersistentContext('./browser-data', {
    headless: false, // 扩展需要非无头模式
    args: [
      `--disable-extensions-except=${CAPSOLVER_EXTENSION_PATH}`,
      `--load-extension=${CAPSOLVER_EXTENSION_PATH}`,
    ],
  });

  // 获取第一个页面并用 AgentQL 包装
  const page = agentql.wrap(context.pages()[0]);

  // 导航到目标页面
  await page.goto('https://example.com/protected-page');

  // 等待 CapSolver 处理任何验证码
  await page.waitForTimeout(30000);

  // 使用 AgentQL 查询进行交互 —— 验证码已解决
  const response = await page.queryElements(`{
    submit_button
  }`);

  await response.submit_button.click();

  // 提取结果数据
  const result = await page.queryData(`{
    confirmation_message
  }`);

  console.log('结果:', result.confirmation_message);

  await context.close();
})();

page.goto("https://example.com/contact")

# 使用 AgentQL 查询填写表单
response = page.query_elements("""
{
    contact_form {
        name_field
        email_field
        message_field
        submit_button
    }
}
""")

response.contact_form.name_field.fill("John Doe")
response.contact_form.email_field.fill("john@example.com")
response.contact_form.message_field.fill("Hello, I have a question about your services.")

# 等待 CapSolver 解决验证码
time.sleep(30)

# 提交 —— 验证码令牌已注入
response.contact_form.submit_button.click()

page.goto("https://example.com/login")

# 等待 CapSolver 解决 Turnstile 挑战
time.sleep(25)

# 使用 AgentQL 查找登录表单元素
response = page.query_elements("""
{
    login_form {
        email_input
        password_input
        login_button
    }
}
""")

# 填写表单 —— Turnstile 已处理
response.login_form.email_input.fill("me@example.com")
response.login_form.password_input.fill("mypassword123")

# 点击登录
response.login_form.login_button.click()

page.goto("https://example.com/data")

# 等待任何验证码挑战清除
time.sleep(30)

# 使用 AgentQL 提取结构化数据
data = page.query_data("""
{
    products[] {
        name
        price
        rating
        availability
    }
}
""")

for product in data['products']:
    print(f"{product['name']}: ${product['price']} ({product['rating']} 星)")

你的 AgentQL 脚本
───────────────────────────────────────────────────
page.goto("https://...")       ──►  Chromium 加载页面
                                           │
                                           ▼
                               ┌─────────────────────────────┐
                               │  包含验证码小部件的页面     │
                               │                               │
                               │  CapSolver 扩展:         │
                               │  1. 内容脚本检测页面上的验证码 │
                               │  2. 服务工作线程调用 CapSolver API │
                               │  3. 收到令牌            │
                               │  4. 将令牌注入隐藏表单字段  │
                               └─────────────────────────────┘
                                           │
                                           ▼
time.sleep(30)                   扩展解决验证码...
                                           │
                                           ▼
page.query_elements(...)         AgentQL 找到表单元素
submit_button.click()            表单提交 —— 有效令牌已注入
                                           │
                                           ▼
                               "验证成功！"

import agentql
from playwright.sync_api import sync_playwright
import os

# 配置
CAPSOLVER_EXTENSION_PATH = os.path.expanduser("~/capsolver-extension")
USER_DATA_DIR = "./browser-data"

with sync_playwright() as p:
    context = p.chromium.launch_persistent_context(
        user_data_dir=USER_DATA_DIR,
        headless=False,
        args=[
            f"--disable-extensions-except={CAPSOLVER_EXTENSION_PATH}",
            f"--load-extension={CAPSOLVER_EXTENSION_PATH}",
        ],
    )
    page = agentql.wrap(context.pages[0])
    # ... 你的自动化代码在这里
    context.close()

pip install capsolver

import time
import requests
import capsolver
from playwright.sync_api import sync_playwright


TINYFISH_API_KEY = "YOUR TINYFISH API KEY"
capsolver.api_key = "YOUR CAPSOLVER API KEY"


website_url = "https://www.google.com/recaptcha/api2/demo"
website_key = "6Le-wvkSAAAAAPBMRTvw0Q4Muexq9bi0DJwx_mJ-"
captcha_type = "ReCaptchaV2TaskProxyLess"


def create_tinyfish_browser(url):
    resp = requests.post(
        "https://api.browser.tinyfish.ai",
        headers={
            "X-API-Key": TINYFISH_API_KEY,
            "Content-Type": "application/json",
        },
        json={
            "url": url,
            "timeout_seconds": 300,
        },
        timeout=90,
    )
    resp.raise_for_status()
    return resp.json()


def close_tinyfish_browser(session_id):
    try:
        requests.delete(
            f"https://api.browser.tinyfish.ai/{session_id}",
            headers={"X-API-Key": TINYFISH_API_KEY},
            timeout=30,
        )
    except Exception as exc:
        print("   Failed to close tinyfish browser session:", repr(exc))


def solve_recaptcha_v2():
    print("3. Solving reCAPTCHA with CapSolver")
    solution = capsolver.solve({
        "type": captcha_type,
        "websiteURL": website_url,
        "websiteKey": website_key,
    })

    token = solution.get("gRecaptchaResponse")
    if not token:
        raise RuntimeError(f"CapSolver did not return a reCAPTCHA token: {solution}")

    print("   CapSolver task solved：", token)
    return token


def inject_recaptcha_token(page, token):
    page.evaluate(
        """
        (token) => {
            const textarea = document.getElementById('g-recaptcha-response');
            if (textarea) {
                textarea.value = token;
            }
        }
        """,
        token,
    )
    print("   CapSolver token injected")
    page.click('input[type="submit"]')
    print("   Submit button clicked")


def main():
    session_id = None
    try:
        print("1. Creating tinyfish browser session")
        session = create_tinyfish_browser(website_url)

        session_id = session["session_id"]
        cdp_url = session["cdp_url"]
        print("   Tinyfish session id:", session_id)
        print("   Tinyfish cdp url:", cdp_url)

        with sync_playwright() as p:
            print("2. Connecting to tinyfish browser via cdp")
            browser = p.chromium.connect_over_cdp(cdp_url)
            context = browser.contexts[0]
            page = context.pages[0] if context.pages else context.new_page()
            page.wait_for_load_state("domcontentloaded", timeout=60000)
            print("   Current page url:", page.url)

            token = solve_recaptcha_v2()

            print("4. Injecting the capsolver token")
            inject_recaptcha_token(page, token)

            page.wait_for_load_state("domcontentloaded", timeout=30000)
            time.sleep(3)

            body_text = page.locator("body").inner_text(timeout=10000)
            print("5. Page result:", " ".join(body_text.split()))

            print("6. Closing browser sessions")
            browser.close()
    finally:
        if session_id:
            close_tinyfish_browser(session_id)


if __name__ == "__main__":
    main()

# ✅ 正确 —— 持久化上下文，有头模式
context = p.chromium.launch_persistent_context(
    user_data_dir="./browser-data",
    headless=False,
    args=[...extension args...]
)

# ❌ 错误 —— 常规上下文（扩展不会加载）
browser = p.chromium.launch()
context = browser.new_context()

# 安装Xvfb
sudo apt-get install xvfb

# 启动虚拟显示
Xvfb :99 -screen 0 1280x720x24 &

# 设置DISPLAY
export DISPLAY=:99

# 安装Playwright的Chromium（推荐）
npx playwright install chromium

# 或下载Chrome for Testing
# 访问：https://googlechromelabs.github.io/chrome-for-testing/