Metadata-Version: 2.4
Name: playwright-ez
Version: 3.4.0
Summary: A simplified Playwright wrapper with multi-tab support, element highlighting, and smart waiting
Author-email: xx <xx@qq.com>
Requires-Python: >=3.11
Description-Content-Type: text/markdown
Requires-Dist: playwright>=1.61.0
Requires-Dist: loguru>=0.7.0
Requires-Dist: croniter>=1.3.0
Provides-Extra: dev
Requires-Dist: pytest>=7.0.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.21.0; extra == "dev"
Requires-Dist: requests>=2.28.0; extra == "dev"
Provides-Extra: dual
Requires-Dist: requests>=2.28.0; extra == "dual"
Provides-Extra: ai
Requires-Dist: aiohttp>=3.9.0; extra == "ai"
Provides-Extra: ocr
Requires-Dist: pytesseract>=0.3.10; extra == "ocr"
Requires-Dist: Pillow>=10.0.0; extra == "ocr"
Provides-Extra: visual
Requires-Dist: Pillow>=10.0.0; extra == "visual"
Requires-Dist: numpy>=1.24.0; extra == "visual"

# Playwright-EZ

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

[![PyPI](https://img.shields.io/pypi/v/playwright-ez)](https://pypi.org/project/playwright-ez/)
[![Python](https://img.shields.io/pypi/pyversions/playwright-ez)](https://pypi.org/project/playwright-ez/)
[![License](https://img.shields.io/badge/license-MIT-blue)](LICENSE)

## 核心特性

| 特性 | 说明 |
|------|------|
| **多标签页管理** | 创建、切换、关闭标签页，并行/顺序执行操作 |
| **元素高亮** | 操作元素时自动高亮显示，方便调试 |
| **智能等待** | 13 种等待条件，自定义条件，带重试执行 |
| **智能选择器** | CSS / XPath / Text / Role / Placeholder / Label / 链式选择器 |
| **Cookie & Session** | 保存/加载 Cookie，会话状态管理 |
| **网络监控 & 拦截** | 请求日志、API 追踪、拦截、Mock 响应、HAR 录制 |
| **文件操作** | 上传/下载文件、批量操作、PDF 转换 |
| **反检测** | 隐藏自动化特征、人类化操作模拟 |
| **表单处理** | 智能填充、验证、登录表单快捷操作 |
| **页面对象模型** | `BasePage` / `PageFactory` / 步骤装饰器 |
| **性能分析** | 页面加载指标、Web Vitals、资源分析、性能预算 |
| **移动端模拟** | 20+ 预定义设备、GPS 定位、网络条件模拟 |
| **视觉测试** | 截图对比、差异检测、报告生成 |
| **操作录制回放** | 录制操作、回放、代码自动生成 |
| **数据提取** | 表格/列表提取、JSON-LD、结构化数据、智能爬虫 |
| **智能爬虫** | `SmartScraperV2` 自动检测页面结构、翻页抓取、增量爬取 |
| **RPA 任务流** | 步骤编排、并行/条件/子流程、重试、容错 |
| **双引擎模式** | 浏览器 + HTTP 请求双引擎，智能切换 |
| **AI 辅助** | 元素定位、页面分析、自动修复、LLM 集成 |
| **缓存优化** | TTL 缓存、LFU 缓存、Memoize、性能监控 |
| **验证码处理** | 验证码识别辅助、滑块验证码、第三方求解器 |
| **WebSocket 监控** | WS 连接监控、消息拦截、录制回放 |
| **自动登录** | 登录模板匹配、多策略登录、会话恢复 |
| **弹窗处理** | 广告弹窗自动关闭、规则管理 |
| **安全扫描** | XSS 检测、Header 检查、安全扫描 |
| **SEO 检查** | SEO 诊断、报告生成 |
| **诊断工具** | 控制台监控、错误分析、页面快照对比 |
| **上下文隔离** | 多用户会话池、独立浏览器上下文 |
| **智能恢复** | 自动错误分类、恢复策略链、优雅降级 |
| **并发执行** | 浏览器池、任务队列、负载测试、基准测试 |
| **同步版** | 所有异步功能均有同步版对应，无需 `asyncio` |

## 安装

```bash
pip install playwright-ez
playwright install chromium
```

可选依赖：

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

## 快速开始

### 异步版（推荐）

```python
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())
```

### 同步版

```python
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)
```

### 一键式便捷函数

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

```python
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 共用）：

| 写法 | 含义 |
|---|---|
| `"a"` | 子元素 `a` 的**文本** |
| `"a@href"` | 子元素 `a` 的 **href 属性** |
| `".price"` / `"[data-x]"` | 任意 CSS 选择器，`@attr` 后缀取属性 |
| `":scope"`（或 `":"`/`"self"`） | **列表项自身**的文本 |
| `"@href"` | **列表项自身**的 href 属性（省略 `:scope` 的简写） |

```python
# 列表页：每条记录自己就是一个 <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（历史行为不变）。

## 主要功能详解

### 智能爬虫

```python
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 任务流编排

```python
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    # 成功步骤列表
```

### 双引擎模式

```python
from playwright_ez import DualEngine

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

### EazySession 持久化

```python
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")
```

### 数据验证 & 限流

```python
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 辅助

```python
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")
```

### 缓存优化

```python
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):
    ...
```

### 安全扫描

```python
from playwright_ez import SecurityScanner

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

### SEO 检查

```python
from playwright_ez import SEOChecker

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

## 模块一览

| 模块 | 异步版 | 同步版 | 说明 |
|------|--------|--------|------|
| `browser` | `EasyBrowser` | `SyncEasyBrowser` | 浏览器核心（多标签页） |
| `wait` | `SmartWait` | `SyncSmartWait` | 智能等待 |
| `highlight` | `ElementHighlighter` | `SyncElementHighlighter` | 元素高亮 |
| `selector` | `SmartSelector` | `SyncSmartSelector` | 智能选择器 |
| `element` | `Element` | `SyncElement` | 元素代理 |
| `session` | `CookieManager` / `SessionManager` | `SyncCookieManager` | Cookie & Session |
| `network` | `NetworkMonitor` / `Interceptor` | `SyncNetworkMonitor` | 网络监控 & 拦截 |
| `file_handler` | `FileHandler` | `SyncFileHandler` | 文件操作 |
| `anti_detection` | `AntiDetection` | `SyncAntiDetection` | 反检测 |
| `form` | `FormHandler` | `SyncFormHandler` | 表单处理 |
| `smart_form` | `SmartFormFiller` | `SyncSmartFormFiller` | 智能表单 |
| `pom` | `BasePage` / `PageFactory` | `SyncBasePage` | POM |
| `performance` | `PerformanceAnalyzer` | `SyncPerformanceAnalyzer` | 性能分析 |
| `mobile` | `DeviceEmulator` | `SyncDeviceEmulator` | 移动端模拟 |
| `visual` | `VisualTester` | `SyncVisualTester` | 视觉测试 |
| `recorder` | `ActionRecorder` | `SyncActionRecorder` | 录制回放 |
| `debug` | `Debugger` | `SyncDebugger` | 调试工具 |
| `extract` | `SmartScraper` / `DataExtractor` | `SyncSmartScraper` | 数据提取 |
| `scraper` | `SmartScraperV2` / `RateLimiter` | `SyncSmartScraperV2` | 智能爬虫 |
| `api` | `APIClient` / `MockServer` | `SyncAPIClient` | API 测试 |
| `websocket` | `WebSocketMonitor` | `SyncWebSocketMonitor` | WebSocket |
| `rpa` | `RPATaskFlow` | `SyncRPATaskFlow` | RPA 任务流 |
| `dual_engine` | `DualEngine` | `SyncDualEngine` | 双引擎模式 |
| `easy` | `EazySession` / 一键函数 | `SyncEazySession` | 便捷函数 |
| `autologin` | `AutoLogin` | `SyncAutoLogin` | 自动登录 |
| `captcha` | `CaptchaHandler` | `SyncCaptchaHandler` | 验证码处理 |
| `ai` | `AIHelper` | `SyncAIHelper` | AI 辅助 |
| `cache` | `AsyncTTLCache` / `memoize` | `SyncTTLCache` | 缓存优化 |
| `context` | `ContextManager` / `SessionPool` | `SyncContextManager` | 上下文隔离 |
| `recovery` | `SmartRecovery` | `SyncSmartRecovery` | 智能恢复 |
| `popup` | `PopupHandler` | `SyncPopupHandler` | 弹窗处理 |
| `automation` | `InfiniteScroll` / `PaginationNav` | `SyncInfiniteScroll` | 自动化增强 |
| `concurrent` | `BrowserPool` / `LoadTester` | — | 并发执行 |
| `scheduler` | `TaskScheduler` | `SyncTaskScheduler` | 任务调度 |
| `extensions` | `ExtensionManager` | `SyncExtensionManager` | 扩展管理 |
| `state` | `PageStateManager` | `SyncPageStateManager` | 页面状态 |
| `video` | `VideoManager` | `SyncVideoManager` | 视频录制 |
| `tracing` | `TraceCollector` | `SyncTraceCollector` | 追踪 |
| `diagnostics` | `ConsoleMonitor` | — | 诊断工具 |
| `monitoring` | `SEOChecker` | — | 监控 |
| `security` | `SecurityScanner` | — | 安全扫描 |
| `test_helpers` | `AccessibilityChecker` | — | 测试辅助 |

## Pythonic 操作

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

```python
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 工具

```bash
# 截图
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），确认后再写盘：

```bash
# 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：

```python
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）
```

转换效果示例：

```diff
-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/`](examples/convert_demo/README.md)，设计文档见
[`docs/design/2026-09-30-playwright-to-ez-converter-design.md`](docs/design/2026-09-30-playwright-to-ez-converter-design.md)。

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

```python
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/`（同步）是一一对应的两套示例，都可直接运行：

```bash
./.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 服务。
   ```python
   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/`：
   ```python
   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 但示例自己把异常吞掉打出来」的软失败）：

```bash
./.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）
```

配套工具：

| 脚本 | 用途 |
|---|---|
| `scripts/run_all_examples.py` | 批量跑 + 失败分类（按异常指纹归类：MISSING_API / TYPE_ERROR / NETWORK …） |
| `scripts/analyze_example_timing.py` | 按日志时间戳切出**静默段**（间隔 >2s），定位某个示例「178s 到底花在哪」 |
| `scripts/scan_missing_stdlib_import.py` | 静态扫「用了 stdlib 名字却没 import」——抓 `await asyncio.sleep()` 这类只在特定分支才炸的 `NameError` |
| `scripts/scan_shadowing.py` | 扫「模块级导入名被局部变量遮蔽」——`url` 当循环变量会让同函数里更早的 `url()` 炸 `UnboundLocalError` |

```bash
./.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 集成

```python
# 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
