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

航班状态监控需要在多个航空公司和机场门户网站上进行频繁检查。这些门户使用 Cloudflare Turnstile、reCAPTCHA 和自定义验证码保护其实时数据，尤其是当它们检测到重复的自动查询时。以下是如何在构建可靠的航班跟踪工具的同时处理验证码。

验证码出现的位置

飞行监控架构

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

多航空公司监控 (JavaScript)

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());
  }
}

监控频率和验证码率

会话优化

通过维护会话状态来减少验证码遭遇：

故障排除

常问问题

我应该多久检查一次航班状态？

通常每 5-15 分钟一次。更频繁的检查会触发更多验证码，并可能导致 IP 封锁。CaptchaAI 处理 Turnstile 的成功率为 100%，因此限制因素是门户的速率限制，而不是验证码解决。

我可以同时监控多家航空公司的航班吗？

是的。每个航空公司使用单独的会话，并为每个航空公司独立解决验证码。 CaptchaAI 处理跨不同站点的并发请求。

航空公司移动 API 是否有验证码？

移动 API 通常使用不同的身份验证（API 密钥、OAuth）而不是验证码。但是，他们服务的 Web 端点可能仍具有 Cloudflare 保护。

相关文章

• Geetest 与 Cloudflare Turnstile 比较
• Cloudflare Turnstile 403 令牌修复后
• Cloudflare Turnstile 小部件模式解释

下一步

建立可靠的飞行监控 -获取您的 CaptchaAI API 密钥并自动处理航空公司验证码。