playwright-scraper.py

import asyncio
from playwright.async_api import async_playwright
import random
import time
import json
import logging
from bs4 import BeautifulSoup

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class PlaywrightScraper:
    def __init__(self, start_url, headless=False):
        self.start_url = start_url
        self.headless = headless
        self.all_products = []

    async def initialize_browser(self):
        """Initialize the Playwright browser with stealth settings."""
        self.playwright = await async_playwright().start()
        
        # Use Firefox for a different browser fingerprint than Chrome
        # This sometimes helps bypass blocks that specifically target Chrome
        browser_type = self.playwright.firefox
        
        # Create a browser context with specific options to appear more human-like
        self.browser = await browser_type.launch(
            headless=self.headless,
            slow_mo=50  # Add slight delays between actions to appear more human
        )
        
        # Create a context with specific viewport and user agent
        self.context = await self.browser.new_context(
            viewport={"width": 1920, "height": 1080},
            user_agent="Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            locale="en-US",
            timezone_id="America/New_York",
            has_touch=False,
            java_script_enabled=True,
            extra_http_headers={
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
                "Accept-Language": "en-US,en;q=0.9",
                "Referer": "https://www.google.com/",
                "DNT": "1",
            },
        )
        
        # Create a page
        self.page = await self.context.new_page()
        
        # Add human-like behaviors
        await self._add_stealth_scripts()
        
    async def _add_stealth_scripts(self):
        """Add scripts to help avoid detection."""
        # Add scripts to make the browser appear more human-like
        await self.page.add_init_script("""
        // Override the navigator.webdriver property
        Object.defineProperty(navigator, 'webdriver', {
            get: () => false,
        });
        
        // Override properties that might reveal automation
        const originalQuery = window.navigator.permissions.query;
        window.navigator.permissions.query = (parameters) => {
            if (parameters.name === 'notifications') {
                return Promise.resolve({state: Notification.permission});
            }
            return originalQuery(parameters);
        };
        
        // Add additional properties to make fingerprinting more consistent
        Object.defineProperty(navigator, 'plugins', {
            get: () => {
                return [
                    {
                        0: {type: "application/pdf"},
                        description: "Portable Document Format",
                        filename: "internal-pdf-viewer",
                        name: "Chrome PDF Plugin"
                    },
                    {
                        0: {type: "application/pdf"},
                        description: "Portable Document Format",
                        filename: "internal-pdf-viewer",
                        name: "Chrome PDF Viewer"
                    }
                ];
            }
        });
        """)
    
    async def navigate_with_retry(self, url, max_retries=3):
        """Navigate to a URL with retry mechanism."""
        for attempt in range(max_retries):
            try:
                # Random delay before navigation
                await asyncio.sleep(random.uniform(1, 3))
                
                # Set various headers to appear more like a regular browser
                await self.page.set_extra_http_headers({
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8",
                    "Accept-Language": "en-US,en;q=0.9",
                    "Referer": "https://www.google.com/",
                    "Cache-Control": "max-age=0",
                    "Connection": "keep-alive"
                })
                
                # First visit the homepage if it's the first attempt
                if attempt == 0 and "product-category" in url:
                    logger.info("Visiting homepage first to set cookies and establish session...")
                    await self.page.goto("https://www.htgsupply.com/", wait_until="domcontentloaded")
                    
                    # Perform some random scrolling on the homepage
                    await self._perform_human_browsing()
                    
                    # Wait a bit after visiting homepage
                    await asyncio.sleep(random.uniform(3, 5))
                
                logger.info(f"Navigating to {url} (attempt {attempt+1}/{max_retries})")
                
                # Navigate to the target URL
                response = await self.page.goto(url, wait_until="domcontentloaded")
                
                # Wait for the content to load
                await asyncio.sleep(3)
                
                # Check if we were blocked
                current_url = self.page.url
                page_content = await self.page.content()
                
                if "captcha" in page_content.lower() or "blocked" in page_content.lower() or "forbidden" in page_content.lower():
                    logger.warning(f"Detected blocking page on attempt {attempt+1}")
                    
                    if not self.headless:
                        logger.info("Browser is in visible mode - please solve the CAPTCHA manually if present")
                        # Wait for potential manual CAPTCHA solving
                        await asyncio.sleep(30)
                        
                        # Check if URL changed (indicating successful CAPTCHA)
                        if self.page.url != current_url:
                            logger.info("URL changed, seems CAPTCHA was solved")
                            return True
                    
                    # Try clearing cookies and cache for next attempt
                    await self.context.clear_cookies()
                    
                    # Wait longer before next attempt
                    await asyncio.sleep(random.uniform(5, 10))
                    continue
                
                # Check for successful navigation
                if response.status in [200, 301, 302]:
                    logger.info(f"Successfully navigated to {url}")
                    
                    # Perform some random scrolling to simulate human behavior
                    await self._perform_human_browsing()
                    
                    return True
                else:
                    logger.warning(f"Navigation failed with status {response.status}")
            
            except Exception as e:
                logger.error(f"Error navigating to {url}: {e}")
            
            # Wait before retrying
            await asyncio.sleep(random.uniform(5, 10))
        
        logger.error(f"Failed to navigate to {url} after {max_retries} attempts")
        return False
    
    async def _perform_human_browsing(self):
        """Perform random scrolling and mouse movements to appear human-like."""
        # Get page height
        page_height = await self.page.evaluate("document.body.scrollHeight")
        viewport_height = await self.page.evaluate("window.innerHeight")
        
        # Perform 2-5 random scroll actions
        num_scrolls = random.randint(2, 5)
        
        for _ in range(num_scrolls):
            # Random scroll position
            scroll_y = random.randint(viewport_height, page_height)
            
            # Scroll with a smooth behavior
            await self.page.evaluate(f"window.scrollTo({{top: {scroll_y}, behavior: 'smooth'}})")
            
            # Random wait between scrolls
            await asyncio.sleep(random.uniform(0.5, 2))
            
        # Return to a random position in the page
        middle_pos = random.randint(viewport_height, max(viewport_height, page_height // 2))
        await self.page.evaluate(f"window.scrollTo({{top: {middle_pos}, behavior: 'smooth'}})")
        
        # Final wait to let page respond
        await asyncio.sleep(random.uniform(1, 3))
    
    async def parse_products(self, html):
        """Parse product data from the HTML content."""
        soup = BeautifulSoup(html, 'html.parser')
        products = []
        
        # Find the content with sidebar body section
        content_section = soup.select_one('.content-with-sidebar-body')
        if not content_section:
            logger.warning("Content section '.content-with-sidebar-body' not found")
            # Try alternative containers that might hold products
            content_section = soup.select_one('ul.products')
            if not content_section:
                logger.warning("Alternative product container 'ul.products' not found")
                # Try the main content area as last resort
                content_section = soup.select_one('main') or soup.select_one('#main') or soup
        
        # Find all product items using different possible selectors
        product_items = content_section.select('li.product') or content_section.select('.product') or content_section.select('.product-item')
        
        if not product_items:
            logger.warning("No product items found with standard selectors")
            # Try broader selectors that might match product cards
            product_items = content_section.select('[class*="product"]') or content_section.select('.item')
        
        if not product_items:
            logger.error("Failed to identify product items on page")
            return products, None
            
        logger.info(f"Found {len(product_items)} potential product items on page")
        
        for item in product_items:
            product = {}
            
            # Product title - try various possible selectors
            title_selectors = [
                '.woocommerce-loop-product__title', 'h2', 'h3', '.product-title', 
                '[class*="product-title"]', '[class*="title"]', '.name'
            ]
            
            title_element = None
            for selector in title_selectors:
                title_element = item.select_one(selector)
                if title_element:
                    break
                    
            if title_element:
                product['title'] = title_element.text.strip()
                logger.debug(f"Found product: {product['title']}")
            else:
                # If we still can't find a title, skip this product
                continue
                
            # Product URL - try various link selectors
            link_selectors = [
                'a.woocommerce-LoopProduct-link', 'a.product-link', '.product-title a', 
                'h2 a', 'h3 a', '[class*="title"] a', 'a:first-child'
            ]
            
            link_element = None
            for selector in link_selectors:
                link_element = item.select_one(selector)
                if link_element:
                    break
                    
            if not link_element:
                # If we can't find a specific link, try any link
                link_element = item.select_one('a')
                
            if link_element:
                href = link_element.get('href')
                # Handle relative URLs
                if href and href.startswith('/'):
                    product['url'] = f"https://www.htgsupply.com{href}"
                else:
                    product['url'] = href
            
            # Product image - look for image elements
            img_element = item.select_one('img')
            if img_element:
                # Try different image attributes
                for attr in ['data-src', 'data-lazy-src', 'src']:
                    img_src = img_element.get(attr)
                    if img_src:
                        product['image_url'] = img_src
                        break
                        
                product['image_alt'] = img_element.get('alt', '')
            
            # Product price - try different price selectors
            price_selectors = [
                '.price', '[class*="price"]', '.amount', '.woocommerce-Price-amount'
            ]
            
            price_element = None
            for selector in price_selectors:
                price_element = item.select_one(selector)
                if price_element:
                    break
            
            if price_element:
                product['price'] = price_element.text.strip()
                
                # Try to identify sale vs. regular price
                sale_element = item.select_one('.sale-price') or item.select_one('.price ins') or item.select_one('[class*="sale"]')
                regular_element = item.select_one('.regular-price') or item.select_one('.price del') or item.select_one('[class*="regular"]')
                
                if regular_element:
                    product['regular_price'] = regular_element.text.strip()
                if sale_element:
                    product['sale_price'] = sale_element.text.strip()
            
            products.append(product)
        
        # Find next page link
        next_page_selectors = ['a.next.page-numbers', '.next', '.pagination .next', '[rel="next"]']
        next_page_link = None
        
        for selector in next_page_selectors:
            next_page_link = soup.select_one(selector)
            if next_page_link:
                break
                
        next_page_url = next_page_link.get('href') if next_page_link else None
        
        return products, next_page_url
    
    async def scrape(self, max_pages=10):
        """Main method to scrape all products with pagination."""
        try:
            await self.initialize_browser()
            
            current_url = self.start_url
            page_number = 1
            
            while current_url and page_number <= max_pages:
                logger.info(f"Scraping page {page_number}: {current_url}")
                
                # Navigate to the page
                success = await self.navigate_with_retry(current_url)
                
                if not success:
                    logger.error(f"Failed to navigate to page {page_number}")
                    break
                
                # Get the page content
                html = await self.page.content()
                
                # Parse products
                products, next_page_url = await self.parse_products(html)
                
                if products:
                    logger.info(f"Found {len(products)} products on page {page_number}")
                    self.all_products.extend(products)
                else:
                    logger.warning(f"No products found on page {page_number}")
                    if page_number == 1:
                        # If first page has no products, something is wrong
                        logger.error("No products found on first page - might be blocked or incorrect selectors")
                        # Try to save the HTML for debugging
                        with open(f"debug_page_{page_number}.html", "w", encoding="utf-8") as f:
                            f.write(html)
                        break
                
                if not next_page_url:
                    logger.info("No more pages to scrape.")
                    break
                
                current_url = next_page_url
                page_number += 1
                
                # Add a larger delay between pages to reduce chance of being blocked
                await asyncio.sleep(random.uniform(5, 10))
                
            return self.all_products
            
        finally:
            # Ensure browser is closed even if an error occurs
            if hasattr(self, 'browser'):
                await self.browser.close()
            if hasattr(self, 'playwright'):
                await self.playwright.stop()
    
    def save_to_json(self, filename="htg_grow_lights_products.json"):
        """Save scraped products to a JSON file."""
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.all_products, f, indent=4, ensure_ascii=False)
        logger.info(f"Saved {len(self.all_products)} products to {filename}")


async def main():
    """Main function to run the scraper."""
    start_url = "https://www.htgsupply.com/product-category/grow-lights/"
    
    # Try first with headless mode
    logger.info("Starting scraping with headless Playwright browser...")
    scraper = PlaywrightScraper(start_url, headless=True)
    products = await scraper.scrape(max_pages=5)  # Limit to 5 pages for testing
    
    # If headless fails, try with visible browser for manual CAPTCHA solving
    if not products:
        logger.info("Headless mode failed. Trying with visible browser...")
        logger.info("If a CAPTCHA appears, please solve it manually when the browser opens")
        scraper = PlaywrightScraper(start_url, headless=False)
        products = await scraper.scrape(max_pages=5)
    
    # Save products if any were found
    if products:
        scraper.save_to_json()
        logger.info(f"Scraping completed successfully. Total products scraped: {len(products)}")
    else:
        logger.error("Failed to scrape any products.")
        logger.info("Consider using a specialized web scraping service like ScrapingBee or BrightData")


if __name__ == "__main__":
    # Install required packages: pip install playwright bs4
    # Also need to install playwright browsers: python -m playwright install firefox
    asyncio.run(main())