Inteligencia competitiva con IA: monitorea a tu competencia en automático

import requests
import anthropic
import json
from datetime import date

def get_ahrefs_data(domain: str, api_key: str) -> dict:
    """Obtiene las métricas SEO principales de un dominio con la Ahrefs API v3.
    La API está disponible en los planes de pago. Revisa los parámetros obligatorios y los límites en docs.ahrefs.com:
    los parámetros de los endpoints cambian."""
    
    headers = {"Authorization": f"Bearer {api_key}"}
    base = "https://api.ahrefs.com/v3"
    
    # Métricas orgánicas
    metrics_resp = requests.get(
        f"{base}/site-explorer/metrics",
        headers=headers,
        params={"target": domain, "mode": "domain", "date": date.today().isoformat()}
    )
    
    # Páginas principales
    pages_resp = requests.get(
        f"{base}/site-explorer/top-pages",
        headers=headers,
        params={"target": domain, "mode": "domain", "limit": 10}
    )
    
    return {
        "metrics": metrics_resp.json() if metrics_resp.ok else {},
        "top_pages": pages_resp.json() if pages_resp.ok else {}
    }

# Análisis con Claude
client = anthropic.Anthropic()

competitors = ["competitor1.com", "competitor2.com", "competitor3.com"]
# AHREFS_API_KEY = os.getenv("AHREFS_API_KEY")

# Para la demostración, la estructura de datos que vas a recibir:
demo_data = {
    "competitor1.com": {
        "organic_traffic": 45200,
        "organic_keywords": 3840,
        "top_pages": [
            {"url": "/blog/ai-automation-guide", "traffic": 8400},
            {"url": "/pricing", "traffic": 4200},
        ]
    }
}

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1500,
    messages=[{
        "role": "user",
        "content": f"""Analiza los datos SEO de la competencia:

{json.dumps(demo_data, ensure_ascii=False, indent=2)}

Mi sitio es [tu dominio], nicho: herramientas de IA para pequeños negocios.

Responde:
1. ¿Quién lidera en orgánico y por qué?
2. ¿Qué temas o páginas generan más tráfico?
3. ¿Qué palabras clave me conviene atacar primero?
4. ¿Qué se les escapa a los competidores (huecos de contenido)?"""
    }]
)
print("".join(b.text for b in message.content if b.type == "text"))

pip install playwright
playwright install chromium

import asyncio
from playwright.async_api import async_playwright
import json
from datetime import datetime

async def scrape_pricing_page(url: str) -> dict:
    """Extrae la página de precios de un competidor"""
    
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        
        # Nos presentamos como un navegador normal; no evadimos la protección del sitio
        await page.set_extra_http_headers({
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
        })
        
        await page.goto(url, wait_until="networkidle", timeout=30000)
        
        # Extraemos el texto de la página
        content = await page.inner_text("body")
        
        # Buscamos los precios con JavaScript
        prices = await page.evaluate("""
            () => {
                const priceElements = document.querySelectorAll('[class*="price"], [class*="plan"], [class*="tier"]');
                return Array.from(priceElements).map(el => el.innerText.trim()).filter(t => t.length > 0);
            }
        """)
        
        await browser.close()
        
        return {
            "url": url,
            "scraped_at": datetime.now().isoformat(),
            "page_text": content[:5000],  # los primeros 5000 caracteres
            "price_elements": prices[:20]
        }

