如何在 Scrapy 中用一个内联请求识别 ALTCHA

要在 Scrapy 中识别 ALTCHA，先从 altcha-widget 元素上读出挑战地址，在同一个回调里用内联请求（inline request）去取它，把拿到的 JSON 交给 AsyncCapSkip 的 altcha 方法，然后提交表单，把 token 放进小组件指定的字段里，这个字段默认叫 altcha。ALTCHA 是工作量证明，不是图片，所以整个过程都不需要浏览器，一个普通的 Scrapy spider 就够了。Scrapy 特有的坑只有一个，却很容易漏掉：挑战接口每次都在同一个 URL 上返回一份新文档，而 Scrapy 的去重过滤器会悄无声息地丢掉发往它的第二个请求。本指南讲小组件、获取、识别和提交，并附上一个可以直接运行的 spider。

你需要什么

• Scrapy 2.14 或更高版本，下文用到的内联请求方法需要它。从 2.13 起，Scrapy 默认运行在 asyncio reactor 上，正是这一点让回调可以 await 异步客户端。文末附近有一节讲更早的版本。
• capskip Python 包 1.2.0 或更高版本，altcha 方法就是在这个版本中加入的；以及在 Windows 机器上运行的 CapSkip 1.3.0 或更高版本。
• form2request 包，Scrapy 现在推荐用它来构建表单提交。
• 带有表单的那个页面。其余一切，包括挑战在内，都来自这个页面，第 1 步会告诉你在哪里。
• 识别工具的地址。Local 模式只在 127.0.0.1 上响应，仅供本机使用；Server 模式监听你的网络地址或公网 IP，这样另一台机器上的爬虫就能通过 API 调用它。两者都位于 连接设置中，下文有一节会说明何时该切换。

# pip install scrapy capskip form2request
pip install scrapy capskip form2request

第 1 步：从小组件上读出挑战

ALTCHA 没有 sitekey。在 Scrapy 中识别 ALTCHA，你需要的是挑战本身，或者挑战的来源地址，而小组件元素上带着的就是其中之一。具体由哪个属性承载，取决于小组件的版本，所以三个属性都要读。

# Inside a spider callback; response is the page with the form.
widget = response.css("altcha-widget")
source = (widget.attrib.get("challenge")
          or widget.attrib.get("challengejson")
          or widget.attrib.get("challengeurl"))

# The token goes in a field named by the widget, altcha by default.
field = widget.attrib.get("name", "altcha")

# A v3 challenge or a v1/v2 challengejson carries the document inline.
inline = source.lstrip().startswith("{")

还要留意 name 属性。小组件会把载荷写进一个以此命名的隐藏 input，默认值是 altcha，但站点可以改掉它。读取它只需要一行代码，却能避免一次悄悄让真正字段留空的提交。少数部署是在页面自己的脚本里生成挑战，而不是由接口提供，这时就没有属性可读了。看一下 Network 标签页，就知道你遇到的是哪一种。

第 2 步：在回调里获取挑战

如果小组件指向的是一个接口，就在你当前所在的回调里，用引擎的 download_async 方法去取它。Scrapy 文档把这个方法放在 内联请求一节中介绍。这个请求会经过下载器中间件，所以 spider 的 cookie jar、user agent 和代理设置都会作用于它，就像作用于页面一样。它不经过调度器，而这正是关键所在。

# The page's jar and proxy; no scheduler, no dupefilter.
meta = {"dont_cache": True, "allow_offsite": True}
for key in ("cookiejar", "proxy"):
    if key in response.meta:
        meta[key] = response.meta[key]
reply = await self.crawler.engine.download_async(
    scrapy.Request(response.urljoin(source), meta=meta,
                   headers={"Referer": response.url})
)
if reply.status != 200:
    raise ValueError(f"challenge endpoint answered {reply.status}")
challenge_json = reply.text

跳过调度器之所以重要，原因如下。最直观的替代做法，是为挑战 yield 一个带独立回调的 Request，这样第一个表单没问题，第二个就丢了。每次请求这个接口都会返回新的挑战，但 URL 始终不变，所以 Scrapy 的去重过滤器会把第二个请求当成已经发过的请求，直接丢弃。不会报任何错。过滤器只在 debug 级别记录一次它的第一次丢弃，在爬取统计里把一个计数器加一，而它背后的那个表单就这样永远不会被提交。内联请求则根本不会经过这个过滤器。

