Use Cases

用于航班状态监控的验证码处理

跨境机票代理、差旅平台和物流团队最常撞见的难题,就是航空公司官网在识别出高频轮询后甩出的 Cloudflare Turnstile 或 reCAPTCHA——不同于国内更常见的极验(GeeTest),海外航司站点大多靠它们挡自动化流量。

思路很简单:控制请求节奏、复用会话,把验证码当成流程里的正常环节,而不是意外。

验证码会卡在哪几个环节

  • 航空公司航班状态页——Cloudflare Turnstile,同一 IP 请求过于频繁
  • 机场到达/出发信息板——Cloudflare 验证流程,触发机器人检测
  • 航班搜索引擎——reCAPTCHA v2/v3,提交搜索表单
  • 订票状态查询——reCAPTCHA v2,展示行程前的校验
  • API 限流页面——自定义验证码,超出请求频率限制

监控脚本的基础架构

FlightMonitor 是方案骨架:发请求、判断是否命中验证码、分流去解,再解析成结构化数据:

import requests
import time
from datetime import datetime

class FlightMonitor:
    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 check_flight(self, airline_url, flight_number):
        """Check flight status, handling CAPTCHAs if encountered."""
        response = self.session.get(
            f"{airline_url}/flight-status/{flight_number}"
        )

        if self._is_captcha_page(response):
            response = self._solve_and_retry(response, airline_url)

        return self._parse_flight_data(response.text)

    def _is_captcha_page(self, response):
        return (
            response.status_code == 403 or
            "cf-turnstile" in response.text or
            "g-recaptcha" in response.text
        )

    def _solve_and_retry(self, response, url):
        import re

        # Detect CAPTCHA type
        if "cf-turnstile" in response.text:
            match = re.search(r'data-sitekey="(0x[^"]+)"', response.text)
            token = self._solve_turnstile(match.group(1), url)
            field = "cf-turnstile-response"
        else:
            match = re.search(r'data-sitekey="([^"]+)"', response.text)
            token = self._solve_recaptcha(match.group(1), url)
            field = "g-recaptcha-response"

        return self.session.post(url, data={field: token})

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

    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"]
        return self._poll_result(task_id)

    def _poll_result(self, task_id):
        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("CAPTCHA solve timed out")

    def _parse_flight_data(self, html):
        # Parse flight status from 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 {
            "status": text_or_none(soup.select_one(".flight-status")),
            "departure": text_or_none(soup.select_one(".departure-time")),
            "arrival": text_or_none(soup.select_one(".arrival-time")),
            "gate": text_or_none(soup.select_one(".gate-info")),
            "checked_at": datetime.now().isoformat()
        }

让脚本按周期自动巡检

架子搭好后,按固定周期跑起来,状态变化时提示一下:

def monitor_flight(monitor, airline_url, flight_number, 
                   interval_seconds=300, max_checks=48):
    """Monitor a flight every N seconds, handling CAPTCHAs as needed."""
    history = []

    for check_num in range(max_checks):
        try:
            status = monitor.check_flight(airline_url, flight_number)
            history.append(status)

            # Alert on changes
            if len(history) > 1 and status["status"] != history[-2]["status"]:
                print(f"Status changed: {history[-2]['status']} → {status['status']}")

            print(f"Check {check_num + 1}: {status['status']} "
                  f"(Gate: {status.get('gate', 'Coming soon')})")

        except Exception as e:
            print(f"Check {check_num + 1} failed: {e}")

        time.sleep(interval_seconds)

    return history

# Usage
monitor = FlightMonitor("YOUR_API_KEY")
monitor_flight(monitor, "https://airline.example.com", "AA1234")

Node.js 版本:并行盯多家航司

监控多家航空公司时,用一个 Map 管理各自的会话和历史更清晰:

class FlightTracker {
  constructor(apiKey) {
    this.apiKey = apiKey;
    this.flights = new Map();
  }

  async addFlight(airline, flightNumber, checkUrl) {
    this.flights.set(flightNumber, {
      airline,
      url: checkUrl,
      history: [],
      lastCheck: null
    });
  }

  async checkAll() {
    const results = [];

    for (const [flightNum, flight] of this.flights) {
      try {
        const status = await this.checkFlight(flight.url, flightNum);
        flight.history.push(status);
        flight.lastCheck = new Date();
        results.push({ flight: flightNum, ...status });
      } catch (error) {
        results.push({ flight: flightNum, error: error.message });
      }
    }

    return results;
  }

  async checkFlight(url, flightNumber) {
    const response = await fetch(`${url}/status/${flightNumber}`);
    const html = await response.text();

    // Check for CAPTCHA
    if (html.includes('cf-turnstile') || response.status === 403) {
      return this.solveAndRetry(url, flightNumber, html);
    }

    return this.parseStatus(html);
  }

  async solveAndRetry(url, flightNumber, html) {
    const siteKeyMatch = html.match(/data-sitekey="(0x[^"]+)"/);
    if (!siteKeyMatch) throw new Error('No sitekey found');

    const token = await this.solveTurnstile(siteKeyMatch[1], url);

    const response = await fetch(`${url}/status/${flightNumber}`, {
      method: 'POST',
      body: new URLSearchParams({ 'cf-turnstile-response': token })
    });

    return this.parseStatus(await response.text());
  }
}

先把会话稳住,再谈检查频率

验证码大多是被请求模式触发,而非单次请求。

把这几件事做对,遇到验证码的概率会明显下降:

  • 保留 cookie——qa_session_cookie 一般有效 15–30 分钟
  • 固定 User-Agent——频繁更换 UA 易触发新挑战
  • 保持出口 IP 一致——同一 IP 更不易被判定可疑
  • 均匀分布请求节奏——突发模式容易撞上限流

检查频率怎么选

间隔越短,验证码出现的概率越高:

  • 每 1 分钟——高(50–80%),过于激进
  • 每 5 分钟——中等(10–30%),优先级航班适用
  • 每 15 分钟——低(5–10%),日常监控平衡点
  • 每 30 分钟——很低(<5%),长期跟踪首选
  • 每小时——极低(<1%),基本不会触发

常见故障排查

  • 每次检查都遇到验证码——会话未复用,跨请求复用同一个 requests.Session()
  • Cloudflare 拦截(错误码 1020)——请求过于密集,增大检查间隔
  • 验证码解出后数据依旧过期——求解耗时过长,token 已过期,缩短等待窗口,即时求解
  • 抓到的数据和浏览器不一致——页面依赖 JavaScript 渲染,重 JS 站点改用浏览器自动化

常见问题

轮询间隔多长比较合适?

5–15 分钟比较常见。

间隔太短会推高触发率、招来 IP 封锁;真正卡吞吐量的往往是站点的速率限制,而非验证码求解——CaptchaAI 处理 Turnstile 通常 10 秒内完成,成功率较高。

一直卡在 Cloudflare Turnstile 怎么办?

先确认页面确实是 Turnstile(data-sitekey0x 开头),再检查会话是否复用——每次都新建 session 几乎必然反复被挑战。

requests.Session() 跨请求保留即可。

会话多久需要刷新一次?

qa_session_cookie 一般 15–30 分钟后失效,之后大概率重新触发验证码——这也是间隔控制在 5–15 分钟更划算的原因。

能同时监控多家航空公司吗?

可以,给每家分配独立 session、独立求解验证码,互不影响,CaptchaAI 支持跨站点并发请求。

相关文章

下一步

把监控脚本从“时不时被验证码打断”变成“稳定巡检”——注册 CaptchaAI 拿到 API Key,按上面的架构接入即可。

该文章已禁用评论。