async def monitor_competitors(competitor_urls: list) -> list:
    """Vigila una lista de competidores en paralelo"""
    tasks = [scrape_pricing_page(url) for url in competitor_urls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    return [r for r in results if not isinstance(r, Exception)]

# Arranque
competitors_pricing = [
    "https://competitor1.com/pricing",
    "https://competitor2.com/plans",
]

results = asyncio.run(monitor_competitors(competitors_pricing))

# Guardamos con la fecha (para el análisis diferencial)
filename = f"competitor_data_{datetime.now().strftime('%Y-%m-%d')}.json"
with open(filename, 'w', encoding='utf-8') as f:
    json.dump(results, f, ensure_ascii=False, indent=2)

print(f"Datos guardados: {filename}")

import json
from datetime import datetime, timedelta

def find_changes(old_file: str, new_file: str) -> dict:
    """Compara dos fotos de los datos de la competencia"""
    
    with open(old_file) as f:
        old_data = json.load(f)
    with open(new_file) as f:
        new_data = json.load(f)
    
    changes = []
    
    old_by_url = {r['url']: r for r in old_data}
    new_by_url = {r['url']: r for r in new_data}
    
    for url in new_by_url:
        if url in old_by_url:
            old_text = old_by_url[url].get('page_text', '')
            new_text = new_by_url[url].get('page_text', '')
            
            # Comparación simple; en producción usa difflib
            if old_text != new_text:
                changes.append({
                    "url": url,
                    "change_detected": True,
                    "old_length": len(old_text),
                    "new_length": len(new_text),
                    "delta": len(new_text) - len(old_text)
                })
    
    return changes

ChangeDetection.io detectó un cambio
→ Webhook POST a n8n
→ n8n: petición a Claude con el texto de "antes" y "después"
→ Claude analiza: ¿qué cambió en lo estratégico?
→ Te llega un aviso por Telegram (o por correo, o por Slack) con el análisis

[LUNES 06:00, automático]
    │
    ├─ Scraper con Playwright: reúne los datos de precios de la competencia
    ├─ ChangeDetection: revisa cambios en los sitios
    └─ Ahrefs API: descarga las métricas SEO
    │
    ▼
[Análisis de Claude]
    Junta todos los datos
    Compara con la semana pasada
    Arma un reporte con conclusiones
    │
    ▼
[Bot de Telegram, 08:00]
    📊 Reporte semanal de la competencia
    Competidor 1: sin cambios
    Competidor 2: ⚠️ Bajó el precio de $49 → $39
    Competidor 3: 🆕 Página nueva /ai-assistant
    
    Análisis de Claude: [conclusiones estratégicas]

import asyncio
import anthropic
import json
import os
from datetime import datetime
from playwright.async_api import async_playwright
import requests

ANTHROPIC_API_KEY = os.getenv("ANTHROPIC_API_KEY")
TELEGRAM_TOKEN = os.getenv("TELEGRAM_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")

COMPETITORS = [
    {
        "name": "Competidor A",
        "pricing_url": "https://competitorA.com/pricing",
        "homepage": "https://competitorA.com"
    },
    {
        "name": "Competidor B",
        "pricing_url": "https://competitorB.com/plans",
        "homepage": "https://competitorB.com"
    },
]

async def scrape_page_text(url: str) -> str:
    """Devuelve el contenido de texto de la página"""
    try:
        async with async_playwright() as p:
            browser = await p.chromium.launch(headless=True)
            page = await browser.new_page()
            await page.goto(url, wait_until="domcontentloaded", timeout=20000)
            text = await page.inner_text("body")
            await browser.close()
            return text[:8000]
    except Exception as e:
        return f"Error de scraping en {url}: {e}"

def load_previous_data(filename: str) -> dict:
    try:
        with open(filename) as f:
            return json.load(f)
    except FileNotFoundError:
        return {}

def save_current_data(data: dict, filename: str):
    with open(filename, 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def send_telegram(message: str):
    url = f"https://api.telegram.org/bot{TELEGRAM_TOKEN}/sendMessage"
    requests.post(url, json={
        "chat_id": TELEGRAM_CHAT_ID,
        "text": message,
        "parse_mode": "Markdown"
    })

async def generate_weekly_report():
    print(f"Empiezo a reunir datos: {datetime.now()}")
    
    # Cargamos los datos anteriores
    prev_data = load_previous_data("competitor_snapshot.json")
    current_data = {}
    
    # Hacemos scraping de todos los competidores
    for competitor in COMPETITORS:
        print(f"  Extrayendo {competitor['name']}...")
        pricing_text = await scrape_page_text(competitor['pricing_url'])
        homepage_text = await scrape_page_text(competitor['homepage'])
        
        current_data[competitor['name']] = {
            "scraped_at": datetime.now().isoformat(),
            "pricing": pricing_text,
            "homepage": homepage_text
        }
    
    # Guardamos la foto actual
    save_current_data(current_data, "competitor_snapshot.json")
    
    # Armamos el contexto para Claude
    analysis_input = []
    for name, data in current_data.items():
        prev = prev_data.get(name, {})
        analysis_input.append({
            "competitor": name,
            "current_pricing": data['pricing'][:2000],
            "current_homepage": data['homepage'][:1000],
            "previous_pricing": prev.get('pricing', 'Sin datos')[:2000],
            "change_detected": data['pricing'] != prev.get('pricing', '')
        })
    
    # Análisis con Claude
    client = anthropic.Anthropic(api_key=ANTHROPIC_API_KEY)
    
    message = client.messages.create(
        model="claude-sonnet-5-5",
        max_tokens=1500,
        messages=[{
            "role": "user",
            "content": f"""Este es el reporte semanal de inteligencia competitiva.

Datos de los competidores (semana actual vs. la anterior):
{json.dumps(analysis_input, ensure_ascii=False, indent=2)}

Escribe un reporte breve (hasta 500 palabras) para un chat, con emojis:
1. 🔍 Qué cambió en cada competidor
2. ⚠️ Señales importantes que requieren que reaccionemos
3. 💡 Una recomendación estratégica para esta semana

Si no hubo cambios, dilo también de forma directa."""
        }]
    )
    
    report = "".join(b.text for b in message.content if b.type == "text")
    
    # Enviamos a Telegram
    header = f"📊 *Inteligencia semanal*: {datetime.now().strftime('%d/%m/%Y')}\n\n"
    send_telegram(header + report)
    
    print("Reporte enviado a Telegram")

# Arranque
asyncio.run(generate_weekly_report())

# crontab -e
0 6 * * 1 cd /path/to/project && python competitive_monitor.py >> logs/ci_$(date +\%Y\%m\%d).log 2>&1

mkdir competitive-intel && cd competitive-intel
python3 -m venv venv && source venv/bin/activate
pip install playwright anthropic python-dotenv requests
playwright install chromium

# Crea el archivo .env:
# ANTHROPIC_API_KEY=sk-ant-...
# TELEGRAM_TOKEN=...
# TELEGRAM_CHAT_ID=...

python first_snapshot.py

# test_analysis.py
import anthropic
import json

client = anthropic.Anthropic()

# Cargamos los datos extraídos
with open('competitor_snapshot.json') as f:
    data = json.load(f)

# Les damos formato para Claude
summary = ""
for name, info in data.items():
    summary += f"\n### {name}\n"
    summary += f"Página de precios (primeros 500 caracteres):\n{info['pricing'][:500]}\n"

message = client.messages.create(
    model="claude-sonnet-5-5",
    max_tokens=1000,
    messages=[{
        "role": "user",
        "content": f"""Estos son los datos de los sitios de la competencia:

{summary}

Dime:
1. ¿Qué modelo de precios usa cada uno?
2. ¿Quién es el más barato y quién el más caro?
3. ¿Qué parece lo más fuerte de su oferta?
4. ¿Qué puntos débiles ves?"""
    }]
)

print("".join(b.text for b in message.content if b.type == "text"))

# Abrimos el crontab
crontab -e

# Agregamos la línea (cada lunes a las 07:00)
0 7 * * 1 /Users/tu-usuario/competitive-intel/venv/bin/python /Users/tu-usuario/competitive-intel/competitive_monitor.py