playwright-scraping-advanced

# Core — always required
pip install playwright
playwright install chromium        # preferred for CDP + stealth

# Stealth plugin (Python)
pip install playwright-stealth

# Node.js stealth alternative
npm install playwright-extra playwright-extra-plugin-stealth

# Optional helpers
pip install httpx faker            # proxy rotation, realistic UAs

import asyncio
from playwright.async_api import async_playwright, Page, BrowserContext

STEALTH_ARGS = [
    "--no-sandbox",
    "--disable-blink-features=AutomationControlled",   # removes webdriver flag
    "--disable-infobars",
    "--disable-dev-shm-usage",
]

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(
            headless=True,
            args=STEALTH_ARGS,
        )
        context: BrowserContext = await browser.new_context(
            viewport={"width": 1366, "height": 768},
            user_agent=(
                "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
                "AppleWebKit/537.36 (KHTML, like Gecko) "
                "Chrome/124.0.0.0 Safari/537.36"
            ),
            locale="en-US",
            timezone_id="America/New_York",
        )
        page: Page = await context.new_page()

        # 1. Stealth patches (see Pillar 3)
        # 2. Route handlers (see Pillar 1)
        # 3. CDP session (see Pillar 2)
        # 4. Navigate last
        await page.goto("https://target.com", wait_until="networkidle")

        await browser.close()

asyncio.run(main())

import json

captured_data = []

async def intercept(route, request):
    response = await route.fetch()           # fetch from real server
    body = await response.body()
    try:
        captured_data.append(json.loads(body))
    except Exception:
        pass
    await route.fulfill(response=response)   # pass through to page

await page.route("**/api/products*", intercept)
await page.goto("https://shop.example.com/catalog")
await page.wait_for_load_state("networkidle")
print(f"Captured {len(captured_data)} API responses")

# Simpler for read-only — no route needed, no risk of hanging
responses = []
page.on("response", lambda resp: responses.append(resp) if "api/items" in resp.url else None)

await page.goto("https://example.com")
await page.wait_for_load_state("networkidle")

for resp in responses:
    if resp.ok:
        data = await resp.json()
        print(data)

all_pages = []

async def capture_page(route, request):
    response = await route.fetch()
    data = await response.json()
    all_pages.append(data)
    await route.fulfill(response=response)

await page.route("**/api/items*", capture_page)
await page.goto("https://example.com/items")

# Click "Load More" until it disappears
while await page.query_selector("#load-more"):
    await page.click("#load-more")
    await page.wait_for_load_state("networkidle")

print(f"Total pages captured: {len(all_pages)}")

BLOCKED_TYPES   = {"image", "media", "font", "stylesheet"}
BLOCKED_DOMAINS = {"google-analytics.com", "doubleclick.net", "hotjar.com", "facebook.com"}

async def smart_block(route, request):
    if request.resource_type in BLOCKED_TYPES:
        await route.abort()
        return
    if any(d in request.url for d in BLOCKED_DOMAINS):
        await route.abort()
        return
    await route.continue_()

await page.route("**/*", smart_block)

async def ws_handler(ws_route):
    messages = []

    async def on_message(message):
        messages.append(message)       # log inbound frames
        await ws_route.send(message)   # forward to page unchanged

    ws_route.on("message", on_message)
    await ws_route.connect()           # connect to real server

await page.route_web_socket("wss://stream.example.com/**", ws_handler)
await page.goto("https://example.com/live")
await page.wait_for_timeout(10_000)    # collect 10s of frames
print(f"Captured {len(messages)} WS frames")

# RECORD — capture all API traffic to HAR
context = await browser.new_context(
    record_har_path="capture.har",
    record_har_url_filter="**/api/**",    # only API traffic
    record_har_content="attach",          # embed response bodies
)
page = await context.new_page()
await page.goto("https://example.com")
await page.wait_for_load_state("networkidle")
await context.close()    # HAR written on close

