Monitoramento de tarifas de viagem com tratamento CAPTCHA

Um monitor de tarifas só tem valor se rodar sozinho por semanas. E o que costuma derrubar esses scripts quase nunca é o parser de preços: é a tela de verificação que aparece na terceira consulta do dia, quando o site percebe que as requisições chegam sempre no mesmo ritmo. A resposta curta é tratar o desafio como mais uma etapa previsível do fluxo — detectar, enviar para a API da CaptchaAI, receber o token e seguir na mesma sessão. É isso que este guia monta, para monitoramento autorizado: desafios por categoria de site, monitor em Python, custo por thread e agendamento.

Antes do código: onde o monitoramento é legítimo

Monitoramento defensável é o que você faz sobre preços públicos, dentro dos termos de uso do site, em ambiente próprio ou de staging, e em volume compatível com o de um usuário real: uma agência acompanhando as rotas que revende, um time de QA validando o próprio funil de reserva, uma pesquisa autorizada.

No Brasil, se a coleta for guardada junto com dado de pessoa física — o e-mail de quem recebe o alerta, por exemplo —, as obrigações da LGPD passam a valer sobre essa base; em Portugal, o equivalente é o RGPD. É decisão de arquitetura, não jurídica: guarde apenas o par rota/preço/horário e mantenha o registro de destinatários separado.

Que tipo de desafio aparece em cada categoria de site

Nem todo site de viagem protege a busca da mesma forma, e isso muda o custo do monitor:

Duas leituras práticas saem daí. A primeira: o reCAPTCHA v3 não exibe widget nenhum, então seu código precisa reconhecer a assinatura do script na página em vez de procurar um quadrado para clicar. A segunda: reCAPTCHA v2/v3, Cloudflare Turnstile, Cloudflare Challenge, GeeTest v3 e CAPTCHAs de imagem/OCR são exatamente os tipos que a CaptchaAI resolve. Já hCaptcha e FunCaptcha não são suportados — se uma rota cair nesses, ela fica fora do monitor, e isso precisa estar claro no plano antes da primeira linha de código.

Implementação do monitor de tarifas

O monitor abaixo faz três coisas em sequência: busca a página da rota, detecta e resolve o desafio pela API, e só então extrai os preços. A resolução acontece dentro da mesma sessão HTTP — o token só vale naquele contexto, e reaproveitar a sessão evita cair de novo na verificação logo depois. O intervalo de 3 s entre rotas também não é enfeite: espaçar as requisições reduz quantos desafios você vai pagar para resolver.

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 lançamentos de teste 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()

A função de resolução envia a tarefa para in.php e consulta res.php até o resultado ficar pronto. Enquanto a resposta for CAPCHA_NOT_READY, o polling continua; qualquer outro texto que não comece com OK| é erro real e deve interromper apenas aquela rota, nunca o monitor inteiro.

Para páginas que só montam o preço depois do JavaScript, troque a busca simples por Selenium ou Playwright e mantenha a chamada da API igual: o navegador entrega o HTML final e a API entrega o token.

Agendamento e ritmo de coleta

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

A frequência certa depende do uso. Para acompanhamento pessoal, verificar a cada 4 a 6 horas já captura quase todas as variações relevantes: as companhias aéreas reajustam preços em lotes, não continuamente. Para uma operação comercial, 1 a 2 horas é um teto razoável; abaixo disso, o retorno cai e a taxa de desafios sobe.

Distribua as rotas ao longo da janela em vez de disparar todas no minuto zero: com 50 rotas, espalhe-as em blocos de poucos minutos. O volume diário é o mesmo, mas some o pico instantâneo, que é o que aciona as defesas. Rodar os workers em uma região de nuvem próxima ao público (São Paulo, para rotas domésticas brasileiras) ajuda pelo mesmo motivo.

Quanto custa manter o monitor rodando

A conta tem duas partes: infraestrutura e resolução de desafios. Para estimar a segunda, comece pelo número de verificações por dia:

O que muda o planejamento é o modelo de cobrança: a CaptchaAI não cobra por desafio resolvido, e sim por thread simultânea, com resoluções ilimitadas no plano. O BASIC (US$ 15/mês, 5 threads) sustenta com folga um monitor pessoal ou de agência pequena, porque cada thread fica livre assim que a resolução termina. Quando o gargalo passa a ser o paralelismo, o passo seguinte é o STANDARD (US$ 30/mês, 15 threads) ou o ADVANCE (US$ 90/mês, 50 threads). Você dimensiona pela concorrência do agendador, não pelo total mensal de desafios.

Como validar o monitor antes de soltá-lo

Antes de apontar o script para produção, rode o fluxo contra um formulário próprio em staging: você valida a detecção do widget, o envio para a API e a submissão do token sem depender de terceiros. Se o monitor funciona no curl mas falha no navegador automatizado, o problema quase sempre está no contexto da sessão, e não na API — compare cookies, cabeçalhos e URL de origem antes de suspeitar da resolução.

Perguntas frequentes

Qual intervalo entre verificações gera menos desafios?

Intervalos irregulares acima de 3 s entre rotas, com a coleta distribuída ao longo da hora. Rajadas simultâneas são o padrão que mais aciona verificação.

O mesmo script serve para monitorar preços de hotéis?

Sim. Muda apenas a expressão de extração, porque páginas de hospedagem exibem valor por noite, taxas e total separadamente. O restante do fluxo continua igual.

Quais tipos de desafio ficam de fora do monitor?

hCaptcha e FunCaptcha não são suportados, e o GeeTest v4 aparece apenas como "em breve". São suportados reCAPTCHA v2/v3, Cloudflare Turnstile e Challenge, GeeTest v3 e CAPTCHAs de imagem/OCR, além de CaptchaFox (beta), Friendly Captcha (beta) e Lemin (beta).

Preciso de um plano maior conforme adiciono rotas?

Não necessariamente. Como a cobrança é por thread simultânea com resoluções ilimitadas, o upgrade só entra quando muitas rotas passam a ser verificadas ao mesmo tempo.

Como faço o monitor sobreviver a mudanças de layout do site?

Trate a extração como código frágil: registre o HTML bruto quando nenhum preço for encontrado e alerte quando uma rota voltar vazia várias vezes. É assim que você distingue queda real de oferta de uma quebra de parser.

Guias relacionados

• Monitoramento de preços no e-commerce
• Coleta de dados estável sem bloqueios
• Rotação de proxy em coleta com CAPTCHA