कैसे स्वचालित करें AI ओवरव्यू संदर्भ निगरानी

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

export CAPSOLVER_API_KEY="आपका-capsolver-api-key"

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

@dataclass
class AIOverviewCitation:
    """एक एआई ओवरव्यू में एक उद्धरण।"""
    position: int          # उद्धरण सूची में 1-आधारित स्थिति
    url: str               # संदर्भित पृष्ठ URL
    domain: str            # संदर्भित पृष्ठ का डोमेन
    anchor_text: str       # स्रोत के लिए उपयोग किया गया पाठ
    snippet_context: str   # एआई ओवरव्यू में आसपास का पाठ

@dataclass
class AIOverviewResult:
    """एक कीवर्ड के लिए पूर्ण एआई ओवरव्यू निगरानी परिणाम।"""
    keyword: str
    has_ai_overview: bool
    citations: List[AIOverviewCitation] = field(default_factory=list)
    overview_text: str = ""
    timestamp: float = 0.0
    your_domain_cited: bool = False
    your_citation_position: Optional[int] = None

# एआई ओवरव्यूज को ट्रिगर करने वाले कीवर्ड
PRIORITY_KEYWORDS = [
    # जानकारीपूर्ण प्रश्न (एआई ओवरव्यूज की दर सबसे अधिक है)
    "जीनरेटिव इंजन ओप्टिमाइजेशन क्या है",
    "एआई खोज के लिए ओप्टिमाइज कैसे करें",
    "एआई संदर्भ के लिए सबसे अच्छी प्रथाएं",
    # व्यावसायिक जांच प्रश्न
    "सबसे अच्छा CAPTCHA हल करने वाला API",
    "2025 में शीर्ष AI एजेंट फ्रेमवर्क",
    # तुलना प्रश्न
    "reCAPTCHA vs Cloudflare Turnstile",
]

from bs4 import BeautifulSoup
import re

class AIOverviewExtractor:
    """गूगल SERP HTML से एआई ओवरव्यूज उद्धरण निकालें।"""
    
    def extract(self, html: str, keyword: str, your_domain: str = "") -> AIOverviewResult:
        """HTML को पार्स करें और एआई ओवरव्यूज उद्धरण डेटा निकालें।"""
        soup = BeautifulSoup(html, 'html.parser')
        
        # एआई ओवरव्यूज उपस्थिति की पहचान करें
        # गूगल विभिन्न कंटेनर वर्गों का उपयोग एआई ओवरव्यूज के लिए करता है
        ai_containers = soup.select(
            '[data-attrid*="ai"], '
            '.ai-overview, '
            '[class*="aiOverview"], '
            '.kp-wholepage [data-md-type]'
        )
        
        if not ai_containers:
            return AIOverviewResult(
                keyword=keyword,
                has_ai_overview=False,
                timestamp=time.time()
            )
        
        # एआई ओवरव्यूज पाठ निकालें
        overview_container = ai_containers[0]
        overview_text = overview_container.get_text(separator=" ", strip=True)[:2000]
        
        # उद्धरण (एआई ओवरव्यू में स्रोत लिंक)
        citations = []
        source_links = overview_container.select('a[href^="http"]')
        
        for i, link in enumerate(source_links, 1):
            url = link.get('href', '')
            if not url or 'google.com' in url:
                continue
            
            domain = self._extract_domain(url)
            anchor = link.get_text(strip=True)
            
            # आसपास के संदर्भ
            parent = link.parent
            context = parent.get_text(strip=True)[:200] if parent else ""
            
            citations.append(AIOverviewCitation(
                position=i,
                url=url,
                domain=domain,
                anchor_text=anchor,
                snippet_context=context
            ))
        
        # यह देखें कि क्या आपका डोमेन उद्धृत है
        your_cited = any(your_domain in c.domain for c in citations) if your_domain else False
        your_position = next(
            (c.position for c in citations if your_domain in c.domain), None
        ) if your_domain else None
        
        return AIOverviewResult(
            keyword=keyword,
            has_ai_overview=True,
            citations=citations,
            overview_text=overview_text[:500],
            timestamp=time.time(),
            your_domain_cited=your_cited,
            your_citation_position=your_position
        )
    
    def _extract_domain(self, url: str) -> str:
        """URL से डोमेन निकालें।"""
        match = re.search(r'https?://([^/]+)', url)
        return match.group(1) if match else url

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

