Python爬虫实战：Playwright/Selenium 动态渲染采集实战指南：电商、评论区深度爬取（附CSV导出 + SQLite持久化存储）！

㊙️本期内容已收录至专栏《Python爬虫实战》，持续完善知识体系与项目实战，建议先订阅收藏，后续查阅更方便～持续更新中！ ㊗️爬虫难度指数：⭐⭐⭐ 🚫声明：本数据&代码仅供学习交流，严禁用于商业用途、倒卖数据或违反目标站点的服务条款等，一切后果皆由使用者本人承担。公开榜单数据一般允许访问，但请务必遵守“君子协议”，技术无罪，责任在人。

全文目录：

• • 🌟 开篇语
  • 📋 摘要（Abstract）
  • 🎯 背景与需求（Why）
  • • 为什么需要动态渲染采集？
    • Playwright vs Selenium：如何选择？
  • 🛠️ 环境准备与项目初始化
  • • Python 版本要求
    • 完整依赖清单
    • 安装步骤
    • Docker Compose 配置
    • 项目目录结构
  • 🌐 核心实现：Playwright 引擎封装
  • • 完整代码实现
    • 代码关键点深度解析
    • • 1. 反检测核心技术详解
      • 2. 智能等待机制原理
      • 3. 无限滚动处理策略
  • 🔧 Selenium 引擎封装
  • • 完整代码实现
    • Selenium vs Playwright 实战对比
  • 🛒 实战案例1：京东商品爬虫
  • • 需求分析
    • 完整实现
    • 使用示例
    • 代码关键点解析
    • • 1. 动态价格获取技巧
      • 2. 评论分页处理
      • 3. 图片懒加载处理
  • 🗨️ 实战案例2：知乎问题爬虫
  • • 需求分析
    • 完整实现
  • 🧩 高级动态加载处理器
  • • 懒加载图片处理
    • 弹窗处理器
    • 分页处理器
  • 🤖 验证码处理
  • • 滑块验证码
    • 图片验证码（OCR）
  • 🌐 代理池管理
  • 🚀 分布式部署架构
  • • Celery 任务队列
    • Docker Compose 完整配置
    • Dockerfile
  • 📊 运行结果展示
  • • 终端输出示例
    • Celery Worker 日志
    • MongoDB 存储数据示例
  • 🐛 常见问题与解决方案
  • • 问题1：浏览器内存泄漏
    • 问题2：频繁触发验证码
    • 问题3：动态内容未加载
  • 🎓 总结与最佳实践
  • • 技术选型建议
    • 反检测最佳实践
    • 性能优化清单
    • 数据质量保证
  • 📚 延伸学习资源
  • 🎉 结语
  • 🌟 文末
  • • 📌 专栏持续更新中｜建议收藏 + 订阅
    • ✅ 互动征集

🌟 开篇语

哈喽，各位小伙伴们你们好呀～我是【喵手】。 运营社区： C站 / 掘金 / 腾讯云 / 阿里云 / 华为云 / 51CTO 欢迎大家常来逛逛，一起学习，一起进步～🌟

  我长期专注 Python 爬虫工程化实战，主理专栏 《Python爬虫实战》：从采集策略到反爬对抗，从数据清洗到分布式调度，持续输出可复用的方法论与可落地案例。内容主打一个“能跑、能用、能扩展”，让数据价值真正做到——抓得到、洗得净、用得上。

  📌 专栏食用指南（建议收藏）

• ✅ 入门基础：环境搭建 / 请求与解析 / 数据落库
• ✅ 进阶提升：登录鉴权 / 动态渲染 / 反爬对抗
• ✅ 工程实战：异步并发 / 分布式调度 / 监控与容错
• ✅ 项目落地：数据治理 / 可视化分析 / 场景化应用

📣 专栏推广时间：如果你想系统学爬虫，而不是碎片化东拼西凑，欢迎订阅/关注专栏👉《Python爬虫实战》👈    💕订阅后更新会优先推送，按目录学习更高效💯～

📋 摘要（Abstract）

本文将带你深入掌握 Playwright 和 Selenium 两大主流浏览器自动化框架，通过实战项目学习如何采集动态渲染的复杂网页，包括无限滚动列表、延迟加载详情、AJAX 分页、Shadow DOM、反爬虫对抗等高级场景。我们将以电商平台商品列表、用户评论区为实例，构建生产级爬虫系统。

读完本文你将收获：

• 掌握 Playwright 与 Selenium 的核心差异与选型策略
• 学会应对无限滚动、懒加载、动态 Token、频率限制等复杂场景
• 获得可直接应用于生产环境的完整爬虫架构（包含异常处理、断点续解反爬虫检测机制及 20+ 种对抗技巧

适用场景：

• 电商平台价格监控与竞品分析
• 社交媒体舆情采集与情感分析
• 在线评论挖掘与用户画像构建
• SaaS 产品数据迁移与备份

技术栈：

• Python 3.10+
• Playwright 1.40+ / Selenium 4.16+
• Redis（分布式任务队列）
• MongoDB（海量数据存储）
• Docker（容器化部署）

🎯 背景与需求（Why）

为什么需要动态渲染采集？

现代 Web 应用普遍采用前后端分离架构，数据通过 JavaScript 动态加载，传统的静态爬虫（如 requests + BeautifulSoup）无法获取这些内容。

典型场景对比：

真实案例分析：

# 案例1：京东商品详情页
# 静态请求返回的 HTML：
<div id="detail">
<p class="loading">正在加载...</p>
</div>

# JavaScript 异步加载后的实际内容（需动态渲染才能获取）：
<div id="detail">
<h1>Apple iPhone 15 Pro Max 256GB</h1>
<span class="price">¥9999</span>
<div class="reviews">
<!–– 10,000+ 条评论，需滚动加载 ––>
</div>
</div>

# 案例2：知乎问题的答案列表
# 首次加载只返回 5 个答案，剩余答案通过无限滚动加载
# 静态爬虫：只能获取 5 个答案
# 动态爬虫：自动滚动到底部，获取全部 500+ 个答案

Playwright vs Selenium：如何选择？

核心差异对比：

选型建议：

# 选择 Playwright 的场景：
✅ 新项目，追求高性能
✅ 需要网络拦截（如过滤广告请求）
✅ 需要采集移动端页面（WebKit）
✅ 需要绕过高级反爬虫检测
✅ 项目主要用 Python/JavaScript

# 选择 Selenium 的场景：
✅ 已有 Selenium 技术栈
✅ 需要 Safari 浏览器支持
✅ 团队熟悉 Java 生态
✅ 需要大量第三方扩展（如验证码识别插件）
✅ 需要在老旧系统上运行

性能测试对比：

# 测试任务：爬取京东某商品的 1000 条评论
# 机器配置：8 核 16GB 内存

Playwright (headless):
– 总耗时: 45 秒
– 内存占用: 280 MB
– CPU 峰值: 35%

Selenium (headless):
– 总耗时: 78 秒
– 内存占用: 450 MB
– CPU 峰值: 62%

结论：Playwright 速度快 42%，内存占用少 38%

🛠️ 环境准备与项目初始化

Python 版本要求

• 推荐版本：Python 3.10+（支持 match-case 语法、类型提示增强）
• 最低版本：Python 3.8

完整依赖清单

# requirements.txt

# 核心框架
playwright==1.40.0
selenium==4.16.0
webdriver-manager==4.0.1

# 异步支持
asyncio==3.4.3
aiohttp==3.9.1

# 数据库
pymongo==4.6.0
redis==5.0.1
motor==3.3.2 # MongoDB 异步驱动

# 数据处理
pandas==2.1.4
beautifulsoup4==4.12.3
lxml==5.1.0

# 图像处理（验证码识别）
Pillow==10.1.0
opencv-python==4.8.1.78
ddddocr==1.4.11 # OCR 验证码识别

# 代理池
requests==2.31.0
fake-useragent==1.4.0

# 任务队列
celery==5.3.4
kombu==5.3.4

# 监控与日志
loguru==0.7.2
prometheus-client==0.19.0

# 工具库
python-dotenv==1.0.0
pyyaml==6.0.1
click==8.1.7
tqdm==4.66.1
tenacity==8.2.3

# 测试
pytest==7.4.3
pytest-asyncio==0.21.1
pytest-playwright==0.4.3

安装步骤

# 1. 创建虚拟环境
python3.10 -m venv venv
source venv/bin/activate # Windows: venv\\Scripts\\activate

# 2. 安装 Python 依赖
pip install -r requirements.txt

# 3. 安装 Playwright 浏览器（约 300MB）
playwright install chromium
# 可选：安装 Firefox 和 WebKit
# playwright install firefox webkit

# 4. 安装 Selenium 浏览器驱动（自动管理）
# webdriver-manager 会自动下载 ChromeDriver

# 5. 配置环境变量
cp .env.example .env
# 编辑 .env，填入 MongoDB、Redis 等配置

# 6. 启动依赖服务（使用 Docker）
docker-compose up -d

Docker Compose 配置

# docker-compose.yml
version: '3.8'

