Мониторинг стоимости поездок с обработкой CAPTCHA

Мониторинг тарифов ломается не на разборе HTML, а на первой же проверке CAPTCHA: скрипт забирает не таблицу рейсов, а страницу с проверкой, и в истории цен появляется дыра. Решается это одним шагом — токен запрашивается у CaptchaAI через API и подставляется в тот же запрос, после чего сбор продолжается без ручного вмешательства. Ниже — рабочий монитор на Python, который различает reCAPTCHA v2, reCAPTCHA v3 и Cloudflare Turnstile, хранит историю цен и отмечает падения тарифа.

Где именно рассыпается сбор тарифов

Почти у любого монитора цен есть три типовые точки отказа, и библиотекой для парсинга закрывается только одна из них:

1. Проверка CAPTCHA появляется нерегулярно. На десяти запросах её может не быть вообще, а на одиннадцатом приходит страница с виджетом. Код должен уметь обработать оба случая в одной ветке логики.
2. Цена рисуется на клиенте. Многие метапоисковики отдают пустой каркас страницы, а числа подгружают отдельными запросами.
3. Частота опроса выбрана наугад. Слишком редко — вы пропускаете короткие акции; слишком часто — растёт число проверок и время работы монитора без выигрыша в точности.

Порядок работы тоже важен: сначала разберитесь, какая именно проверка стоит на целевой странице, и только потом пишите извлечение цен.

Какие проверки CAPTCHA встречаются на туристических сайтах

Все перечисленные типы — reCAPTCHA v2 и v3, Cloudflare Turnstile, Cloudflare Challenge, графическая (OCR) и текстовая CAPTCHA, GeeTest v3 — входят в поддерживаемые CaptchaAI. Если на сайте стоит тип из другого списка (например, hCaptcha или FunCaptcha), мониторинг этого источника придётся строить иначе: эти два типа сервис не поддерживает. Проверить тип стоит до написания парсера: от него зависят метод задачи и имя поля, в которое подставляется токен.

Сценарий: турагентство из Алматы и 40 направлений

Типичная задача для небольшого агентства: есть 40 пар «город — дата», которые нужно перепроверять несколько раз в сутки, чтобы менеджер утром видел актуальную витрину, а не вчерашние цифры. Команда из двух разработчиков держит монитор на одном арендованном сервере в европейском регионе и запускает его по cron.

При четырёх проверках на маршрут это около 160 загрузок страниц в сутки, из которых реальная проверка CAPTCHA выпадает далеко не каждый раз. Потоков при этом нужно немного: задачи идут последовательно, и параллельно в любой момент времени решается одна-две проверки. Формально такому монитору хватило бы и BASIC ($15/мес, 5 потоков); STANDARD ($30/мес, 15 потоков) берут с запасом — чтобы позже развести маршруты по нескольким воркерам и не упереться в потолок. Счёт при этом не вырастет, даже если проверок станет вдвое больше: оплачиваются потоки, а не отдельные решения. Фиксированная месячная сумма в USD удобна и для планирования бюджета: вы заранее знаете потолок расходов на месяц.

Монитор тарифов на Python

Класс ниже делает четыре вещи: загружает страницу маршрута, определяет тип проверки по разметке, получает токен через in.php/res.php и складывает цены в историю. Обратите внимание на две детали: разделение v2 и v3 по наличию recaptcha/api.js?render= и разные имена полей для токена — g-recaptcha-response для reCAPTCHA и cf-turnstile-response для Turnstile. API-ключ берётся из переменной окружения, а не из исходного кода.

import requests
import time
import re
import json
import os
from datetime import datetime, timedelta

API_KEY = os.environ["CAPTCHAAI_API_KEY"]


def solve_captcha(params):
    params["key"] = API_KEY
    resp = requests.get("https://ocr.captchaai.com/in.php", params=params)
    if not resp.text.startswith("OK|"):
        raise Exception(f"Submit: {resp.text}")

    task_id = resp.text.split("|")[1]
    for _ in range(60):
        time.sleep(5)
        result = requests.get("https://ocr.captchaai.com/res.php", params={
            "key": API_KEY, "action": "get", "id": task_id,
        })
        if result.text == "CAPCHA_NOT_READY":
            continue
        if result.text.startswith("OK|"):
            return result.text.split("|", 1)[1]
        raise Exception(f"Solve: {result.text}")
    raise TimeoutError()


