Cómo desarrollar el monitoreo de características de SERP para agentes de búsqueda de inteligencia artificial

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install aiohttp beautifulsoup4 pandas schedule

export CAPSOLVER_API_KEY="tu-clave-de-api-de-capsolver"

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum

class SERPFeature(Enum):
    FEATURED_SNIPPET = "featured_snippet"
    PEOPLE_ALSO_ASK = "people_also_ask"
    AI_OVERVIEW = "ai_overview"
    LOCAL_PACK = "local_pack"
    KNOWLEDGE_PANEL = "knowledge_panel"
    VIDEO_CAROUSEL = "video_carousel"
    IMAGE_PACK = "image_pack"
    TOP_STORIES = "top_stories"
    SHOPPING_RESULTS = "shopping_results"

@dataclass
class SERPResult:
    keyword: str
    features_detected: List[SERPFeature]
    featured_snippet_url: Optional[str]
    featured_snippet_text: Optional[str]
    paa_questions: List[str]
    ai_overview_present: bool
    ai_overview_sources: List[str]
    organic_positions: List[dict]  # [{url, title, position}]
    timestamp: float

class SERPFeatureDetector:
    """Detectar características de SERP a partir de páginas de resultados de búsqueda analizadas."""
    
    def detect_features(self, html_content: str, keyword: str) -> SERPResult:
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html_content, 'html.parser')
        
        features = []
        
        # Detección de fragmento destacado
        snippet_div = soup.select_one('[data-attrid="wa:/description"], .xpdopen')
        featured_url = None
        featured_text = None
        if snippet_div:
            features.append(SERPFeature.FEATURED_SNIPPET)
            featured_text = snippet_div.get_text(strip=True)[:500]
            link = snippet_div.find('a')
            featured_url = link['href'] if link else None
        
        # Detección de "Preguntas también preguntadas"
        paa_questions = []
        paa_section = soup.select('[data-q]')
        if paa_section:
            features.append(SERPFeature.PEOPLE_ALSO_ASK)
            paa_questions = [q.get('data-q', '') for q in paa_section[:8]]
        
        # Detección de resúmenes de IA
        ai_overview = soup.select_one('[data-attrid*="ai"], .ai-overview-container')
        ai_sources = []
        if ai_overview:
            features.append(SERPFeature.AI_OVERVIEW)
            source_links = ai_overview.select('a[href]')
            ai_sources = [a['href'] for a in source_links[:5]]
        
        # Detección de paquetes locales
        if soup.select_one('.VkpGBb, [data-attrid*="local"]'):
            features.append(SERPFeature.LOCAL_PACK)
        
        return SERPResult(
            keyword=keyword,
            features_detected=features,
            featured_snippet_url=featured_url,
            featured_snippet_text=featured_text,
            paa_questions=paa_questions,
            ai_overview_present=SERPFeature.AI_OVERVIEW in features,
            ai_overview_sources=ai_sources,
            organic_positions=self._extract_organic(soup),
            timestamp=time.time()
        )
    
    def _extract_organic(self, soup) -> list:
        results = []
        for i, item in enumerate(soup.select('.g, [data-sokoban-container]')[:10], 1):
            link = item.select_one('a[href^="http"]')
            title = item.select_one('h3')
            if link and title:
                results.append({
                    "position": i,
                    "url": link['href'],
                    "title": title.get_text(strip=True)
                })
        return results

