Panel de análisis de la competencia con CaptchaAI en Python

La pregunta que responde este tutorial es concreta: ¿bajó alguien el precio esta semana, y quién fue primero? Para contestarla no basta con mirar la web de tu competidor hoy; hace falta una lectura fechada cada día, acumulada durante meses. Eso es lo que vas a montar aquí con Python, SQLite y la API de CaptchaAI: unas 300 líneas que corren en un cron diario y van llenando un histórico consultable.

El obstáculo real no es el HTML: las páginas de precios de tu competencia suelen estar detrás de un reCAPTCHA v2, y un script que se detiene ahí devuelve datos vacíos justo cuando más los necesitas. CaptchaAI resuelve ese desafío por API y te devuelve el token; el resto del pipeline ni se entera.

Qué vas a construir: la arquitectura

Competitor Sites ──> CAPTCHA Solver ──> Data Extractors
                                             │
                                        SQLite Store
                                             │
                                      Dashboard Report

Cuatro módulos, cada uno con una responsabilidad: models.py (persistencia), solver.py (CAPTCHA), scraper.py (extracción) y report.py (comparativas). El orquestador en main.py los une.

Por qué importa el histórico: si vendes en un marketplace regional como MercadoLibre o en Amazon.es, los precios de tus rivales se mueven varias veces por semana. Una foto puntual no dice nada; treinta lecturas diarias te dicen quién bajó primero, quién sigue al líder y quién solo ajusta los fines de semana. Por eso todo se guarda con marca de tiempo.

Antes de seguir: rastrea únicamente páginas públicas y respeta los términos de servicio del sitio y la normativa de protección de datos aplicable (GDPR y LOPDGDD en España, LFPDPPP en México).

Paso 1: modelo de datos e histórico en SQLite

La tabla es deliberadamente simple: una fila por métrica observada, con competitor, metric, value, un numeric_value para poder ordenar y comparar, y scraped_at. Nunca actualizamos filas; solo insertamos. Así el histórico queda intacto y el informe siempre puede reconstruir la serie.

# models.py
import sqlite3
from datetime import datetime
from dataclasses import dataclass
from typing import Optional


@dataclass
class CompetitorData:
    competitor: str
    metric: str
    value: str
    numeric_value: Optional[float] = None
    url: str = ""
    scraped_at: str = ""

    def __post_init__(self):
        if not self.scraped_at:
            self.scraped_at = datetime.now().isoformat()


class CompetitorDB:
    def __init__(self, path="competitor_data.db"):
        self.conn = sqlite3.connect(path)
        self._init()

    def _init(self):
        self.conn.execute("""
            CREATE TABLE IF NOT EXISTS metrics (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                competitor TEXT,
                metric TEXT,
                value TEXT,
                numeric_value REAL,
                url TEXT,
                scraped_at TEXT
            )
        """)
        self.conn.commit()

    def save(self, data: CompetitorData):
        self.conn.execute(
            """INSERT INTO metrics
               (competitor, metric, value, numeric_value, url, scraped_at)
               VALUES (?, ?, ?, ?, ?, ?)""",
            (data.competitor, data.metric, data.value,
             data.numeric_value, data.url, data.scraped_at),
        )
        self.conn.commit()

    def get_history(self, competitor, metric, limit=30):
        cursor = self.conn.execute(
            """SELECT value, numeric_value, scraped_at
               FROM metrics
               WHERE competitor = ? AND metric = ?
               ORDER BY scraped_at DESC LIMIT ?""",
            (competitor, metric, limit),
        )
        return cursor.fetchall()

    def latest_comparison(self, metric):
        cursor = self.conn.execute(
            """SELECT competitor, value, numeric_value, MAX(scraped_at) as latest
               FROM metrics WHERE metric = ?
               GROUP BY competitor ORDER BY numeric_value""",
            (metric,),
        )
        return cursor.fetchall()

Dos consultas hacen todo el trabajo del panel: get_history devuelve la serie de un competidor para dibujar tendencias, y latest_comparison agrupa por competidor y se queda con la lectura más reciente de cada uno, que es exactamente la tabla que quieres ver en el informe.

Paso 2: resolver el reCAPTCHA v2 con la API

Aquí entra CaptchaAI. Si el HTML descargado contiene un data-sitekey, hay un reCAPTCHA v2 esperando: extraes el sitekey, envías la tarea a in.php con el método userrecaptcha, sondeas res.php hasta que el estado sea 1 y reenvías el formulario con el token en g-recaptcha-response.

# solver.py
import requests
import time
import re
import os


