enhanced_documentation_extractor.py

#!/usr/bin/env python3
"""
Enhanced Documentation Extractor with Anti-Bot Detection Bypass
Implements multiple strategies to reliably access API documentation
"""

import asyncio
import time
import random
from typing import Dict, Optional, Tuple, List
from playwright.async_api import async_playwright, Page, Browser
import logging

logger = logging.getLogger(__name__)

class EnhancedDocumentationExtractor:
    """Enhanced documentation extractor with anti-bot bypass capabilities"""
    
    def __init__(self):
        self.browser: Optional[Browser] = None
        self.page: Optional[Page] = None
        self.user_agents = [
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
            'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:120.0) Gecko/20100101 Firefox/120.0',
            'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15'
        ]
        
    async def initialize(self):
        """Initialize Playwright browser with stealth configuration"""
        try:
            playwright = await async_playwright().start()
            
            # Launch browser with stealth settings
            self.browser = await playwright.chromium.launch(
                headless=True,
                args=[
                    '--no-sandbox',
                    '--disable-setuid-sandbox',
                    '--disable-dev-shm-usage',
                    '--disable-accelerated-2d-canvas',
                    '--no-first-run',
                    '--no-zygote',
                    '--disable-gpu',
                    '--disable-web-security',
                    '--disable-features=VizDisplayCompositor',
                    '--disable-blink-features=AutomationControlled'
                ]
            )
            
            # Create context with realistic settings
            context = await self.browser.new_context(
                viewport={'width': 1920, 'height': 1080},
                user_agent=random.choice(self.user_agents),
                locale='en-US',
                timezone_id='America/New_York',
                extra_http_headers={
                    'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
                    'Accept-Language': 'en-US,en;q=0.5',
                    'Accept-Encoding': 'gzip, deflate, br',
                    'DNT': '1',
                    'Connection': 'keep-alive',
                    'Upgrade-Insecure-Requests': '1',
                    'Sec-Fetch-Dest': 'document',
                    'Sec-Fetch-Mode': 'navigate',
                    'Sec-Fetch-Site': 'none',
                    'Cache-Control': 'max-age=0'
                }
            )
            
            self.page = await context.new_page()
            
            # Remove webdriver detection
            await self.page.add_init_script("""
                Object.defineProperty(navigator, 'webdriver', {
                    get: () => undefined,
                });
                
                // Remove automation indicators
                delete window.cdc_adoQpoasnfa76pfcZLmcfl_Array;
                delete window.cdc_adoQpoasnfa76pfcZLmcfl_Promise;
                delete window.cdc_adoQpoasnfa76pfcZLmcfl_Symbol;
                
                // Mock plugins
                Object.defineProperty(navigator, 'plugins', {
                    get: () => [1, 2, 3, 4, 5],
                });
                
                // Mock languages
                Object.defineProperty(navigator, 'languages', {
                    get: () => ['en-US', 'en'],
                });
            """)
            
            logger.info("Enhanced documentation extractor initialized with stealth mode")
            
        except Exception as e:
            logger.error(f"Failed to initialize enhanced documentation extractor: {e}")
            raise
            
    async def extract_documentation_with_retry(self, api_name: str, documentation_url: str, max_retries: int = 3) -> Tuple[bool, str, int]:
        """
        Extract documentation with multiple retry strategies
        
        Returns:
            Tuple of (success, content, extraction_time_ms)
        """
        start_time = time.time()
        
        for attempt in range(max_retries):
            try:
                logger.info(f"Extracting documentation for {api_name} (attempt {attempt + 1}/{max_retries})")
                logger.info(f"URL: {documentation_url}")
                
                # Random delay to appear more human-like
                if attempt > 0:
                    delay = random.uniform(2, 5)
                    await asyncio.sleep(delay)
                    
                # Rotate user agent for retries
                if attempt > 0:
                    await self.page.set_extra_http_headers({
                        'User-Agent': random.choice(self.user_agents)
                    })
                
                # Try different navigation strategies
                if attempt == 0:
                    # Standard navigation
                    await self.page.goto(documentation_url, wait_until='networkidle', timeout=30000)
                elif attempt == 1:
                    # Try with domcontentloaded
                    await self.page.goto(documentation_url, wait_until='domcontentloaded', timeout=30000)
                else:
                    # Try with load event
                    await self.page.goto(documentation_url, wait_until='load', timeout=30000)
                
                # Wait for content to load
                await self.page.wait_for_timeout(random.randint(2000, 4000))
                
                # Extract content
                content = await self.extract_relevant_content(api_name)
                
                extraction_time = int((time.time() - start_time) * 1000)
                
                if content and len(content.strip()) > 100:
                    logger.info(f"Successfully extracted {len(content)} characters for {api_name} ({extraction_time}ms)")
                    return True, content, extraction_time
                else:
                    logger.warning(f"Insufficient content extracted for {api_name} on attempt {attempt + 1}")
                    
            except Exception as e:
                logger.warning(f"Attempt {attempt + 1} failed for {api_name}: {e}")
                if attempt == max_retries - 1:
                    extraction_time = int((time.time() - start_time) * 1000)
                    return False, f"All attempts failed. Last error: {str(e)}", extraction_time
                    
        extraction_time = int((time.time() - start_time) * 1000)
        return False, "All retry attempts failed", extraction_time
        
    async def extract_relevant_content(self, api_name: str) -> str:
        """Extract relevant sections from documentation page with enhanced selectors"""
        
        # Get page title and main content
        try:
            title = await self.page.title()
        except:
            title = f"{api_name} Documentation"
        
        # Enhanced content selectors
        content_selectors = [
            'main',
            '.documentation',
            '.docs',
            '.content',
            '.api-docs',
            '#content',
            '.markdown-body',
            'article',
            '.container',
            '.doc-content',
            '.api-reference',
            '.guide',
            '.tutorial',
            '[role="main"]',
            '.main-content'
        ]
        
        extracted_content = []
        extracted_content.append(f"# {title}\n\n")
        
        # Extract from main content areas
        for selector in content_selectors:
            try:
                elements = await self.page.query_selector_all(selector)
                for element in elements[:3]:  # Limit to first 3 matches
                    text = await element.inner_text()
                    if text and len(text.strip()) > 50:
                        extracted_content.append(text.strip())
                        break
            except:
                continue
                
        # Extract code examples with enhanced selectors
        code_selectors = [
            'pre code',
            '.highlight',
            '.code-example',
            '.example',
            'code',
            '.codehilite',
            '.language-python',
            '.language-javascript',
            '.language-curl',
            '.code-block'
        ]
        
        code_examples = []
        for selector in code_selectors:
            try:
                elements = await self.page.query_selector_all(selector)
                for element in elements[:5]:  # Limit to first 5 code examples
                    code = await element.inner_text()
                    if code and len(code.strip()) > 10:
                        code_examples.append(f"```\n{code.strip()}\n```")
            except:
                continue
                
        if code_examples:
            extracted_content.append("\n## Code Examples:\n")
            extracted_content.extend(code_examples[:3])  # Limit to 3 examples
            
        # Extract authentication information with enhanced keywords
        auth_keywords = ['authentication', 'auth', 'api key', 'token', 'authorization', 'bearer', 'oauth', 'credentials']
        auth_content = await self.extract_sections_by_keywords(auth_keywords)
        if auth_content:
            extracted_content.append(f"\n## Authentication:\n{auth_content}")
            
        # Extract endpoint information with enhanced keywords
        endpoint_keywords = ['endpoint', 'url', 'request', 'parameters', 'response', 'api', 'method', 'get', 'post']
        endpoint_content = await self.extract_sections_by_keywords(endpoint_keywords)
        if endpoint_content:
            extracted_content.append(f"\n## API Endpoints:\n{endpoint_content}")
            
        final_content = "\n\n".join(extracted_content)
        
        # Limit content length to prevent token overflow
        if len(final_content) > 8000:
            final_content = final_content[:8000] + "\n\n[Content truncated for length]"
            
        return final_content
        
    async def extract_sections_by_keywords(self, keywords: list) -> str:
        """Extract sections containing specific keywords with enhanced logic"""
        try:
            # Get all text content
            page_text = await self.page.inner_text('body')
            
            # Split into sections (by headers or paragraphs)
            sections = []
            lines = page_text.split('\n')
            current_section = []
            
            for line in lines:
                line = line.strip()
                if not line:
                    if current_section:
                        sections.append('\n'.join(current_section))
                        current_section = []
                else:
                    current_section.append(line)
                    
            if current_section:
                sections.append('\n'.join(current_section))
                
            # Find sections containing keywords
            relevant_sections = []
            for section in sections:
                section_lower = section.lower()
                if any(keyword in section_lower for keyword in keywords):
                    if len(section) > 50 and len(section) < 1000:  # Reasonable length
                        relevant_sections.append(section)
                        
            return '\n\n'.join(relevant_sections[:3])  # Limit to 3 sections
            
        except Exception as e:
            logger.warning(f"Failed to extract sections by keywords: {e}")
            return ""
            
    async def cleanup(self):
        """Clean up browser resources"""
        try:
            if self.browser:
                await self.browser.close()
                logger.info("Enhanced documentation extractor cleaned up")
        except Exception as e:
            logger.warning(f"Cleanup warning: {e}")

# Test function for standalone usage
async def test_enhanced_extraction():
    """Test enhanced documentation extraction with a sample API"""
    extractor = EnhancedDocumentationExtractor()
    
    try:
        await extractor.initialize()
        
        # Test with a known problematic API
        success, content, time_ms = await extractor.extract_documentation_with_retry(
            "OpenWeatherMap",
            "https://openweathermap.org/api/one-call-3"
        )
        
        print(f"Success: {success}")
        print(f"Time: {time_ms}ms")
        print(f"Content length: {len(content)}")
        print(f"Content preview:\n{content[:500]}...")
        
    finally:
        await extractor.cleanup()

if __name__ == "__main__":
    asyncio.run(test_enhanced_extraction())