ChatGPT 搜索品牌提及监控：在AI回答中追踪您的品牌

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

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

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

@dataclass
class BrandMention:
    """ChatGPT答案中的单个品牌提及。"""
    query: str
    brand_mentioned: str
    mention_context: str      # 周围文本
    sentiment: str            # 正面、中性、负面
    position: int             # 提及顺序（第1、第2等）
    co_mentioned_brands: List[str]
    timestamp: float

@dataclass
class MonitoringResult:
    """单个查询的完整监控结果。"""
    query: str
    your_brand_mentioned: bool
    your_mention_position: Optional[int]
    competitor_mentions: List[str]
    answer_text: str
    sources_cited: List[str]
    timestamp: float

import asyncio
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class ChatGPTBrandMonitor:
    """监控ChatGPT Search答案中的品牌提及。"""
    
    def __init__(self, api_key: str, your_brand: str, competitors: list):
        self.cap = create_capsolver(api_key=api_key)
        self.your_brand = your_brand
        self.competitors = competitors
        self.history = []
    
    async def check_query(self, query: str) -> MonitoringResult:
        """检查单个查询中的品牌提及。"""
        # 访问ChatGPT Search（受Cloudflare保护）
        answer = await self._fetch_chatgpt_answer(query)
        
        # 分析品牌提及
        your_mentioned = self.your_brand.lower() in answer.lower()
        competitor_mentions = [c for c in self.competitors if c.lower() in answer.lower()]
        
        result = MonitoringResult(
            query=query,
            your_brand_mentioned=your_mentioned,
            your_mention_position=self._find_position(answer, self.your_brand),
            competitor_mentions=competitor_mentions,
            answer_text=answer[:1000],
            sources_cited=self._extract_sources(answer),
            timestamp=time.time()
        )
        
        self.history.append(result)
        return result
    
    async def _fetch_chatgpt_answer(self, query: str) -> str:
        """获取ChatGPT Search答案并处理Cloudflare。"""
        # 解决Turnstile挑战
        info = CaptchaInfo(
            type=CaptchaType.CLOUDFLARE,
            website_url="https://chatgpt.com",
            website_key="0x4AAAAAAAx..."  # ChatGPT的Turnstile密钥
        )
        solution = await self.cap.solve(info)
        # 使用令牌访问搜索结果
        # 实现取决于访问方式
        return ""  # 解析后的答案文本
    
    async def monitor_batch(self, queries: list) -> dict:
        """监控一批查询并生成报告。"""
        results = []
        for query in queries:
            result = await self.check_query(query)
            results.append(result)
            await asyncio.sleep(10)  # 限制频率
        
        return self._generate_report(results)
    
    def _generate_report(self, results: list) -> dict:
        mentioned_count = sum(1 for r in results if r.your_brand_mentioned)
        return {
            "total_queries": len(results),
            "your_brand_mentioned": mentioned_count,
            "mention_rate": f"{mentioned_count/len(results)*100:.1f}%",
            "top_competitors": self._rank_competitors(results),
            "queries_without_mention": [r.query for r in results if not r.your_brand_mentioned]
        }
    
    async def close():
        await self.cap.aclose()