class CaptchaSolver:
    def __init__(self):
        self.api_key = os.environ["CAPTCHAAI_API_KEY"]

    def solve_if_needed(self, session, url, html):
        if "data-sitekey" not in html:
            return html

        match = re.search(r'data-sitekey="([^"]+)"', html)
        if not match:
            return html

        sitekey = match.group(1)
        resp = requests.post("https://ocr.captchaai.com/in.php", data={
            "key": self.api_key,
            "method": "userrecaptcha",
            "googlekey": sitekey,
            "pageurl": url,
            "json": 1,
        }, timeout=30)
        task_id = resp.json()["request"]

        time.sleep(15)
        for _ in range(24):
            resp = requests.get("https://ocr.captchaai.com/res.php", params={
                "key": self.api_key, "action": "get",
                "id": task_id, "json": 1,
            }, timeout=15)
            data = resp.json()
            if data.get("status") == 1:
                post_resp = session.post(url, data={
                    "g-recaptcha-response": data["request"],
                }, timeout=30)
                return post_resp.text
            if data["request"] != "CAPCHA_NOT_READY":
                raise RuntimeError(data["request"])
            time.sleep(5)

        raise TimeoutError("CAPTCHA solve timeout")

Tres detalles que ahorran depuración:

• El primer sondeo espera 15 segundos. Consultar antes solo gasta llamadas: la respuesta será CAPCHA_NOT_READY de todas formas.
• CAPCHA_NOT_READY se escribe así, sin la "T". Es el literal que devuelve la API; no lo "corrijas".
• El bucle tiene techo. 24 reintentos de 5 segundos dan unos dos minutos antes de lanzar TimeoutError, en lugar de dejar el proceso colgado toda la noche.

Si tu objetivo usa Cloudflare Turnstile o GeeTest v3, la estructura es idéntica: solo cambia el método de envío, y el sondeo y el manejo de errores se reutilizan tal cual.

Paso 3: extractores por competidor

El scraper hace tres cosas sobre la misma sesión HTTP: precios, listas de funcionalidades y tamaño de catálogo. Cada método recibe los selectores CSS del competidor: no hay dos páginas de precios iguales, y mantener los selectores fuera de la lógica evita reescribir la clase cada trimestre.

# scraper.py
import requests
import re
from bs4 import BeautifulSoup
from solver import CaptchaSolver
from models import CompetitorData


