Playwright-EZ

一个功能全面、简单易用的 Playwright 封装库，让浏览器自动化变得简单高效。

核心特性

安装

pip install playwright-ez
playwright install chromium

可选依赖：

pip install playwright-ez[dual]    # 双引擎模式（requests）
pip install playwright-ez[ocr]     # OCR 验证码识别
pip install playwright-ez[visual]  # 视觉测试（Pillow + numpy）

快速开始

异步版（推荐）

import asyncio
from playwright_ez import EasyBrowser

async def main():
    async with EasyBrowser() as browser:
        await browser.navigate("https://example.com")

        # 操作（自动高亮 + 智能等待）
        await browser.click_element("button.submit")
        await browser.type_text("input.email", "test@example.com")

        # 获取内容
        text = await browser.get_text("h1")
        print(text)

asyncio.run(main())

同步版

from playwright_ez.sync import SyncEasyBrowser

with SyncEasyBrowser() as browser:
    browser.navigate("https://example.com")
    browser.click_element("button.submit")
    text = browser.get_text("h1")
    print(text)

一键式便捷函数

无需手动管理浏览器生命周期：

import playwright_ez as ez

# 一键截图
await ez.screenshot("https://example.com", "output.png")

# 一键抓取
data = await ez.scrape("https://example.com", ".product", max_items=10)

# 一键提取链接
links = await ez.extract_links("https://example.com")

# 一键提取图片
images = await ez.extract_images("https://example.com")

# 一键抓取表格
table = await ez.scrape_table("https://example.com", "table.data")

# Session 持久化
async with ez.session("my_session") as s:
    await s.goto("https://example.com")
    await s.login(username="user", password="pass")

字段选择器：取决于 "@属性" 这套约定

scrape / scrape_list 的 fields 参数统一按一条规则解析（实现在 playwright_ez.field_selector，async / sync 共用）：

# 列表页：每条记录自己就是一个 <li>，想要"自身的文本 + 自身的 href"
items = await ez.scrape_list("https://example.com/list", "li.item", {
    "title": "h3",          # 子元素的文本
    "author": ".author",    # 子元素的文本
    "text":  ":scope",      # 列表项自身的文本
    "href":  "@href",       # 列表项自身的 href
})

⚠️ 只有列表级的 scrape_list 支持「取自身」；页面级的 scrape 没有这个概念， 传 :scope 会按原样交给 Playwright（历史行为不变）。

主要功能详解

智能爬虫

from playwright_ez import EasyBrowser

async with EasyBrowser() as browser:
    await browser.navigate("https://news.site.com")

    # 自动检测页面结构并抓取
    scraper = browser.smart_scraper()
    result = await scraper.scrape_list(max_items=20)

    # 翻页抓取
    paginated = await scraper.scrape_paginated(
        next_page_selector="a.next",
        max_pages=5,
    )

    # @attr 语法提取属性
    data = await ez.scrape("https://shop.com", "img@src")

    # 一键导出
    result.export("data.json")
    result.export("data.csv")

RPA 任务流编排

from playwright_ez import RPATaskFlow, FlowContext

flow = RPATaskFlow("数据采集流程")

# 添加步骤
flow.add_step("打开页面", lambda ctx: ctx.browser.navigate("https://example.com"))
flow.add_step("提取数据", extract_data, on_error="retry")
flow.add_step("保存结果", save_data)

# 条件分支
flow.add_conditional_step("验证数据", [
    {"condition": lambda ctx: ctx["count"] > 0, "step": "保存结果"},
    {"condition": lambda ctx: True, "step": "日志记录"},
])

# 并行执行
flow.add_parallel_steps([
    ("抓取列表", scrape_list),
    ("抓取详情", scrape_detail),
])

# 子流程嵌套
flow.add_subflow("登录流程", login_subflow, on_error="skip")

# 执行
result = await flow.run(browser=browser)
print(result)           # FlowResult 支持完整 Pythonic 操作
print(result.summary()) # 人类可读摘要
result.failed_steps     # 失败步骤列表
result.success_steps    # 成功步骤列表

双引擎模式

from playwright_ez import DualEngine

engine = DualEngine()
# 静态页面 → HTTP 请求（快速）
# 动态页面 → 浏览器渲染（准确）
response = await engine.smart_get("https://example.com")

EazySession 持久化

from playwright_ez import EazySession

async with EazySession("my_session") as s:
    # 自动保存/恢复 Cookie 和状态
    await s.goto("https://example.com")
    await s.login(username="user", password="pass")
    await s.scrape(".product", export="products.json")

数据验证 & 限流

from playwright_ez import DataValidator, BuiltInRules, RateLimiter

validator = DataValidator()
validator.add_field_rules("email", [BuiltInRules.not_empty, BuiltInRules.is_email])
validator.add_field_rules("price", [BuiltInRules.is_number])
result, cleaned = validator.validate_and_clean(data)