services:
mongodb:
image: mongo:7.0
container_name: scraper_mongodb
ports:
– "27017:27017"
environment:
MONGO_INITDB_ROOT_USERNAME: admin
MONGO_INITDB_ROOT_PASSWORD: password123
volumes:
– mongodb_data:/data/db
restart: unless–stopped

redis:
image: redis:7.2–alpine
container_name: scraper_redis
ports:
– "6379:6379"
command: redis–server ––appendonly yes
volumes:
– redis_data:/data
restart: unless–stopped

# 可选：RabbitMQ（用于 Celery）
rabbitmq:
image: rabbitmq:3.12–management–alpine
container_name: scraper_rabbitmq
ports:
– "5672:5672"
– "15672:15672"
environment:
RABBITMQ_DEFAULT_USER: guest
RABBITMQ_DEFAULT_PASS: guest
volumes:
– rabbitmq_data:/var/lib/rabbitmq
restart: unless–stopped

# 可选：Prometheus + Grafana 监控
prometheus:
image: prom/prometheus:latest
container_name: scraper_prometheus
ports:
– "9090:9090"
volumes:
– ./prometheus.yml:/etc/prometheus/prometheus.yml
– prometheus_data:/prometheus
restart: unless–stopped

volumes:
mongodb_data:
redis_data:
rabbitmq_data:
prometheus_data:

项目目录结构

dynamic_scraper/
├── config/
│ ├── __init__.py
│ ├── settings.py # 全局配置
│ ├── browsers.yaml # 浏览器配置
│ └── proxies.yaml # 代理池配置
│
├── core/ # 核心引擎
│ ├── __init__.py
│ ├── playwright_engine.py # Playwright 封装
│ ├── selenium_engine.py # Selenium 封装
│ ├── base_scraper.py # 爬虫基类
│ └── session_manager.py # 会话管理（登录态保持）
│
├── scrapers/ # 具体爬虫实现
│ ├── __init__.py
│ ├── ecommerce/
│ │ ├── jd_scraper.py # 京东爬虫
│ │ ├── taobao_scraper.py # 淘宝爬虫
│ │ └── amazon_scraper.py # 亚马逊爬虫
│ ├── social/
│ │ ├── zhihu_scraper.py # 知乎爬虫
│ │ ├── weibo_scraper.py # 微博爬虫
│ │ └── twitter_scraper.py # Twitter 爬虫
│ └── review/
│ ├── dianping_scraper.py # 大众点评
│ └── meituan_scraper.py # 美团
│
├── handlers/ # 动态加载处理器
│ ├── __init__.py
│ ├── infinite_scroll.py # 无限滚动处理
│ ├── lazy_load.py # 懒加载处理
│ ├── pagination.py # 分页处理
│ ├── modal_handler.py # 弹窗处理
│ └── captcha_handler.py # 验证码处理
│
├── anti_detection/ # 反检测模块
│ ├── __init__.py
│ ├── stealth.py # 隐身技术
│ ├── fingerprint.py # 指纹伪装
│ ├── user_behavior.py # 人类行为模拟
│ └── proxy_pool.py # 代理池管理
│
├── parsers/ # 数据解析
│ ├── __init__.py
│ ├── product_parser.py # 商品解析
│ ├── review_parser.py # 评论解析
│ └── user_parser.py # 用户信息解析
│
├── storage/ # 数据存储
│ ├── __init__.py
│ ├── mongodb_storage.py # MongoDB
│ ├── redis_cache.py # Redis 缓存
│ └── file_exporter.py # 文件导出
│
├── tasks/ # 分布式任务
│ ├── __init__.py
│ ├── celery_app.py # Celery 配置
│ ├── scraper_tasks.py # 爬取任务
│ └── scheduler.py # 定时调度
│
├── utils/ # 工具模块
│ ├── __init__.py
│ ├── logger.py # 日志配置
│ ├── retry.py # 重试装饰器
│ ├── rate_limiter.py # 速率限制
│ ├── fingerprint_generator.py # 指纹生成器
│ └── validators.py # 数据验证
│
├── tests/ # 测试
│ ├── test_scrapers.py
│ ├── test_handlers.py
│ └── fixtures/
│
├── scripts/ # 脚本工具
│ ├── init_db.py # 初始化数据库
│ ├── run_scraper.py # 运行爬虫
│ └── export_data.py # 导出数据
│
├── output/ # 输出目录
│ ├── screenshots/ # 调试截图
│ ├── exports/ # 导出文件
│ └── logs/ # 日志文件
│
├── docker/ # Docker 配置
│ ├── Dockerfile
│ ├── docker–compose.yml
│ └── entrypoint.sh
│
├── .env # 环境变量（不提交）
├── .env.example # 环境变量模板
├── requirements.txt
├── README.md
└── main.py # 主入口

🌐 核心实现：Playwright 引擎封装

Playwright 是微软开发的现代浏览器自动化工具，提供了强大的异步 API 和智能等待机制。

完整代码实现

# core/playwright_engine.py
import asyncio
from typing import Optional, Dict, List, Callable, Any
from playwright.async_api import (
async_playwright,
Browser,
BrowserContext,
Page,
Playwright,
Error as PlaywrightError,
TimeoutError as PlaywrightTimeoutError
)
from loguru import logger
import random
import time
from pathlib import Path

class PlaywrightEngine:
"""
Playwright 浏览器自动化引擎

核心功能：
1. 浏览器生命周期管理（启动/关闭/重用）
2. 反检测配置（User-Agent、WebGL、Canvas 指纹等）
3. 智能等待与重试机制
4. 网络请求拦截与修改
5. 截图与性能监控

设计模式：
– 单例模式：确保全局只有一个 Playwright 实例
– 上下文管理器：自动资源清理
– 异步优先：充分利用 Python 异步特性
"""

def __init__(self, config: Dict = None):
"""
初始化 Playwright 引擎

Args:
config: 配置字典
– headless: 是否无头模式（默认 True）
– browser_type: 浏览器类型（chromium/firefox/webkit）
– viewport: 视口大小 {"width": 1920, "height": 1080}
– user_agent: 自定义 User-Agent
– proxy: 代理配置
– slow_mo: 慢动作延时（毫秒，用于调试）
– timeout: 默认超时时间（毫秒）
"""
self.config = config or {}

# 浏览器配置
self.headless = self.config.get('headless', True)
self.browser_type = self.config.get('browser_type', 'chromium')
self.viewport = self.config.get('viewport', {'width': 1920, 'height': 1080})
self.user_agent = self.config.get('user_agent', self._generate_user_agent())
self.proxy = self.config.get('proxy')
self.slow_mo = self.config.get('slow_mo', 0)
self.timeout = self.config.get('timeout', 30000)

# 状态管理
self.playwright: Optional[Playwright] = None
self.browser: Optional[Browser] = None
self.context: Optional[BrowserContext] = None
self.page: Optional[Page] = None

# 性能监控
self.request_count = 0
self.start_time = None

logger.info(f"Playwright 引擎初始化完成 [{self.browser_type}]")

async def start(self):
"""
启动 Playwright 和浏览器

工作流程：
1. 启动 Playwright 实例
2. 启动浏览器（应用反检测配置）
3. 创建浏览器上下文（Context）
4. 创建新页面
"""
try:
# 1. 启动 Playwright
self.playwright = await async_playwright().start()

# 2. 选择浏览器类型
if self.browser_type == 'chromium':
browser_launcher = self.playwright.chromium
elif self.browser_type == 'firefox':
browser_launcher = self.playwright.firefox
elif self.browser_type == 'webkit':
browser_launcher = self.playwright.webkit
else:
raise ValueError(f"不支持的浏览器类型: {self.browser_type}")

# 3. 构建启动参数
launch_options = {
'headless': self.headless,
'slow_mo': self.slow_mo,
'args': self._get_browser_args(),
}

if self.proxy:
launch_options['proxy'] = {
'server': self.proxy.get('server'),
'username': self.proxy.get('username'),
'password': self.proxy.get('password'),
}

# 4. 启动浏览器
self.browser = await browser_launcher.launch(**launch_options)

# 5. 创建浏览器上下文（隔离的环境，类似隐身模式）
context_options = {
'viewport': self.viewport,
'user_agent': self.user_agent,
'locale': 'zh-CN',
'timezone_id': 'Asia/Shanghai',
'permissions': ['geolocation'], # 授予权限
'geolocation': {'latitude': 39.9042, 'longitude': 116.4074}, # 北京
'color_scheme': 'light',
'device_scale_factor': 1,
}

self.context = await self.browser.new_context(**context_options)

# 6. 应用反检测脚本
await self._apply_stealth_scripts(self.context)

# 7. 创建新页面
self.page = await self.context.new_page()

# 8. 配置页面超时
self.page.set_default_timeout(self.timeout)

# 9. 注册事件监听
self._register_event_listeners(self.page)

self.start_time = time.time()

logger.info(f"浏览器启动成功 [headless={self.headless}]")

except Exception as e:
logger.error(f"启动浏览器失败: {e}", exc_info=True)
await self.close()
raise

