aiohttp + CaptchaAI：异步验证码解决

脚本要同时处理几十个页面的验证码时，同步的 requests 会让 CPU 大部分时间耗在等网络响应上；aiohttp 能在等待 token 时继续处理其他请求，这是异步识别的核心优势。本文用一个可复用客户端，演示单任务、批量并发、限流和 Turnstile 场景。

环境准备

pip install aiohttp

国内装包慢可加 -i 走镜像，例如清华 TUNA：pip install -i https://pypi.tuna.tsinghua.edu.cn/simple aiohttp。

封装一个异步 CaptchaAI 客户端

CaptchaAI 接口是标准的提交 + 轮询模式：参数发到 in.php 拿任务 ID，轮询 res.php 直到变成 OK。下面 AsyncCaptchaAI 类把流程封装成协程，配合 aiohttp.ClientSession 复用连接池，四个方法覆盖日常需要。

import aiohttp
import asyncio


class AsyncCaptchaAI:
    def __init__(self, api_key):
        self.api_key = api_key
        self.base_url = "https://ocr.captchaai.com"

    async def submit(self, session, params):
        """Submit a CAPTCHA task and return the task ID."""
        params["key"] = self.api_key
        async with session.get(
            f"{self.base_url}/in.php", params=params
        ) as resp:
            text = await resp.text()

        if not text.startswith("OK|"):
            raise Exception(f"Submit failed: {text}")

        return text.split("|")[1]

    async def poll(self, session, task_id, timeout=300):
        """Poll for the result with a timeout."""
        params = {
            "key": self.api_key,
            "action": "get",
            "id": task_id,
        }
        deadline = asyncio.get_event_loop().time() + timeout

        while asyncio.get_event_loop().time() < deadline:
            await asyncio.sleep(5)

            async with session.get(
                f"{self.base_url}/res.php", params=params
            ) as resp:
                text = await resp.text()

            if text == "CAPCHA_NOT_READY":
                continue
            if text.startswith("OK|"):
                return text.split("|", 1)[1]
            raise Exception(f"Solve failed: {text}")

        raise TimeoutError(f"Task {task_id} timed out after {timeout}s")

    async def solve(self, session, params, timeout=300):
        """Submit and poll in one call."""
        task_id = await self.submit(session, params)
        return await self.poll(session, task_id, timeout)

    async def get_balance(self, session):
        """Check account balance."""
        params = {"key": self.api_key, "action": "getbalance"}
        async with session.get(
            f"{self.base_url}/res.php", params=params
        ) as resp:
            return float(await resp.text())

poll 用 asyncio.sleep(5) 做轮询间隔，接近实际识别耗时，不需要更激进的重试。

单任务示例：识别一个 reCAPTCHA v2

先跑通最简单的场景：查询余额，再识别一个页面的 reCAPTCHA v2。googlekey 是页面 sitekey，pageurl 需与浏览器实际加载地址一致，对不上通常直接报错。

import asyncio
import os

async def main():
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Check balance
        balance = await solver.get_balance(session)
        print(f"Balance: ${balance:.2f}")

        # Solve reCAPTCHA v2
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": "6Le-wvkS...",
            "pageurl": "https://example.com",
        })
        print(f"Token: {token[:50]}...")

asyncio.run(main())

批量并发：一次性处理多个页面的验证码

把 solve 协程包进 asyncio.gather，多个 URL 并发提交、并发轮询，总耗时接近单个任务而非 URL 数乘以单任务耗时。return_exceptions=True 让单个任务失败不中断整批请求。

举例：跨境电商团队核对 50 个商品页面的 reCAPTCHA 集成，同步脚本要跑好几分钟，换成下面的并发代码，几秒就能拿到结果。

async def solve_batch(urls, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        tasks = [
            solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })
            for url in urls
        ]

        results = await asyncio.gather(*tasks, return_exceptions=True)

        for url, result in zip(urls, results):
            if isinstance(result, Exception):
                print(f"FAILED {url}: {result}")
            else:
                print(f"SOLVED {url}: {len(result)} chars")

        return results


urls = [
    "https://example.com/page1",
    "https://example.com/page2",
    "https://example.com/page3",
    "https://example.com/page4",
    "https://example.com/page5",
]
asyncio.run(solve_batch(urls, "6Le-wvkS..."))