# REPLAY — serve all requests from HAR, no real network
context = await browser.new_context()
await context.route_from_har(
    "capture.har",
    url="**/api/**",
    update=False,         # strict mode — abort on miss
)
page = await context.new_page()
await page.goto("https://example.com")   # fully offline

# Open session scoped to a specific page
client = await page.context.new_cdp_session(page)

# Enable a domain before using it
await client.send("Network.enable")

# Send a command and get the result
result = await client.send("Network.getAllCookies")
cookies = result["cookies"]

# Listen to events
client.on("Network.requestWillBeSent", lambda p: print(p["request"]["url"]))

# Detach when done (optional — auto-closes with context)
await client.detach()

client = await page.context.new_cdp_session(page)
await client.send("Network.enable")

requests: dict = {}

def on_request(params):
    requests[params["requestId"]] = {
        "url":     params["request"]["url"],
        "method":  params["request"]["method"],
        "headers": params["request"]["headers"],
    }

def on_response(params):
    rid = params["requestId"]
    if rid in requests:
        requests[rid]["status"] = params["response"]["status"]
        requests[rid]["mime"]   = params["response"]["mimeType"]

client.on("Network.requestWillBeSent", on_request)
client.on("Network.responseReceived",  on_response)

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

# Get response body via CDP (bypasses Playwright's body restrictions)
async def get_body(request_id: str) -> bytes:
    result = await client.send("Network.getResponseBody", {"requestId": request_id})
    if result.get("base64Encoded"):
        import base64
        return base64.b64decode(result["body"])
    return result["body"].encode()

# Performance metrics
await client.send("Performance.enable")
metrics_result = await client.send("Performance.getMetrics")
metrics = {m["name"]: m["value"] for m in metrics_result["metrics"]}
print(f"DOM Content Loaded: {metrics.get('DOMContentLoaded', 0):.2f}ms")
print(f"JS Heap Used: {metrics.get('JSHeapUsedSize', 0) / 1e6:.1f} MB")

# JS coverage — which code actually ran?
await client.send("Profiler.enable")
await client.send("Profiler.startPreciseCoverage", {"callCount": True, "detailed": True})

await page.goto("https://example.com")
await page.wait_for_load_state("networkidle")

result = await client.send("Profiler.takePreciseCoverage")
for script in result["result"]:
    total   = sum(r["endOffset"] - r["startOffset"] for r in script["functions"])
    covered = sum(
        r["endOffset"] - r["startOffset"]
        for fn in script["functions"]
        for r in fn["ranges"] if r["count"] > 0
    )
    pct = (covered / total * 100) if total else 0
    print(f"{script['url']}: {pct:.1f}% executed")

# CPU throttling — simulate mid-range Android device (4× slowdown)
await client.send("Emulation.setCPUThrottlingRate", {"rate": 4})

# Network throttling — simulate 3G connection
await client.send("Network.emulateNetworkConditions", {
    "offline":             False,
    "downloadThroughput":  1.5 * 1024 * 1024 / 8,   # 1.5 Mbps
    "uploadThroughput":    750 * 1024 / 8,            # 750 Kbps
    "latency":             150,                       # 150ms RTT
})

# Simulate offline
await client.send("Network.emulateNetworkConditions", {
    "offline": True, "downloadThroughput": -1, "uploadThroughput": -1, "latency": 0,
})

# Override geolocation
await client.send("Emulation.setGeolocationOverride", {
    "latitude":  1.3521,    # Singapore
    "longitude": 103.8198,
    "accuracy":  100,
})

STEALTH_ARGS = [
    "--no-sandbox",
    "--disable-blink-features=AutomationControlled",    # removes webdriver flag
    "--disable-infobars",
    "--disable-dev-shm-usage",
    "--disable-browser-side-navigation",
    "--disable-features=IsolateOrigins,site-per-process",
    "--flag-switches-begin",
    "--disable-site-isolation-trials",
    "--flag-switches-end",
]

browser = await p.chromium.launch(headless=True, args=STEALTH_ARGS)

# Always set a matching real user-agent
context = await browser.new_context(
    user_agent=(
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
        "AppleWebKit/537.36 (KHTML, like Gecko) "
        "Chrome/124.0.0.0 Safari/537.36"
    ),
)

