旅游和航空公司网站经常使用验证码来限制自动票价检查。检查多条路线票价的价格监控系统将遇到 reCAPTCHA 挑战、Cloudflare 插页式广告和速率限制机制。 CaptchaAI 处理验证码解决步骤，以便您的监控管道可以继续收集票价数据。
本指南展示了如何构建票价监控工作流程，以在价格检查期间检测和解决验证码问题。
监控工作流程
Schedule check → Request fare page → CAPTCHA detected?
↓ Yes
Solve via CaptchaAI → Inject token → Retry request
↓ No
Parse fare data → Store → Alert on price change
你需要什么
| 要求 | 细节 |
|---|---|
| CaptchaAI API 密钥 | 验证码网站 |
| Python 3.8+ | 使用requests |
| 代理人 | 旅游网站的自有服务器基础设施 |
pip install requests
CaptchaAI 求解器助手
import requests
import time
API_KEY = "YOUR_API_KEY"
def solve_recaptcha_v2(sitekey, pageurl):
"""Solve reCAPTCHA v2 and return the token."""
submit = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY, "method": "userrecaptcha",
"googlekey": sitekey, "pageurl": pageurl, "json": 1
}).json()
if submit.get("status") != 1:
raise RuntimeError(f"Submit error: {submit.get('request')}")
task_id = submit["request"]
time.sleep(20)
for _ in range(30):
result = requests.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get", "id": task_id, "json": 1
}).json()
if result.get("status") == 1:
return result["request"]
if result.get("request") != "CAPCHA_NOT_READY":
raise RuntimeError(f"Solve error: {result['request']}")
time.sleep(5)
raise TimeoutError("Solve timed out")
def solve_turnstile(sitekey, pageurl):
"""Solve Cloudflare Turnstile and return the token."""
submit = requests.post("https://ocr.captchaai.com/in.php", data={
"key": API_KEY, "method": "turnstile",
"sitekey": sitekey, "pageurl": pageurl, "json": 1
}).json()
if submit.get("status") != 1:
raise RuntimeError(f"Submit error: {submit.get('request')}")
task_id = submit["request"]
time.sleep(10)
for _ in range(30):
result = requests.get("https://ocr.captchaai.com/res.php", params={
"key": API_KEY, "action": "get", "id": task_id, "json": 1
}).json()
if result.get("status") == 1:
return result["request"]
if result.get("request") != "CAPCHA_NOT_READY":
raise RuntimeError(f"Solve error: {result['request']}")
time.sleep(5)
raise TimeoutError("Solve timed out")
通过验证码处理进行票价监控
import json
from datetime import datetime
class FareMonitor:
def __init__(self, proxy=None):
self.session = requests.Session()
if proxy:
self.session.proxies = {
"http": f"http://{proxy}",
"https": f"http://{proxy}"
}
self.fare_history = {}
def check_fare(self, route):
"""Check fare for a route, solving CAPTCHAs if needed."""
url = route["url"]
response = self.session.get(url)
# Detect CAPTCHA in response
if self._has_recaptcha(response.text):
sitekey = self._extract_sitekey(response.text)
token = solve_recaptcha_v2(sitekey, url)
response = self.session.post(url, data={
"g-recaptcha-response": token,
**route.get("params", {})
})
elif self._has_turnstile(response.text):
sitekey = self._extract_turnstile_key(response.text)
token = solve_turnstile(sitekey, url)
response = self.session.post(url, data={
"cf-turnstile-response": token,
**route.get("params", {})
})
return self._parse_fare(response.text, route)
def _has_recaptcha(self, html):
return "g-recaptcha" in html or "recaptcha/api" in html
def _has_turnstile(self, html):
return "cf-turnstile" in html or "turnstile" in html
def _extract_sitekey(self, html):
# Extract data-sitekey from reCAPTCHA div
if 'data-sitekey="' in html:
start = html.index('data-sitekey="') + 14
end = html.index('"', start)
return html[start:end]
return None
def _extract_turnstile_key(self, html):
if 'data-sitekey="' in html:
idx = html.index("cf-turnstile")
start = html.index('data-sitekey="', idx) + 14
end = html.index('"', start)
return html[start:end]
return None
def _parse_fare(self, html, route):
"""Parse fare data from the response. Customize per target site."""
# Placeholder — implement per site
return {
"route": route["name"],
"timestamp": datetime.now().isoformat(),
"raw_length": len(html)
}
def monitor_routes(self, routes):
"""Check all routes and report price changes."""
results = []
for route in routes:
try:
fare = self.check_fare(route)
results.append(fare)
print(f"[OK] {route['name']}: checked")
except Exception as e:
print(f"[ERROR] {route['name']}: {e}")
return results
# Usage
routes = [
{
"name": "NYC-LAX",
"url": "https://example-airline.com/search?from=JFK&to=LAX&date=2025-08-15",
"params": {"adults": 1}
},
{
"name": "SFO-ORD",
"url": "https://example-airline.com/search?from=SFO&to=ORD&date=2025-08-20",
"params": {"adults": 1}
}
]
monitor = FareMonitor(proxy="user:pass@proxy.example.com:8080")
results = monitor.monitor_routes(routes)
for r in results:
print(json.dumps(r, indent=2))
安排检查
使用 cron 或任务调度程序按计划运行监视器：
# Check fares every 6 hours
0 */6 * * * cd /path/to/project && python fare_monitor.py >> /var/log/fares.log 2>&1
票价捕获清单
- Capture base fare, taxes, currency, and fare family in one payload so later comparisons stay usable.
- 至少保留两个选择器后备，并在解析失败时存储原始 HTML 快照。
- 记录每个解决事件的路线、出发日期、集合时间和验证码延迟。
故障排除
| 问题 | 原因 | 处理方式 |
|---|---|---|
| 频繁验证码 | 来自同一 IP 的请求过多 | 使用轮换自有服务器基础设施 |
| 陈旧的价格 | 缓存页面 | 添加缓存破坏标头或随机化请求参数 |
| IP 被封锁 | 速率限制 | 增加检查之间的延迟，QA 测试会话 |
| 验证码解决失败 | 站点密钥提取错误 | 验证 sitekey 与页面上的验证码匹配 |
常问问题
我应该多久检查一次票价？
通常每 4 至 8 小时检查一次。更频繁的检查会增加验证码遇到次数和代理成本。
航空公司网站使用哪些验证码类型？
最常见的是 reCAPTCHA v2、Cloudflare Turnstile 或挑战页面，偶尔也会使用图像验证码。
我需要自有服务器基础设施吗？
是的。旅游网站主动阻止数据中心 IP。住宅或自有服务器基础设施的成功率明显更高。
我可以监控多家航空公司吗？
是的。为每个航空公司的响应格式自定义 _parse_fare 方法，并为每个站点添加路线。
如何处理 Cloudflare 验证流程 页面？
将 method=cloudflare_challenge 与代理一起使用。返回的 qa_session_cookie cookie 授予对站点的访问权限。请参阅Cloudflare 验证流程指南。
获取您的 CaptchaAI API 密钥
开始监控机票价格验证码网站。在价格跟踪工作流程中自动处理验证码。
相关指南
- 如何使用API解决reCAPTCHA v2
- 如何使用API解决Cloudflare Turnstile
- 使用验证码处理进行财务数据抓取
- 代理设置最佳实践