def _get_browser_args(self) –> List[str]:
"""
获取浏览器启动参数（绕过反爬虫检测）

这些参数可以：
– 禁用自动化检测特征
– 模拟真实用户环境
– 提升性能

Returns:
启动参数列表
"""
args = [
# 核心反检测参数
'–disable-blink-features=AutomationControlled', # 隐藏 window.navigator.webdriver
'–disable-dev-shm-usage', # 避免共享内存不足
'–no-sandbox', # 禁用沙箱（Docker 环境需要）
'–disable-setuid-sandbox',

# 性能优化
'–disable-gpu', # 禁用 GPU（无头模式不需要）
'–disable-web-security', # 禁用同源策略（谨慎使用）
'–disable-features=IsolateOrigins,site-per-process',

# 隐私与安全
'–disable-background-networking',
'–disable-background-timer-throttling',
'–disable-backgrounding-occluded-windows',
'–disable-breakpad',
'–disable-component-extensions-with-background-pages',
'–disable-extensions',
'–disable-features=TranslateUI',
'–disable-ipc-flooding-protection',
'–disable-renderer-backgrounding',

# 窗口大小（避免被检测为无头模式）
f'–window-size={self.viewport["width"]},{self.viewport["height"]}',

# 语言和地区
'–lang=zh-CN',

# 禁用不需要的功能（减少资源消耗）
'–disable-default-apps',
'–disable-sync',
'–no-first-run',
'–no-default-browser-check',
'–disable-popup-blocking',

# 日志相关
'–log-level=3', # 只显示致命错误
'–silent',
]

return args

async def _apply_stealth_scripts(self, context: BrowserContext):
"""
应用反检测脚本

核心技术：
1. 修改 navigator.webdriver（最重要）
2. 伪装 Chrome 对象
3. 伪装 Permissions API
4. 伪装插件列表
5. 伪装 WebGL 指纹
6. 伪装 Canvas 指纹
7. 伪装 AudioContext 指纹

参考：https://github.com/berstend/puppeteer-extra/tree/master/packages/puppeteer-extra-plugin-stealth
"""
stealth_js = """
// 1. 隐藏 webdriver 特征
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});

// 2. 伪装 Chrome 对象（Chromium 专用）
window.chrome = {
runtime: {},
loadTimes: function() {},
csi: function() {},
app: {}
};

// 3. 伪装 Permissions API
const originalQuery = window.navigator.permissions.query;
window.navigator.permissions.query = (parameters) => (
parameters.name === 'notifications' ?
Promise.resolve({ state: Notification.permission }) :
originalQuery(parameters)
);

// 4. 伪装插件列表
Object.defineProperty(navigator, 'plugins', {
get: () => [
{
0: {type: "application/x-google-chrome-pdf", suffixes: "pdf", description: "Portable Document Format"},
description: "Portable Document Format",
filename: "internal-pdf-viewer",
length: 1,
name: "Chrome PDF Plugin"
},
{
0: {type: "application/pdf", suffixes: "pdf", description: ""},
description: "",
filename: "mhjfbmdgcfjbbpaeojofohoefgiehjai",
length: 1,
name: "Chrome PDF Viewer"
}
]
});

// 5. 伪装语言列表
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-CN', 'zh', 'en-US', 'en']
});

// 6. 伪装平台
Object.defineProperty(navigator, 'platform', {
get: () => 'Win32'
});

// 7. 伪装 WebGL Vendor
const getParameter = WebGLRenderingContext.prototype.getParameter;
WebGLRenderingContext.prototype.getParameter = function(parameter) {
if (parameter === 37445) { // UNMASKED_VENDOR_WEBGL
return 'Intel Inc.';
}
if (parameter === 37446) { // UNMASKED_RENDERER_WEBGL
return 'Intel Iris OpenGL Engine';
}
return getParameter.apply(this, arguments);
};

// 8. 伪装 Canvas 指纹（添加轻微噪声）
const originalToDataURL = HTMLCanvasElement.prototype.toDataURL;
HTMLCanvasElement.prototype.toDataURL = function(type) {
const context = this.getContext('2d');
const imageData = context.getImageData(0, 0, this.width, this.height);

// 添加随机噪声（不可察觉但改变指纹）
for (let i = 0; i < imageData.data.length; i += 4) {
imageData.data[i] += Math.floor(Math.random() * 5) – 2;
}

context.putImageData(imageData, 0, 0);
return originalToDataURL.apply(this, arguments);
};

// 9. 伪装 AudioContext 指纹
const audioContext = window.AudioContext || window.webkitAudioContext;
if (audioContext) {
const originalCreateAnalyser = audioContext.prototype.createAnalyser;
audioContext.prototype.createAnalyser = function() {
const analyser = originalCreateAnalyser.apply(this, arguments);
const originalGetFloaudioContext.prototype.createAnalyser = function() {
const analyser = originalCreateAnalyser.apply(this, arguments);
const originalGetFloatFrequencyData = analyser.getFloatFrequencyData;

analyser.getFloatFrequencyData = function(array) {
originalGetFloatFrequencyData.apply(this, arguments);
// 添加轻微噪声
for (let i = 0; i < array.length; i++) {
array[i] += Math.random() * 0.0001;
}
};

return analyser;
};
}

// 10. 伪装电池 API（移动设备特征）
if (navigator.getBattery) {
navigator.getBattery = () => Promise.resolve({
charging: true,
chargingTime: 0,
dischargingTime: Infinity,
level: 0.85
});
}

// 11. 移除 Headless Chrome 特征
delete navigator.__proto__.webdriver;

// 12. 伪装 Connection API
Object.defineProperty(navigator, 'connection', {
get: () => ({
effectiveType: '4g',
rtt: 50,
downlink: 10,
saveData: false
})
});
"""

# 在所有新页面创建时注入脚本
await context.add_init_script(stealth_js)

logger.debug("反检测脚本注入成功")

def _register_event_listeners(self, page: Page):
"""
注册页面事件监听器

监听事件：
– request: 网络请求（用于统计、拦截）
– response: 网络响应（用于捕获 AJAX 数据）
– console: 控制台消息（用于调试）
– dialog: 弹窗（自动处理 alert/confirm）
– crash: 页面崩溃
"""
# 监听请求
page.on('request', self._on_request)

# 监听响应
page.on('response', self._on_response)

# 监听控制台（调试用）
page.on('console', lambda msg: logger.debug(f"Console: {msg.text}"))

# 自动处理弹窗
page.on('dialog', lambda dialog: asyncio.create_task(dialog.accept()))

# 监听页面崩溃
page.on('crash', lambda: logger.error("页面崩溃！"))

logger.debug("事件监听器注册完成")

def _on_request(self, request):
"""请求事件处理器"""
self.request_count += 1

# 记录 AJAX 请求（用于分析 API）
if request.resource_type in ['xhr', 'fetch']:
logger.debug(f"XHR/Fetch: {request.method} {request.url}")

async def _on_response(self, response):
"""响应事件处理器"""
# 捕获 JSON 响应（可能包含数据）
if 'application/json' in response.headers.get('content-type', ''):
try:
json_data = await response.json()
logger.debug(f"JSON Response: {response.url[:100]}… → {len(str(json_data))} bytes")
except Exception:
pass

def _generate_user_agent(self) –> str:
"""
生成随机 User-Agent

策略：
– 使用最新的主流浏览器版本
– 模拟 Windows/macOS 平台
– 包含真实的 Chrome/Safari 特征

Returns:
User-Agent 字符串
"""
templates = [
# Windows Chrome
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

# macOS Chrome
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",

# macOS Safari
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.1 Safari/605.1.15",

# Windows Edge
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
]

return random.choice(templates)

async def goto(self, url: str, wait_until: str = 'networkidle', timeout: int = None) –> bool:
"""
导航到指定 URL

Args:
url: 目标 URL
wait_until: 等待策略
– load: 等待 load 事件
– domcontentloaded: 等待 DOM 加载
– networkidle: 等待网络空闲（推荐，适合 SPA）
timeout: 超时时间（毫秒），None 使用默认值

Returns:
是否成功导航
"""
if not self.page:
raise RuntimeError("页面未初始化，请先调用 start()")

try:
logger.info(f"正在访问: {url}")

response = await self.page.goto(
url,
wait_until=wait_until,
timeout=timeout or self.timeout
)

if response:
status = response.status
logger.info(f"页面加载完成 [HTTP {status}]")
return status < 400

return True

except PlaywrightTimeoutError:
logger.error(f"页面加载超时: {url}")
return False
except PlaywrightError as e:
logger.error(f"导航失败: {e}")
return False

async def wait_for_selector(
self,
selector: str,
timeout: int = None,
state: str = 'visible'
) –> bool:
"""
等待元素出现

Args:
selector: CSS 选择器
timeout: 超时时间（毫秒）
state: 等待状态
– attached: 元素已添加到 DOM
– detached: 元素已从 DOM 移除
– visible: 元素可见
– hidden: 元素隐藏

Returns:
是否成功等待到元素
"""
try:
await self.page.wait_for_selector(
selector,
timeout=timeout or self.timeout,
state=state
)
logger.debug(f"元素已出现: {selector}")
return True

except PlaywrightTimeoutError:
logger.warning(f"等待元素超时: {selector}")
return False

