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

export CAPSOLVER_API_KEY="your-capsolver-api-key"

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:
    """检测Google搜索结果随时间的变化。"""
    
    def __init__(self):
        self.snapshots = {}  # keyword -> list of SERPSnapshot
    
    def detect_changes(self, keyword: str, current: SERPSnapshot) -> List[SERPChange]:
        """将当前快照与之前进行比较并检测变化。"""
        changes = []
        previous_list = self.snapshots.get(keyword, [])
        
        if not previous_list:
            self.snapshots[keyword] = [current]
            return []
        
        previous = previous_list[-1]
        
        # 检测跟踪URL的排名变化
        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"
                ))
        
        # 检测精选摘要变化
        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"
            ))
        
        # 存储当前快照
        self.snapshots[keyword].append(current)
        self.snapshots[keyword] = self.snapshots[keyword][-30:]  # 保留30天
        
        return changes

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class SERPCollector:
    """收集SERP数据并自动处理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:
        """收集关键词的完整SERP快照。"""
        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]:
        """监控一批关键词并检测变化。"""
        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:
    """为重要的SERP变化生成警报。"""
    
    def __init__(self, tracked_domains: list):
        self.tracked_domains = tracked_domains
    
    def filter_relevant_changes(self, changes: List[SERPChange]) -> List[SERPChange]:
        """过滤与跟踪域名相关的更改。"""
        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:
        """生成可读的警报消息。"""
        if not changes:
            return "未检测到显著的SERP变化。"
        
        lines = [f"🔔 检测到{len(changes)}个SERP变化:\n"]
        for c in changes:
            if c.change_type == "ranking_gained":
                lines.append(f"  📈 [{c.keyword}] {c.details['url']} 从#{c.details['old']}上升至#{c.details['new']}")
            elif c.change_type == "ranking_lost":
                lines.append(f"  📉 [{c.keyword}] {c.details['url']} 从#{c.details['old']}下降至#{c.details['new']}")
            elif c.change_type == "snippet_gained":
                lines.append(f"  ⭐ [{c.keyword}] 获得精选摘要")
            elif c.change_type == "new_competitor":
                lines.append(f"  🆕 [{c.keyword}] 新竞争对手出现在#{c.details['position']}：{c.details['url']}")
        return "\n".join(lines)