Scrapy + CaptchaAI 集成指南

Scrapy 抓到一半撞上验证码，最省力的接入点是 downloader middleware，而不是 spider。响应进入 parse() 之前先过中间件：命中 sitekey 就调用 CaptchaAI 识别，token 写进 request.meta，spider 照旧只管解析。整套实现只依赖 Scrapy 和 requests，Scrapy 2.5 以上可直接使用。识别是同步调用，会占住 Twisted 事件循环，验证码密集的域名建议单独跑一个 spider。

环境与依赖

pip install scrapy requests

国内拉包慢可以加镜像参数，例如 -i https://pypi.tuna.tsinghua.edu.cn/simple。

第一步：封装 CaptchaAI 识别模块

项目根目录新建 captcha_solver.py：任务提交到 in.php 拿任务 ID，再每 5 秒轮询一次 res.php，直到返回 OK|<token>。reCAPTCHA v2 走 userrecaptcha，图片验证码走 base64。

import requests
import time


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

    def solve_recaptcha(self, site_key, page_url, timeout=300):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": page_url,
        })

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

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

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

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

    def solve_image(self, image_base64, timeout=120):
        resp = requests.get(f"{self.base_url}/in.php", params={
            "key": self.api_key,
            "method": "base64",
            "body": image_base64,
        })

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

        task_id = resp.text.split("|")[1]
        deadline = time.time() + timeout

        while time.time() < deadline:
            time.sleep(5)
            result = requests.get(f"{self.base_url}/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
            })

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

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

CAPCHA_NOT_READY（拼写就是少一个 T）是正常中间态，代表任务还在排队，别当成错误抛出去。

第二步：中间件检测并识别验证码

新建 middlewares.py，中间件只做三件事：

1. 启动时从 settings 读 API Key，缺失就直接报错。
2. 在 process_response 里判断页面有没有验证码：正则匹配 data-sitekey，或取 img#captcha-image 里的 base64 图片。
3. 把识别结果写进 request.meta，交给 spider。

import base64
import re
from scrapy import signals
from scrapy.http import HtmlResponse
from captcha_solver import CaptchaAISolver


class CaptchaAIMiddleware:
    """Scrapy downloader middleware that detects and solves CAPTCHAs."""

    def __init__(self, api_key):
        self.solver = CaptchaAISolver(api_key)

    @classmethod
    def from_crawler(cls, crawler):
        api_key = crawler.settings.get("CAPTCHAAI_API_KEY")
        if not api_key:
            raise ValueError("CAPTCHAAI_API_KEY setting is required")
        return cls(api_key)

    def process_response(self, request, response, spider):
        # Check for reCAPTCHA on the page
        site_key = self._find_recaptcha_key(response.text)
        if site_key:
            spider.logger.info(f"reCAPTCHA detected on {response.url}")
            token = self.solver.solve_recaptcha(site_key, response.url)
            request.meta["captcha_token"] = token
            spider.logger.info("CAPTCHA solved successfully")

        # Check for image CAPTCHA
        captcha_img = self._find_image_captcha(response)
        if captcha_img:
            spider.logger.info(f"Image CAPTCHA detected on {response.url}")
            text = self.solver.solve_image(captcha_img)
            request.meta["captcha_text"] = text
            spider.logger.info(f"Image CAPTCHA solved: {text}")

        return response

    def _find_recaptcha_key(self, html):
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', html
        )
        return match.group(1) if match else None

    def _find_image_captcha(self, response):
        img = response.css("img#captcha-image::attr(src)").get()
        if img and img.startswith("data:image"):
            return img.split(",", 1)[1]
        return None

正则取的是页面里第一个 data-sitekey。页面嵌了多个组件时，先用容器选择器收窄范围再取属性。

第三步：在 settings.py 里注册

API Key 从环境变量读，不要写死在仓库里：

import os

CAPTCHAAI_API_KEY = os.environ.get("CAPTCHAAI_API_KEY")

DOWNLOADER_MIDDLEWARES = {
    "myproject.middlewares.CaptchaAIMiddleware": 560,
}

560 落在内置中间件的空位上（RetryMiddleware 550、MetaRefreshMiddleware 580），不撞号即可。注意 process_response 按数值从大到小回流，中间件拿到的是重定向、解压之后的最终响应。

第四步：spider 回填 g-recaptcha-response

spider 只看 response.meta：有 token 就用 FormRequest 连同表单字段一起提交，没有就直接解析。