class CitationMonitorCollector:
    """CAPTCHA हल करने के साथ एआई ओवरव्यू डेटा एकत्र करें।"""
    
    def __init__(self, api_key: str, proxies: list):
        self.cap = create_capsolver(api_key=api_key)
        self.proxies = proxies
        self.proxy_idx = 0
        self.extractor = AIOverviewExtractor()
        self.stats = {"searches": 0, "captchas": 0, "overviews_found": 0}
    
    async def collect_citation_data(
        self, keyword: str, your_domain: str = "", location: str = "us"
    ) -> AIOverviewResult:
        """एक कीवर्ड के लिए एआई ओवरव्यू उद्धरण डेटा एकत्र करें।"""
        proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        url = f"https://www.google.com/search?q={keyword}&gl={location}&hl=en"
        headers = {
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36",
            "Accept": "text/html,application/xhtml+xml",
            "Accept-Language": "en-US,en;q=0.9"
        }
        
        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 "असामान्य ट्रैफिक" in html.lower() or "g-recaptcha" in html:
                self.stats["captchas"] += 1
                html = await self._solve_and_retry(session, url, headers, proxy)
            
            self.stats["searches"] += 1
        
        # उद्धरण डेटा निकालें
        result = self.extractor.extract(html, keyword, your_domain)
        if result.has_ai_overview:
            self.stats["overviews_found"] += 1
        
        return result
    
    async def _solve_and_retry(self, session, url, headers, proxy) -> str:
        """गूगल के reCAPTCHA को हल करें और खोज को दोहराएं।"""
        info = CaptchaInfo(
            type=CaptchaType.RECAPTCHA_V2,
            website_url="https://www.google.com",
            website_key="6LfwuyUTAAAAAOAmoS0fdqijC2PbbdH4kjq62Y1b"
        )
        solution = await self.cap.solve(info)
        
        # हल करने के बाद एक नया प्रॉक्सी उपयोग करें
        new_proxy = self.proxies[self.proxy_idx % len(self.proxies)]
        self.proxy_idx += 1
        
        async with session.get(url, headers=headers, proxy=new_proxy, timeout=30) as resp:
            return await resp.text()
    
    async def monitor_batch(
        self, keywords: list, your_domain: str = "", delay: float = 7.0
    ) -> list:
        """दर्जनों कीवर्ड की निगरानी करें सीमित दर के साथ।"""
        results = []
        for keyword in keywords:
            result = await self.collect_citation_data(keyword, your_domain)
            results.append(result)
            await asyncio.sleep(delay + random.uniform(0, 3))
        return results
    
    async def close(self):
        await self.cap.aclose()

class CitationChangeDetector:
    """समय के साथ एआई ओवरव्यूज उद्धरण में बदलाव की पहचान करें।"""
    
    def __init__(self, your_domain: str):
        self.your_domain = your_domain
        self.history = {}  # कीवर्ड -> AIOverviewResult की सूची
    
    def record_and_detect(self, results: list) -> list:
        """नए परिणामों को रिकॉर्ड करें और पिछली जांच से बदलाव की पहचान करें।"""
        changes = []
        
        for result in results:
            keyword = result.keyword
            previous = self.history.get(keyword, [])
            
            if previous:
                last = previous[-1]
                
                # उद्धरण जीता
                if result.your_domain_cited and not last.your_domain_cited:
                    changes.append({
                        "type": "CITATION_GAINED",
                        "keyword": keyword,
                        "position": result.your_citation_position,
                        "timestamp": result.timestamp
                    })
                
                # उद्धरण हारा
                elif last.your_domain_cited and not result.your_domain_cited:
                    changes.append({
                        "type": "CITATION_LOST",
                        "keyword": keyword,
                        "previous_position": last.your_citation_position,
                        "timestamp": result.timestamp
                    })
                
                # स्थिति बदल गई
                elif (result.your_domain_cited and last.your_domain_cited and 
                      result.your_citation_position != last.your_citation_position):
                    changes.append({
                        "type": "POSITION_CHANGED",
                        "keyword": keyword,
                        "old_position": last.your_citation_position,
                        "new_position": result.your_citation_position,
                        "timestamp": result.timestamp
                    })
                
                # एआई ओवरव्यू दिखाई दिया/गायब हो गया
                if result.has_ai_overview and not last.has_ai_overview:
                    changes.append({
                        "type": "AI_OVERVIEW_APPEARED",
                        "keyword": keyword,
                        "timestamp": result.timestamp
                    })
            
            # इतिहास अपडेट करें
            if keyword not in self.history:
                self.history[keyword] = []
            self.history[keyword].append(result)
            # अंतिम 30 दिनों के इतिहास के साथ रखें
            self.history[keyword] = self.history[keyword][-60:]
        
        return changes
    
    def generate_report(self) -> dict:
        """उद्धरण स्थिति के बारे में एक सारांश रिपोर्ट जनरेट करें।"""
        total_keywords = len(self.history)
        cited_keywords = sum(
            1 for k, h in self.history.items() 
            if h and h[-1].your_domain_cited
        )
        overview_keywords = sum(
            1 for k, h in self.history.items()
            if h and h[-1].has_ai_overview
        )
        
        return {
            "total_keywords_monitored": total_keywords,
            "keywords_with_ai_overview": overview_keywords,
            "keywords_where_cited": cited_keywords,
            "citation_rate": f"{cited_keywords/max(overview_keywords,1)*100:.1f}%",
            "average_citation_position": self._avg_position()
        }
    
    def _avg_position(self) -> float:
        positions = [
            h[-1].your_citation_position 
            for h in self.history.values() 
            if h and h[-1].your_citation_position
]
        return sum(positions) / len(positions) if positions else 0

async def run_daily_monitoring():
    """दैनिक AI समीक्षा संदर्भ मॉनिटरिंग चलाएं।"""
    collector = CitationMonitorCollector(
        api_key="YOUR_CAPSOLVER_API_KEY",
        proxies=RESIDENTIAL_PROXY_LIST
    )
    detector = CitationChangeDetector(your_domain="capsolver.com")
    
    # सभी प्राथमिकता वाले कीवर्ड मॉनिटर करें
    results = await collector.monitor_batch(
        keywords=PRIORITY_KEYWORDS,
        your_domain="capsolver.com",
        delay=8.0
    )
    
    # परिवर्तन का पता लगाएं
    changes = detector.record_and_detect(results)
    
    # रिपोर्ट
    if changes:
        print(f"\n{'='*50}")
        print(f"संदर्भ परिवर्तन पाए गए: {len(changes)}")
        for change in changes:
            print(f"  [{change['type']}] {change['keyword']}")
        print(f"{'='*50}\n")
    
    report = detector.generate_report()
    print(f"सारांश: {report}")
    
    await collector.close()

# दिन में एक बार चलाएं
asyncio.run(run_daily_monitoring())