通过验证码处理进行电子商务价格监控

比价爬虫抓着抓着数据就断更了？十有八九不是网站改版，而是页面被验证码拦住了——脚本拿到的其实是验证码页面，却误以为抓取成功，数据库里悄悄留下一段价格空档。电商平台普遍把验证码当作反爬的第一道闸门。CaptchaAI 提供自动识别 reCAPTCHA、Cloudflare Turnstile 等验证码的 API，让脚本遇到验证码时不用人工介入，直接继续跑。

电商平台的验证码卡在哪

主流平台各自用不同验证码拦截高频请求：

不处理验证码，抓取任务会静默失败——不报错，只是数据没了，等发现价格曲线断了一截，往往已错过几个调价窗口。

请求频率、UA、请求间隔——三个信号叠加，命中风控阈值的概率明显升高。

一个真实场景：跨境卖家盯着海外平台的定价

跨境卖家常盯着亚马逊、沃尔玛的竞品定价，每 15-30 分钟刷一次。三个因素叠加，抓取比想象中更容易被拦：

• Google 托管脚本，国内访问本身不算稳定；
• 平台自己的机器人检测随请求量收紧；
• 固定刷新节奏本身就是自动化信号。

接入 CaptchaAI 后，遇到验证码自动提交识别，不用大改抓取代码。

整体架构：从定时调度到自动识别

思路不复杂：调度器把 URL 放进队列，worker 并发抓取，遇到验证码交给 CaptchaAI 识别后重试，没有就直接解析价格。这套架构解决两个问题：

• 单个 worker 卡住不拖慢整条队列；
• worker 数量可控，请求不会同一秒扎堆。

Scheduler (every 30 min)
    → URL Queue
        → Scraper Workers (5-10 concurrent)
            → Fetch page
            → CAPTCHA detected?
                → Yes → CaptchaAI → Solve → Retry page
                → No → Parse prices
            → Store in database
        → Alert on price changes

落地实现

整套逻辑拆成四步，跟上面的架构图一一对应：

• 检测：扫一遍页面，看是否命中 sitekey；
• 提交：把 sitekey、pageurl 发给 in.php；
• 轮询：每 5 秒查一次 res.php；
• 重试：带 token 重新提交即可。

Python 版本：价格监控器

下面这段代码把上面四步封装成几个函数，接进现有抓取脚本就能用：

import requests
import time
import re
import json
import os
from datetime import datetime

API_KEY = os.environ["CAPTCHAAI_API_KEY"]
BASE_URL = "https://ocr.captchaai.com"