from playwright.async_api import async_playwright
from playwright_stealth import stealth_async

async with async_playwright() as p:
    browser = await p.chromium.launch(headless=True, args=STEALTH_ARGS)
    page    = await browser.new_page()

    await stealth_async(page)   # ← MUST be called before goto()

    await page.goto("https://bot.sannysoft.com")   # detection test page
    await page.screenshot(path="stealth_test.png")

FINGERPRINT_PATCHES = """
// 1. Remove webdriver flag
Object.defineProperty(navigator, 'webdriver', {
    get: () => undefined, configurable: true,
});

// 2. Fake chrome object
window.chrome = {
    app: { isInstalled: false, InstallState: {}, RunningState: {} },
    runtime: {}, loadTimes: function() {}, csi: function() {},
};

// 3. Fake plugins (need at least 3)
Object.defineProperty(navigator, 'plugins', {
    get: () => {
        const plugins = [
            { name: 'Chrome PDF Plugin',  filename: 'internal-pdf-viewer' },
            { name: 'Chrome PDF Viewer',  filename: 'mhjfbmdgcfjbbpaeojofohoefgiehjai' },
            { name: 'Native Client',      filename: 'internal-nacl-plugin' },
        ];
        plugins.__proto__ = PluginArray.prototype;
        return plugins;
    },
});

// 4. Fix permissions API
const _origQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (p) => (
    p.name === 'notifications'
        ? Promise.resolve({ state: Notification.permission })
        : _origQuery(p)
);

// 5. Canvas noise — subtle per-session randomisation
const _origGetContext = HTMLCanvasElement.prototype.getContext;
HTMLCanvasElement.prototype.getContext = function(type, ...args) {
    const ctx = _origGetContext.apply(this, [type, ...args]);
    if (type === '2d' && ctx) {
        const _origGetImageData = ctx.getImageData;
        ctx.getImageData = function(...a) {
            const img = _origGetImageData.apply(this, a);
            for (let i = 0; i < img.data.length; i += 100) {
                img.data[i] = img.data[i] ^ (Math.random() * 3 | 0);
            }
            return img;
        };
    }
    return ctx;
};
"""

await context.add_init_script(FINGERPRINT_PATCHES)  # apply to all pages in context

import math, random, asyncio

# Natural mouse movement along a cubic Bezier curve
async def bezier_move(page, x1, y1, x2, y2, steps=30):
    cx1 = x1 + random.randint(20, 80);  cy1 = y1 + random.randint(-40, 40)
    cx2 = x2 - random.randint(20, 80);  cy2 = y2 + random.randint(-40, 40)
    for i in range(steps + 1):
        t = i / steps
        x = (1-t)**3*x1 + 3*(1-t)**2*t*cx1 + 3*(1-t)*t**2*cx2 + t**3*x2
        y = (1-t)**3*y1 + 3*(1-t)**2*t*cy1 + 3*(1-t)*t**2*cy2 + t**3*y2
        await page.mouse.move(x, y)
        await asyncio.sleep(random.uniform(0.005, 0.015))

# Human-like typing with per-keystroke delays
async def human_type(page, selector: str, text: str):
    await page.click(selector)
    await asyncio.sleep(random.uniform(0.3, 0.7))
    for char in text:
        await page.keyboard.type(char)
        await asyncio.sleep(random.uniform(0.05, 0.18))
    await asyncio.sleep(random.uniform(0.2, 0.5))

# Realistic scroll with jitter
async def scroll_down(page, total_px=2000, steps=15):
    per_step = total_px // steps
    for _ in range(steps):
        await page.mouse.wheel(0, per_step + random.randint(-30, 30))
        await asyncio.sleep(random.uniform(0.05, 0.2))

# Random wait helper — use between every action
async def jitter(min_ms=500, max_ms=2000):
    await asyncio.sleep(random.uniform(min_ms, max_ms) / 1000)

