AI検索エージェントのためのSERP特徴モニタリングの構築方法

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
from typing import List, Optional
from enum import Enum

class SERPFeature(Enum):
    FEATURED_SNIPPET = "featured_snippet"
    PEOPLE_ALSO_ASK = "people_also_ask"
    AI_OVERVIEW = "ai_overview"
    LOCAL_PACK = "local_pack"
    KNOWLEDGE_PANEL = "knowledge_panel"
    VIDEO_CAROUSEL = "video_carousel"
    IMAGE_PACK = "image_pack"
    TOP_STORIES = "top_stories"
    SHOPPING_RESULTS = "shopping_results"

@dataclass
class SERPResult:
    keyword: str
    features_detected: List[SERPFeature]
    featured_snippet_url: Optional[str]
    featured_snippet_text: Optional[str]
    paa_questions: List[str]
    ai_overview_present: bool
    ai_overview_sources: List[str]
    organic_positions: List[dict]  # [{url, title, position}]
    timestamp: float

class SERPFeatureDetector:
    """パースされた検索結果ページからSERPの特徴を検出します。"""
    
    def detect_features(self, html_content: str, keyword: str) -> SERPResult:
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(html_content, 'html.parser')
        
        features = []
        
        # 特徴付きスニペットの検出
        snippet_div = soup.select_one('[data-attrid="wa:/description"], .xpdopen')
        featured_url = None
        featured_text = None
        if snippet_div:
            features.append(SERPFeature.FEATURED_SNIPPET)
            featured_text = snippet_div.get_text(strip=True)[:500]
            link = snippet_div.find('a')
            featured_url = link['href'] if link else None
        
        # People Also Askの検出
        paa_questions = []
        paa_section = soup.select('[data-q]')
        if paa_section:
            features.append(SERPFeature.PEOPLE_ALSO_ASK)
            paa_questions = [q.get('data-q', '') for q in paa_section[:8]]
        
        # AI Overviewの検出
        ai_overview = soup.select_one('[data-attrid*="ai"], .ai-overview-container')
        ai_sources = []
        if ai_overview:
            features.append(SERPFeature.AI_OVERVIEW)
            source_links = ai_overview.select('a[href]')
            ai_sources = [a['href'] for a in source_links[:5]]
        
        # ローカルパックの検出
        if soup.select_one('.VkpGBb, [data-attrid*="local"]'):
            features.append(SERPFeature.LOCAL_PACK)
        
        return SERPResult(
            keyword=keyword,
            features_detected=features,
            featured_snippet_url=featured_url,
            featured_snippet_text=featured_text,
            paa_questions=paa_questions,
            ai_overview_present=SERPFeature.AI_OVERVIEW in features,
            ai_overview_sources=ai_sources,
            organic_positions=self._extract_organic(soup),
            timestamp=time.time()
        )
    
    def _extract_organic(self, soup) -> list:
        results = []
        for i, item in enumerate(soup.select('.g, [data-sokoban-container]')[:10], 1):
            link = item.select_one('a[href^="http"]')
            title = item.select_one('h3')
            if link and title:
                results.append({
                    "position": i,
                    "url": link['href'],
                    "title": title.get_text(strip=True)
                })
        return results

