AI概览 竞争对手可见性跟踪：监控Google在您的领域中引用了谁

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:
    """跟踪Google AI摘要中的竞争对手可见性。"""
    
    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]:
        """跟踪某个查询的AI摘要中出现的竞争对手。"""
        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)
        
        # 过滤已跟踪的竞争对手
        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]:
        """对所有目标查询进行可见性审计。"""
        for query in queries:
            await self.track_query(query)
            await asyncio.sleep(8)
        
        # 生成每个竞争对手的报告
        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:
        """根据引用频率和位置计算可见性分数（0-100）。"""
        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):
    """生成每周竞争情报报告。"""
    reports = await tracker.run_full_audit(queries)
    
    print("=== AI摘要竞争可见性报告 ===\n")
    print(f"分析的查询数量: {len(queries)}")
    print(f"跟踪的竞争对手数量: {len(tracker.competitors)}\n")
    
    for i, report in enumerate(reports, 1):
        print(f"{i}. {report.competitor}")
        print(f"   可见性分数: {report.visibility_score:.1f}/100")
        print(f"   引用次数: {report.total_citations} 涉及 {len(report.queries_cited)} 个查询")
        print(f"   平均位置: {report.avg_position:.1f}")
        print(f"   主要查询: {', '.join(report.queries_cited[:3])}")
        print()