Use Cases

用于拍卖站点监控的验证码处理

跑拍卖监控脚本时,最常见的中断不是网络超时,而是突然弹出的 reCAPTCHA v2。请求模式像脚本,拍卖平台就会拦一下。下面拆解触发规律,给出接入 CaptchaAI API 的 Python 与 JavaScript 示例。

验证码在拍卖网站的常见触发点

  • 搜索或浏览列表(reCAPTCHA v2)——短时间内连续搜索。
  • 查看商品详情(reCAPTCHA v2)——同一 IP 请求量偏高。
  • 查询出价历史(reCAPTCHA v2)——详情页反复加载。
  • 按分类浏览(Cloudflare Turnstile)——导航速度不像人类操作。
  • 价格提醒页面(reCAPTCHA v2)——频繁刷新。

先降低触发概率,再上自动识别

把 CaptchaAI 留给真正绕不开的场景,其余交给请求行为优化:

  • 复用会话 cookie:维持已验证状态。
  • 轮换自有服务器基础设施:分散出口 IP。
  • 请求间隔随机化:避免规律性模式。
  • 使用已登录账户:触发阈值更高。

Python 示例:搭建拍卖监控与验证码自动识别

AuctionMonitor 封装了识别流程:search_listings() 处理搜索,monitor_listing() 抓取出价,track_bids() 轮询并记录变化,检测到 reCAPTCHA 时自动完成识别并提交。

import requests
import time
import re
from datetime import datetime

class AuctionMonitor:
    def __init__(self, api_key):
        self.api_key = api_key
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def search_listings(self, auction_url, query, category=None):
        """Search auction listings, solving CAPTCHAs when triggered."""
        params = {"q": query}
        if category:
            params["category"] = category

        response = self.session.get(
            f"{auction_url}/search", params=params
        )

        if self._has_captcha(response.text):
            site_key = self._extract_site_key(response.text)
            token = self._solve_recaptcha(site_key, f"{auction_url}/search")
            response = self.session.post(
                f"{auction_url}/search",
                data={**params, "g-recaptcha-response": token}
            )

        return self._parse_listings(response.text)

    def monitor_listing(self, auction_url, listing_id):
        """Get current bid and listing details."""
        url = f"{auction_url}/item/{listing_id}"
        response = self.session.get(url)

        if self._has_captcha(response.text):
            site_key = self._extract_site_key(response.text)
            token = self._solve_recaptcha(site_key, url)
            response = self.session.post(url, data={
                "g-recaptcha-response": token
            })

        return self._parse_listing_detail(response.text)

    def track_bids(self, auction_url, listing_ids, interval=60):
        """Track bid changes across multiple listings."""
        history = {lid: [] for lid in listing_ids}

        while True:
            for listing_id in listing_ids:
                try:
                    detail = self.monitor_listing(auction_url, listing_id)
                    previous = history[listing_id]

                    if previous and detail["current_bid"] != previous[-1]["current_bid"]:
                        print(f"Bid change on {listing_id}: "
                              f"${previous[-1]['current_bid']} → ${detail['current_bid']}")

                    history[listing_id].append(detail)
                except Exception as e:
                    print(f"Error checking {listing_id}: {e}")

            time.sleep(interval)

    def _has_captcha(self, html):
        return "g-recaptcha" in html or "recaptcha" in html.lower()

    def _extract_site_key(self, html):
        match = re.search(r'data-sitekey="([^"]+)"', html)
        if match:
            return match.group(1)
        match = re.search(r"sitekey['\"]?\s*[:=]\s*['\"]([^'\"]+)", html)
        if match:
            return match.group(1)
        raise ValueError("Could not find reCAPTCHA site key")

    def _solve_recaptcha(self, site_key, page_url):
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": site_key,
            "pageurl": page_url,
            "json": 1
        })
        task_id = resp.json()["request"]

        for _ in range(60):
            time.sleep(3)
            result = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key,
                "action": "get",
                "id": task_id,
                "json": 1
            })
            data = result.json()
            if data["status"] == 1:
                return data["request"]

        raise TimeoutError("reCAPTCHA solve timed out")

    def _parse_listings(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        def text_or_none(node):
            return node.text.strip() if node and node.text else None

        def attr_or_none(node, attr):
            return node.get(attr) if node else None

        listings = []
        for item in soup.select(".listing-item, .auction-item"):
            listings.append({
                "title": text_or_none(item.select_one(".title")),
                "current_bid": text_or_none(item.select_one(".price, .bid")),
                "time_left": text_or_none(item.select_one(".time-left")),
                "url": attr_or_none(item.select_one("a"), "href")
            })
        return listings

    def _parse_listing_detail(self, html):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html, "html.parser")

        def text_or_none(node):
            return node.text.strip() if node and node.text else None

        return {
            "title": text_or_none(soup.select_one("h1, .item-title")),
            "current_bid": text_or_none(soup.select_one(".current-bid, .price")),
            "bid_count": text_or_none(soup.select_one(".bid-count")),
            "time_left": text_or_none(soup.select_one(".time-remaining")),
            "checked_at": datetime.now().isoformat()
        }

