想让票价监控长期跑下去，脚本里必须有一步“遇到验证码就去识别”。旅行站点的风控比电商更敏感：同一条航线连查三四次，返回的往往是 reCAPTCHA 或 Cloudflare 挑战，而不是价格页。检测到挑战就把 sitekey 与 pageurl 交给 CaptchaAI，拿回 token 填进对应字段，再继续解析。
安全范围： 本文示例面向你自有或已获授权的采集场景。采集前请确认目标站点的 robots 协议与服务条款，并对齐《网络安全法》与 PIPL 的要求。
旅行站点会给你哪几类验证码
先确认要监控的 URL 落在哪一行，这决定脚本要写几个分支。
| 站点类型 | 常见验证码 | 处理难度 |
|---|---|---|
| 航空公司官网（直销） | reCAPTCHA v3、Cloudflare | 中 |
| OTA（Expedia、Booking） | reCAPTCHA v2、Turnstile | 中偏高 |
| 比价聚合（Google Flights、Kayak） | reCAPTCHA v3 | 中 |
| 酒店聚合平台 | Cloudflare Challenge | 高 |
国内 OTA 是另一套体系
携程、去哪儿几乎不用 reCAPTCHA，主流是 GeeTest（极验）滑块，参数是 gt 和 challenge，与 googlekey 不是一回事。CaptchaAI 支持 GeeTest v3；v4 的官方口径是即将支持，现在还不能用。另外 reCAPTCHA 依赖 Google 托管脚本，境内加载不稳定，调试时容易误判成超时。
把验证码识别嵌进抓取循环
脚本只做三件事：取页面、遇到挑战调 CaptchaAI 换 token、解析价格写入历史。提交走 in.php，轮询 res.php。token 落点不能混：reCAPTCHA 填回 g-recaptcha-response，Turnstile 填回 cf-turnstile-response。
import requests
import time
import re
import json
import os
from datetime import datetime, timedelta
API_KEY = os.environ["CAPTCHAAI_API_KEY"]
def solve_captcha(params):
params["key"] = API_KEY
resp = requests.get("https://ocr.captchaai.com/in.php", params=params)
if not resp.text.startswith("OK|"):
raise Exception(f"Submit: {resp.text}")
task_id = resp.text.split("|")[1]
for _ in range(60):
time.sleep(5)
result = requests.get("https://ocr.captchaai.com/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: {result.text}")
raise TimeoutError()
class FareMonitor:
def __init__(self):
self.session = requests.Session()
self.session.headers["User-Agent"] = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 Chrome/120.0.0.0"
)
self.history = []
def fetch_with_captcha(self, url):
"""Fetch a travel page, solving CAPTCHAs if encountered."""
resp = self.session.get(url)
# reCAPTCHA v2/v3
match = re.search(
r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text
)
if match:
site_key = match.group(1)
# Detect v3 vs v2
if "recaptcha/api.js?render=" in resp.text:
token = solve_captcha({
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
"version": "v3",
"action": "search",
})
else:
token = solve_captcha({
"method": "userrecaptcha",
"googlekey": site_key,
"pageurl": url,
})
resp = self.session.post(url, data={
"g-recaptcha-response": token,
})
# Cloudflare Turnstile
if "cf-turnstile" in resp.text:
match = re.search(
r'data-sitekey=["\']([^"\']+)', resp.text
)
if match:
token = solve_captcha({
"method": "turnstile",
"sitekey": match.group(1),
"pageurl": url,
})
resp = self.session.post(url, data={
"cf-turnstile-response": token,
})
return resp.text
def check_fares(self, routes):
"""Check fares for a list of routes."""
results = []
for route in routes:
try:
html = self.fetch_with_captcha(route["url"])
prices = self._extract_prices(html)
result = {
"route": f"{route['origin']}-{route['destination']}",
"date": route["date"],
"prices": prices,
"min_price": min(prices) if prices else None,
"timestamp": datetime.utcnow().isoformat(),
}
results.append(result)
self.history.append(result)
if prices:
print(f" {result['route']} ({route['date']}): "
f"${min(prices)}-${max(prices)}")
else:
print(f" {result['route']}: No prices found")
time.sleep(3) # Respectful delay
except Exception as e:
print(f" {route.get('origin', '?')}-"
f"{route.get('destination', '?')}: ERROR - {e}")
return results
def _extract_prices(self, html):
"""Extract prices from travel page HTML."""
prices = []
# Common price patterns
for match in re.finditer(
r'\$\s*([\d,]+(?:\.\d{2})?)', html
):
price = float(match.group(1).replace(",", ""))
if 20 < price < 10000: # Filter noise
prices.append(price)
return sorted(set(prices))
def detect_price_drops(self, threshold_pct=5):
"""Detect significant price drops in history."""
route_prices = {}
for entry in self.history:
key = f"{entry['route']}_{entry['date']}"
if key not in route_prices:
route_prices[key] = []
if entry["min_price"]:
route_prices[key].append(entry["min_price"])
alerts = []
for key, prices in route_prices.items():
if len(prices) >= 2:
prev = prices[-2]
current = prices[-1]
change_pct = ((current - prev) / prev) * 100
if change_pct < -threshold_pct:
alerts.append({
"route": key,
"previous": prev,
"current": current,
"change": f"{change_pct:.1f}%",
})
return alerts
def export_report(self, filename="fare_report.json"):
"""Export fare history to JSON."""
with open(filename, "w") as f:
json.dump(self.history, f, indent=2)
print(f"Exported {len(self.history)} fare checks to {filename}")
# Define routes to monitor
routes = [
{
"origin": "JFK",
"destination": "LAX",
"date": "2025-03-15",
"url": "https://example-airline.com/flights?from=JFK&to=LAX&date=2025-03-15",
},
{
"origin": "SFO",
"destination": "ORD",
"date": "2025-03-20",
"url": "https://example-airline.com/flights?from=SFO&to=ORD&date=2025-03-20",
},
]
monitor = FareMonitor()
results = monitor.check_fares(routes)
monitor.export_report()
三个容易踩的点
- 先分清 v2 和 v3。 页面里出现
recaptcha/api.js?render=就是 v3，要多带version和action；action与页面真实动作对不上，分数会偏低。 - 轮询节奏按类型给。 官方指标：reCAPTCHA v3 <4 秒，Turnstile <10 秒，GeeTest v3 <12 秒，v2 按 <60 秒预留；
CAPCHA_NOT_READY不是报错，继续轮询即可。 time.sleep(3)不要删。 请求压得越密，触发挑战的概率越高，省下的几秒会以识别耗时还回来。
价格由 JavaScript 渲染时，把取页面换成 Selenium 或 Playwright，识别流程不变。
票价监控多久跑一次：频率对准调价窗口
航司和 OTA 按批次刷新价格缓存，查得更勤很快进入收益递减区间。
# Check fares every 4 hours
0 */4 * * * cd /opt/fare-monitor && python fare_monitor.py
个人盯几条航线，4 到 6 小时一轮足够；做代客比价或分销，1 到 2 小时是上限。值得提频的只有几个窗口：出发前 21 天、14 天、7 天这几个调价节点，以及春运、国庆前后。给重点航线单独配一条更密的任务，比全量提频划算。
套餐怎么选：并发线程决定价格
CaptchaAI 按并发线程数计价，不按识别条数：套餐内次数不限，也不因验证码类型加价。所以该问的是“同一时刻最多几个请求卡在验证码上”。
| 监控规模 | 路线数 | 每条每天检查 | 峰值并发 | 对应套餐 |
|---|---|---|---|---|
| 个人 | 5 | 6 次 | 1–2 | BASIC（$15/月，5 线程） |
| 小型团队 | 50 | 4 次 | 5–10 | STANDARD（$30/月，15 线程） |
| 机构 / 分销 | 500 | 6 次 | 30–50 | ADVANCE（$90/月，50 线程） |
串行执行时并发恒为 1，50 条航线每 4 小时轮一遍也只占 1 个线程；线程池并行铺开才会顶到几十。先定并发模型，再选套餐。
常见问题
跑一段时间后验证码越来越频繁，是被拉黑了吗？
多半不是针对性封禁，而是风控的常规升级：同一会话反复查同一条航线，站点就会插入挑战。拉开间隔、把识别纳入重试流程即可继续。
token 拿到了，页面还是不放行，问题出在哪？
按三步查：pageurl 与实际请求的 URL 是否一致，v3 的 action 是否与页面动作对得上，token 是否填进了正确字段。错一项都会被判为未通过。
遇到 hCaptcha 怎么办？
CaptchaAI 目前不支持 hCaptcha 和 FunCaptcha（Arkose Labs），脚本应把这类页面标记为跳过，别反复重试占用线程。CaptchaFox、Friendly Captcha、Lemin 属于测试版（beta）支持。
监控 500 条航线，是不是必须上 VIP 套餐？
不需要。计费单位是并发线程，不是路线数量。500 条路线串行跑只占个位数线程；只有线程池同时发出几十条时，才需要 ADVANCE（$90/月，50 线程）。