async def scroll_to_bottom(self, max_scrolls: int = 50, delay: float = 1.0):
"""
滚动到页面底部（处理无限滚动）

策略：
1. 获取当前页面高度
2. 滚动到底部
3. 等待新内容加载
4. 检查高度是否变化
5. 重复直到不再加载或达到最大次数

Args:
max_scrolls: 最大滚动次数
delay: 每次滚动后的等待时间（秒）
"""
logger.info("开始滚动到底部…")

last_height = await self.page.evaluate('document.body.scrollHeight')
scroll_count = 0

while scroll_count < max_scrolls:
# 滚动到底部
await self.page.evaluate('window.scrollTo(0, document.body.scrollHeight)')

# 等待加载
await asyncio.sleep(delay)

# 获取新高度
new_height = await self.page.evaluate('document.body.scrollHeight')

if new_height == last_height:
logger.info(f"已到达底部（共滚动 {scroll_count} 次）")
break

last_height = new_height
scroll_count += 1

logger.debug(f"滚动进度: {scroll_count}/{max_scrolls}, 高度: {new_height}")

if scroll_count >= max_scrolls:
logger.warning(f"达到最大滚动次数 ({max_scrolls})")

async def screenshot(self, path: str = None, full_page: bool = True) –> bytes:
"""
截图

Args:
path: 保存路径，None 则返回字节数据
full_page: 是否截取整个页面

Returns:
截图的字节数据
"""
if not self.page:
raise RuntimeError("页面未初始化")

screenshot_data = await self.page.screenshot(
path=path,
full_page=full_page
)

if path:
logger.info(f"截图已保存: {path}")

return screenshot_data

async def get_html(self) –> str:
"""获取当前页面的 HTML"""
if not self.page:
raise RuntimeError("页面未初始化")

return await self.page.content()

async def evaluate(self, script: str) –> Any:
"""
执行 JavaScript 代码

Args:
script: JavaScript 代码字符串

Returns:
执行结果
"""
if not self.page:
raise RuntimeError("页面未初始化")

return await self.page.evaluate(script)

async def close(self):
"""关闭浏览器并清理资源"""
try:
if self.page:
await self.page.close()
self.page = None

if self.context:
await self.context.close()
self.context = None

if self.browser:
await self.browser.close()
self.browser = None

if self.playwright:
await self.playwright.stop()
self.playwright = None

elapsed = time.time() – self.start_time if self.start_time else 0
logger.info(f"浏览器已关闭 [运行时长: {elapsed:.2f}秒, 请求数: {self.request_count}]")

except Exception as e:
logger.error(f"关闭浏览器时出错: {e}")

async def __aenter__(self):
"""异步上下文管理器入口"""
await self.start()
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器退出"""
await self.close()

# 使用示例
async def example_usage():
"""Playwright 引擎使用示例"""

# 方式1：手动管理生命周期
engine = PlaywrightEngine(config={
'headless': False, # 显示浏览器（调试用）
'slow_mo': 100, # 慢动作 100ms（便于观察）
})

await engine.start()

try:
# 访问页面
await engine.goto('https://www.example.com')

# 等待元素
await engine.wait_for_selector('h1')

# 执行 JavaScript
title = await engine.evaluate('document.title')
print(f"页面标题: {title}")

# 截图
await engine.screenshot('example.png')

finally:
await engine.close()

# 方式2：使用上下文管理器（推荐）
async with PlaywrightEngine() as engine:
await engine.goto('https://www.example.com')
html = await engine.get_html()
print(f"HTML 长度: {len(html)}")

if __name__ == '__main__':
asyncio.run(example_usage())

代码关键点深度解析

1. 反检测核心技术详解

# 问题：网站如何检测自动化工具？

# 检测方法1：navigator.webdriver
# 原理：自动化工具会在 navigator 对象上设置 webdriver 属性
if (navigator.webdriver) {
console.log('检测到自动化工具！');
block_request();
}

# 对抗方法：
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined // 返回 undefined 而不是 true
});

# ==========================================

# 检测方法2：Chrome 对象缺失
# 原理：真实 Chrome 浏览器有 window.chrome 对象，自动化工具没有
if (typeof window.chrome === 'undefined') {
console.log('这不是真实的 Chrome！');
}

# 对抗方法：
window.chrome = {
runtime: {}, // 伪装 chrome.runtime API
loadTimes: function() {},
csi: function() {},
};

# ==========================================

# 检测方法3：插件列表为空
# 原理：真实浏览器通常有插件，自动化工具默认没有
if (navigator.plugins.length === 0) {
console.log('没有插件，可疑！');
}

# 对抗方法：
Object.defineProperty(navigator, 'plugins', {
get: () => [
// 伪装 PDF 插件
{name: "Chrome PDF Plugin", filename: "internal-pdf-viewer"},
{name: "Chrome PDF Viewer", filename: "mhjfbmdgcfjbbpaeojofohoefgiehjai"}
]
});

# ==========================================

# 检测方法4：Canvas 指纹
# 原理：不同设备/浏览器绘制 Canvas 的像素略有差异，形成"指纹"
const canvas = document.createElement('canvas');
const ctx = canvas.getContext('2d');
ctx.textBaseline = 'top';
ctx.font = '14px Arial';
ctx.fillText('Hello, world!', 2, 2);
const fingerprint = canvas.toDataURL(); // 每次相同则是机器人

# 对抗方法：添加随机噪声
HTMLCanvasElement.prototype.toDataURL = function() {
const imageData = context.getImageData(0, 0, this.width, this.height);

// 为每个像素添加 ±2 的随机偏移
for (let i = 0; i < imageData.data.length; i += 4) {
imageData.data[i] += Math.floor(Math.random() * 5) – 2;
}

context.putImageData(imageData, 0, 0);
return originalToDataURL.apply(this, arguments);
};

# ==========================================

# 检测方法5：WebGL 指纹
# 原理：GPU 型号和驱动版本会影响 WebGL 渲染，形成唯一指纹
const gl = canvas.getContext('webgl');
const vendor = gl.getParameter(gl.VENDOR); // 如 "Brian Paul"（虚拟GPU）
const renderer = gl.getParameter(gl.RENDERER); // 如 "Mesa OffScreen"

# 对抗方法：伪装成真实 GPU
WebGLRenderingContext.prototype.getParameter = function(param) {
if (param === 37445) return 'Intel Inc.'; // VENDOR
if (param === 37446) return 'Intel Iris OpenGL Engine'; // RENDERER
return originalGetParameter.apply(this, arguments);
};

为什么这些技术有效？

# 真实案例测试：访问淘宝商品详情页

# 测试1：不使用反检测（原生 Playwright）
await page.goto('https://item.taobao.com/item.htm?id=12345')
# 结果：被重定向到滑块验证页面
# 原因：检测到 navigator.webdriver = true

# 测试2：使用反检测（注入 stealth 脚本）
await context.add_init_script(stealth_js)
await page.goto('https://item.taobao.com/item.htm?id=12345')
# 结果：正常显示商品详情
# 原因：navigator.webdriver = undefined，通过检测

2. 智能等待机制原理

# 为什么需要智能等待？

# 错误做法1：固定延时
await page.goto('https://example.com')
time.sleep(5) # ❌ 浪费时间，且不可靠
element = await page.query_selector('.product')

# 问题：
# – 网速快时浪费 3-4 秒
# – 网速慢时 5 秒可能不够
# – 无法适应不同网络环境

# ==========================================

# 错误做法2：轮询检查
while True:
element = await page.query_selector('.product')
if element:
break
await asyncio.sleep(0.1) # ❌ 消耗 CPU，不优雅

# 问题：
# – 高频轮询消耗资源
# – 没有超时机制
# – 代码冗长

# ==========================================

# 正确做法：Playwright 内置智能等待
await page.wait_for_selector('.product', state='visible', timeout=30000)

# 优势：
# 1. 底层使用事件驱动（MutationObserver），零 CPU 消耗
# 2. 元素一出现立即返回（最快）
# 3. 自动超时（30 秒）
# 4. 可指定等待状态（visible/hidden/attached）

智能等待的内部实现（简化版）：

// Playwright 内部如何实现 wait_for_selector

async function waitForSelector(selector, options) {
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
reject(new TimeoutError(`等待元素超时: ${selector}`));
}, options.timeout);

// 使用 MutationObserver 监听 DOM 变化
const observer = new MutationObserver(() => {
const element = document.querySelector(selector);

if (element) {
// 检查是否满足状态要求
if (options.state === 'visible' && isVisible(element)) {
clearTimeout(timeout);
observer.disconnect();
resolve(element);
}
}
});

observer.observe(document.body, {
childList: true, // 监听子节点变化
subtree: true, // 监听所有后代节点
attributes: true // 监听属性变化（如 class, style）
});

// 首次检查（可能元素已存在）
const element = document.querySelector(selector);
if (element && isVisible(element)) {
clearTimeout(timeout);
observer.disconnect();
resolve(element);
}
});
}

function isVisible(element) {
const style = window.getComputedStyle(element);
return (
style.display !== 'none' &&
style.visibility !== 'hidden' &&
style.opacity !== '0' &&
element.offsetWidth > 0 &&
element.offsetHeight > 0
);
}

3. 无限滚动处理策略

# 场景：电商平台的商品列表，滚动加载 1000+ 个商品

async def scroll_to_bottom_naive(page):
"""❌ 错误实现：可能过早停止"""
last_height = await page.evaluate('document.body.scrollHeight')

while True:
await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
await asyncio.sleep(1)

new_height = await page.evaluate('document.body.scrollHeight')

if new_height == last_height:
break # ❌ 问题：如果网络慢，1 秒内没加载完就误判

last_height = new_height

# ==========================================

async def scroll_to_bottom_robust(page, max_scrolls=50, patience=3):
"""✅ 正确实现：更稳健"""
last_height = await page.evaluate('document.body.scrollHeight')
no_change_count = 0 # 连续未变化次数
scroll_count = 0

while scroll_count < max_scrolls:
# 滚动到底部
await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')

# 等待加载（可以根据网络情况调整）
await asyncio.sleep(random.uniform(1.5, 2.5)) # 随机延时，更像人类

# 获取新高度
new_height = await page.evaluate('document.body.scrollHeight')

if new_height == last_height:
no_change_count += 1

if no_change_count >= patience:
# 连续 3 次未变化，真的到底了
logger.info(f"确认到达底部（{scroll_count} 次滚动）")
break
else:
no_change_count = 0 # 重置计数器
last_height = new_height

scroll_count += 1

# 可选：检查是否出现"没有更多"提示
no_more = await page.query_selector('.no-more, .end-tip')
if no_more:
logger.info("检测到'没有更多'提示")
break

return scroll_count

# ==========================================

async def scroll_to_bottom_with_progress(page, target_count=None):
"""✅ 高级实现：带进度检测"""
scroll_count = 0

while True:
# 滚动前统计当前商品数量
before_count = await page.evaluate('''
document.querySelectorAll('.product-item').length
''')

# 滚动
await page.evaluate('window.scrollTo(0, document.body.scrollHeight)')
await asyncio.sleep(2)

# 滚动后统计商品数量
after_count = await page.evaluate('''
document.querySelectorAll('.product-item').length
''')

new_items = after_count – before_count
logger.info(f"本次加载 {new_items} 个商品，当前总数: {after_count}")

scroll_count += 1

# 停止条件1：没有新商品加载
if new_items == 0:
logger.info("没有新商品，停止滚动")
break

# 停止条件2：达到目标数量
if target_count and after_count >= target_count:
logger.info(f"已达到目标数量 {target_count}")
break

# 停止条件3：超过最大滚动次数
if scroll_count >= 100:
logger.warning("达到最大滚动次数")
break

return after_count

为什么需要这么复杂的逻辑？

# 真实场景：京东商品搜索结果页
# URL: https://search.jd.com/Search?keyword=iPhone

# 问题1：加载速度不一致
# – 网速快：0.5 秒加载完
# – 网速慢：3 秒才加载完
# – 服务器繁忙：偶尔 5 秒
# → 解决：使用"耐心等待"策略，连续 N 次未变化才停止

# 问题2：误判到底
# – 页面高度偶尔会抖动（广告加载/移除）
# – 某次滚动刚好遇到抖动，高度未变化
# – 误以为到底，实际还有很多内容
# → 解决：检查商品数量变化，而不仅仅是高度

# 问题3：无限循环
# – 某些网站的"无限滚动"是真的无限（动态生成假数据）
# – 如果不设置上限，爬虫会一直运行
# → 解决：设置 max_scrolls 上限

🔧 Selenium 引擎封装

虽然 Playwright 更现代，但 Selenium 仍然是最成熟的选择，特别是在需要 Safari 支持或已有技术栈的场景。

完整代码实现

# core/selenium_engine.py
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.chrome.options import Options
from selenium.common.exceptions import (
TimeoutException,
NoSuchElementException,
WebDriverException
)
from webdriver_manager.chrome import ChromeDriverManager
from loguru import logger
from typing import Optional, Dict, List, Any
import time
import random

class SeleniumEngine:
"""
Selenium WebDriver 封装类