import scrapy


class ProductSpider(scrapy.Spider):
    name = "products"
    start_urls = ["https://example.com/products"]

    def parse(self, response):
        # If CAPTCHA was solved, the token is in meta
        token = response.meta.get("captcha_token")
        if token:
            # Resubmit the page with the token
            yield scrapy.FormRequest(
                url=response.url,
                formdata={"g-recaptcha-response": token},
                callback=self.parse_products,
            )
        else:
            yield from self.parse_products(response)

    def parse_products(self, response):
        for product in response.css(".product-item"):
            yield {
                "name": product.css("h2::text").get(),
                "price": product.css(".price::text").get(),
                "url": response.urljoin(
                    product.css("a::attr(href)").get()
                ),
            }

        next_page = response.css("a.next-page::attr(href)").get()
        if next_page:
            yield scrapy.Request(response.urljoin(next_page))

翻页逻辑不受影响，验证码只是中间多插了一次识别。

第五步：整页验证码的自动重试

有些站点触发风控时会整页替换成验证码页，光拿到 token 不够，还得把原请求重发一次：

class CaptchaRetryMiddleware:
    """Retry requests that return CAPTCHA challenge pages."""

    max_retries = 3

    def process_response(self, request, response, spider):
        if self._is_captcha_page(response):
            retries = request.meta.get("captcha_retries", 0)
            if retries < self.max_retries:
                request.meta["captcha_retries"] = retries + 1
                spider.logger.info(
                    f"CAPTCHA page detected, retry {retries + 1}"
                )
                return request.copy()

        return response

    def _is_captcha_page(self, response):
        indicators = [
            "g-recaptcha",
            "cf-turnstile",
            "captcha-image",
            "Please verify you are human",
        ]
        return any(ind in response.text for ind in indicators)

四个特征串按你的目标页面改。写得太宽，正常页面也会被判成验证码页。

运行爬虫

export CAPTCHAAI_API_KEY="YOUR_API_KEY"
scrapy crawl products -o products.json

PowerShell 下环境变量写成 $env:CAPTCHAAI_API_KEY="YOUR_API_KEY"。

线程数怎么和爬虫并发对上

CaptchaAI 按并发线程计费，套餐内识别次数不限。一个线程就是一次在途任务，返回后立刻释放。所以线程数取决于同一时刻有几个验证码在等结果，而不是一天抓了多少页。

• BASIC（$15/月，5 线程）：单机一到两个 spider。
• STANDARD（$30/月，15 线程）：多个 spider 并行。
• ADVANCE（$90/月，50 线程）：scrapyd 多节点部署。

本文是同步实现，单进程只有一个识别在途，BASIC 够用。

抓国内站点时要改什么

这套检测按 reCAPTCHA 和图片验证码写，抓海外站点够用。国内主流是 GeeTest（极验）、网易易盾、腾讯防水墙，页面里没有 data-sitekey，正则会一直落空。CaptchaAI 支持 GeeTest v3（方法 geetest）；GeeTest v4 官方标注为即将支持，暂不可用；hCaptcha 和 FunCaptcha（Arkose Labs）不支持；CaptchaFox、Friendly Captcha、Lemin 为测试版。

reCAPTCHA 还要加载 Google 域名下的脚本，国内网络未必稳定，排查时先分清是页面没加载还是识别没返回。

采集范围： 只抓你有权抓取的数据，遵守目标站点的 robots 协议，并留意数据安全法与 PIPL 对个人信息的要求。

排查清单

常见问题

中间件优先级为什么写成 560？

550 是 RetryMiddleware、580 是 MetaRefreshMiddleware，560 正好落在两者之间的空位。换成 555 或 570 一样能跑，关键是排在重试之后。

一个 Scrapy 项目要买几个线程？

看同一时刻在途的识别数，不看总页数。同步实现单进程只占一个线程，BASIC（$15/月，5 线程）就够；多节点部署再上 STANDARD（$30/月，15 线程）。

hCaptcha 和 GeeTest v4 能用这套中间件识别吗？

都不支持。GeeTest v4 官方标注为即将支持。现在可用的是 reCAPTCHA、Cloudflare、GeeTest v3 与图片/九宫格/BLS 这几类。

相关指南

• Python 爬虫验证码处理实战
• HTTPX 调用 CaptchaAI 接口
• aiohttp 异步识别验证码