Playwright 深度实战

# 安装 Python 包
pip install playwright

# 安装浏览器二进制文件（约 300MB，只需一次）
playwright install

# 只安装 Chromium（最常用，体积最小）
playwright install chromium

# 安装系统依赖（Linux 服务器上需要）
playwright install-deps

# ========== 同步 API（适合脚本、简单爬虫） ==========
from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch(headless=True)
    page = browser.new_page()
    page.goto('https://example.com')
    title = page.title()
    print(title)
    browser.close()

# ========== 异步 API（适合高并发爬虫） ==========
import asyncio
from playwright.async_api import async_playwright

async def main():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        page = await browser.new_page()
        await page.goto('https://example.com')
        title = await page.title()
        print(title)
        await browser.close()

asyncio.run(main())

# 基本导航
page.goto('https://example.com')

# 带等待条件导航（等待网络空闲）
page.goto('https://example.com', wait_until='networkidle')

# 等待条件选项
# 'domcontentloaded' — DOM 解析完成（最快）
# 'load'             — 页面完全加载（默认）
# 'networkidle'      — 网络空闲（至少 500ms 无新请求，最安全但最慢）

# 前进/后退/刷新
page.go_back()
page.go_forward()
page.reload()

# Playwright 自动等待元素出现后再点击
page.click('button.submit')  # 如果按钮还没渲染，自动等待最多 30s

# 显式等待特定条件
page.wait_for_selector('.result', state='visible')  # 等待元素可见
page.wait_for_selector('.loading', state='hidden')  # 等待元素消失
page.wait_for_load_state('networkidle')              # 等待网络空闲
page.wait_for_url('**/success')                      # 等待 URL 匹配
page.wait_for_timeout(5000)                          # 固定等待 5s（不推荐，仅调试用）

# 推荐方式：Locator（自动等待+重试）
search = page.locator('input[name="q"]')
search.fill('Python爬虫')
search.press('Enter')

# 各种定位方式
page.locator('text=登录')                        # 文本匹配
page.locator('button:has-text("提交")')          # 包含文本
page.locator('input[type="email"]')              # CSS 选择器
page.locator('[data-testid="submit-btn"]')       # data-testid（最推荐）
page.get_by_role('button', name='登录')           # ARIA Role
page.get_by_text('联系我们')                       # 文本内容
page.get_by_placeholder('请输入关键词')            # placeholder
page.get_by_label('用户名')                        # label 关联
page.get_by_test_id('submit-btn')                 # data-testid

# 点击
page.click('button.submit')
page.dblclick('img.thumbnail')  # 双击

# 输入文本
page.fill('input[name="username"]', 'admin')      # 清空后输入
page.type('input[name="code"]', '1234', delay=100)  # 逐字输入（模拟人类）

# 选择下拉框
page.select_option('select#country', 'China')

# 复选框
page.check('input[type="checkbox"]')
page.uncheck('input[type="checkbox"]')

# 文件上传
page.set_input_files('input[type="file"]', '/path/to/file.pdf')

# 键盘操作
page.press('input', 'Enter')
page.keyboard.press('Control+A')  # 全选
page.keyboard.type('Hello World', delay=50)  # 模拟打字

# 鼠标操作
page.mouse.click(100, 200)
page.mouse.dblclick(100, 200)
page.mouse.move(300, 400)

# 滚动
page.mouse.wheel(0, 1000)  # 向下滚动

# 截图
page.screenshot(path='screenshot.png')
element = page.locator('.chart')
element.screenshot(path='chart.png')  # 元素截图

# 生成 PDF（仅 Chromium）
page.pdf(path='page.pdf')

def block_resources(route):
    """屏蔽图片、字体、CSS 等无关资源，加速爬取"""
    if route.request.resource_type in ('image', 'font', 'stylesheet', 'media'):
        route.abort()
    else:
        route.continue_()

page.route('**/*', block_resources)

# 现在页面加载速度提升 3-5 倍
page.goto('https://example.com')

import json

captured_data = []

def capture_api(route):
    """拦截特定 API 请求，提取响应数据"""
    if 'api.example.com/data' in route.request.url:
        # 继续请求，但监听响应
        response = route.fetch()
        body = response.json()
        captured_data.append(body)
        # 将原始响应返回给页面
        route.fulfill(response=response)
    else:
        route.continue_()

page.route('**/*', capture_api)
page.goto('https://example.com')

print(f"捕获了 {len(captured_data)} 条 API 数据")

def modify_headers(route):
    """为所有请求添加自定义 Header"""
    headers = {**route.request.headers, 'X-Custom-Header': 'value'}
    route.continue_(headers=headers)

page.route('**/*', modify_headers)

def mock_api(route):
    """拦截 API 请求，返回自定义数据"""
    if 'api.example.com/user' in route.request.url:
        route.fulfill(
            status=200,
            content_type='application/json',
            body=json.dumps({'name': 'Mock User', 'id': 999}),
        )
    else:
        route.continue_()

page.route('**/*', mock_api)

# 监听新标签页打开
with page.expect_popup() as popup_info:
    page.click('a[target="_blank"]')  # 点击会打开新标签的链接
popup = popup_info.value

# 在新标签页中操作
popup.wait_for_load_state()
print(popup.title())
popup.close()

# 获取所有页面
pages = browser.contexts[0].pages
for p in pages:
    print(p.url)

from playwright.sync_api import sync_playwright