limiter = RateLimiter(max_requests=10, per_seconds=60)
await limiter.wait()  # 自动限流
print(len(limiter))   # 已处理请求数

AI 辅助

from playwright_ez import AIHelper, AIConfig

ai = AIHelper(AIConfig(provider="claude"))
# 智能元素定位
suggestion = await ai.find_element(page, "登录按钮")
# 自动修复失败选择器
healed = await ai.auto_heal(page, failed_selector="btn.login")

缓存优化

from playwright_ez import AsyncTTLCache, memoize

cache = AsyncTTLCache(default_ttl=300)
await cache.set("key", "value")
result = await cache.get("key")

# 函数级缓存
@memoize(ttl=60)
async def fetch_data(url):
    ...

安全扫描

from playwright_ez import SecurityScanner

scanner = SecurityScanner()
result = await scanner.scan("https://example.com")
print(result)  # XSS、Header 等安全问题

SEO 检查

from playwright_ez import SEOChecker

checker = SEOChecker()
report = await checker.check("https://example.com")
print(report)  # SEO 诊断报告

模块一览

Pythonic 操作

所有数据类均支持丰富的 Pythonic 操作，调试友好：

result = await scraper.scrape_list()
len(result)          # 数据条数
result[0]            # 按索引访问
result["title"]      # 按字段名访问
result.first         # 第一条
result.last          # 最后一条
result.filter(lambda x: x["price"] > 100)  # 过滤
result.pluck("title")   # 提取字段
result.dedup()       # 去重
result + other       # 合并两个结果
result.export("data.json")  # 导出
repr(result)         # 调试友好输出

# RPA FlowResult
len(flow_result)              # 步骤数
flow_result["步骤名"]          # 按名称查找
flow_result.failed_steps      # 失败步骤
flow_result.summary()         # 摘要

# RateLimiter
len(limiter)                  # 已处理请求数
bool(limiter)                 # 是否活跃

# DataValidator
len(validator)                # 已配置字段数
"email" in validator          # 字段是否配置了规则

CLI 工具

# 截图
playwright-ez screenshot https://example.com -o output.png

# 录制操作
playwright-ez record https://example.com -o recording.json

# 回放操作
playwright-ez replay recording.json

# 抓取页面数据
playwright-ez scrape https://example.com --links --images

# 查看可用设备
playwright-ez devices

存量代码迁移（Playwright → Playwright-EZ）

把原生 Playwright 代码（异步脚本 / 同步脚本 / pytest 用例）批量转换成 playwright-ez 风格， 默认 dry-run（只打印 diff），确认后再写盘：

# 1) 预览 diff（不落盘）
playwright-ez convert legacy/crawler.py

# 2) 写盘：生成 legacy/crawler.ez.py；--in-place 则原地改写并留 .bak
playwright-ez convert legacy/crawler.py --write
playwright-ez convert legacy/crawler.py --in-place

# 3) 目录批量 + 转换报告（迁移工作量评估）
playwright-ez convert legacy/ -r --write --report convert_report.md --report-json convert_report.json

# 4) 目标风格
playwright-ez convert legacy/crawler.py --target-style sync   # 强制同步（SyncEasyBrowser）
playwright-ez convert legacy/crawler.py --target-style easy   # 折合为一键 scrape()
playwright-ez convert legacy/crawler.py --preset stealth      # 生成 EasyBrowser.quick("stealth")

# 5) 规则覆盖不到的长尾 API → 交给 LLM 兜底（默认关闭，需显式开启）
playwright-ez convert legacy/ -r --mode hybrid \
    --llm-provider ollama --llm-model qwen2.5-coder

Python API：

from playwright_ez.converter import CodeConverter, ConvertOptions

converter = CodeConverter()
result = converter.convert_file("legacy/crawler.py", ConvertOptions(dry_run=True))
print(result.diff)          # 改写预览
print(result.coverage)      # 覆盖率
print(result.to_dict())     # 结构化报告（rewrites / todos / by_rule / by_origin）

转换效果示例：

-from playwright.async_api import async_playwright
+from playwright_ez import EasyBrowser

-    async with async_playwright() as p:
-        browser = await p.chromium.launch()
-        page = await browser.new_page()
-        await page.goto("https://example.com")
-        title = await page.inner_text("h1")
+    async with EasyBrowser() as browser:
+        await browser.navigate("https://example.com")
+        title = await browser.get_text("h1")

要点：

• 产物始终可运行：规则未覆盖的原生 API 改写为 browser.page.<原生方法>(...) 透传 + 报告记 TODO，绝不静默丢代码；
• 幂等：已是 ez 风格的文件直接跳过（already_ez）；
• 三种模式：rule（默认，纯静态分析）/ hybrid（规则 + LLM 兜底长尾片段）/ llm（当前与 hybrid 同链路）；
• 三种目标风格：auto（跟随源）/ sync（强制同步）/ easy（整段折合成 ez.scrape() 等一键函数）；
• LLM 兜底有护栏：只处理规则标记的片段，逐条 ast.parse 校验，非法回填丢弃并记入报告；
• 退出码：0 全部成功 / 2 存在 TODO 待人工处理 / 1 解析或参数错误。