class FareMonitor:
    def __init__(self):
        self.session = requests.Session()
        self.session.headers["User-Agent"] = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/120.0.0.0"
        )
        self.history = []

    def fetch_with_captcha(self, url):
        """Fetch a travel page, solving CAPTCHAs if encountered."""
        resp = self.session.get(url)

        # reCAPTCHA v2/v3
        match = re.search(
            r'data-sitekey=["\']([A-Za-z0-9_-]+)["\']', resp.text
        )
        if match:
            site_key = match.group(1)

            # Detect v3 vs v2
            if "recaptcha/api.js?render=" in resp.text:
                token = solve_captcha({
                    "method": "userrecaptcha",
                    "googlekey": site_key,
                    "pageurl": url,
                    "version": "v3",
                    "action": "search",
                })
            else:
                token = solve_captcha({
                    "method": "userrecaptcha",
                    "googlekey": site_key,
                    "pageurl": url,
                })

            resp = self.session.post(url, data={
                "g-recaptcha-response": token,
            })

        # Cloudflare Turnstile
        if "cf-turnstile" in resp.text:
            match = re.search(
                r'data-sitekey=["\']([^"\']+)', resp.text
            )
            if match:
                token = solve_captcha({
                    "method": "turnstile",
                    "sitekey": match.group(1),
                    "pageurl": url,
                })
                resp = self.session.post(url, data={
                    "cf-turnstile-response": token,
                })

        return resp.text

    def check_fares(self, routes):
        """Check fares for a list of routes."""
        results = []
        for route in routes:
            try:
                html = self.fetch_with_captcha(route["url"])
                prices = self._extract_prices(html)

                result = {
                    "route": f"{route['origin']}-{route['destination']}",
                    "date": route["date"],
                    "prices": prices,
                    "min_price": min(prices) if prices else None,
                    "timestamp": datetime.utcnow().isoformat(),
                }
                results.append(result)
                self.history.append(result)

                if prices:
                    print(f"  {result['route']} ({route['date']}): "
                          f"${min(prices)}-${max(prices)}")
                else:
                    print(f"  {result['route']}: No prices found")

                time.sleep(3)  # Respectful delay

            except Exception as e:
                print(f"  {route.get('origin', '?')}-"
                      f"{route.get('destination', '?')}: ERROR - {e}")

        return results

    def _extract_prices(self, html):
        """Extract prices from travel page HTML."""
        prices = []
        # Common price patterns
        for match in re.finditer(
            r'\$\s*([\d,]+(?:\.\d{2})?)', html
        ):
            price = float(match.group(1).replace(",", ""))
            if 20 < price < 10000:  # Filter noise
                prices.append(price)
        return sorted(set(prices))

    def detect_price_drops(self, threshold_pct=5):
        """Detect significant price drops in history."""
        route_prices = {}
        for entry in self.history:
            key = f"{entry['route']}_{entry['date']}"
            if key not in route_prices:
                route_prices[key] = []
            if entry["min_price"]:
                route_prices[key].append(entry["min_price"])

        alerts = []
        for key, prices in route_prices.items():
            if len(prices) >= 2:
                prev = prices[-2]
                current = prices[-1]
                change_pct = ((current - prev) / prev) * 100
                if change_pct < -threshold_pct:
                    alerts.append({
                        "route": key,
                        "previous": prev,
                        "current": current,
                        "change": f"{change_pct:.1f}%",
                    })

        return alerts

    def export_report(self, filename="fare_report.json"):
        """Export fare history to JSON."""
        with open(filename, "w") as f:
            json.dump(self.history, f, indent=2)
        print(f"Exported {len(self.history)} fare checks to {filename}")