class CompetitorScraper:
    def __init__(self):
        self.solver = CaptchaSolver()
        self.session = requests.Session()
        self.session.headers["User-Agent"] = (
            "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
            "AppleWebKit/537.36 Chrome/125.0.0.0 Safari/537.36"
        )

    def scrape_pricing(self, competitor_name, url, plan_selector, price_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        plans = soup.select(plan_selector)
        data = []

        for plan in plans:
            name_el = plan.select_one("h3, h2, .plan-name")
            price_el = plan.select_one(price_selector)

            if not name_el or not price_el:
                continue

            price_text = price_el.get_text(strip=True)
            match = re.search(r'[\d,.]+', price_text)
            numeric = float(match.group().replace(",", "")) if match else None

            data.append(CompetitorData(
                competitor=competitor_name,
                metric=f"price_{name_el.get_text(strip=True).lower().replace(' ', '_')}",
                value=price_text,
                numeric_value=numeric,
                url=url,
            ))

        return data

    def scrape_features(self, competitor_name, url, feature_list_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        features = soup.select(f"{feature_list_selector} li")

        return [
            CompetitorData(
                competitor=competitor_name,
                metric="feature",
                value=f.get_text(strip=True),
                url=url,
            )
            for f in features if f.get_text(strip=True)
        ]

    def scrape_product_count(self, competitor_name, url, count_selector):
        html = self._fetch(url)
        soup = BeautifulSoup(html, "html.parser")
        el = soup.select_one(count_selector)

        if el:
            text = el.get_text(strip=True)
            match = re.search(r'[\d,]+', text)
            if match:
                count = int(match.group().replace(",", ""))
                return CompetitorData(
                    competitor=competitor_name,
                    metric="product_count",
                    value=text,
                    numeric_value=count,
                    url=url,
                )
        return None

    def _fetch(self, url):
        resp = self.session.get(url, timeout=20)
        return self.solver.solve_if_needed(self.session, url, resp.text)

Fíjate en _fetch: es el único punto por donde pasa toda solicitud, y ahí se conecta el solver. Añadir un competidor nuevo no toca la lógica de CAPTCHA en absoluto.

La expresión regular [\d,.]+ sobre el texto del precio es tolerante a propósito: capta tanto $14.90 como 1,299. Ojo: numeric_value es un número plano, sin moneda. Si comparas sitios en distintas divisas, guarda la moneda como métrica aparte.

Paso 4: informes comparativos y de tendencia

# report.py
from models import CompetitorDB


def generate_report(db: CompetitorDB, metrics):
    lines = ["=" * 60, "Competitor Analysis Report", "=" * 60, ""]

    for metric in metrics:
        results = db.latest_comparison(metric)
        if not results:
            continue

        lines.append(f"--- {metric.replace('_', ' ').title()} ---")
        for comp, value, numeric, ts in results:
            marker = ""
            if numeric is not None:
                marker = f" (${numeric:,.2f})" if "price" in metric else f" ({numeric:,.0f})"
            lines.append(f"  {comp}: {value}{marker}")
        lines.append("")

    return "\n".join(lines)


def generate_trend(db: CompetitorDB, competitor, metric, periods=10):
    history = db.get_history(competitor, metric, limit=periods)
    if not history:
        return f"No data for {competitor} — {metric}"

    lines = [f"Trend: {competitor} — {metric}", "-" * 40]
    for value, numeric, ts in reversed(history):
        date = ts[:10]
        lines.append(f"  {date}: {value}")

    return "\n".join(lines)

generate_report produce la vista transversal —todos los competidores para una métrica, ordenados— y generate_trend la vista longitudinal de uno solo a lo largo del tiempo. Con esas dos salidas en texto plano ya tienes algo que se pega en un correo o en un canal de Slack sin montar ningún frontend.

Paso 5: orquestador y ejecución diaria

# main.py
import time
from models import CompetitorDB
from scraper import CompetitorScraper
from report import generate_report

COMPETITORS = [
    {
        "name": "Competitor A",
        "pricing_url": "https://competitor-a.example.com/pricing",
        "plan_selector": ".pricing-plan",
        "price_selector": ".price",
    },
    {
        "name": "Competitor B",
        "pricing_url": "https://competitor-b.example.com/pricing",
        "plan_selector": ".plan-card",
        "price_selector": ".plan-price",
    },
]


def main():
    db = CompetitorDB()
    scraper = CompetitorScraper()

    for comp in COMPETITORS:
        print(f"Scraping {comp['name']}...")

        try:
            pricing = scraper.scrape_pricing(
                comp["name"], comp["pricing_url"],
                comp["plan_selector"], comp["price_selector"],
            )
            for p in pricing:
                db.save(p)
                print(f"  {p.metric}: {p.value}")
        except Exception as e:
            print(f"  Error: {e}")

        time.sleep(5)

    # Generate report
    metrics = ["price_basic", "price_pro", "price_enterprise", "product_count"]
    report = generate_report(db, metrics)
    print(report)

    with open("competitor_report.txt", "w") as f:
        f.write(report)


if __name__ == "__main__":
    main()

La pausa de 5 segundos entre competidores no es decoración: reduce la carga sobre el sitio ajeno y baja la probabilidad de toparte un CAPTCHA en cada página. Cada competidor va en su propio try, para que un selector roto no tumbe la ejecución completa.

Consumo y coste. CaptchaAI factura por thread concurrente, no por resolución, y aquí las solicitudes van en serie: con 10 o 20 competidores, el plan BASIC ($15/mes, 5 threads) sobra. Si vigilas cientos de URLs en paralelo varias veces al día, STANDARD ($30/mes, 15 threads) o ADVANCE ($90/mes, 50 threads) dan margen sin tocar este código. El coste mensual es fijo en USD, más fácil de presupuestar que un modelo de pago por resolución.

Solución de problemas

Preguntas frecuentes

¿Cuánto cuesta mantener este panel funcionando cada mes?

Depende de la concurrencia, no del número de resoluciones: un cron diario sobre una docena de competidores cabe de sobra en BASIC ($15/mes, 5 threads). Consulta los precios vigentes en captchaai.com/pricing.

¿Sirve si mi competidor usa hCaptcha en lugar de reCAPTCHA?

No con este flujo. CaptchaAI no resuelve hCaptcha ni FunCaptcha (Arkose Labs) actualmente. Sí cubre reCAPTCHA v2 y v3, Cloudflare Turnstile y Cloudflare Challenge, GeeTest v3 e imagen/OCR; GeeTest v4 figura como próximamente.

¿Con qué frecuencia debería ejecutar el scraper?

Una vez al día cubre la mayoría de los casos de precios y catálogos. Subir a cada hora multiplica el consumo y la probabilidad de encontrarte con un CAPTCHA sin aportar señal nueva, salvo en campañas puntuales o lanzamientos.

¿Puedo llevar los datos a un panel visual en lugar de texto?

Sí. La tabla metrics es SQLite plano: exporta a CSV y conéctalo a Google Sheets, Metabase o Grafana, o dibuja las series de get_history con matplotlib. El módulo de informes no necesita cambios.

¿Es legal rastrear las páginas de mis competidores?

Este panel lee páginas públicas de precios y catálogo, práctica habitual de inteligencia competitiva, pero las reglas cambian según la jurisdicción y los términos del sitio. Revísalos antes de desplegar; no accedas a áreas autenticadas ni a datos personales.

Guías relacionadas

• Monitorizar cambios de contenido en sitios con CAPTCHA

Empieza a vigilar los precios de tu competencia: crea tu cuenta en CaptchaAI.