# Usage
await bezier_move(page, 100, 200, 400, 350)
await page.mouse.click(400, 350)
await jitter(300, 800)
await human_type(page, "#search", "playwright scraping")

PROXIES = [
    "http://user:pass@proxy1.example.com:8080",
    "http://user:pass@proxy2.example.com:8080",
    "socks5://user:pass@proxy3.example.com:1080",
]

async def scrape_with_rotation(urls: list[str]):
    async with async_playwright() as p:
        results = []
        for i, url in enumerate(urls):
            proxy_url = PROXIES[i % len(PROXIES)]
            browser = await p.chromium.launch(
                proxy={"server": proxy_url},
                args=STEALTH_ARGS,
            )
            context = await browser.new_context(
                proxy={"server": proxy_url},
                user_agent="Mozilla/5.0 ...",
            )
            page = await context.new_page()
            await page.goto(url)
            results.append(await page.content())
            await browser.close()
        return results

# Residential proxy (e.g. Bright Data)
PROXY_CONFIG = {
    "server":   "http://brd.superproxy.io:22225",
    "username": "YOUR_ZONE_USERNAME",
    "password": "YOUR_PASSWORD",
}
context = await browser.new_context(proxy=PROXY_CONFIG)

import httpx

async def flaresolverr_get(url: str) -> str:
    async with httpx.AsyncClient() as client:
        r = await client.post("http://localhost:8191/v1", json={
            "cmd":        "request.get",
            "url":        url,
            "maxTimeout": 60000,
        })
        return r.json()["solution"]["response"]

html = await flaresolverr_get("https://cloudflare-protected-site.com")

context = await browser.new_context(
    extra_http_headers={
        "Accept-Language":    "en-US,en;q=0.9",
        "Sec-Ch-Ua":          '"Chromium";v="124", "Google Chrome";v="124"',
        "Sec-Ch-Ua-Mobile":   "?0",
        "Sec-Ch-Ua-Platform": '"Windows"',
    }
)

async def full_stack_scrape(url: str) -> list:
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True, args=STEALTH_ARGS)
        context = await browser.new_context(
            viewport={"width": 1366, "height": 768},
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) ...",
        )
        page = await context.new_page()

        # 1. Stealth patches — BEFORE anything else
        await stealth_async(page)

        # 2. Network interception — registered before goto()
        captured = []
        async def capture_api(route, request):
            if "api/v2/products" in request.url:
                response = await route.fetch()
                captured.append(await response.json())
                await route.fulfill(response=response)
            else:
                await route.continue_()
        await page.route("**/*", capture_api)

        # 3. CDP session — after context creation
        client = await context.new_cdp_session(page)
        await client.send("Network.enable")

        # 4. Navigate — always last
        await jitter(500, 1200)                      # pre-load delay
        await page.goto(url, wait_until="networkidle")
        await jitter(2000, 4000)                     # post-load human pause

        # Scroll to trigger lazy-loaded API calls
        await scroll_down(page, total_px=3000)
        await page.wait_for_load_state("networkidle")

        await browser.close()
        return captured

TEST_URLS = [
    "https://bot.sannysoft.com",                     # webdriver, plugins, chrome object
    "https://fingerprintjs.com/demo/",               # fingerprint hash
    "https://deviceandbrowserinfo.com/",             # detailed browser signals
    "https://abrahamjuliot.github.io/creepjs/",      # overall creak score
]

for url in TEST_URLS:
    await page.goto(url)
    await page.screenshot(path=f"test_{url.split('/')[2]}.png")

# Inline JS assertions — run after stealth applied
checks = await page.evaluate("""() => ({
    webdriver: navigator.webdriver,
    hasChrome:  !!window.chrome,
    plugins:    navigator.plugins.length,
    ua:         navigator.userAgent,
    platform:   navigator.platform,
    langs:      navigator.languages,
})""")

assert checks["webdriver"] is None or checks["webdriver"] is False
assert checks["hasChrome"]  is True
assert checks["plugins"]    >= 3
assert "HeadlessChrome" not in checks["ua"]
print("✓ All stealth checks passed")