那段代码的其余部分，处理的是调度器平时会替你处理的细节。每个请求自己的状态不会自动沿用，所以如果页面是带着 cookiejar 或 proxy 键抓取的，循环会把它从页面的 meta 里复制过来。Referer 请求头是手动设置的，因为负责添加它的中间件运行在 spider 一侧。如果挑战接口在另一个域名上，比如托管的 ALTCHA 服务，而 spider 又设置了 allowed_domains，offsite 过滤器就会拦下这次请求，allow_offsite 则让这一个请求放行。之所以要检查状态码，是因为接口返回的 403 在这里会作为普通响应回来：把错误状态码转成失败的那个中间件运行在 spider 一侧，而内联请求会跳过它。至于 dont_cache，如果你开启了 Scrapy 的 HTTP 缓存，它能让缓存别碰挑战。缓存下来的挑战就是过期的挑战，而 CapSkip 遇到过期的内联挑战会立即拒绝，不会把 CPU 浪费在一个站点注定会拒收的 token 上。没有这个标志，第二次运行时，所有共用该接口的页面都会拿到同一个缓存下来的挑战。表单页面也是同样的道理。缓存的页面会重放旧的防伪造 token 和会话 cookie，如果挑战是内联的，还会重放旧的挑战，所以开启缓存时，也要给表单页面加上 dont_cache，或者干脆对这个 spider 关闭缓存。

你也可以把接口作为 challenge_url 传入，让 CapSkip 自己去取挑战。对公开接口来说，这样没问题。但 CapSkip 的请求不带你 spider 的任何 cookie，而有些站点只向加载了表单的那个会话下发挑战。通过 Scrapy 获取，能让整个流程始终保持同一个会话。

第 3 步：识别挑战，不阻塞爬取

使用 AsyncCapSkip，并把回调写成 async def。在 Python 包里，这个类是真正的 asyncio 客户端，而不是一个别名，所以 await 它时，控制权会在识别进行期间交还给 Scrapy，其他所有请求都照常推进。同步的 CapSkip 类则会在每次识别期间让整个爬取停下来，这个问题的详细解释见 Scrapy 验证码中间件指南.

from capskip import AsyncCapSkip

solver = AsyncCapSkip(host="127.0.0.1", port=8080)

# Pass the document exactly as the endpoint sent it.
result = await solver.altcha(url=response.url, challenge_json=challenge_json)

token = result["token"]   # base64 payload for the form field

因为挑战是内联传入的，CapSkip 完全不会发出网络请求。它一直做哈希运算，直到找到答案，通常只需几毫秒；客户端会立即轮询一次，四分之一秒后再轮询一次，所以一次典型的识别大约四分之一秒就能返回。这次调用最多等待 defaultTimeout，即 120 秒，因为这是 CPU 计算，而不是浏览器会话。有两种挑战算法，Argon2id 和 scrypt，会被直接拒绝而不是尝试，这种拒绝会在一秒之内以 ApiException 的形式返回。更常见的 PBKDF2 和 SHA 方案都受支持，包括 ALTCHA 推荐的默认算法。

把响应的文本原样传入，不需要解析。真正不能改动的是 token：它是一份 JSON 文档的 base64，其中的字段都经过站点服务器的签名，所以被截断、解码或重新编码过的 token 都无法通过验证。

第 4 步：带上 token 提交表单

小组件要填的那个隐藏 input，在 Scrapy 下载到的 HTML 里并不存在。它是小组件在浏览器中运行之后才创建的，所以任何读取服务器返回 HTML 的东西都找不到它。你要自己把它加上，字段名用第 1 步读到的那个。示例用 form2request 构建提交，Scrapy 推荐用它代替 FormRequest.from_response：从 Scrapy 2.16 起，那个旧方法每次调用都会记录一条弃用警告。

from form2request import form2request

form = response.xpath("//form[.//altcha-widget]")
data = {"email": "[email protected]", field: token}
yield form2request(form, data).to_scrapy(
    callback=self.after_submit,
    priority=10,
    meta={"handle_httpstatus_all": True},
)

form2request 会保留页面已经放进表单的内容，比如防伪造 token，并按下第一个提交按钮，所以这个请求看起来就和浏览器会发出的一样。XPath 选中的是包含小组件的那个表单，这在同时带有搜索框的页面上很重要。handle_httpstatus_all 让回调能看到被拒的提交，否则 Scrapy 会在它到达回调之前就把它丢掉。提高优先级能让提交排在调度器里所有等待中的请求前面，因为挑战会过期。有些窗口只有短短两分钟，一个排在长时间爬取后面等待的 token，可能还没发出去就过期了。

每个 token 只当作一次提交有效。如果站点拒绝了表单，就从第 2 步重新开始，换一个新挑战，而不是重试同一个请求。有些集成是把载荷放在 JSON 请求体或 cookie 里发送，而不是表单字段，所以请打开 DevTools，手动提交一次表单，然后照搬页面发送的内容。

把识别工具放到别处运行

示例使用 127.0.0.1，因为当 spider 和 CapSkip 在同一台机器上时，这样写是对的。部署到另一台机器上的 Scrapyd、VPS 或托管 Scrapy 平台上的爬虫，回环地址指向的是它自己，第一次识别就会抛出 NetworkException。把 CapSkip 切换到 Server 模式，它就会监听你的网络地址或公网 IP，这样 spider 就能从你允许的任何地方通过 API 访问到它。如果链路要经过公网，请使用静态公网 IP，打开 API 密钥校验，并用一条 Windows Firewall 规则把端口限制在你预期的地址上。它仍然是你自己的 Windows 机器，识别也仍然不计量。

