Los sitios de viajes y aerolíneas utilizan con frecuencia CAPTCHA para limitar la verificación automática de tarifas. Un sistema de seguimiento de precios que verifique las tarifas en múltiples rutas enfrentará desafíos reCAPTCHA, anuncios intersticiales de Cloudflare y mecanismos de limitación de tarifas. CaptchaAI maneja el paso de resolución de CAPTCHA para que su canal de monitoreo pueda continuar recopilando datos de tarifas.
Esta guía muestra cómo crear un flujo de trabajo de monitoreo de tarifas que detecte y resuelva CAPTCHA durante las verificaciones de precios.
El flujo de trabajo de monitoreo
Schedule check → Request fare page → CAPTCHA detected?
↓ Yes
Solve via CaptchaAI → Inject token → Retry request
↓ No
Parse fare data → Store → Alert on price change
Lo que necesitas
| Requisito | Detalles |
|---|---|
| API key de CaptchaAI | captchaai.com |
| Python 3.8+ | Con requests |
| Proxy | Red residencial autorizada para sitios de viajes |
pip install requests
Ayudante del solucionador 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")
Monitoreo de tarifas con manejo 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 common airline result layouts."""
from bs4 import BeautifulSoup
import re
soup = BeautifulSoup(html, "html.parser")
selectors = [
"[data-testid='price-value']",
".fare-amount",
".price-total",
".booking-price",
]
price_text = ""
for selector in selectors:
el = soup.select_one(selector)
if el and el.get_text(strip=True):
price_text = el.get_text(strip=True)
break
amount_match = re.search(r"(\d[\d,\.]+)", price_text or html)
currency_match = re.search(r"(USD|EUR|GBP|\$|€|£)", price_text or html)
return {
"route": route["name"],
"carrier": route.get("carrier", "unknown"),
"price_text": price_text or None,
"amount": amount_match.group(1) if amount_match else None,
"currency": currency_match.group(1) if currency_match else route.get("currency"),
"captured_at": datetime.now().isoformat(),
"source_url": route["url"],
}
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))
Verificaciones de programación
Ejecute el monitor según una programación utilizando cron o un programador de tareas:
# Check fares every 6 hours
0 */6 * * * cd /path/to/project && python fare_monitor.py >> /var/log/fares.log 2>&1
Solución de problemas
| Problema | causa | Solución |
|---|---|---|
| CAPTCHA frecuentes | Demasiadas solicitudes de la misma IP | Utilice proxies residenciales rotativos |
| Precios obsoletos | Páginas almacenadas en caché | Agregue encabezados anti-caché o aleatorice los parámetros de solicitud |
| IP bloqueada | Limitación de velocidad | Aumentar los retrasos entre controles, rotar proxies |
| La resolución de CAPTCHA falla | Extracción de clave de sitio incorrecta | Verifique que la clave del sitio coincida con el CAPTCHA de la página |
Preguntas frecuentes
¿Con qué frecuencia debo consultar las tarifas?
Lo normal es cada 4 a 8 horas. Las comprobaciones más frecuentes aumentan los encuentros con CAPTCHA y los costos de proxy.
¿Qué tipos de CAPTCHA utilizan los sitios de las aerolíneas?
Más comúnmente, reCAPTCHA v2, Cloudflare Turnstile o páginas de desafío y, ocasionalmente, CAPTCHA de imágenes.
¿Necesito proxies residenciales?
Sí. Los sitios de viajes bloquean activamente las direcciones IP de los centros de datos. Los proxies residenciales o móviles tienen tasas de éxito significativamente más altas.
¿Puedo monitorear varias aerolíneas?
Sí. Personalice el método _parse_fare para el formato de respuesta de cada aerolínea y agregue rutas para cada sitio.
¿Cómo manejo las páginas Cloudflare Challenge?
Utilice method=cloudflare_challenge con un proxy. La cookie qa_validation_cookie devuelta otorga acceso al sitio. Ver la guía de Cloudflare Challenge.
Obtenga su clave API CaptchaAI
Comienza a monitorear las tarifas aéreas en captchaai.com. Maneja CAPTCHA automáticamente en tu flujo de trabajo de seguimiento de precios.
Guías relacionadas
- Cómo resolver reCAPTCHA v2 usando API
- Cómo resolver Cloudflare Turnstile usando API
- Extracción de datos financieros con manejo de CAPTCHA
- Monitoreo de viajes
