AI Overview Competitor Visibility Tracking: Monitor Who Google Cites in Your Niche

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

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

import asyncio
import time
from dataclasses import dataclass, field
from typing import List, Optional
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

@dataclass
class CompetitorCitation:
    domain: str
    position: int
    query: str
    context: str
    timestamp: float

@dataclass
class CompetitorVisibilityReport:
    competitor: str
    total_citations: int
    avg_position: float
    queries_cited: List[str]
    content_types_cited: List[str]
    visibility_score: float  # 0-100

class CompetitorVisibilityTracker:
    """Track competitor visibility in Google AI Overviews."""
    
    def __init__(self, api_key: str, proxies: list, competitors: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.competitors = competitors
        self.citation_history = []
    
    async def track_query(self, query: str) -> List[CompetitorCitation]:
        """Track which competitors appear in AI Overview for a query."""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx = getattr(self, 'proxy_idx', 0) + 1
        
        html = await self._search_google(query, 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_search(query, proxy)
        
        citations = self._extract_ai_overview_citations(html)
        
        # Filter for tracked competitors
        competitor_citations = []
        for citation in citations:
            for comp in self.competitors:
                if comp in citation.get("domain", ""):
                    competitor_citations.append(CompetitorCitation(
                        domain=comp,
                        position=citation["position"],
                        query=query,
                        context=citation.get("context", ""),
                        timestamp=time.time()
                    ))
        
        self.citation_history.extend(competitor_citations)
        return competitor_citations
    
    async def run_full_audit(self, queries: list) -> List[CompetitorVisibilityReport]:
        """Run visibility audit across all target queries."""
        for query in queries:
            await self.track_query(query)
            await asyncio.sleep(8)
        
        # Generate per-competitor reports
        reports = []
        for comp in self.competitors:
            comp_citations = [c for c in self.citation_history if c.domain == comp]
            if comp_citations:
                reports.append(CompetitorVisibilityReport(
                    competitor=comp,
                    total_citations=len(comp_citations),
                    avg_position=sum(c.position for c in comp_citations) / len(comp_citations),
                    queries_cited=list(set(c.query for c in comp_citations)),
                    content_types_cited=self._analyze_content_types(comp_citations),
                    visibility_score=self._calc_visibility_score(comp_citations, len(queries))
                ))
        
        return sorted(reports, key=lambda r: r.visibility_score, reverse=True)
    
    def _calc_visibility_score(self, citations: list, total_queries: int) -> float:
        """Calculate visibility score (0-100) based on citation frequency and position."""
        if not citations or total_queries == 0:
            return 0
        frequency_score = (len(citations) / total_queries) * 50
        position_score = sum(max(0, 10 - c.position) for c in citations) / len(citations) * 5
        return min(100, frequency_score + position_score)
    
    async def close(self):
        await self.cap.aclose()

async def weekly_competitive_report(tracker, queries):
    """Generate weekly competitive intelligence report."""
    reports = await tracker.run_full_audit(queries)
    
    print("=== AI Overview Competitive Visibility Report ===\n")
    print(f"Queries analyzed: {len(queries)}")
    print(f"Competitors tracked: {len(tracker.competitors)}\n")
    
    for i, report in enumerate(reports, 1):
        print(f"{i}. {report.competitor}")
        print(f"   Visibility Score: {report.visibility_score:.1f}/100")
        print(f"   Citations: {report.total_citations} across {len(report.queries_cited)} queries")
        print(f"   Avg Position: {report.avg_position:.1f}")
        print(f"   Top queries: {', '.join(report.queries_cited[:3])}")
        print()