import asyncio
import aiohttp
import time
import random
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class SERPCollector:
    """自動CAPTCHA処理を備えたSERPデータの収集。"""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_index = 0
        self.stats = {"queries": 0, "captchas_solved": 0, "failures": 0}
    
    def _get_proxy(self) -> str:
        proxy = self.proxies[self.proxy_index % len(self.proxies)]
        self.proxy_index += 1
        return proxy
    
    async def search(self, keyword: str, location: str = "us") -> str:
        """CAPTCHA処理を備えたGoogle検索を実行します。"""
        proxy = self._get_proxy()
        url = f"https://www.google.com/search?q={keyword}&gl={location}&hl=en"
        
        headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
            "Accept-Language": "en-US,en;q=0.9",
            "Accept": "text/html,application/xhtml+xml"
        }
        
        async with aiohttp.ClientSession() as session:
            async with session.get(url, headers=headers, proxy=proxy, timeout=30) as resp:
                html = await resp.text()
                
                # CAPTCHAチャレンジをチェック
                if self._is_captcha_page(html):
                    html = await self._solve_and_retry(keyword, location, proxy, session)
                
                self.stats["queries"] += 1
                return html
    
    def _is_captcha_page(self, html: str) -> bool:
        """応答がCAPTCHAチャレンジページであるかを検出します。"""
        captcha_indicators = [
            "unusual traffic from your computer",
            "g-recaptcha",
            "recaptcha/api",
            "sorry/index",
            "captcha"
        ]
        return any(indicator in html.lower() for indicator in captcha_indicators)
    
    async def _solve_and_retry(self, keyword: str, location: str, proxy: str, session) -> str:
        """CAPTCHAを解決し、再試行します。"""
        # GoogleはそのチャレンジページでreCAPTCHA v2を使用します
        info = CaptchaInfo(
            type=CaptchaType.RECAPTCHA_V2,
            website_url="https://www.google.com/sorry/index",
            website_key="6LfwuyUTAAAAAOAmoS0fdqijC2PbbdH4kjq62Y1b"
        )
        
        solution = await self.cap.solve(info)
        self.stats["captchas_solved"] += 1
        
        # 解決されたトークンを含めて再試行
        retry_url = f"https://www.google.com/search?q={keyword}&gl={location}&hl=en"
        headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
        
        # CAPTCHAトークンをリトライリクエストに含めます
        async with session.get(retry_url, headers=headers, proxy=proxy) as resp:
            return await resp.text()
    
    async def close():
        await self.cap.aclose()

class SERPMonitoringPipeline:
    """完全なSERP機能モニタリングパイプライン。"""
    
    def __init__(self, api_key: str, proxies: list):
        self.collector = SERPCollector(api_key, proxies)
        self.detector = SERPFeatureDetector()
        self.results_history = []
    
    async def monitor_keywords(self, keywords: list, location: str = "us") -> list:
        """SERP機能変化を検出するためにキーワードのバッチをモニタリングします。"""
        results = []
        
        for keyword in keywords:
            try:
                # SERPデータを収集
                html = await self.collector.search(keyword, location)
                
                # 機能を検出
                serp_result = self.detector.detect_features(html, keyword)
                results.append(serp_result)
                
                # レートリミット—5〜10秒のランダムな遅延
                await asyncio.sleep(random.uniform(5, 10))
                
            except Exception as e:
                results.append(SERPResult(
                    keyword=keyword, features_detected=[], 
                    featured_snippet_url=None, featured_snippet_text=None,
                    paa_questions=[], ai_overview_present=False,
                    ai_overview_sources=[], organic_positions=[],
                    timestamp=time.time()
                ))
        
        self.results_history.extend(results)
        return results
    
    def detect_changes(self, current: list, previous: list) -> list:
        """現在の結果と以前の結果を比較して変化を検出します。"""
        changes = []
        prev_map = {r.keyword: r for r in previous}
        
        for result in current:
            prev = prev_map.get(result.keyword)
            if not prev:
                continue
            
            # 特徴付きスニペットの取得/喪失
            had_snippet = SERPFeature.FEATURED_SNIPPET in prev.features_detected
            has_snippet = SERPFeature.FEATURED_SNIPPET in result.features_detected
            if has_snippet and not had_snippet:
                changes.append(f"[取得] '{result.keyword}'の特徴付きスニペット: {result.featured_snippet_url}")
            elif had_snippet and not has_snippet:
                changes.append(f"[喪失] '{result.keyword}'の特徴付きスニペット")
            
            # AI Overviewの出現/消失
            if result.ai_overview_present and not prev.ai_overview_present:
                changes.append(f"[新規] '{result.keyword}'にAI Overviewが出現")
            elif prev.ai_overview_present and not result.ai_overview_present:
                changes.append(f"[削除] '{result.keyword}'のAI Overviewが削除")
        
        return changes