通过验证码处理监控航空公司票价

机票价格监控脚本最怕的不是网络波动，而是验证码——同一条航线短时间内被反复请求，多数航空公司网站就会弹出 reCAPTCHA v2 或 Cloudflare Turnstile。CaptchaAI 把求解做成一次 API 调用：识别类型、提交任务、拿到 token。本文用可运行的 Python 示例，演示如何检测并求解验证码、提交 token，再用定时任务长期采集票价。

真实场景：跨境比价为什么总撞上验证码

假设你在做跨境机票比价，每天要抓取上海–洛杉矶、北京–纽约等多条中美航线的实时票价。同一 IP 短时间内反复搜索，目标网站就会插入验证码——识别验证码因此是常规步骤，而不是异常分支。

监控流程一览

一次完整的票价检查包含四步：请求页面、判断验证码、必要时求解重试、解析存储。

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，其余请求直接进入解析环节。

环境准备

pip install requests

用 CaptchaAI 求解 reCAPTCHA v2 与 Turnstile

两个函数分别提交 reCAPTCHA v2 和 Turnstile 任务，再轮询 res.php 拿 token，遇到 CAPCHA_NOT_READY 就继续等，其他非 1 状态直接抛错。

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")

把验证码求解接入票价抓取逻辑

FareMonitor 检测页面里的 reCAPTCHA 或 Turnstile 标记，命中就求解并把 token 通过对应字段提交；_parse_fare 是占位实现，需按每家航空公司页面结构重写。

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:[email protected]: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

6 小时一次是常见起点，可按验证码触发率调整。

票价数据采集清单

• 把票价、税费、币种和运价家族存进同一条记录，方便后续比价。
• 保留两个选择器后备，解析失败时存原始 HTML 快照。
• 记录每次求解的航线、日期和验证码耗时。

常见故障排查

常见问题

检查频率多高比较合适？

每 4~8 小时一次即可，频率越高越容易触发验证码。

脚本一直报 TimeoutError 怎么排查？

先看轮询间隔和重试次数——reCAPTCHA v2 示例留了约 170 秒，站点求解慢时调大重试次数更有效。

验证码 token 能不能缓存复用？

不能，token 与 sitekey、页面上下文绑定，每条航线都要重新求解。

Cloudflare 验证挑战页面怎么处理？

用 method=cloudflare_challenge 提交任务并搭配自有服务器基础设施；返回的 qa_session_cookie 用于换取访问权限，详见 Cloudflare 验证挑战指南。

立即获取 CaptchaAI API 密钥

前往 CaptchaAI 官网 注册账号，把验证码求解接入你的机票监控流程。

相关阅读

• reCAPTCHA v2 API 求解教程
• Cloudflare Turnstile API 求解教程
• 用验证码处理做财务数据抓取
• 自有服务器基础设施配置最佳实践