# Usage
monitor = AuctionMonitor("YOUR_API_KEY")
listings = monitor.search_listings(
    "https://auctions.example.com",
    "vintage electronics",
    category="collectibles"
)

JavaScript 示例:价格提醒系统的自动识别流程

AuctionTrackercheckAll() 判断出价是否逼近设定价位,命中就提醒;遇到验证码时 solveAndFetch() 自动识别再提交。

class AuctionTracker {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.watchList = new Map();
  }

  addWatch(listingId, url, maxPrice) {
    this.watchList.set(listingId, { url, maxPrice, history: [] });
  }

  async checkAll() {
    const alerts = [];

    for (const [id, watch] of this.watchList) {
      try {
        const detail = await this.fetchListing(watch.url);
        watch.history.push(detail);

        const price = parseFloat(detail.currentBid.replace(/[^0-9.]/g, ''));
        if (price >= watch.maxPrice * 0.9) {
          alerts.push({
            listing: id,
            price,
            threshold: watch.maxPrice,
            message: `Price approaching limit: $${price} / $${watch.maxPrice}`
          });
        }
      } catch (error) {
        alerts.push({ listing: id, error: error.message });
      }
    }

    return alerts;
  }

  async fetchListing(url) {
    const response = await fetch(url);
    const html = await response.text();

    if (html.includes('g-recaptcha')) {
      return this.solveAndFetch(url, html);
    }

    return this.parseDetail(html);
  }

  async solveAndFetch(url, html) {
    const siteKeyMatch = html.match(/data-sitekey="([^"]+)"/);
    if (!siteKeyMatch) throw new Error('No reCAPTCHA site key found');

    const submitResp = await fetch('https://ocr.captchaai.com/in.php', {
      method: 'POST',
      body: new URLSearchParams({
        key: this.apiKey,
        method: 'userrecaptcha',
        googlekey: siteKeyMatch[1],
        pageurl: url,
        json: '1'
      })
    });

    const { request: taskId } = await submitResp.json();

    for (let i = 0; i < 60; i++) {
      await new Promise(r => setTimeout(r, 3000));
      const result = await fetch(
        `https://ocr.captchaai.com/res.php?key=${this.apiKey}&action=get&id=${taskId}&json=1`
      );
      const data = await result.json();
      if (data.status === 1) {
        // Resubmit with token
        const response = await fetch(url, {
          method: 'POST',
          body: new URLSearchParams({ 'g-recaptcha-response': data.request })
        });
        return this.parseDetail(await response.text());
      }
    }

    throw new Error('reCAPTCHA solve timed out');
  }

  parseDetail(html) {
    // Parse auction listing details from HTML
    return {
      currentBid: html.match(/current.?bid[^>]*>([^<]+)/i)?.[1]?.trim(),
      bidCount: html.match(/(\d+)\s*bids?/i)?.[1],
      timeLeft: html.match(/time.?(?:left|remaining)[^>]*>([^<]+)/i)?.[1]?.trim(),
      checkedAt: new Date().toISOString()
    };
  }
}

// Usage
const tracker = new AuctionTracker('YOUR_API_KEY');
tracker.addWatch('item-123', 'https://auctions.example.com/item/123', 500);
tracker.addWatch('item-456', 'https://auctions.example.com/item/456', 200);
const alerts = await tracker.checkAll();

监控频率该怎么选

查得越勤,遇到验证码的概率越高,按用途选频率更省调用量:

  • 每 30 秒:截止前的实时出价,概率高,建议搭配自有服务器基础设施。
  • 每 5 分钟:持续跟踪进行中的拍卖,概率适中。
  • 每 15 分钟:关注列表的常规巡检,概率低。
  • 每小时:长期价格走势研究,概率极低。

常见故障排查

问题 原因 处理方式
每次请求都触发验证码 未保持会话 复用 requests.Session()
reCAPTCHA token 被拒绝 token 已过期(2 分钟有效期) 识别后立即提交,不缓存复用
列表页返回 0 条结果 验证码静默过滤了结果 检查隐藏的验证码元素
多次触发验证码后 IP 被封 超出速率限制 轮换自有服务器基础设施,拉长间隔

常见问题

CaptchaAI 识别 reCAPTCHA v2 要多久?

通常 60 秒以内完成,成功率较高。token 只有 2 分钟有效期,识别后应立即提交,不要缓存复用。

同时监控几十个商品,需要多少线程?

按并发线程计费:偶尔看几个商品,BASIC($15/月,5 线程)够用;分钟级并发几十个 listing,建议 ADVANCE($90/月,50 线程)。

国内网络环境监控海外拍卖网站,速度会受影响吗?

会,但那是网络可达性问题:reCAPTCHA 依赖 Google 托管脚本,内地网络加载可能不稳定,建议把脚本部署在海外服务器。

怎样才不会被判定为异常流量?

站点主要看请求模式。间隔随机化、轮换 User-Agent、避开低峰时段高频抓取都更有效。

相关文章

下一步

把拍卖监控做成自动化流程 —— 获取 CaptchaAI API 密钥,让 reCAPTCHA v2 识别自动完成。

该文章已禁用评论。