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

export CAPSOLVER_API_KEY="sua-chave-da-api-do-capsolver"

from dataclasses import dataclass, field
from typing import List, Optional
import time

@dataclass
class SERPSnapshot:
    keyword: str
    timestamp: float
    organic_results: List[dict]  # [{position, url, title}]
    featured_snippet: Optional[dict]
    ai_overview_present: bool
    paa_questions: List[str]

@dataclass
class SERPChange:
    keyword: str
    change_type: str  # ranking_gained, ranking_lost, snippet_gained, snippet_lost, new_competitor
    details: dict
    detected_at: float
    severity: str  # high, medium, low

class SERPChangeDetector:
    """Detecta mudanças nos resultados da pesquisa do Google ao longo do tempo."""
    
    def __init__(self):
        self.snapshots = {}  # keyword -> lista de SERPSnapshot
    
    def detect_changes(self, keyword: str, current: SERPSnapshot) -> List[SERPChange]:
        """Compara o snapshot atual com o anterior e detecta mudanças."""
        changes = []
        previous_list = self.snapshots.get(keyword, [])
        
        if not previous_list:
            self.snapshots[keyword] = [current]
            return []
        
        previous = previous_list[-1]
        
        # Detectar mudanças de classificação para URLs monitoradas
        prev_positions = {r["url"]: r["position"] for r in previous.organic_results}
        curr_positions = {r["url"]: r["position"] for r in current.organic_results}
        
        for url, curr_pos in curr_positions.items():
            prev_pos = prev_positions.get(url)
            if prev_pos is None:
                changes.append(SERPChange(
                    keyword=keyword, change_type="new_competitor",
                    details={"url": url, "position": curr_pos},
                    detected_at=time.time(), severity="medium"
                ))
            elif curr_pos < prev_pos and (prev_pos - curr_pos) >= 3:
                changes.append(SERPChange(
                    keyword=keyword, change_type="ranking_gained",
                    details={"url": url, "old": prev_pos, "new": curr_pos},
                    detected_at=time.time(), severity="high"
                ))
            elif curr_pos > prev_pos and (curr_pos - prev_pos) >= 3:
                changes.append(SERPChange(
                    keyword=keyword, change_type="ranking_lost",
                    details={"url": url, "old": prev_pos, "new": curr_pos},
                    detected_at=time.time(), severity="high"
                ))
        
        # Detectar mudanças em trechos destacados
        if current.featured_snippet and not previous.featured_snippet:
            changes.append(SERPChange(
                keyword=keyword, change_type="snippet_gained",
                details=current.featured_snippet,
                detected_at=time.time(), severity="high"
            ))
        elif previous.featured_snippet and not current.featured_snippet:
            changes.append(SERPChange(
                keyword=keyword, change_type="snippet_lost",
                details=previous.featured_snippet,
                detected_at=time.time(), severity="high"
            ))
        
        # Armazenar o snapshot atual
        self.snapshots[keyword].append(current)
        self.snapshots[keyword] = self.snapshots[keyword][-30:]  # Manter 30 dias
        
        return changes

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class SERPCollector:
    """Coleta dados do SERP com tratamento automatizado de CAPTCHA."""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_idx = 0
    
    async def collect_serp(self, keyword: str, location: str = "us") -> SERPSnapshot:
        """Coleta um snapshot completo do SERP para uma palavra-chave."""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        html = await self._search(keyword, location, proxy)
        
        if self._is_captcha(html):
            info = CaptchaInfo(
                type=CaptchaType.RECAPTCHA_V2,
                website_url="https://www.google.com",
                website_key="6LfwuyUTAAAAAOAmoS0fdqijC2PbbdH4kjq62Y1b"
            )
            solution = await self.cap.solve(info)
            html = await self._retry(keyword, location, proxy)
        
        return self._parse_serp(html, keyword)
    
    async def monitor_keywords(self, keywords: list, detector: SERPChangeDetector) -> List[SERPChange]:
        """Monitora um lote de palavras-chave e detecta mudanças."""
        all_changes = []
        for keyword in keywords:
            snapshot = await self.collect_serp(keyword)
            changes = detector.detect_changes(keyword, snapshot)
            all_changes.extend(changes)
            await asyncio.sleep(7)
        return all_changes
    
    async def close(self):
        await self.cap.aclose()

class SERPAlertSystem:
    """Gera alertas para mudanças significativas no SERP."""
    
    def __init__(self, tracked_domains: list):
        self.tracked_domains = tracked_domains
    
    def filter_relevant_changes(self, changes: List[SERPChange]) -> List[SERPChange]:
        """Filtrar mudanças relevantes para domínios monitorados."""
        relevant = []
        for change in changes:
            url = change.details.get("url", "")
            if any(domain in url for domain in self.tracked_domains):
                relevant.append(change)
            elif change.change_type == "new_competitor" and change.severity == "high":
                relevant.append(change)
        return relevant
    
    def generate_alert_message(self, changes: List[SERPChange]) -> str:
        """Gera uma mensagem de alerta legível para humanos."""
        if not changes:
            return "Nenhuma mudança significativa no SERP detectada."
        
        lines = [f"🔔 {len(changes)} mudança(s) no SERP detectada(s):\n"]
        for c in changes:
            if c.change_type == "ranking_gained":
                lines.append(f"  📈 [{c.keyword}] {c.details['url']} subiu de #{c.details['old']} para #{c.details['new']}")
            elif c.change_type == "ranking_lost":
                lines.append(f"  📉 [{c.keyword}] {c.details['url']} caiu de #{c.details['old']} para #{c.details['new']}")
            elif c.change_type == "snippet_gained":
                lines.append(f"  ⭐ [{c.keyword}] Trecho destacado ganho")
            elif c.change_type == "new_competitor":
                lines.append(f"  🆕 [{c.keyword}] Novo concorrente na posição #{c.details['position']}: {c.details['url']}")
        return "\n".join(lines)