跑拍卖监控脚本时,最常见的中断不是网络超时,而是突然弹出的 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 示例:价格提醒系统的自动识别流程
AuctionTracker 的 checkAll() 判断出价是否逼近设定价位,命中就提醒;遇到验证码时 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、避开低峰时段高频抓取都更有效。
相关文章
- reCAPTCHA v2 回调识别教程
- reCAPTCHA v2 与 Turnstile 同站处理
- reCAPTCHA v2 回调机制详解
下一步
把拍卖监控做成自动化流程 —— 获取 CaptchaAI API 密钥,让 reCAPTCHA v2 识别自动完成。