完整演示见 examples/convert_demo/，设计文档见 docs/design/2026-09-30-playwright-to-ez-converter-design.md。

录制直出 ez 代码（与转换器共用同一份映射表）：

from playwright_ez.recorder import CodeGenerator

code = CodeGenerator.generate_ez(actions, use_async=True)   # → EasyBrowser 风格脚本
code = CodeGenerator.generate_ez(actions, use_async=False)  # → SyncEasyBrowser 风格脚本

跑示例

examples/（异步）与 examples/sync/（同步）是一一对应的两套示例，都可直接运行：

./.venv/Scripts/python.exe examples/basic_usage.py         # 异步
./.venv/Scripts/python.exe examples/sync/basic_usage.py    # 同步

三条约定，写新示例时请遵守：

1. 不要打第三方站点。示例统一跑 examples/_site/ 的本地演示站点，由 examples/demo_site.py 惰性起一个只监听 127.0.0.1 随机端口的 HTTP 服务。

   from demo_site import url, site_url, download_url   # sync 侧同名 shim
   
   browser.navigate(url())              # http://127.0.0.1:<port>/index.html
   browser.navigate(url("login.html"))
   

   为什么不用 example.com：IANA 已改版该页，只剩一个 <p> 和一个 <a>， 没有 h1 / form / input，示例会白等到超时；百度这类站点则会对无头 UA 下发另一套布局，选择器漂移。把正确性押在别人的页面上，示例就会周期性变红。

   为什么是 HTTP 而不是 file://：file:// 没有 origin，浏览器拒绝写 Cookie， 网络监控 / 性能 / 会话类示例就无数据可演示。

2. 输出产物一律走 tmp_path()，落到系统临时目录的 playwright-ez-demos/：

   from demo_paths import tmp_path
   
   await browser.save_screenshot(tmp_path("home.png"))
   

   别写 /tmp/x.png（Windows 上会被解释成当前盘符根下的 D:\tmp\x.png）， 也别写裸文件名（会污染仓库工作区）。

3. 同步版别用线程。Playwright 同步 API 基于 greenlet 且绑定创建它的线程， 在别的线程里碰 page / locator / download 必然抛 greenlet.error: cannot switch to a different thread。需要「后台监控」之类的能力， 就做成协作式的 poll() / watch(seconds)，由调用方在主线程驱动 （参考 SyncAutoPopupWatcher）。

批量巡检全部示例（每个示例跑在独立子进程里，记录退出码 + 输出指纹 + 耗时， 并检测「退出码 0 但示例自己把异常吞掉打出来」的软失败）：

./.venv/Scripts/python.exe scripts/run_all_examples.py            # 全部
./.venv/Scripts/python.exe scripts/run_all_examples.py sync       # 只跑同步版
./.venv/Scripts/python.exe scripts/run_all_examples.py <文件>...        # 跑指定文件
# 参数可混给：sync examples/demo_profile.py  →  跑完一大片后只重跑失败项
# 产出 .examples-sweep/report.json 与 .examples-sweep/logs/*.log（已 gitignore）

配套工具：

./.venv/Scripts/python.exe scripts/analyze_example_timing.py demo_usability
./.venv/Scripts/python.exe scripts/scan_missing_stdlib_import.py

💡 这些脚本要落成文件再跑，别用 python -c "import re; re.match(r'\d')"： Git Bash 会把 \d 当路径转换掉（→ /d:），正则静默失效，结论是错的。

Pytest 集成

# conftest.py
pytest_plugins = ["playwright_ez"]

# 测试中使用
async def test_page(browser, page):
    await browser.navigate("https://example.com")
    title = await browser.get_text("h1")
    assert title == "Example Domain"

项目结构

src/playwright_ez/
├── browser.py          # EasyBrowser 核心
├── config.py           # 全局配置
├── easy.py             # 一键式便捷函数 + EazySession
├── rpa.py              # RPA 任务流编排
├── scraper/            # 智能爬虫
│   ├── smart_scraper.py
│   ├── data_validator.py
│   ├── rate_limiter.py
│   └── incremental_crawler.py
├── extract/            # 数据提取
│   ├── basic.py / scraper.py / pagination.py
│   ├── dynamic.py / form.py / template.py
│   ├── exporter.py / types.py
├── ai/                 # AI 辅助模块
│   ├── helper.py / finder.py / analyzer.py / healer.py
│   ├── llm/            # LLM 提供商（Claude/OpenAI/Ollama）
│   └── engine/         # 规则引擎 / 混合引擎
├── concurrent/         # 并发执行（浏览器池/负载测试）
├── sync/               # 同步版（与异步版完全对应）
│   ├── easy.py / rpa.py / network.py ...
│   ├── scraper/ / ai/ / extract/ ...
└── ... (40+ 模块)

License

MIT