功能：
1. 自动管理 ChromeDriver（无需手动下载）
2. 反检测配置
3. 显式等待封装
4. Cookie 管理
5. 代理支持
"""

def __init__(self, config: Dict = None):
"""
初始化 Selenium 引擎

Args:
config: 配置字典
– headless: 是否无头模式
– proxy: 代理配置
– user_data_dir: 用户数据目录（保存登录态）
– disable_images: 是否禁用图片加载
– timeout: 默认超时时间（秒）
"""
self.config = config or {}

self.headless = self.config.get('headless', True)
self.proxy = self.config.get('proxy')
self.user_data_dir = self.config.get('user_data_dir')
self.disable_images = self.config.get('disable_images', False)
self.timeout = self.config.get('timeout', 30)

self.driver: Optional[webdriver.Chrome] = None
self.wait: Optional[WebDriverWait] = None

logger.info("Selenium 引擎初始化完成")

def start(self):
"""启动 Selenium WebDriver"""
try:
# 1. 配置 Chrome Options
chrome_options = Options()

# 无头模式
if self.headless:
chrome_options.add_argument('–headless=new') # 新版无头模式
chrome_options.add_argument('–disable-gpu')

# 反检测参数（关键！）
chrome_options.add_argument('–disable-blink-features=AutomationControlled')
chrome_options.add_experimental_option('excludeSwitches', ['enable-automation'])
chrome_options.add_experimental_option('useAutomationExtension', False)

# 性能优化
chrome_options.add_argument('–no-sandbox')
chrome_options.add_argument('–disable-dev-shm-usage')
chrome_options.add_argument('–disable-extensions')

# 禁用图片（提速）
if self.disable_images:
prefs = {
'profile.managed_default_content_settings.images': 2,
'profile.default_content_setting_values': {
'images': 2
}
}
chrome_options.add_experimental_option('prefs', prefs)

# 用户数据目录（保存 Cookie）
if self.user_data_dir:
chrome_options.add_argument(f'–user-data-dir={self.user_data_dir}')

# 代理设置
if self.proxy:
chrome_options.add_argument(f'–proxy-server={self.proxy["server"]}')

# 窗口大小
chrome_options.add_argument('–window-size=1920,1080')

# 语言设置
chrome_options.add_argument('–lang=zh-CN')

# User-Agent
user_agent = self._generate_user_agent()
chrome_options.add_argument(f'user-agent={user_agent}')

# 2. 使用 webdriver-manager 自动管理 ChromeDriver
service = Service(ChromeDriverManager().install())

# 3. 启动浏览器
self.driver = webdriver.Chrome(
service=service,
options=chrome_options
)

# 4. 应用反检测脚本
self._apply_stealth_scripts()

# 5. 设置隐式等待
self.driver.implicitly_wait(10)

# 6. 创建显式等待对象
self.wait = WebDriverWait(self.driver, self.timeout)

logger.info(f"WebDriver 启动成功 [headless={self.headless}]")

except Exception as e:
logger.error(f"启动 WebDriver 失败: {e}", exc_info=True)
raise

def _apply_stealth_scripts(self):
"""应用反检测脚本"""
# 执行 CDP 命令（Chrome DevTools Protocol）
self.driver.execute_cdp_cmd('Page.addScriptToEvaluateOnNewDocument', {
'source': '''
// 隐藏 webdriver 特征
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});

// 伪装 Chrome 对象
window.chrome = {
runtime: {}
};

// 伪装插件
Object.defineProperty(navigator, 'plugins', {
get: () => [1, 2, 3, 4, 5]
});

// 伪装语言
Object.defineProperty(navigator, 'languages', {
get: () => ['zh-CN', 'zh', 'en']
});
'''
})

logger.debug("反检测脚本注入成功")

def _generate_user_agent(self) –> str:
"""生成随机 User-Agent"""
templates = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
]
return random.choice(templates)

def get(self, url: str, wait_time: float = 0) –> bool:
"""
访问 URL

Args:
url: 目标 URL
wait_time: 额外等待时间（秒）

Returns:
是否成功加载
"""
try:
logger.info(f"正在访问: {url}")
self.driver.get(url)

if wait_time > 0:
time.sleep(wait_time)

logger.info("页面加载完成")
return True

except WebDriverException as e:
logger.error(f"页面加载失败: {e}")
return False

def wait_for_element(
self,
locator: tuple,
timeout: int = None,
condition: str = 'presence'
) –> bool:
"""
等待元素出现

Args:
locator: 定位器元组，如 (By.CSS_SELECTOR, '.product')
timeout: 超时时间（秒）
condition: 等待条件
– presence: 元素存在于 DOM（可能不可见）
– visibility: 元素可见
– clickable: 元素可点击

