Giám sát giá vé máy bay bằng cách xử lý CAPTCHA

Các trang web du lịch và hàng không thường xuyên sử dụng CAPTCHA để hạn chế việc kiểm tra giá vé tự động. Hệ thống giám sát giá kiểm tra giá vé trên nhiều tuyến đường sẽ gặp phải thách thức reCAPTCHA, quảng cáo xen kẽ Cloudflare và cơ chế giới hạn tỷ lệ. CaptchaAI xử lý bước giải CAPTCHA để quy trình giám sát của bạn có thể tiếp tục thu thập dữ liệu giá vé.

Hướng dẫn này chỉ ra cách xây dựng quy trình giám sát giá vé nhằm phát hiện và giải CAPTCHA trong quá trình kiểm tra giá.

Quy trình làm việc giám sát

Schedule check → Request fare page → CAPTCHA detected?
                                         ↓ Yes
                                    Solve via CaptchaAI → Inject token → Retry request
                                         ↓ No
                                    Parse fare data → Store → Alert on price change

Những gì bạn cần

Yêu cầu	Chi tiết
Khóa API CaptchaAI	captchaai.com
Python 3.8+	Với `requests`
Ủy quyền	đa dạng nguồn yêu cầu cho các trang web du lịch

pip install requests

Trình trợ giúp giải CaptchaAI

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

Giám sát giá vé bằng cách xử lý CAPTCHA

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:pass@proxy.example.com:8080")
results = monitor.monitor_routes(routes)

for r in results:
    print(json.dumps(r, indent=2))

Lập kế hoạch kiểm tra

Chạy màn hình theo lịch bằng cron hoặc trình lập lịch tác vụ:

# Check fares every 6 hours
0 */6 * * * cd /path/to/project && python fare_monitor.py >> /var/log/fares.log 2>&1

Danh sách kiểm tra thu thập giá vé

Ghi lại giá vé cơ sở, thuế, đơn vị tiền tệ và nhóm giá vé trong một tải trọng để các phép so sánh sau này vẫn có thể sử dụng được.
Giữ ít nhất hai dự phòng bộ chọn và lưu trữ ảnh chụp nhanh HTML thô bất cứ khi nào phân tích cú pháp không thành công.
Ghi lại lộ trình, ngày khởi hành, thời gian thu thập và độ trễ CAPTCHA với mỗi sự kiện giải quyết.

Khắc phục sự cố

Vấn đề	Nguyên nhân	Cách xử lý
CAPTCHA thường xuyên	Quá nhiều yêu cầu từ cùng một IP	Sử dụng đa dạng nguồn yêu cầu luân phiên
Giá cũ	Các trang được lưu trong bộ nhớ đệm	Thêm tiêu đề chặn bộ nhớ đệm hoặc ngẫu nhiên hóa các tham số yêu cầu
IP bị chặn	Giới hạn tỷ lệ	Tăng độ trễ giữa các lần kiểm tra, xoay proxy
Giải quyết CAPTCHA không thành công	Trích xuất khóa trang web sai	Xác minh khóa trang web khớp với CAPTCHA trên trang

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

Tôi nên kiểm tra giá vé bao lâu một lần?

Thông thường là 4–8 giờ một lần. Việc kiểm tra thường xuyên hơn sẽ làm tăng số lần gặp CAPTCHA và chi phí proxy.

Các trang web hàng không sử dụng loại CAPTCHA nào?

Phổ biến nhất là các trang reCAPTCHA v2, Cloudflare Turnstile hoặc Thử thách và đôi khi có hình ảnh CAPTCHA.

Tôi có cần đa dạng nguồn yêu cầu không?

Vâng. Các trang web du lịch tích cực chặn IP của trung tâm dữ liệu. đa dạng nguồn yêu cầu hoặc di động có tỷ lệ thành công cao hơn đáng kể.

Tôi có thể giám sát nhiều hãng hàng không?

Vâng. Tùy chỉnh phương pháp _parse_fare cho định dạng phản hồi của từng hãng hàng không và thêm đường bay cho từng trang web.

Làm cách nào để xử lý các trang Cloudflare Challenge?

Sử dụng method=cloudflare_challenge với proxy. Cookie <staging-session-cookie> được trả về sẽ cấp quyền truy cập vào trang web. XemHướng dẫn Cloudflare Challenge.

Nhận khóa API CaptchaAI của bạn

Bắt đầu theo dõi giá vé máy bay tạicaptchaai.com. Tự động xử lý CAPTCHA trong quy trình theo dõi giá của bạn.

Hướng dẫn liên quan

Cách giải reCAPTCHA v2 bằng API
Cách giải Cloudflare Turnstile bằng API
Quét dữ liệu tài chính bằng cách xử lý CAPTCHA
Các phương pháp hay nhất về thiết lập proxy