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回答における1つのブランドの言及。"""
    query: str
    brand_mentioned: str
    mention_context: str      # 周囲のテキスト
    sentiment: str            # ポジティブ、ニュートラル、ネガティブ
    position: int             # 言及の順序（1番目、2番目など）
    co_mentioned_brands: List[str]
    timestamp: float

@dataclass
class MonitoringResult:
    """1つのクエリに対する完全なモニタリング結果。"""
    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:
        """1つのクエリにおけるブランド言及をチェックします。"""
        # 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:
        """Cloudflare処理を伴うChatGPT Searchの回答を取得します。"""
        # インターフェースがチャレンジされた場合に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()