Returns:
是否成功等待到元素
"""
try:
wait = WebDriverWait(self.driver, timeout or self.timeout)

if condition == 'presence':
wait.until(EC.presence_of_element_located(locator))
elif condition == 'visibility':
wait.until(EC.visibility_of_element_located(locator))
elif condition == 'clickable':
wait.until(EC.element_to_be_clickable(locator))

logger.debug(f"元素已出现: {locator}")
return True

except TimeoutException:
logger.warning(f"等待元素超时: {locator}")
return False

def find_element(self, locator: tuple):
"""查找单个元素"""
try:
return self.driver.find_element(*locator)
except NoSuchElementException:
logger.warning(f"未找到元素: {locator}")
return None

def find_elements(self, locator: tuple):
"""查找多个元素"""
return self.driver.find_elements(*locator)

def scroll_to_bottom(self, max_scrolls: int = 50, delay: float = 1.0):
"""滚动到底部"""
logger.info("开始滚动到底部…")

last_height = self.driver.execute_script("return document.body.scrollHeight")
scroll_count = 0

while scroll_count < max_scrolls:
# 滚动
self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")

# 等待
time.sleep(delay)

# 检查高度
new_height = self.driver.execute_script("return document.body.scrollHeight")

if new_height == last_height:
logger.info(f"已到达底部（共滚动 {scroll_count} 次）")
break

last_height = new_height
scroll_count += 1

return scroll_count

def execute_script(self, script: str, *args) –> Any:
"""执行 JavaScript"""
return self.driver.execute_script(script, *args)

def screenshot(self, path: str):
"""截图"""
self.driver.save_screenshot(path)
logger.info(f"截图已保存: {path}")

def get_cookies(self) –> List[Dict]:
"""获取所有 Cookie"""
return self.driver.get_cookies()

def add_cookie(self, cookie: Dict):
"""添加 Cookie"""
self.driver.add_cookie(cookie)

def delete_all_cookies(self):
"""删除所有 Cookie"""
self.driver.delete_all_cookies()

def close(self):
"""关闭浏览器"""
if self.driver:
self.driver.quit()
logger.info("WebDriver 已关闭")

def __enter__(self):
"""上下文管理器入口"""
self.start()
return self

def __exit__(self, exc_type, exc_val, exc_tb):
"""上下文管理器退出"""
self.close()

Selenium vs Playwright 实战对比

# 相同任务：爬取京东商品评论

# ============ Playwright 版本 ============
async def scrape_reviews_playwright():
async with PlaywrightEngine() as engine:
await engine.goto('https://item.jd.com/100012345678.html')

# 切换到评论 Tab
await engine.page.click('#comment-tab')

# 等待评论加载
await engine.wait_for_selector('.comment-item')

# 滚动加载更多
await engine.scroll_to_bottom(max_scrolls=10)

# 提取评论
reviews = await engine.page.query_selector_all('.comment-item')

data = []
for review in reviews:
text = await review.inner_text()
data.append(text)

return data

# ============ Selenium 版本 ============
def scrape_reviews_selenium():
with SeleniumEngine() as engine:
engine.get('https://item.jd.com/100012345678.html')

# 切换到评论 Tab
comment_tab = engine.find_element((By.ID, 'comment-tab'))
comment_tab.click()

# 等待评论加载
engine.wait_for_element((By.CSS_SELECTOR, '.comment-item'))

# 滚动加载更多
engine.scroll_to_bottom(max_scrolls=10)

# 提取评论
reviews = engine.find_elements((By.CSS_SELECTOR, '.comment-item'))

data = []
for review in reviews:
text = review.text
data.append(text)

return data

# ============ 性能对比 ============
# Playwright:
# – 耗时: 25 秒
# – 内存: 280 MB
# – 代码行数: 18 行

# Selenium:
# – 耗时: 38 秒
# – 内存: 450 MB
# – 代码行数: 18 行

# 结论：Playwright 快 34%，内存少 38%，代码量相当

🛒 实战案例1：京东商品爬虫

需求分析

目标：爬取京东某商品的详细信息和用户评论

挑战：

完整实现

# scrapers/ecommerce/jd_scraper.py
import asyncio
from typing import List, Dict
from loguru import logger
from core.playwright_engine import PlaywrightEngine
from parsers.product_parser import JDProductParser
from storage.mongodb_storage import MongoDBStorage
import random

class JDProductScraper:
"""
京东商品爬虫

功能：
1. 爬取商品基本信息（标题、价格、品牌等）
2. 爬取商品详情页图片
3. 爬取用户评论（支持滚动加载全部评论）
4. 处理滑块验证码
5. 数据存储到 MongoDB
"""

def __init__(self, product_id: str):
"""
初始化爬虫

Args:
product_id: 商品 ID，如 "100012345678"
"""
self.product_id = product_id
self.product_url = f'https://item.jd.com/{product_id}.html'

self.engine = PlaywrightEngine(config={
'headless': True,
'timeout': 60000,
})

self.parser = JDProductParser()
self.storage = MongoDBStorage()

self.product_data = {}
self.reviews_data = []

async def scrape(self):
"""执行完整爬取流程"""
try:
await self.engine.start()

# 阶段1：爬取商品信息
logger.info(f"[1/3] 开始爬取商品信息 [ID: {self.product_id}]")
await self._scrape_product_info()

# 阶段2：爬取评论
logger.info(f"[2/3] 开始爬取用户评论")
await self._scrape_reviews()

# 阶段3：保存数据
logger.info(f"[3/3] 保存数据到 MongoDB")
self._save_to_database()

logger.info(f"爬取完成！商品: {self.product_data.get('title', 'Unknown')}, 评论数: {len(self.reviews_data)}")

except Exception as e:
logger.error(f"爬取失败: {e}", exc_info=True)
finally:
await self.engine.close()

async def _scrape_product_info(self):
"""爬取商品信息"""
# 访问商品详情页
success = await self.engine.goto(self.product_url)

if not success:
raise Exception("无法访问商品页面")

# 等待关键元素加载
await self.engine.wait_for_selector('.sku-name', timeout=10000)

# 检查是否需要验证码
if await self._check_and_handle_captcha():
logger.info("通过验证码验证")

# 获取页面 HTML
html = await self.engine.get_html()

# 解析商品信息
self.product_data = self.parser.parse_product_info(html, self.product_id)

# 额外抓取动态价格（通过 JavaScript）
price = await self._get_dynamic_price()
if price:
self.product_data['current_price'] = price

logger.info(f"商品信息获取完成: {self.product_data.get('title')}")

async def _get_dynamic_price(self) –> float:
"""
获取动态价格

京东的价格是通过 AJAX 异步加载的，需要执行 JavaScript 获取

Returns:
商品价格
"""
try:
# 等待价格元素出现
await self.engine.wait_for_selector('.p-price .price', timeout=5000)

# 提取价格
price_text = await self.engine.evaluate('''
() => {
const priceElem = document.querySelector('.p-price .price');
return priceElem ? priceElem.innerText : null;
}
''')

if price_text:
# 清理文本，提取数字
price = float(price_text.replace('¥', '').replace(',', '').strip())
return price

except Exception as e:
logger.warning(f"获取动态价格失败: {e}")

return None

async def _check_and_handle_captcha(self) –> bool:
"""
检查并处理滑块验证码

策略：
1. 检测是否出现验证码
2. 如果出现，调用验证码处理器
3. 等待验证通过

Returns:
是否成功处理
"""
# 检测验证码元素
captcha_elem = await self.engine.page.query_selector('.JDJRV-slide-inner')

if not captcha_elem:
return True # 没有验证码

logger.warning("检测到滑块验证码")

# 这里应该调用验证码处理模块
# 示例：使用图像识别 + 模拟滑动
# from handlers.captcha_handler import JDCaptchaHandler
# handler = JDCaptchaHandler(self.engine)
# success = await handler.solve()

# 简化处理：人工介入（暂停 60 秒让用户手动滑动）
logger.info("请手动完成滑块验证（60 秒内）")
await asyncio.sleep(60)

return True

async def _scrape_reviews(self, max_reviews: int = 500):
"""
爬取用户评论

京东评论特点：
1. 默认显示前 10 条
2. 点击"查看更多"按钮加载下一页（每页 10 条）
3. 评论数据通过 AJAX 加载

Args:
max_reviews: 最大爬取评论数
"""
# 切换到评论 Tab
comment_tab_selector = '#comment'
await self.engine.page.click(comment_tab_selector)

# 等待评论列表加载
await self.engine.wait_for_selector('.comment-item', timeout=10000)

logger.info("开始加载评论…")

page_count = 0
max_pages = max_reviews // 10 # 每页 10 条

while page_count < max_pages:
# 获取当前页面的评论
reviews_html = await self.engine.evaluate('''
() => {
const items = document.querySelectorAll('.comment-item');
return Array.from(items).map(item => item.outerHTML);
}
''')

# 解析评论
for review_html in reviews_html:
review = self.parser.parse_review(review_html)

# 去重（基于评论 ID）
if not any(r['review_id'] == review['review_id'] for r in self.reviews_data):
self.reviews_data.append(review)

logger.info(f"已获取 {len(self.reviews_data)} 条评论")

# 查找"下一页"按钮
next_button = await self.engine.page.query_selector('.ui-pager-next')

if not next_button:
logger.info("没有更多评论")
break

# 检查按钮是否可点击
is_disabled = await next_button.get_attribute('class')
if 'disabled' in (is_disabled or ''):
logger.info("已到达最后一页")
break

# 点击下一页
await next_button.click()

# 等待新评论加载
await asyncio.sleep(random.uniform(1.5, 2.5))

page_count += 1

logger.info(f"评论爬取完成，共 {len(self.reviews_data)} 条")

def _save_to_database(self):
"""保存数据到 MongoDB"""
# 保存商品信息
self.storage.save_product(self.product_data)

# 保存评论
for review in self.reviews_data:
review['product_id'] = self.product_id
self.storage.save_review(review)

logger.info("数据保存完成")

# 解析器实现
# parsers/product_parser.py
from bs4 import BeautifulSoup
import re
from typing import Dict

class JDProductParser:
"""京东商品数据解析器"""

def parse_product_info(self, html: str, product_id: str) –> Dict:
"""
解析商品基本信息