# Define routes to monitor
routes = [
    {
        "origin": "JFK",
        "destination": "LAX",
        "date": "2025-03-15",
        "url": "https://example-airline.com/flights?from=JFK&to=LAX&date=2025-03-15",
    },
    {
        "origin": "SFO",
        "destination": "ORD",
        "date": "2025-03-20",
        "url": "https://example-airline.com/flights?from=SFO&to=ORD&date=2025-03-20",
    },
]

monitor = FareMonitor()
results = monitor.check_fares(routes)
monitor.export_report()

Что стоит добавить сразу, пока монитор ещё маленький:

• запись сырого HTML в файл при пустом списке цен — без этого невозможно понять, сломалась разметка или страница вернула проверку;
• отдельный счётчик задач CAPTCHA на маршрут — он показывает, какие источники реально требуют решения;
• хранение истории не в памяти, а в SQLite или построчном JSON, чтобы перезапуск не терял предыдущие замеры.

Если цены появляются только после выполнения JavaScript, страницу забирает Playwright или Selenium в headless-режиме, а токен по-прежнему приходит из API и подставляется в форму перед отправкой.

Запуск по расписанию

# Check fares every 4 hours
0 */4 * * * cd /opt/fare-monitor && python fare_monitor.py

Четырёхчасовой шаг — разумный старт для личных задач. Для коммерческого мониторинга имеет смысл сократить интервал до 1–2 часов и разнести маршруты по расписанию, а не запускать их все одним залпом: ровная нагрузка даёт более предсказуемое время решения и меньше повторов.

Сколько потоков закладывать под мониторинг

Тарификация идёт по одновременным потокам, а не по числу решённых задач: внутри тарифа количество решений за месяц не ограничено. Поэтому планируйте не «сколько проверок в месяц», а «сколько задач решается одновременно в пиковый час». Последовательный монитор по определению использует один поток; если список маршрутов обрабатывают восемь воркеров — до восьми.

Как снизить долю неудачных проверок

• Тайм-ауты и повторы. Долгий опрос res.php — нормальное поведение под нагрузкой; один повтор с экспоненциальной задержкой закрывает большую часть таких случаев.
• Логируйте код ошибки целиком. Строка вида ERROR_... из ответа говорит больше, чем собственное сообщение «не удалось решить».
• Разносите запросы во времени. Пауза в несколько секунд между маршрутами снижает частоту появления проверок и нагрузку на чужой сайт.
• Собирайте только публичные данные. Цены и расписания — открытая витрина, персональные данные пассажиров — нет. Если вы работаете с российскими клиентами, учитывайте требования 152-ФЗ «О персональных данных» и храните в истории только то, что вправе обрабатывать. Это ваша зона ответственности, а не сервиса решения CAPTCHA.
• Проверяйте интеграцию на staging. Собственный тестовый стенд с тем же типом проверки даёт стабильный контур для отладки без лишних запросов к боевому сайту.

Для Turnstile отдельно сверьте параметры задачи с документацией CaptchaAI: метод turnstile требует sitekey и pageurl, а результат возвращается в поле cf-turnstile-response.

Частые вопросы

Как понять, что на странице reCAPTCHA v3, а не v2?

По подключённому скрипту. Если в HTML есть recaptcha/api.js?render=, это v3, и в задачу нужно передать version и action. Видимый виджет с data-sitekey без этого параметра — обычно v2.

Что делать, если res.php долго возвращает CAPCHA_NOT_READY?

Это штатный ответ «задача ещё в работе». Продолжайте опрос с интервалом в несколько секунд и ставьте общий тайм-аут на задачу; по его истечении лучше пропустить маршрут и вернуться к нему в следующем цикле, чем держать поток занятым.

Можно ли держать мониторинг авиабилетов и отелей в одном процессе?

Да, логика решения CAPTCHA общая для обоих типов источников. Различаются только правила извлечения цены и валюты, поэтому держите парсеры в отдельных модулях, а клиент API — один.

Нужно ли менять тариф, если число маршрутов выросло вдвое?

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

Связанные материалы

• Мониторинг цен в интернет-торговле
• Устойчивый парсинг без потери данных
• Ротация прокси при сборе данных