50 Browsers in Parallel

pip install httpx playwright
playwright install chromium  # Only for type stubs

export SANDBOX_API_KEY="ab0t_sk_live_YOUR_KEY"

"""
Fan out 50 cloud browsers to scrape 50 websites simultaneously.
Each browser navigates to a URL, extracts pricing data, and returns results.
Total cost: ~$0.20 for 50 browsers x 5 minutes each.
"""

import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import Optional
import httpx
from playwright.async_api import async_playwright

SANDBOX_URL = "https://sandbox.dev.ab0t.com"
API_KEY = "ab0t_sk_live_YOUR_KEY"
HEADERS = {"Authorization": f"Bearer {API_KEY}"}

# Maximum concurrent browsers. Start with 10, scale up to 50+.
MAX_CONCURRENT = 50
# Seconds to wait for container to become ready
READY_TIMEOUT = 45
# Seconds for page navigation timeout
PAGE_TIMEOUT = 30000


@dataclass
class ScrapeResult:
    url: str
    success: bool = False
    data: dict = field(default_factory=dict)
    error: Optional[str] = None
    duration_ms: int = 0
    container_id: Optional[str] = None


async def launch_browser(http: httpx.AsyncClient) -> dict:
    """Launch a cloud browser and return connection details."""
    resp = await http.post(
        f"{SANDBOX_URL}/api/browsers",
        headers=HEADERS,
        json={
            "browser_type": "headless-shell",  # Cheapest option for scraping
            "idle_timeout_minutes": 10,
        },
    )
    resp.raise_for_status()
    return resp.json()


