Мониторинг тарифов авиакомпаний с обработкой CAPTCHA

Сайты путешествий и авиакомпаний часто используют CAPTCHA, чтобы ограничить автоматическую проверку тарифов. Система мониторинга цен, которая проверяет тарифы на нескольких маршрутах, столкнется с проблемами reCAPTCHA, межстраничными объявлениями Cloudflare и механизмами ограничения скорости. CaptchaAI выполняет этап решения CAPTCHA, поэтому ваш конвейер мониторинга может продолжать собирать данные о тарифах.

В этом руководстве показано, как создать рабочий процесс мониторинга тарифов, который обнаруживает и решает CAPTCHA во время проверки цен.

Рабочий процесс мониторинга

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

Что вам нужно

pip install requests

Помощник по решению 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")

Мониторинг тарифов с обработкой 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))

Планирование проверок

Запускайте монитор по расписанию с помощью cron или планировщика задач:

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

Поиск неисправностей

Контрольный список сбора платы за проезд

• Соберите базовый тариф, налоги, валюту и семейство тарифов в одной полезной нагрузке, чтобы можно было использовать последующие сравнения.
• Сохраняйте как минимум два резервных селектора и сохраняйте необработанный снимок HTML в случае сбоя анализа.
• Записывайте маршрут, дату отправления, время сбора и задержку CAPTCHA при каждом событии решения.

Часто задаваемые вопросы

Как часто мне следует проверять тарифы?

Обычно каждые 4–8 часов. Более частые проверки увеличивают количество случаев CAPTCHA и затраты на прокси.

Какие типы CAPTCHA используют сайты авиакомпаний?

Чаще всего это страницы reCAPTCHA v2, Cloudflare Turnstile или Challenge, а иногда и CAPTCHA с изображениями.

Нужны ли мне авторизованный сетевой выход?

Да. Туристические сайты активно блокируют IP-адреса центров обработки данных. Резидентные или авторизованный сетевой выход имеют значительно более высокие показатели успеха.

Могу ли я контролировать несколько авиакомпаний?

Да. Настройте метод _parse_fare для формата ответа каждой авиакомпании и добавьте маршруты для каждого сайта.

Как мне обрабатывать страницы страница Cloudflare-защиты в staging?

Используйте method=turnstile с прокси. Возвращенный файл cookie qa_validation_cookie предоставляет доступ к сайту. См.Руководство по страница Cloudflare-защиты в staging.

Получите API-ключ CaptchaAI

Начните отслеживать тарифы авиакомпаний наcaptchaai.com. Автоматически обрабатывайте CAPTCHA в рабочем процессе отслеживания цен.

Связанные руководства

• Как решить reCAPTCHA v2 с помощью API
• Как решить Cloudflare Turnstile с помощью API
• Парсинг финансовых данных с обработкой CAPTCHA
• Рекомендации по настройке прокси-сервера