客户端不会自己读取环境变量。像完整示例那样，在 spider 里读取 CAPSKIP_HOST 和 CAPSKIP_API_KEY 并传给构造函数，这样同一份代码在你的桌面机和服务器上都能运行。

完整可运行示例

# pip install scrapy capskip form2request
import os

import scrapy
from capskip import AsyncCapSkip, CapSkipError
from form2request import form2request


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

    solver = AsyncCapSkip(
        apiKey=os.environ.get("CAPSKIP_API_KEY", "capskip"),
        host=os.environ.get("CAPSKIP_HOST", "127.0.0.1"),
        port=8080,
    )

    async def parse(self, response):
        widget = response.css("altcha-widget")
        source = (widget.attrib.get("challenge")
                  or widget.attrib.get("challengejson")
                  or widget.attrib.get("challengeurl"))
        if not source:
            self.logger.warning("no ALTCHA challenge on %s", response.url)
            return
        field = widget.attrib.get("name", "altcha")

        if source.lstrip().startswith("{"):
            challenge_json = source
        else:
            # Inline request: the page's jar and proxy, no dupefilter.
            meta = {"dont_cache": True, "allow_offsite": True}
            for key in ("cookiejar", "proxy"):
                if key in response.meta:
                    meta[key] = response.meta[key]
            reply = await self.crawler.engine.download_async(
                scrapy.Request(response.urljoin(source), meta=meta,
                               headers={"Referer": response.url})
            )
            if reply.status != 200:
                self.logger.error("challenge endpoint answered %s", reply.status)
                return
            challenge_json = reply.text

        try:
            result = await self.solver.altcha(
                url=response.url, challenge_json=challenge_json)
        except CapSkipError as exc:
            # Expired challenge, unsupported algorithm, or CapSkip unreachable.
            self.logger.error("ALTCHA not solved on %s: %r", response.url, exc)
            return

        form = response.xpath("//form[.//altcha-widget]")
        data = {"email": "[email protected]", field: result["token"]}
        yield form2request(form, data).to_scrapy(
            callback=self.after_submit,
            priority=10,
            meta={"handle_httpstatus_all": True},
        )

    def after_submit(self, response):
        yield {"url": response.url, "status": response.status}

用 scrapy runspider 加文件名运行它，或者把这个类放进一个项目里。这个 spider 能处理两代小组件，在页面的会话中获取挑战，并为每个提交的表单 yield 一个 item，附带站点返回的状态码，被拒的也包括在内。由于获取挑战绕过了调度器，共用同一个挑战接口的页面都会各自拿到自己的挑战。altcha 方法背后的原始接口记录在 API 参考文档中；不套 Scrapy、直接用 Python 调用的写法，见 Python ALTCHA 指南.

在低于 2.14 的 Scrapy 上

2.14 之前没有 download_async，但从 Scrapy 2.6 起，同样的内联获取可以通过 engine.download 实现，再用一个辅助函数把它的结果转换成协程可以等待的对象，然后 await 它。

from scrapy.utils.defer import maybe_deferred_to_future

# Scrapy 2.6 to 2.13: the same inline fetch, through engine.download.
reply = await maybe_deferred_to_future(self.crawler.engine.download(
    scrapy.Request(response.urljoin(source),
                   meta={"dont_cache": True, "allow_offsite": True})
))

在这些版本上，FormRequest.from_response 构建提交时不会有警告，所以你可以用它代替 form2request：把同一个 XPath 作为 formxpath 传入，把 token 放进 formdata。如果你改为让挑战走调度器，作为一个带独立回调的普通请求，就要给它设置 dont_filter，否则去重过滤器会丢掉第一次之后的每一次获取。

在低于 2.13 的 Scrapy 上，AsyncCapSkip 还需要在 settings.py 里把 TWISTED_REACTOR 设为 asyncio reactor。从 Scrapy 2.7 起生成的项目已经自带这一行。

常见错误及其含义

常见问题

简短版结论

要在 Scrapy 中识别 ALTCHA，先从 altcha-widget 元素上读出 challenge、challengejson 或 challengeurl 属性，以及 name 属性。用 download_async 加 dont_cache 获取挑战，让它沿用 spider 的会话，并且永远碰不到去重过滤器。把原样收到的 JSON 传给 AsyncCapSkip 的 altcha 方法并 await 它，然后 yield 一个 form2request 提交，数据里放上 token，并提高优先级。一个挑战，一个 token，一次提交；spider 在别处运行时，就用 Server 模式。

• ALTCHA 如何运作、识别工具覆盖哪些内容： ALTCHA 验证码识别页面.
• 整套爬虫技术栈中的验证码处理： 网络爬虫验证码识别页面.
• Python 包提供的所有方法： Python 验证码识别页面.

关于用量，最后再说一点。一个提交成千上万个表单的爬虫，要识别成千上万个挑战；而当 验证码识别工具 就运行在你自己的机器上时，这些花掉的是 CPU 时间，而不是按次计费的费用。