import asyncio
import aiohttp
import time
import random
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class SERPCollector:
    """Recopilar datos de SERP con manejo automático de CAPTCHA."""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_index = 0
        self.stats = {"queries": 0, "captchas_solved": 0, "failures": 0}
    
    def _get_proxy(self) -> str:
        proxy = self.proxies[self.proxy_index % len(self.proxies)]
        self.proxy_index += 1
        return proxy
    
    async def search(self, keyword: str, location: str = "us") -> str:
        """Ejecutar una búsqueda de Google con manejo de CAPTCHA."""
        proxy = self._get_proxy()
        url = f"https://www.google.com/search?q={keyword}&gl={location}&hl=en"
        
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept": "text/html,application/xhtml+xml"
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, proxy=proxy, timeout=30) as resp:
                html = await resp.text()
                
                # Verificar si hay una página de CAPTCHA
                if self._is_captcha_page(html):
                    html = await self._solve_and_retry(keyword, location, proxy, session)
                
                self.stats["queries"] += 1
                return html
    
    def _is_captcha_page(self, html: str) -> bool:
        """Detectar si la respuesta es una página de desafío de CAPTCHA."""
        captcha_indicators = [
            "tráfico inusual desde tu computadora",
            "g-recaptcha",
            "recaptcha/api",
            "sorry/index",
            "captcha"
        ]
        return any(indicator in html.lower() for indicator in captcha_indicators)
    
    async def _solve_and_retry(self, keyword: str, location: str, proxy: str, session) -> str:
        """Resolver el CAPTCHA y reintentar la búsqueda."""
        # Google usa reCAPTCHA v2 en sus páginas de desafío
        info = CaptchaInfo(
            type=CaptchaType.RECAPTCHA_V2,
            website_url="https://www.google.com/sorry/index",
            website_key="6LfwuyUTAAAAAOAmoS0fdqijC2PbbdH4kjq62Y1b"
        )
        
        solution = await self.cap.solve(info)
        self.stats["captchas_solved"] += 1
        
        # Reintentar búsqueda con token resuelto
        retry_url = f"https://www.google.com/search?q={keyword}&gl={location}&hl=en"
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        # Incluir el token de CAPTCHA en la solicitud de reintentar
        async with session.get(retry_url, headers=headers, proxy=proxy) as resp:
            return await resp.text()
    
    async def close():
        await self.cap.aclose()

class SERPMonitoringPipeline:
    """Canalización completa de monitoreo de características de SERP."""
    
    def __init__(self, api_key: str, proxies: list):
        self.collector = SERPCollector(api_key, proxies)
        self.detector = SERPFeatureDetector()
        self.results_history = []
    
    async def monitor_keywords(self, keywords: list, location: str = "us") -> list:
        """Monitorear un lote de palabras clave en busca de cambios en características de SERP."""
        results = []
        
        for keyword in keywords:
            try:
                # Recopilar datos de SERP
                html = await self.collector.search(keyword, location)
                
                # Detectar características
                serp_result = self.detector.detect_features(html, keyword)
                results.append(serp_result)
                
                # Límite de velocidad: retardo aleatorio de 5-10 segundos
                await asyncio.sleep(random.uniform(5, 10))
                
            except Exception as e:
                results.append(SERPResult(
                    keyword=keyword, features_detected=[], 
                    featured_snippet_url=None, featured_snippet_text=None,
                    paa_questions=[], ai_overview_present=False,
                    ai_overview_sources=[], organic_positions=[],
                    timestamp=time.time()
                ))
        
        self.results_history.extend(results)
        return results
    
    def detect_changes(self, current: list, previous: list) -> list:
        """Comparar resultados actuales y anteriores para detectar cambios."""
        changes = []
        prev_map = {r.keyword: r for r in previous}
        
        for result in current:
            prev = prev_map.get(result.keyword)
            if not prev:
                continue
            
            # Fragmento destacado ganado/perdido
            had_snippet = SERPFeature.FEATURED_SNIPPET in prev.features_detected
            has_snippet = SERPFeature.FEATURED_SNIPPET in result.features_detected
            if has_snippet and not had_snippet:
                changes.append(f"[GANADO] Fragmento destacado para '{result.keyword}': {result.featured_snippet_url}")
            elif had_snippet and not has_snippet:
                changes.append(f"[PERDIDO] Fragmento destacado para '{result.keyword}'")
            
            # Resumen de IA aparecido/desaparecido
            if result.ai_overview_present and not prev.ai_overview_present:
                changes.append(f"[NUEVO] Resumen de IA apareció para '{result.keyword}'")
            elif prev.ai_overview_present and not result.ai_overview_present:
                changes.append(f"[REMOVIDO] Resumen de IA eliminado para '{result.keyword}'")
        
        return changes