返回博客教程

用Python解决reCAPTCHA v3:分步教程

学习如何使用异步请求、重试逻辑和生产就绪的错误处理将验证码解决集成到Python脚本中。

reGOTCHA团队2025年12月19日6分钟 阅读
用Python解决reCAPTCHA v3:分步教程

前提条件

  • Python 3.8 或更高版本
  • 带有 API 密钥的 reGOTCHA 账户
  • 对异步编程的基本理解

安装

terminal
pip install httpx asyncio

基本同步示例

从简单的同步实现开始:

example.py
import httpx
import time

API_KEY = "your-api-key"
BASE_URL = "https://api.regotcha.com"

def solve_recaptcha(website_url: str, website_key: str, action: str = "submit"):
    # 创建任务
    response = httpx.post(
        f"{BASE_URL}/createTask",
        json={
            "clientKey": API_KEY,
            "task": {
                "type": "ReCaptchaV3EnterpriseTaskProxyless",
                "websiteURL": website_url,
                "websiteKey": website_key,
                "pageAction": action
            }
        }
    )
    task_id = response.json()["taskId"]

    # 轮询结果
    while True:
        result = httpx.post(
            f"{BASE_URL}/getTaskResult",
            json={"clientKey": API_KEY, "taskId": task_id}
        ).json()

        if result["status"] == "ready":
            return result["solution"]["gRecaptchaResponse"]

        time.sleep(2)

# 使用
token = solve_recaptcha(
    "https://example.com",
    "6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
    "login"
)
print(f"Token: {token}")

生产就绪的异步实现

对于生产工作负载,使用带有正确错误处理的异步:

example.py
import httpx
import asyncio
from typing import Optional

class RecaptchaSolver:
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.regotcha.com"

    async def solve(
        self,
        website_url: str,
        website_key: str,
        action: str = "submit",
        timeout: int = 120
    ) -> Optional[str]:
        async with httpx.AsyncClient() as client:
            # 创建任务
            response = await client.post(
                f"{self.base_url}/createTask",
                json={
                    "clientKey": self.api_key,
                    "task": {
                        "type": "ReCaptchaV3EnterpriseTaskProxyless",
                        "websiteURL": website_url,
                        "websiteKey": website_key,
                        "pageAction": action
                    }
                }
            )
            data = response.json()

            if data.get("errorId"):
                raise Exception(f"API 错误: {data.get('errorDescription')}")

            task_id = data["taskId"]

            # 带超时轮询
            start = asyncio.get_event_loop().time()
            while asyncio.get_event_loop().time() - start < timeout:
                result = await client.post(
                    f"{self.base_url}/getTaskResult",
                    json={"clientKey": self.api_key, "taskId": task_id}
                )
                result_data = result.json()

                if result_data["status"] == "ready":
                    return result_data["solution"]["gRecaptchaResponse"]

                if result_data.get("errorId"):
                    raise Exception(f"解决错误: {result_data.get('errorDescription')}")

                await asyncio.sleep(2)

            raise TimeoutError("验证码解决超时")

# 使用
async def main():
    solver = RecaptchaSolver("your-api-key")
    token = await solver.solve(
        "https://example.com",
        "6LcR_RIpAAAAANGDwXhTz8nO4N_NijF_Pj2iO4L6",
        "checkout"
    )
    print(f"Token: {token}")

asyncio.run(main())

批量处理

并发解决多个验证码:

example.py
async def solve_batch(solver, tasks):
    """并发解决多个验证码"""
    return await asyncio.gather(
        *[solver.solve(**task) for task in tasks],
        return_exceptions=True
    )

# 示例:并行解决 10 个验证码
tasks = [
    {"website_url": "https://site1.com", "website_key": "key1", "action": "login"},
    {"website_url": "https://site2.com", "website_key": "key2", "action": "signup"},
    # ... 更多任务
]

results = await solve_batch(solver, tasks)
性能提示:对于高容量工作负载,保持 AsyncClient 连接 并在请求之间复用以减少连接开销。
Python教程API代码

准备大规模破解验证码?

免费获取 50 积分开始使用。无需信用卡。