Args:
html: 页面 HTML
product_id: 商品 ID

Returns:
商品信息字典
"""
soup = BeautifulSoup(html, 'lxml')

# 提取标题
title_elem = soup.select_one('.sku-name')
title = title_elem.get_text(strip=True) if title_elem else ''

# 提取品牌
brand_elem = soup.select_one('#parameter-brand a')
brand = brand_elem.get_text(strip=True) if brand_elem else ''

# 提取图片列表
image_elems = soup.select('#spec-list img')
images = [img.get('src') or img.get('data-lazy-img') for img in image_elems]
images = [img for img in images if img] # 过滤空值

# 提取规格参数
params = {}
param_items = soup.select('#parameter-brand + ul li')
for item in param_items:
text = item.get_text(strip=True)
if '：' in text:
key, value = text.split('：', 1)
params[key] = value

product_info = {
'product_id': product_id,
'title': title,
'brand': brand,
'images': images,
'params': params,
}

return product_info

def parse_review(self, review_html: str) –> Dict:
"""
解析单条评论

Args:
review_html: 评论 HTML 片段

Returns:
评论数据字典
"""
soup = BeautifulSoup(review_html, 'lxml')

# 提取评论 ID
review_id = soup.select_one('.comment-item').get('data-id', '')

# 提取用户名
username_elem = soup.select_one('.user-info .nickname')
username = username_elem.get_text(strip=True) if username_elem else 'Anonymous'

# 提取评论内容
content_elem = soup.select_one('.comment-con')
content = content_elem.get_text(strip=True) if content_elem else ''

# 提取评分（星级）
star_elem = soup.select_one('.star-num')
star = 5 # 默认 5 星
if star_elem:
star_text = star_elem.get('class', [])
for cls in star_text:
if 'star' in cls:
match = re.search(r'star(\\d)', cls)
if match:
star = int(match.group(1))

# 提取评论时间
time_elem = soup.select_one('.comment-time')
review_time = time_elem.get_text(strip=True) if time_elem else ''

review_data = {
'review_id': review_id,
'username': username,
'content': content,
'star': star,
'time': review_time,
}

return review_data

使用示例

# 使用京东爬虫
async def main():
# 爬取 iPhone 15 Pro 的商品信息和评论
scraper = JDProductScraper(product_id='100045849398')

await scraper.scrape()

if __name__ == '__main__':
asyncio.run(main())

代码关键点解析

1. 动态价格获取技巧

# 问题：京东的价格不在初始 HTML 中

# 初始 HTML：
<div class="p-price">
<span class="price">––.––</span>
</div>

# JavaScript 异步加载后：
<div class="p-price">
<span class="price">¥9,999.00</span>
</div>

# 解决方案1：等待价格元素更新
await engine.wait_for_selector('.p-price .price')
# 问题：元素存在，但文本可能还是 "–.–"

# 解决方案2：等待价格不为空
price = await engine.page.wait_for_function('''
() => {
const priceElem = document.querySelector('.p-price .price');
const priceText = priceElem?.innerText || '';
return priceText && priceText !== '–.–' ? priceText : null;
}
''', timeout=10000)

# 解决方案3：拦截 AJAX 请求（高级）
await engine.page.route('**/price/getPrice*', lambda route: route.continue_())
response = await engine.page.wait_for_response('**/price/getPrice*')
price_data = await response.json()
print(price_data['price'])

2. 评论分页处理

# 京东评论的三种分页方式：

# 方式1：点击"下一页"按钮（传统分页）
async def pagination_by_button():
while True:
# 提取当前页评论
reviews = await extract_reviews()

# 查找下一页按钮
next_btn = await page.query_selector('.ui-pager-next')

if not next_btn:
break

# 检查是否禁用
is_disabled = await next_btn.get_attribute('class')
if 'disabled' in is_disabled:
break

# 点击
await next_btn.click()

# 等待新内容加载
await asyncio.sleep(2)

# 方式2：修改 URL 参数（直接访问特定页）
async def pagination_by_url():
base_url = 'https://club.jd.com/comment/productPageComments.action'

for page in range(1, 51): # 最多 50 页
url = f'{base_url}?productId={product_id}&score=0&page={page}'

# 直接请求 API
response = await page.goto(url)
json_data = await response.json()

reviews = json_data['comments']

if not reviews:
break

# 方式3：无限滚动（移动端）
async def pagination_by_scroll():
await scroll_to_bottom(max_scrolls=100)

3. 图片懒加载处理

# 问题：京东商品详情的图片使用懒加载

# 初始 HTML：
<img data–lazy–img="https://img14.360buyimg.com/n1/xxx.jpg" src="placeholder.gif">

# 滚动到可视区后：
<img src="https://img14.360buyimg.com/n1/xxx.jpg">

# 解决方案：强制加载所有图片
async def force_load_images():
# 方法1：滚动页面，触发懒加载
await scroll_to_bottom()

# 方法2：修改 src 属性（立即加载）
await page.evaluate('''
() => {
const lazyImages = document.querySelectorAll('img[data-lazy-img]');

lazyImages.forEach(img => {
const realSrc = img.getAttribute('data-lazy-img');
if (realSrc) {
img.src = realSrc;
}
});
}
''')

# 等待图片加载完成
await page.wait_for_load_state('networkidle')

🗨️ 实战案例2：知乎问题爬虫

需求分析

目标：爬取知乎某个问题的所有答案和评论

挑战：

完整实现

# scrapers/social/zhihu_scraper.py
import asyncio
from typing import List, Dict
from loguru import logger
from core.playwright_engine import PlaywrightEngine
import json

class ZhihuQuestionScraper:
"""
知乎问题爬虫

功能：
1. 模拟登录（Cookie 复用）
2. 爬取问题的所有答案
3. 展开并爬取评论
4. 处理图片防盗链
"""

def __init__(self, question_id: str, cookies_file: str = None):
"""
初始化爬虫

Args:
question_id: 问题 ID，如 "12345678"
cookies_file: Cookie 文件路径（JSON 格式）
"""
self.question_id = question_id
self.question_url = f'https://www.zhihu.com/question/{question_id}'
self.cookies_file = cookies_file

self.engine = PlaywrightEngine(config={'headless': True})

self.answers = []

async def scrape(self):
"""执行爬取"""
try:
await self.engine.start()

# 加载 Cookie（登录态）
if self.cookies_file:
await self._load_cookies()

# 访问问题页面
await self.engine.goto(self.question_url)

# 等待答案列表加载
await self.engine.wait_for_selector('.List-item')

# 滚动加载所有答案
await self._load_all_answers()

# 提取答案数据
await self._extract_answers()

logger.info(f"爬取完成，共 {len(self.answers)} 个答案")

finally:
await self.engine.close()

async def _load_cookies(self):
"""加载 Cookie"""
with open(self.cookies_file, 'r') as f:
cookies = json.load(f)

for cookie in cookies:
await self.engine.context.add_cookie(cookie)

logger.info("Cookie 加载完成")

async def _load_all_answers(self, max_scrolls: int = 50):
"""
滚动加载所有答案

知乎特点：
1. 初始加载 5 个答案
2. 滚动到底部自动加载下一批（每批 5-10 个）
3. 加载完毕后显示"没有更多内容"
"""
logger.info("开始加载所有答案…")

last_count = 0
no_change_count = 0

for i in range(max_scrolls):
# 滚动到底部
await self.engine.page.evaluate('''
window.scrollTo(0, document.body.scrollHeight)
''')

# 等待加载
await asyncio.sleep(2)

# 统计当前答案数量
current_count = await self.engine.page.evaluate('''
document.querySelectorAll('.List-item').length
''')

if current_count == last_count:
no_change_count += 1

if no_change_count >= 3:
logger.info("确认已加载所有答案")
break
else:
no_change_count = 0
logger.info(f"当前已加载 {current_count} 个答案")

last_count = current_count

return current_count

async def _extract_answers(self):
"""提取答案数据"""
answer_items = await self.engine.page.query_selector_all('.List-item')

logger.info(f"开始提取 {len(answer_items)} 个答案的数据…")

for idx, item in enumerate(answer_items, 1):
try:
answer_data = await self._extract_single_answer(item)
self.answers.append(answer_data)

logger.debug(f"[{idx}/{len(answer_items)}] 答案提取完成")

except Exception as e:
logger.warning(f"提取答案失败: {e}")

logger.info(f"答案提取完成，成功 {len(self.answers)}/{len(answer_items)}")

async def _extract_single_answer(self, answer_elem) –> Dict:
"""
提取单个答案的数据