def solve_captcha(method, params):
    params["key"] = API_KEY
    params["method"] = method

    resp = requests.get(f"{BASE_URL}/in.php", params=params)
    if not resp.text.startswith("OK|"):
        raise Exception(f"Submit failed: {resp.text}")

    task_id = resp.text.split("|")[1]

    for _ in range(60):
        time.sleep(5)
        result = requests.get(f"{BASE_URL}/res.php", params={
            "key": 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("CAPTCHA solve timed out")


def fetch_with_captcha(url, session):
    """Fetch a page, solving CAPTCHAs if encountered."""
    resp = session.get(url)

    # Check for reCAPTCHA
    match = re.search(r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text)
    if match:
        site_key = match.group(1)
        token = solve_captcha("userrecaptcha", {
            "googlekey": site_key,
            "pageurl": url,
        })
        resp = session.post(url, data={"g-recaptcha-response": token})

    # Check for Turnstile
    match = re.search(
        r'class="cf-turnstile"[^>]*data-sitekey=["\']([^"\']+)', resp.text
    )
    if match:
        site_key = match.group(1)
        token = solve_captcha("turnstile", {
            "sitekey": site_key,
            "pageurl": url,
        })
        resp = session.post(url, data={"cf-turnstile-response": token})

    return resp


def extract_price(html, selectors):
    """Extract price from HTML using regex patterns."""
    for pattern in selectors:
        match = re.search(pattern, html)
        if match:
            price_str = match.group(1).replace(",", "")
            return float(price_str)
    return None


def monitor_prices(products):
    """Monitor prices for a list of products."""
    session = requests.Session()
    session.headers["User-Agent"] = (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 Chrome/120.0.0.0"
    )

    results = []
    for product in products:
        try:
            resp = fetch_with_captcha(product["url"], session)
            price = extract_price(resp.text, product["selectors"])

            results.append({
                "name": product["name"],
                "url": product["url"],
                "price": price,
                "timestamp": datetime.utcnow().isoformat(),
                "status": "ok",
            })
            print(f"  {product['name']}: ${price}")

        except Exception as e:
            results.append({
                "name": product["name"],
                "url": product["url"],
                "price": None,
                "timestamp": datetime.utcnow().isoformat(),
                "status": f"error: {e}",
            })
            print(f"  {product['name']}: ERROR - {e}")

    return results


# Define products to monitor
products = [
    {
        "name": "Wireless Headphones",
        "url": "https://example.com/product/headphones",
        "selectors": [
            r'class="price"[^>]*>\$?([\d,]+\.?\d*)',
            r'itemprop="price" content="([\d.]+)"',
        ],
    },
    {
        "name": "Bluetooth Speaker",
        "url": "https://example.com/product/speaker",
        "selectors": [
            r'class="price"[^>]*>\$?([\d,]+\.?\d*)',
        ],
    },
]

print("Starting price check...")
results = monitor_prices(products)

# Save results
with open("prices.json", "w") as f:
    json.dump(results, f, indent=2)

Node.js 版本

检测、提交、轮询、重试四步逻辑不变，只是换成 axios + cheerio 实现。

const axios = require("axios");
const cheerio = require("cheerio");

const API_KEY = process.env.CAPTCHAAI_API_KEY;

async function solveCaptcha(method, params) {
  params.key = API_KEY;
  params.method = method;

  const submit = await axios.get("https://ocr.captchaai.com/in.php", {
    params,
  });
  const taskId = String(submit.data).split("|")[1];

  for (let i = 0; i < 60; i++) {
    await new Promise((r) => setTimeout(r, 5000));
    const poll = await axios.get("https://ocr.captchaai.com/res.php", {
      params: { key: API_KEY, action: "get", id: taskId },
    });
    const text = String(poll.data);
    if (text === "CAPCHA_NOT_READY") continue;
    if (text.startsWith("OK|")) return text.split("|").slice(1).join("|");
    throw new Error(text);
  }
  throw new Error("Timeout");
}

async function monitorPrice(url) {
  const resp = await axios.get(url);
  const $ = cheerio.load(resp.data);

  // Check for reCAPTCHA
  const siteKey = $(".g-recaptcha").attr("data-sitekey");
  if (siteKey) {
    const token = await solveCaptcha("userrecaptcha", {
      googlekey: siteKey,
      pageurl: url,
    });
    // Re-fetch with token
    const formResp = await axios.post(url, { "g-recaptcha-response": token });
    return cheerio.load(formResp.data);
  }

  const price = $('[itemprop="price"]').attr("content") || $(".price").text();
  return parseFloat(price.replace(/[^0-9.]/g, ""));
}

调度：多久跑一次合适

两种主流做法：

• cron 定时任务：适合大多数场景，每 30 分钟跑一次足够；
• 常驻进程：脚本一直跑，用 schedule 库自行控制节奏。

# crontab -e
*/30 * * * * cd /opt/monitor && python price_monitor.py >> /var/log/prices.log 2>&1

import schedule

schedule.every(30).minutes.do(lambda: monitor_prices(products))

while True:
    schedule.run_pending()
    time.sleep(60)

成本怎么算

验证码识别按次计费，费用随监控频率和商品数量放大：

实际花费通常低于表中估算，原因有两个：

• 不是每次刷新都会碰到验证码，命中率随平台和时段波动；
• 综合下来通常比估算低 50%-70%。

常见问题

抓取频率怎么设置合理？

取决于波动速度：大促类目 15-30 分钟一次，长尾商品每小时甚至更低频也够用，按类目分层设置即可。

验证码识别失败要不要重试？重试节奏怎么设计？

要重试，但别立即重试。轮询直到有结果，出错时用指数退避（5 秒、15 秒、45 秒）。

reCAPTCHA 和 Cloudflare Turnstile 混用，一套代码能一起处理吗？

可以。按类型分别检测 sitekey，method 参数不同（userrecaptcha / turnstile），提交、轮询、拿 token 的流程完全一样。

同时监控多个店铺和平台，会不会触发更严格的验证码策略？

并发量越高，越容易被判定为异常流量。给不同平台的请求错开时间，比拉满并发更稳；验证码触发变多也没关系。

国内网络下监控海外平台，超时要怎么设置？

超时设置宽松一些，几秒到十几秒的等待是正常的链路波动，跟识别环节无关。

相关指南

• 网页抓取中的验证码处理
• 抓取不被封锁的实战经验
• 市场调研数据采集指南