async def wait_ready(http: httpx.AsyncClient, container_id: str) -> bool:
    """Poll until container is ready. Returns True if ready, False if timeout."""
    for _ in range(READY_TIMEOUT // 2):
        try:
            resp = await http.get(
                f"{SANDBOX_URL}/api/containers/{container_id}/status",
                headers=HEADERS,
            )
            state = resp.json().get("state", "")
            if state in ("ready", "assigned"):
                return True
        except:
            pass
        await asyncio.sleep(2)
    return False


async def terminate(http: httpx.AsyncClient, container_id: str) -> None:
    """Terminate a browser container."""
    try:
        await http.delete(
            f"{SANDBOX_URL}/api/containers/{container_id}",
            headers=HEADERS,
        )
    except:
        pass  # Best-effort cleanup


async def scrape_one(
    url: str,
    extraction_js: str,
    semaphore: asyncio.Semaphore,
    http: httpx.AsyncClient,
    pw: "AsyncPlaywrightContextManager",
) -> ScrapeResult:
    """Scrape a single URL using a dedicated cloud browser."""
    result = ScrapeResult(url=url)
    start = time.monotonic()

    async with semaphore:
        container_id = None
        try:
            # 1. Launch a browser
            browser_info = await launch_browser(http)
            container_id = browser_info["container_id"]
            result.container_id = container_id
            cdp_url = browser_info["cdp_url"]

            # 2. Wait for ready
            ready = await wait_ready(http, container_id)
            if not ready:
                result.error = "Container did not become ready"
                return result

            # 3. Connect Playwright
            browser = await pw.chromium.connect_over_cdp(cdp_url)
            context = browser.contexts[0]
            page = context.pages[0] if context.pages else await context.new_page()

            # 4. Navigate
            await page.goto(url, timeout=PAGE_TIMEOUT, wait_until="domcontentloaded")
            await page.wait_for_load_state("networkidle")

            # 5. Extract data
            data = await page.evaluate(extraction_js)
            result.data = data if isinstance(data, dict) else {"raw": data}
            result.success = True

            await browser.close()

        except Exception as e:
            result.error = str(e)[:200]

        finally:
            # 6. Always terminate the browser
            if container_id:
                await terminate(http, container_id)
            result.duration_ms = int((time.monotonic() - start) * 1000)

    return result


async def scrape_many(
    urls: list[str],
    extraction_js: str,
    max_concurrent: int = MAX_CONCURRENT,
) -> list[ScrapeResult]:
    """Scrape multiple URLs in parallel using cloud browsers."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async with httpx.AsyncClient(timeout=60) as http:
        async with async_playwright() as pw:
            tasks = [
                scrape_one(url, extraction_js, semaphore, http, pw)
                for url in urls
            ]
            results = await asyncio.gather(*tasks)

    return list(results)


# ========================================
# Example: Scrape 50 pricing pages
# ========================================

PRICING_URLS = [
    "https://vercel.com/pricing",
    "https://netlify.com/pricing",
    "https://render.com/pricing",
    "https://railway.app/pricing",
    "https://fly.io/pricing",
    "https://www.heroku.com/pricing",
    "https://www.digitalocean.com/pricing",
    "https://aws.amazon.com/pricing/",
    # ... add up to 50 URLs
]

# JavaScript that runs inside each browser to extract pricing data
EXTRACT_PRICING = """() => {
    const title = document.title;
    const h1 = document.querySelector('h1')?.textContent?.trim() || '';

    // Try to find pricing cards/plans
    const plans = Array.from(
        document.querySelectorAll('[class*="price"], [class*="plan"], [class*="tier"]')
    ).slice(0, 10).map(el => ({
        text: el.textContent?.trim()?.slice(0, 200),
    }));

    // Extract any dollar amounts on the page
    const priceRegex = /\\$[\\d,.]+(?:\\/\\w+)?/g;
    const allText = document.body.innerText;
    const prices = [...new Set(allText.match(priceRegex) || [])].slice(0, 20);

    return {
        title,
        heading: h1,
        prices,
        plan_count: plans.length,
        plans: plans.slice(0, 5),
        url: window.location.href,
    };
}"""


async def main():
    print(f"Scraping {len(PRICING_URLS)} pricing pages in parallel...")
    start = time.monotonic()

    results = await scrape_many(PRICING_URLS, EXTRACT_PRICING, max_concurrent=20)

    elapsed = time.monotonic() - start
    successes = [r for r in results if r.success]
    failures = [r for r in results if not r.success]

    print(f"\n{'='*60}")
    print(f"Completed in {elapsed:.1f}s")
    print(f"  Success: {len(successes)}/{len(results)}")
    print(f"  Failed:  {len(failures)}/{len(results)}")
    print(f"{'='*60}\n")

    # Print results
    for r in successes:
        d = r.data
        prices_str = ", ".join(d.get("prices", [])[:5])
        print(f"  {d.get('title', '?'):50s}  {prices_str}")

    if failures:
        print(f"\nFailed URLs:")
        for r in failures:
            print(f"  {r.url}: {r.error}")

    # Save full results as JSON
    output = [{
        "url": r.url,
        "success": r.success,
        "data": r.data,
        "error": r.error,
        "duration_ms": r.duration_ms,
    } for r in results]
    with open("pricing-comparison.json", "w") as f:
        json.dump(output, f, indent=2)
    print(f"\nFull results saved to pricing-comparison.json")

    # Cost estimate
    total_minutes = sum(r.duration_ms for r in results) / 60000
    cost = total_minutes * (0.10 / 60)  # browser containers at $0.1/hr
    print(f"Estimated cost: ${cost:.4f}")


asyncio.run(main())

from collections import defaultdict
from urllib.parse import urlparse

# Per-domain semaphores: max 3 concurrent requests to the same domain
domain_semaphores: dict[str, asyncio.Semaphore] = defaultdict(
    lambda: asyncio.Semaphore(3)
)

async def scrape_with_domain_limit(url: str, ...) -> ScrapeResult:
    domain = urlparse(url).netloc
    async with domain_semaphores[domain]:
        return await scrape_one(url, ...)

# Now even if you fan out 100 URLs, no single domain gets
# more than 3 concurrent requests. Different domains run
# at full concurrency.

async def scrape_with_retry(
    url: str, extraction_js: str,
    max_retries: int = 3,
    **kwargs,
) -> ScrapeResult:
    """Retry failed scrapes with exponential backoff."""
    for attempt in range(max_retries):
        result = await scrape_one(url, extraction_js, **kwargs)
        if result.success:
            return result

        # Don't retry on 404 or known-bad URLs
        if result.error and ("404" in result.error or "403" in result.error):
            return result

        # Exponential backoff: 2s, 4s, 8s
        await asyncio.sleep(2 ** attempt)
        print(f"  Retry {attempt + 1}/{max_retries}: {url}")

    return result  # Return last failure

# At scale, expect some failures. Design for partial success.
results = await scrape_many(urls, extraction_js)

successes = [r for r in results if r.success]
failures = [r for r in results if not r.success]

print(f"Success rate: {len(successes)}/{len(results)} ({len(successes)*100//len(results)}%)")

# A 90% success rate across 50 URLs is normal. Sites go down,
# page structures change, JavaScript errors happen. Don't fail
# the entire run because 5 out of 50 URLs had issues.

# Optionally retry just the failures
if failures:
    retry_results = await scrape_many(
        [r.url for r in failures],
        extraction_js,
        max_concurrent=5,  # Gentler on retry
    )
    # Merge retry successes into main results

# In scrape_one, before the except block:
except Exception as e:
    # Take a screenshot for debugging
    try:
        await page.screenshot(path=f"errors/{container_id}.png")
    except:
        pass
    result.error = str(e)[:200]

# Now you have a visual record of what the page looked like
# when the extraction failed. Was it a CAPTCHA? A different layout?
# A JavaScript error popup? The screenshot tells you instantly.

Without warm pools:
  50 browsers x 20 sec startup = 50 sec before first data

With warm pools (20 warm):
  First 20 browsers: instant (<1 sec)
  Next 30: 20 sec startup (overlapped with first 20 working)
  Net time: ~25 sec total vs ~50 sec

Cost of warm pool:
  20 idle browsers x $0.1/hr = ~$2.00/hr
  Worth it if you run 10+ times per hour.
  Not worth it for once-a-day jobs.