在异步爬虫中接入验证码识别

典型流程：先请求页面，检查有没有验证码标记，有才调用 CaptchaAI，没有则直接返回，省掉不必要的请求。token 按目标表单字段名提交，reCAPTCHA v2 对应字段固定是 g-recaptcha-response。

async def scrape_with_captcha(url, site_key):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        # Fetch the page
        async with session.get(url) as resp:
            html = await resp.text()

        # Check if page has a CAPTCHA
        if "g-recaptcha" not in html:
            return html  # No CAPTCHA, return content

        # Solve the CAPTCHA
        token = await solver.solve(session, {
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": url,
        })

        # Submit with solved token
        async with session.post(url, data={
            "g-recaptcha-response": token,
        }) as resp:
            return await resp.text()

用信号量控制并发，避免打满 API

URL 列表一大，直接用 asyncio.gather 会瞬间把几十上百个请求打到 CaptchaAI，容易拖慢响应。asyncio.Semaphore 把同时在跑的任务数锁定在上限内，其余排队，既保证吞吐量，也不打满连接池。

async def solve_with_limit(urls, site_key, max_concurrent=10):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])
    semaphore = asyncio.Semaphore(max_concurrent)

    async def solve_one(session, url):
        async with semaphore:
            return await solver.solve(session, {
                "method": "userrecaptcha",
                "googlekey": site_key,
                "pageurl": url,
            })

    async with aiohttp.ClientSession() as session:
        tasks = [solve_one(session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

    solved = sum(1 for r in results if not isinstance(r, Exception))
    print(f"Solved {solved}/{len(urls)} CAPTCHAs")
    return results

max_concurrent=10 是比较稳妥的起点，再结合账号线程数往上调。

Turnstile 场景：sitekey 换 token

Turnstile 调用方式和 reCAPTCHA 类似，只是 method 换成 turnstile，参数名从 googlekey 变成 sitekey。token 提交到对应表单字段（cf-turnstile-response），上面批量、限流代码可直接复用这个方法。

async def solve_turnstile(url, sitekey):
    solver = AsyncCaptchaAI(os.environ["CAPTCHAAI_API_KEY"])

    async with aiohttp.ClientSession() as session:
        token = await solver.solve(session, {
            "method": "turnstile",
            "sitekey": sitekey,
            "pageurl": url,
        })
        return token

常见报错与处理方式

下面几类报错在 aiohttp + CaptchaAI 组合里最常见，多数问题都能在表里对上号。

目标页面若用 reCAPTCHA（依赖 Google 托管脚本），国内网络下偶尔会连接偏慢或超时，先排查链路本身，别急着改代码。

常见问题

aiohttp 和 httpx 该选哪个用于异步验证码识别？

aiohttp 是 Python 最成熟的异步 HTTP 库，高并发下性能更稳定。已用 httpx 的项目也能照搬同样的提交 + 轮询逻辑，参考httpx 异步集成指南。

CaptchaAI 异步客户端支持哪些验证码类型？

换掉 method 即可覆盖 reCAPTCHA v2/v3、Cloudflare Turnstile、GeeTest v3（极验）、图片/OCR 验证码。hCaptcha、FunCaptcha（Arkose Labs）不支持，GeeTest v4 暂未开放，只能算即将支持。

asyncio.gather 里某个 URL 识别失败，会不会影响其他 URL？

不会。传了 return_exceptions=True 后，异常会变成结果列表里的一项，其余任务照常跑完，用 isinstance(result, Exception) 逐个判断即可。

并发数量设多少合适，会不会触发限流？

从 max_concurrent=10 起步比较稳妥，再结合线程数往上调；数字设得比线程数高很多，多出的任务只会排队，不会更快。

国内环境请求验证码相关接口经常超时，是代码写错了吗？

大概率不是。reCAPTCHA 部分脚本托管在 Google，国内访问本身就不稳定；先确认 ocr.captchaai.com 能否连通，再排查线路问题。

相关阅读

• 用 httpx 实现异步验证码识别
• 并发识别验证码的整体思路
• Scrapy 项目接入 CaptchaAI