Xử lý CAPTCHA để theo dõi trạng thái chuyến bay

Giám sát tình trạng chuyến bay yêu cầu kiểm tra thường xuyên trên nhiều cổng thông tin của hãng hàng không và sân bay. Các cổng này bảo vệ dữ liệu thời gian thực của họ bằng Cloudflare Turnstile, reCAPTCHA và CAPTCHA tùy chỉnh — đặc biệt khi chúng phát hiện các truy vấn tự động lặp lại. Dưới đây là cách xử lý CAPTCHA trong khi xây dựng các công cụ theo dõi chuyến bay đáng tin cậy.

Nơi CAPTCHA xuất hiện

Kiến trúc giám sát chuyến bay

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

Giám sát định kỳ bằng cách xử lý CAPTCHA

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

Giám sát nhiều hãng hàng không (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());
  }
}

Giám sát tần suất và tỷ lệ CAPTCHA

Tối ưu hóa phiên

Giảm số lần gặp CAPTCHA bằng cách duy trì trạng thái phiên:

Khắc phục sự cố

Câu hỏi thường gặp

Tôi nên kiểm tra tình trạng chuyến bay bao lâu một lần?

Thông thường cứ 5–15 phút một lần. Việc kiểm tra thường xuyên hơn sẽ kích hoạt nhiều CAPTCHA hơn và có thể dẫn đến chặn IP. CaptchaAI xử lý Turnstile với tỷ lệ thành công 100%, vì vậy yếu tố hạn chế là giới hạn tốc độ của cổng chứ không phải giải CAPTCHA.

Tôi có thể theo dõi các chuyến bay từ nhiều hãng hàng không cùng một lúc không?

Vâng. Sử dụng các phiên riêng biệt cho mỗi hãng hàng không và giải CAPTCHA một cách độc lập cho từng hãng. CaptchaAI xử lý các yêu cầu đồng thời trên các trang web khác nhau.

API di động của hãng hàng không có CAPTCHA không?

API di động thường sử dụng xác thực khác nhau (khóa API, OAuth) thay vì CAPTCHA. Tuy nhiên, các điểm cuối web mà chúng phục vụ vẫn có thể được bảo vệ bằng Cloudflare.

bài viết liên quan

• So sánh Geetest và Cloudflare Turnstile
• Cloudflare Turnstile 403 Sau khi sửa mã thông báo
• Giải thích về các chế độ widget Cloudflare Turnstile

Các bước tiếp theo

Xây dựng hệ thống giám sát chuyến bay đáng tin cậy —lấy khóa API CaptchaAI của bạnvà tự động xử lý CAPTCHA của hãng hàng không.