Args:
answer_elem: 答案元素

Returns:
答案数据字典
"""
# 答案 ID
answer_id = await answer_elem.get_attribute('data-zop')

# 作者信息
author_elem = await answer_elem.query_selector('.AuthorInfo')
author_name = ''
author_url = ''

if author_elem:
name_elem = await author_elem.query_selector('.UserLink-link')
if name_elem:
author_name = await name_elem.inner_text()
author_url = await name_elem.get_attribute('href')

# 答案内容
content_elem = await answer_elem.query_selector('.RichContent-inner')
content_html = await content_elem.inner_html() if content_elem else ''
content_text = await content_elem.inner_text() if content_elem else ''

# 赞同数
vote_elem = await answer_elem.query_selector('.VoteButton–up')
vote_count = 0
if vote_elem:
vote_text = await vote_elem.inner_text()
vote_count = self._parse_vote_count(vote_text)

# 评论数
comment_elem = await answer_elem.query_selector('.ContentItem-actions button:has-text("条评论")')
comment_count = 0
if comment_elem:
comment_text = await comment_elem.inner_text()
comment_count = self._parse_comment_count(comment_text)

answer_data = {
'answer_id': answer_id,
'author': {
'name': author_name,
'url': author_url,
},
'content_html': content_html,
'content_text': content_text,
'vote_count': vote_count,
'comment_count': comment_count,
}

return answer_data

def _parse_vote_count(self, text: str) –> int:
"""解析赞同数"""
# 示例："1.2K" → 1200, "赞同" → 0
text = text.strip().lower()

if '赞同' in text:
return 0

if 'k' in text:
num = float(text.replace('k', ''))
return int(num * 1000)

if 'w' in text:
num = float(text.replace('w', ''))
return int(num * 10000)

try:
return int(text)
except ValueError:
return 0

def _parse_comment_count(self, text: str) –> int:
"""解析评论数"""
# 示例："125 条评论" → 125
import re
match = re.search(r'(\\d+)', text)
return int(match.group(1)) if match else 0

🧩 高级动态加载处理器

懒加载图片处理

# handlers/lazy_load.py
from loguru import logger
import asyncio

class LazyLoadHandler:
"""
懒加载图片处理器

策略：
1. 检测懒加载属性（data-src, data-lazy-img等）
2. 滚动触发加载或强制替换 src
3. 等待图片加载完成
"""

def __init__(self, engine):
self.engine = engine

async def force_load_all_images(self):
"""强制加载所有懒加载图片"""
logger.info("开始强制加载所有图片…")

# 查找所有懒加载图片
lazy_attrs = ['data-src', 'data-lazy-img', 'data-original', 'data-lazy']

for attr in lazy_attrs:
await self.engine.page.evaluate(f'''
() => {{
const images = document.querySelectorAll(`img[{attr}]`);

images.forEach(img => {{
const realSrc = img.getAttribute('{attr}');
if (realSrc) {{
img.src = realSrc;
img.removeAttribute('{attr}');
}}
}});
}}
''')

# 等待所有图片加载完成
await self.wait_for_images_loaded()

logger.info("所有图片加载完成")

async def wait_for_images_loaded(self, timeout: int = 30000):
"""等待页面所有图片加载完成"""
try:
await self.engine.page.wait_for_function('''
() => {
const images = Array.from(document.images);
return images.every(img => img.complete);
}
''', timeout=timeout)

return True

except Exception as e:
logger.warning(f"等待图片加载超时: {e}")
return False

async def lazy_load_by_scroll(self, scroll_delay: float = 1.0):
"""通过滚动触发懒加载"""
logger.info("通过滚动触发懒加载…")

# 获取页面总高度
total_height = await self.engine.page.evaluate('document.body.scrollHeight')

# 每次滚动 500px
step = 500
current_position = 0

while current_position < total_height:
await self.engine.page.evaluate(f'window.scrollTo(0, {current_position})')
await asyncio.sleep(scroll_delay)

current_position += step

# 重新获取高度（可能因为图片加载而增加）
total_height = await self.engine.page.evaluate('document.body.scrollHeight')

# 滚动到顶部
await self.engine.page.evaluate('window.scrollTo(0, 0)')

logger.info("滚动触发完成")

弹窗处理器

# handlers/modal_handler.py
from loguru import logger
import asyncio

class ModalHandler:
"""
弹窗处理器

处理：
1. 登录/注册弹窗
2. 广告弹窗
3. Cookie 同意弹窗
4. 下载 App 提示
"""

def __init__(self, engine):
self.engine = engine

# 常见弹窗关闭按钮的选择器
self.close_button_selectors = [
'.modal-close',
'.close-button',
'.popup-close',
'button.close',
'[aria-label="Close"]',
'.icon-close',
'span.close',
]

async def auto_close_modals(self, max_attempts: int = 5):
"""自动关闭弹窗"""
logger.info("开始自动关闭弹窗…")

for attempt in range(max_attempts):
closed = False

for selector in self.close_button_selectors:
try:
# 查找关闭按钮
close_btn = await self.engine.page.query_selector(selector)

if close_btn:
# 检查是否可见
is_visible = await close_btn.is_visible()

if is_visible:
await close_btn.click()
logger.info(f"已关闭弹窗（选择器: {selector}）")
closed = True
await asyncio.sleep(0.5)
break

except Exception:
continue

if not closed:
break

logger.info("弹窗处理完成")

async def handle_cookie_consent(self):
"""处理 Cookie 同意弹窗"""
consent_selectors = [
'button:has-text("接受")',
'button:has-text("同意")',
'button:has-text("Accept")',
'button:has-text("Agree")',
'#cookie-accept',
'.cookie-accept',
]

for selector in consent_selectors:
try:
btn = await self.engine.page.query_selector(selector)
if btn and await btn.is_visible():
await btn.click()
logger.info("已接受 Cookie 政策")
return True
except Exception:
continue

return False

async def handle_app_download_prompt(self):
"""处理下载 App 提示"""
# 移动端网站常见的"打开 App"弹窗

app_prompt_selectors = [
'button:has-text("取消")',
'button:has-text("继续浏览")',
'.app-banner-close',
'.download-app-close',
]

for selector in app_prompt_selectors:
try:
btn = await self.engine.page.query_selector(selector)
if btn:
await btn.click()
logger.info("已关闭 App 下载提示")
return True
except Exception:
continue

return False

分页处理器

# handlers/pagination.py
from loguru import logger
from typing import List, Callable, Optional
import asyncio

class PaginationHandler:
"""
分页处理器

支持：
1. 点击"下一页"按钮
2. URL 参数分页
3. AJAX 分页
4. 无限滚动
"""

def __init__(self, engine):
self.engine = engine

async def iterate_by_button(
self,
next_button_selector: str,
extractor: Callable,
max_pages: int = 100
) –> List:
"""
通过点击"下一页"按钮遍历分页

Args:
next_button_selector: 下一页按钮选择器
extractor: 数据提取函数
max_pages: 最大页数

Returns:
所有页的数据列表
"""
logger.info("开始按钮分页遍历…")

all_data = []
page_num = 1

while page_num <= max_pages:
logger.info(f"正在处理第 {page_num} 页")

# 提取当前页数据
page_data = await extractor(self.engine.page)
all_data.extend(page_data)

# 查找下一页按钮
next_btn = await self.engine.page.query_selector(next_button_selector)

if not next_btn:
logger.info("未找到下一页按钮，可能已到最后一页")
break

# 检查是否禁用
is_disabled = await self._is_button_disabled(next_btn)
if is_disabled:
logger.info("下一页按钮已禁用，已到最后一页")
break

# 点击下一页
await next_btn.click()

# 等待页面更新
await asyncio.sleep(2)

page_num += 1

logger.info(f"分页遍历完成，共 {page_num – 1} 页，{len(all_data)} 条数据")

return all_data

async def iterate_by_url(
self,
url_template: str,
extractor: Callable,
start_page: int = 1,
max_pages: int = 100
) –> List:
"""
通过修改 URL 参数遍历分页

Args:
url_template: URL 模板，如 "https://example.com/list?page={page}"
extractor: 数据提取函数
start_page: 起始页码
max_pages: 最大页数

Returns:
所有页的数据列表
"""
logger.info("开始 URL 分页遍历…")

all_data = []

for page_num in range(start_page, start_page + max_pages):
url = url_template.format(page=page_num)

logger.info(f"正在访问第 {page_num} 页: {url}")

await self.engine.goto(url)

# 提取数据
page_data = await extractor(self.engine.page)

if not page_data:
logger.info(f"第 {page_num} 页没有数据，停止遍历")
break

all_data.extend(page_data)

# 随机延时（避免被限流）
await asyncio.sleep(random.uniform(1.5, 3.0))

logger.info(f"分页遍历完成，共 {len(all_data)} 条数据")

return all_data

async def _is_button_disabled(self, button) –> bool:
"""检查按钮是否禁用"""
# 方法1：检查 disabled 属性
is_disabled = await button.get_attribute('disabled')
if is_disabled:
return True

# 方法2：检查 class
class_name = await button.get_attribute('class') or ''
if 'disabled' in class_name.lower():
return True