Pyppeteer: Finally Got Puppeteer Working in Python

import asyncio
from pyppeteer import launch

async def force_chromium_download():
    """
    Force Pyppeteer to download Chromium
    Run this once before using pyppeteer
    """
    import pyppeteer.chromium_downloader

    print("Downloading Chromium...")
    chromium_path = await pyppeteer.chromium_downloader.download_chromium()

    print(f"Chromium downloaded to: {chromium_path}")
    return chromium_path

async def scrape_with_pyppeteer():
    """
    Scrape with explicit executable path
    """
    # Get the correct Chromium path
    import pyppeteer.chromium_downloader
    chromium_path = pyppeteer.chromium_downloader.get_chromium_executable_path()

    print(f"Using Chromium at: {chromium_path}")

    # Launch with explicit path
    browser = await launch(
        executablePath=chromium_path,
        headless=True,
        args=['--no-sandbox', '--disable-setuid-sandbox']
    )

    page = await browser.newPage()

    await page.goto('https://example.com')

    title = await page.title()
    print(f"Page title: {title}")

    await browser.close()

# Run
if __name__ == '__main__':
    # First time: download Chromium
    # asyncio.run(force_chromium_download())

    # Then use it
    asyncio.run(scrape_with_pyppeteer())

# Fix 1: Use download mirror for China users
import os
os.environ['PYPPETEER_DOWNLOAD_HOST'] = 'https://cdn.npmmirror.com/binraries'

# Fix 2: Manual download and configuration
"""
Manual Chromium download:

1. Find your Pyppeteer version's Chromium revision:
   python -c "import pyppeteer.__main__; print(pyppeteer.__main__.CHROMIUM_REVISION)"

2. Download Chromium from:
   https://chromiumbrowserdl.appspot.com/REVISION/Linux_x64.zip

3. Extract to:
   ~/.local/share/pyppeteer/local-chromium/REVISION/

4. Verify path with get_chromium_executable_path()
"""

async def setup_chromium_manually():
    """
    Configure manually downloaded Chromium
    """
    import os
    from pyppeteer import launch

    # Option 1: Use environment variable
    os.environ['PYPPETEER_CHROMIUM_REVISION'] = '588429'

    # Option 2: Specify custom executable path
    custom_chrome_path = '/usr/bin/google-chrome'  # System Chrome

    browser = await launch(
        executablePath=custom_chrome_path,
        headless=True,
        args=['--no-sandbox']
    )

    return browser

# Fix 3: Use system Chrome (simpler, more compatible)
async def use_system_chrome():
    """
    Use system-installed Chrome instead of downloaded Chromium
    """
    # Check available Chrome installations
    chrome_paths = [
        '/usr/bin/google-chrome',
        '/usr/bin/google-chrome-stable',
        '/usr/bin/chromium-browser',
        '/Applications/Google Chrome.app/Contents/MacOS/Google Chrome',  # macOS
        'C:\\Program Files\\Google\\Chrome\\Application\\chrome.exe'  # Windows
    ]

    import os
    for path in chrome_paths:
        if os.path.exists(path):
            print(f"Found Chrome at: {path}")
            return path

    print("No system Chrome found, will download Chromium")
    return None

import asyncio
from pyppeteer import launch
from typing import List
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class AsyncPyppeteerCrawler:
    """
    Production async crawler with Pyppeteer
    """

    def __init__(self, concurrency: int = 5):
        """
        Args:
            concurrency: Number of concurrent pages to scrape
        """
        self.concurrency = concurrency
        self.browser = None

    async def init(self):
        """Initialize browser (call once)"""
        self.browser = await launch(
            headless=True,
            args=['--no-sandbox', '--disable-dev-shm-usage']
        )
        logger.info("Browser launched")

    async def close(self):
        """Close browser (call when done)"""
        if self.browser:
            await self.browser.close()
            logger.info("Browser closed")

    async def scrape_page(self, url: str) -> dict:
        """
        Scrape a single page

        Args:
            url: URL to scrape

        Returns:
            Scraped data
        """
        if not self.browser:
            raise RuntimeError("Call init() first")

        page = None
        try:
            # Create new page
            page = await self.browser.newPage()

            # Set viewport and user agent
            await page.setViewport({'width': 1920, 'height': 1080})
            await page.setUserAgent('Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36')

            # Navigate with timeout
            await page.goto(url, {'timeout': 30000, 'waitUntil': 'networkidle0'})

            # Wait for content
            await page.waitForSelector('body', {'timeout': 10000})

            # Extract data
            data = await page.evaluate('''() => ({
                title: document.title,
                url: window.location.href,
                bodyLength: document.body.innerText.length
            })''')

            logger.info(f"Scraped: {url}")
            return data

        except Exception as e:
            logger.error(f"Error scraping {url}: {e}")
            return {'error': str(e), 'url': url}

        finally:
            # Always close page to free memory
            if page:
                await page.close()

    async def scrape_urls(self, urls: List[str]) -> List[dict]:
        """
        Scrape multiple URLs concurrently

        Args:
            urls: List of URLs to scrape

        Returns:
            List of scraped data
        """
        # Create semaphore to limit concurrency
        semaphore = asyncio.Semaphore(self.concurrency)

        async def scrape_with_semaphore(url):
            async with semaphore:
                return await self.scrape_page(url)

        # Scrape all URLs concurrently
        results = await asyncio.gather(
            *[scrape_with_semaphore(url) for url in urls],
            return_exceptions=True
        )

        # Filter exceptions
        return [r for r in results if not isinstance(r, Exception)]

# Usage
async def main():
    crawler = AsyncPyppeteerCrawler(concurrency=3)

    try:
        await crawler.init()

        urls = [
            'https://example.com/page1',
            'https://example.com/page2',
            'https://example.com/page3',
            'https://example.com/page4',
            'https://example.com/page5'
        ]

        results = await crawler.scrape_urls(urls)

        for result in results:
            if 'error' not in result:
                print(f"{result['url']}: {result['title']}")

    finally:
        await crawler.close()

# Run
if __name__ == '__main__':
    asyncio.run(main())