with sync_playwright() as p:
    browser = p.chromium.launch()

    # 用户 A 的隔离会话
    context_a = browser.new_context()
    page_a = context_a.new_page()
    page_a.goto('https://example.com/login')
    page_a.fill('input[name="user"]', 'user_a')
    page_a.click('button[type="submit"]')

    # 用户 B 的隔离会话（Cookie/缓存完全独立）
    context_b = browser.new_context()
    page_b = context_b.new_page()
    page_b.goto('https://example.com/login')
    page_b.fill('input[name="user"]', 'user_b')
    page_b.click('button[type="submit"]')

    # 两个用户看到的页面内容不同
    print(page_a.locator('.username').text_content())  # user_a
    print(page_b.locator('.username').text_content())  # user_b

    browser.close()

# 获取所有 Cookie
cookies = context.cookies()

# 添加 Cookie
context.add_cookies([{
    'name': 'session_id',
    'value': 'abc123',
    'domain': 'example.com',
    'path': '/',
}])

# 清除 Cookie
context.clear_cookies()

# Cookie 持久化（保存登录状态）
import json

# 保存
storage = context.storage_state(path='auth.json')
# auth.json 包含 cookies 和 localStorage

# 恢复（下次启动时无需重新登录）
context = browser.new_context(storage_state='auth.json')
page = context.new_page()
page.goto('https://example.com/dashboard')  # 直接进入已登录页面

browser = chromium.launch(
    headless=True,
    args=[
        '--disable-blink-features=AutomationControlled',  # 隐藏 webdriver 标记
        '--disable-features=IsolateOrigins,site-per-process',
    ],
)

context = browser.new_context(
    user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 '
               '(KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
    viewport={'width': 1920, 'height': 1080},
    locale='zh-CN',
    timezone_id='Asia/Shanghai',
    geolocation={'latitude': 39.9042, 'longitude': 116.4074},
    permissions=['geolocation'],
)

page = context.new_page()

# 注入 stealth 脚本
page.add_init_script("""
    // 隐藏 webdriver
    Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
    // 伪造 plugins
    Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] });
    // 伪造 languages
    Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh', 'en'] });
""")

context = browser.new_context(
    # 模拟真实设备
    user_agent='...',
    viewport={'width': 1920, 'height': 1080},
    device_scale_factor=1,
    is_mobile=False,
    has_touch=False,
    # Canvas 指纹混淆
    color_scheme='light',
    reduced_motion='no-preference',
)

import json
from playwright.sync_api import sync_playwright

def scrape_spa_site(base_url: str, max_pages: int = 10) -> list[dict]:
    """
    爬取 SPA 单页应用 —— 无限滚动加载 + API 拦截
    """
    captured_data = []

    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context(
            user_agent='Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/125.0.0.0',
            viewport={'width': 1920, 'height': 1080},
        )
        page = context.new_page()

        # 拦截 API 请求，提取数据
        def capture_api(route):
            if 'api.example.com/items' in route.request.url:
                response = route.fetch()
                try:
                    data = response.json()
                    for item in data.get('items', []):
                        captured_data.append(item)
                except Exception:
                    pass
                route.fulfill(response=response)
            else:
                route.continue_()

        # 屏蔽无关资源加速加载
        def block_resources(route):
            if route.request.resource_type in ('image', 'font', 'stylesheet'):
                route.abort()
            else:
                route.continue_()

        page.route('**/api.example.com/**', capture_api)
        page.route('**/*', block_resources)

        # 访问页面
        page.goto(base_url, wait_until='networkidle')

        # 模拟无限滚动加载更多数据
        for i in range(max_pages):
            # 滚动到底部
            page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
            # 等待新数据加载
            page.wait_for_timeout(2000)
            print(f"已滚动 {i+1} 次，累计捕获 {len(captured_data)} 条数据")

        browser.close()

    return captured_data

# 运行
data = scrape_spa_site('https://example.com/items')
print(f"总共捕获 {len(data)} 条数据")
with open('items.json', 'w', encoding='utf-8') as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# scrapy_playwright_middleware.py
import scrapy
from scrapy.http import HtmlResponse
from playwright.sync_api import sync_playwright


class PlaywrightMiddleware:
    """Playwright 中间件：用浏览器渲染页面后返回 HtmlResponse"""

    def __init__(self):
        self.pw = sync_playwright().start()
        self.browser = self.pw.chromium.launch(headless=True)

    def process_request(self, request, spider):
        # 只对标记了 playwright=True 的请求使用浏览器渲染
        if not request.meta.get('playwright'):
            return None

        context = self.browser.new_context(
            user_agent='Mozilla/5.0 ...',
        )
        page = context.new_page()

        try:
            page.goto(request.url, wait_until='networkidle', timeout=30000)

            # 可选：等待特定元素出现
            wait_selector = request.meta.get('playwright_wait_for')
            if wait_selector:
                page.wait_for_selector(wait_selector, timeout=15000)

            # 可选：滚动加载更多内容
            if request.meta.get('playwright_scroll'):
                for _ in range(request.meta.get('playwright_scroll_count', 3)):
                    page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
                    page.wait_for_timeout(1000)

            body = page.content().encode('utf-8')
            return HtmlResponse(
                url=request.url,
                body=body,
                request=request,
                encoding='utf-8',
            )
        except Exception as e:
            spider.logger.error(f'Playwright error: {e}')
            return HtmlResponse(url=request.url, status=503, request=request)
        finally:
            context.close()

    def __del__(self):
        self.browser.close()
        self.pw.stop()

class MySpider(scrapy.Spider):
    name = 'my_spider'

    def start_requests(self):
        yield scrapy.Request(
            url='https://spa-example.com/list',
            meta={
                'playwright': True,                    # 启用 Playwright
                'playwright_wait_for': '.item-card',   # 等待元素
                'playwright_scroll': True,              # 启用滚动
                'playwright_scroll_count': 5,           # 滚动 5 次
            },
        )

    def parse(self, response):
        for item in response.css('.item-card'):
            yield {
                'title': item.css('h2::text').get(),
                'price': item.css('.price::text').get(),
            }