网页自动化与爬虫实战：Playwright vs Selenium vs Puppeteer 选型与无人值守流水线

1
import asyncio
2
from playwright.async_api import async_playwright, expect
3

4
async def modern_locator_demo():
5
    async with async_playwright() as p:
6
        browser = await p.chromium.launch(headless=True)
7
        page = await browser.new_page()
8
        await page.goto("https://example.com/login")
9

10
        # 推荐 1: get_by_role (最符合用户真实心智模型，抗重构能力极强)
11
        login_btn = page.get_by_role("button", name="登录")
12

13
        # 推荐 2: get_by_label (精准匹配带 <label> 关联的表单输入框)
14
        username_input = page.get_by_label("用户名或电子邮箱")
15
        await username_input.fill("antigravity_engineer")
16

17
        # 推荐 3: get_by_placeholder
18
        password_input = page.get_by_placeholder("请输入 8 位以上包含字母与数字的密码")
19
        await password_input.fill("SecretPassword2026!")
20

21
        # 推荐 4: get_by_test_id (面向企业级研发规范，前端埋设 data-testid)
22
        submit_anchor = page.get_by_test_id("submit-verify-code")
23

24
        # 动作自动等待
25
        await login_btn.click()
26

27
        # Web-First 断言：expect 会自动轮询重试直到条件达成，绝不偶发报错
28
        await expect(page.get_by_role("heading", name="个人仪表盘")).to_be_visible(timeout=10000)
29

30
        await browser.close()
31

32
if __name__ == "__main__":
33
    asyncio.run(modern_locator_demo())

1
import asyncio
2
import json
3
from playwright.async_api import async_playwright
4

5
async def intercept_api_json_demo():
6
    async with async_playwright() as p:
7
        browser = await p.chromium.launch(headless=True)
8
        page = await browser.new_page()
9

10
        extracted_products = []
11

12
        # 定义网络响应拦截器回调
13
        async def handle_response(response):
14
            # 精准嗅探目标商品接口
15
            if "/api/v2/products/search" in response.url and response.status == 200:
16
                try:
17
                    data = await response.json()
18
                    items = data.get("data", {}).get("items", [])
19
                    for item in items:
20
                        extracted_products.append({
21
                            "id": item["productId"],
22
                            "title": item["name"],
23
                            "price": item["price"],
24
                            "stock": item["inventoryStock"],
25
                            "brand": item.get("brandName", "")
26
                        })
27
                    print(f"[API 捕获] 成功捕获本批次商品 {len(items)} 条！")
28
                except Exception as err:
29
                    print(f"[解析异常] {err}")
30

31
        # 挂载监听
32
        page.on("response", handle_response)
33

34
        print("[导航开始] 访问电商大盘...")
35
        await page.goto("https://example.com/mall/search?category=electronics")
36

37
        # 模拟真实向下滚动页面，触发前端分页拉取后续 API
38
        for scroll_idx in range(5):
39
            print(f"[滚动加载] 触发第 {scroll_idx + 1} 次触底加载...")
40
            await page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
41
            # 优雅等待网络处于闲置状态（没有超过 2 个网络连接持续 500ms）
42
            await page.wait_for_load_state("networkidle")
43

44
        print(f"\n[抓取总结] 共截获入库有效高价值商品数据: {len(extracted_products)} 条！")
45
        await browser.close()
46

47
if __name__ == "__main__":
48
    asyncio.run(intercept_api_json_demo())

1
async def setup_high_speed_interception(page):
2
    """
3
    配置极致性能路由过滤：物理屏蔽所有媒体、字体与分析追踪脚本
4
    """
5
    # 定义禁止下载的资源类型黑名单
6
    BLOCKED_RESOURCE_TYPES = ["image", "media", "font", "stylesheet"]
7

8
    # 定义第三方广告与风控遥测域名通配符
9
    BLOCKED_DOMAINS = [
10
        "google-analytics.com",
11
        "doubleclick.net",
12
        "facebook.net",
13
        "sentry.io",
14
        "hotjar.com"
15
    ]
16

17
    async def route_filter(route):
18
        req = route.request
19
        # 1. 检查资源类型
20
        if req.resource_type in BLOCKED_RESOURCE_TYPES:
21
            # 直接在本地丢弃，不发生任何真实网络出海握手
22
            await route.abort()
23
            return
24

25
        # 2. 检查遥测域名
26
        if any(domain in req.url for domain in BLOCKED_DOMAINS):
27
            await route.abort()
28
            return
29

30
        # 3. 动态改写请求头（例如注入自定义安全凭证或修改 Referer）
31
        headers = {
32
            **req.headers,
33
            "X-Client-Platform": "Web-Desktop",
34
            "Referer": "https://example.com/portal"
35
        }
36
        await route.continue_(headers=headers)
37

38
    # 全局挂载路由拦截匹配通配符
39
    await page.route("**/*", route_filter)
40
    print("[路由提速] 媒体与遥测物理拦截门禁已成功部署！")

1
async def listen_websocket_stream(page):
2
    """
3
    实时监听并转录页面中的 WebSocket 数据帧
4
    """
5
    def on_web_socket(ws):
6
        print(f"[WebSocket 握手成功] 连接地址: {ws.url}")
7

8
        # 监听从服务器推送下发的消息帧
9
        def on_frame_received(payload):
10
            try:
11
                # 尝试反序列化 JSON 行情帧
12
                msg = json.loads(payload)
13
                print(f"[行情下推] 标的: {msg.get('symbol')} | 最新价: {msg.get('price')}")
14
            except Exception:
15
                pass
16

17
        ws.on("framereceived", on_frame_received)
18

19
    page.on("websocket", on_web_socket)

1
import asyncio
2
from playwright.async_api import async_playwright
3

4
STEALTH_INJECTION_JS = """
5
// 1. 彻底抹除 navigator.webdriver 标志位
6
Object.defineProperty(navigator, 'webdriver', {
7
    get: () => undefined
8
});
9

10
// 2. 伪装 Chrome 原生运行时对象 window.chrome
11
window.chrome = {
12
    app: { isInstalled: false, InstallState: { DISABLED: 'DISABLED', INSTALLED: 'INSTALLED', NOT_INSTALLED: 'NOT_INSTALLED' }, RunningState: { CANNOT_RUN: 'CANNOT_RUN', READY_TO_RUN: 'READY_TO_RUN', RUNNING: 'RUNNING' } },
13
    runtime: { OnInstalledReason: {}, OnRestartRequiredReason: {}, PlatformArch: {}, PlatformNaclArch: {}, PlatformOs: {}, RequestUpdateCheckStatus: {} },
14
    loadTimes: function() {},
15
    csi: function() {}
16
};
17

18
// 3. 伪装真实的 Plugins 插件列表与语言
19
Object.defineProperty(navigator, 'plugins', {
20
    get: () => [1, 2, 3, 4, 5]
21
});
22
Object.defineProperty(navigator, 'languages', {
23
    get: () => ['zh-CN', 'zh', 'en-US', 'en']
24
});
25

26
// 4. 深度伪装 WebGL 渲染器：将 SwiftShader 伪装为真实主流独立显卡
27
const getParameterProxyHandler = {
28
    apply: function(target, thisArg, argumentsList) {
29
        const param = argumentsList[0];
30
        // 37445: UNMASKED_VENDOR_WEBGL
31
        if (param === 37445) {
32
            return 'Google Inc. (NVIDIA)';
33
        }
34
        // 37446: UNMASKED_RENDERER_WEBGL
35
        if (param === 37446) {
36
            return 'ANGLE (NVIDIA, NVIDIA GeForce RTX 4070 Direct3D11 vs_5_0 ps_5_0, D3D11)';
37
        }
38
        return Reflect.apply(target, thisArg, argumentsList);
39
    }
40
};
41

42
const origGetParameter = WebGLRenderingContext.prototype.getParameter;
43
WebGLRenderingContext.prototype.getParameter = new Proxy(origGetParameter, getParameterProxyHandler);
44

45
// 5. 修复 Notification 权限逻辑冲突
46
const origQuery = window.navigator.permissions.query;
47
window.navigator.permissions.query = (parameters) => (
48
    parameters.name === 'notifications' ?
49
        Promise.resolve({ state: Notification.permission }) :
50
        origQuery(parameters)
51
);
52
"""
53

54
async def launch_stealth_browser():
55
    async with async_playwright() as p:
56
        # 针对 Chromium 启动参数进行深层安全加固
57
        launch_args = [
58
            "--disable-blink-features=AutomationControlled", # 核心加固：禁用 Blink 自动化标志
59
            "--disable-infobars",
60
            "--no-sandbox",
61
            "--disable-setuid-sandbox",
62
            "--disable-dev-shm-usage",
63
            "--window-size=1920,1080"
64
        ]
65

66
        browser = await p.chromium.launch(
67
            headless=True,
68
            args=launch_args
69
        )
70

71
        # 创建上下文时绑定高拟真度 User-Agent 与物理视口
72
        context = await browser.new_context(
73
            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
74
            viewport={"width": 1920, "height": 1080},
75
            device_scale_factor=1,
76
            is_mobile=False,
77
            has_touch=False,
78
            locale="zh-CN",
79
            timezone_id="Asia/Shanghai"
80
        )
81

82
        # 在任何页面加载前注入全局伪装
83
        await context.add_init_script(STEALTH_INJECTION_JS)
84

85
        page = await context.new_page()
86
        # 访问指纹检测权威评估站点
87
        await page.goto("https://bot.sannysoft.com")
88
        await page.screenshot(path="stealth_result.png", full_page=True)
89
        print("[指纹伪装成功] 诊断测试截图已生成至 stealth_result.png！")
90
        await browser.close()

1
async def create_geo_aligned_context(browser, proxy_server, proxy_user, proxy_pass):
2
    """
3
    创建地理位置、时区、语言与 IP 严格对齐的高拟真度会话
4
    """
5
    context = await browser.new_context(
6
        proxy={
7
            "server": proxy_server,       # 例如: "http://gateway.residential-proxy.com:8000"
8
            "username": proxy_user,
9
            "password": proxy_pass
10
        },
11
        # 假设使用的是日本东京住宅 IP：
12
        locale="ja-JP",
13
        timezone_id="Asia/Tokyo",
14
        geolocation={"latitude": 35.6762, "longitude": 139.6503},
15
        permissions=["geolocation"]  # 自动授权地理位置弹窗，防止人工交互阻断
16
    )
17
    return context

1
import random
2
import math
3

4
def generate_human_track(distance):
5
    """
6
    模拟人类手部肌肉生理学特性的非对称加减速滑动轨迹生成器
7
    """
8
    track = []
9
    current = 0
10
    # 模拟人手习惯性稍稍滑过目标点，再微调回退
11
    overshoot = random.randint(3, 8)
12
    target = distance + overshoot
13

14
    # 设定初速度与加速度拐点
15
    mid = distance * (random.uniform(0.6, 0.8))
16
    t = 0.2
17
    v = 0
18

19
    while current < target:
20
        if current < mid:
21
            # 前半段爆发加速
22
            a = random.uniform(2.5, 4.5)
23
        else:
24
            # 后半段预判接近，急剧减速
25
            a = -random.uniform(3.0, 5.0)
26

27
        v0 = v
28
        v = v0 + a * t
29
        move = v0 * t + 0.5 * a * (t ** 2)
30
        current += move
31
        # Y 轴引入轻微人类手部上下生理抖动（-1 到 +2 像素）
32
        track.append((round(move), random.choice([-1, 0, 0, 1, 2])))
33

34
    # 执行回退修正（Overshoot Compensation）
35
    back_moves = [(-1, 0), (-2, 0), (-1, 0)]
36
    for bm in back_moves[:overshoot]:
37
        track.append(bm)
38

39
    return track
40

41
async def perform_human_slider_drag(page, slider_locator, target_distance):
42
    """
43
    在 Playwright 页面中执行高拟真度的人类拖拽交互
44
    """
45
    box = await slider_locator.bounding_box()
46
    start_x = box["x"] + box["width"] / 2
47
    start_y = box["y"] + box["height"] / 2
48

49
    # 1. 鼠标移至滑块并悬停
50
    await page.mouse.move(start_x, start_y)
51
    await asyncio.sleep(random.uniform(0.2, 0.5))
52

53
    # 2. 按下鼠标左键
54
    await page.mouse.down()
55
    await asyncio.sleep(random.uniform(0.1, 0.3))
56

57
    # 3. 按轨迹步进移动
58
    tracks = generate_human_track(target_distance)
59
    curr_x, curr_y = start_x, start_y
60
    for dx, dy in tracks:
61
        curr_x += dx
62
        curr_y += dy
63
        await page.mouse.move(curr_x, curr_y)
64
        # 随机微小微秒级延时
65
        await asyncio.sleep(random.uniform(0.008, 0.025))
66

67
    # 4. 到达终点短暂迟疑后松开鼠标
68
    await asyncio.sleep(random.uniform(0.3, 0.6))
69
    await page.mouse.up()
70
    print(f"[验证码攻关] 已完成 {target_distance} 像素的人类拟真滑动尝试！")

1
import asyncio
2
from pathlib import Path
3
from playwright.async_api import async_playwright, expect
4

5
AUTH_FILE = Path("./credentials/auth_session.json")
6

7
async def manual_login_and_save_session():
8
    """
9
    一次性交互式登录引导脚本：人工扫码或输入验证码后，永久保存状态
10
    """
11
    AUTH_FILE.parent.mkdir(parents=True, exist_ok=True)
12
    async with async_playwright() as p:
13
        # 调试阶段必须开启图形界面，方便人工操作
14
        browser = await p.chromium.launch(headless=False)
15
        context = await browser.new_context()
16
        page = await context.new_page()
17

18
        print("[引导] 请在打开的浏览器窗口中完成账号登录（可扫码或输入短信）...")
19
        await page.goto("https://example.com/login")
20

21
        # 智能等待登录成功的标志性元素（例如个人头像或用户控制台）
22
        # 此处设置超长 120 秒等待，留给人工充裕的扫码时间
23
        dashboard_elem = page.locator(".user-avatar, #dashboard-header")
24
        await expect(dashboard_elem).to_be_visible(timeout=120000)
25

26
        # 将全部 Cookie 与 LocalStorage 物理固化为 JSON
27
        await context.storage_state(path=str(AUTH_FILE))
28
        print(f"[成功] 登录态已安全持久化至: {AUTH_FILE.resolve()}")
29
        await browser.close()
30

31
async def run_unattended_scraper_with_session():
32
    """
33
    无人值守自动化主任务：加载持久化凭证，静默直达受限后台
34
    """
35
    if not AUTH_FILE.exists():
36
        print("[错误] 未检测到持久化凭证，请先执行 manual_login_and_save_session()")
37
        return
38

39
    async with async_playwright() as p:
40
        browser = await p.chromium.launch(headless=True)
41

42
        # 关键核心：创建上下文时直接注入 storage_state
43
        context = await browser.new_context(storage_state=str(AUTH_FILE))
44
        page = await context.new_page()
45

46
        print("[导航] 正在以已登录身份直达高价值订单后台...")
47
        await page.goto("https://example.com/admin/orders")
48

49
        # 验证是否直接处于登录态
50
        page_title = await page.title()
51
        print(f"[验证成功] 当前已登录页面标题: {page_title}")
52

53
        # 开展数据抓取业务...
54
        await browser.close()

1
import duckdb
2
import asyncio
3

4
class DuckDBDataSink:
5
    def __init__(self, db_path="scraped_warehouse.duckdb"):
6
        self.con = duckdb.connect(db_path)
7
        # 初始化高可用表结构
8
        self.con.execute("""
9
            CREATE TABLE IF NOT EXISTS product_catalog (
10
                product_id VARCHAR PRIMARY KEY,
11
                title VARCHAR,
12
                price DOUBLE,
13
                brand VARCHAR,
14
                captured_at TIMESTAMP
15
            )
16
        """)
17
        self.buffer = []
18
        self.lock = asyncio.Lock()
19

20
    async def insert_record(self, record):
21
        async with self.lock:
22
            self.buffer.append(record)
23
            if len(self.buffer) >= 200:
24
                await self.flush()
25

26
    async def flush(self):
27
        if not self.buffer:
28
            return
29
        # 使用 DuckDB 极速批量 INSERT OR REPLACE / IGNORE
30
        records_to_write = self.buffer.copy()
31
        self.buffer.clear()
32

33
        # 转换为列表元组进行批量执行
34
        param_list = [
35
            (r["id"], r["title"], r["price"], r.get("brand", ""), r["captured_at"])
36
            for r in records_to_write
37
        ]
38
        self.con.executemany("""
39
            INSERT OR REPLACE INTO product_catalog VALUES (?, ?, ?, ?, ?)
40
        """, param_list)
41
        print(f"[数据落盘] 批量归档 {len(records_to_write)} 条记录至 DuckDB 列存引擎！")

1
import asyncio
2
import random
3
import time
4
import datetime
5
from pathlib import Path
6
from playwright.async_api import async_playwright
7
import duckdb
8

9
# ================= 全局工程配置 =================
10
CONCURRENCY_LIMIT = 5            # 异步受控并发度
11
BATCH_FLUSH_THRESHOLD = 50       # 数据落盘批量大小
12
TARGET_BASE_URL = "https://example.com/mall"
13
AUTH_STATE_PATH = "./credentials/session.json"
14
DB_OUTPUT_PATH = "./data/scraped_products.duckdb"
15

16
# 极客级无头浏览器底层指纹伪装 JavaScript 补丁
17
STEALTH_INJECT_CODE = """
18
Object.defineProperty(navigator, 'webdriver', { get: () => undefined });
19
window.chrome = { runtime: {}, loadTimes: function() {}, csi: function() {} };
20
Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3] });
21
Object.defineProperty(navigator, 'languages', { get: () => ['zh-CN', 'zh'] });
22
"""
23

24
class ProductionScrapingEngine:
25
    def __init__(self, concurrency=5):
26
        self.semaphore = asyncio.Semaphore(concurrency)
27
        self.db = duckdb.connect(DB_OUTPUT_PATH)
28
        self.init_database()
29
        self.write_buffer = []
30
        self.lock = asyncio.Lock()
31
        self.total_captured = 0
32

33
    def init_database(self):
34
        self.db.execute("""
35
            CREATE TABLE IF NOT EXISTS scraped_catalog (
36
                sku_id VARCHAR PRIMARY KEY,
37
                product_name VARCHAR,
38
                price_cny DOUBLE,
39
                category VARCHAR,
40
                scraped_at TIMESTAMP
41
            )
42
        """)
43

44
    async def buffer_data(self, record):
45
        """线程/协程安全的数据写入缓冲区"""
46
        async with self.lock:
47
            self.write_buffer.append(record)
48
            self.total_captured += 1
49
            if len(self.write_buffer) >= BATCH_FLUSH_THRESHOLD:
50
                await self.flush_buffer()
51

52
    async def flush_buffer(self):
53
        """将内存缓冲批量写入 DuckDB 列式存储"""
54
        if not self.write_buffer:
55
            return
56
        records = self.write_buffer.copy()
57
        self.write_buffer.clear()
58

59
        tuples = [
60
            (r["sku_id"], r["name"], r["price"], r.get("category", ""), r["scraped_at"])
61
            for r in records
62
        ]
63
        self.db.executemany("""
64
            INSERT OR REPLACE INTO scraped_catalog VALUES (?, ?, ?, ?, ?)
65
        """, tuples)
66
        print(f"[持久化] 成功事务落盘 {len(records)} 条数据到 DuckDB！")
67

68
    async def setup_page_interception(self, page):
69
        """配置网络拦截：极速丢弃静态垃圾资源，提速 300%"""
70
        BLOCKED_TYPES = ["image", "media", "font", "stylesheet"]
71

72
        async def intercept_route(route):
73
            if route.request.resource_type in BLOCKED_TYPES:
74
                await route.abort()
75
            else:
76
                await route.continue_()
77

78
        await page.route("**/*", intercept_route)
79

80
    async def scrape_category_task(self, browser, category_slug):
81
        """单个分类的多页面深度采集协程任务"""
82
        async with self.semaphore:
83
            print(f"[任务启动] 正在调度分类: {category_slug}")
84
            # 创建沙盒化的独立上下文，如果存在登录凭据则注入
85
            context_kwargs = {
86
                "viewport": {"width": 1920, "height": 1080},
87
                "user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
88
                "locale": "zh-CN",
89
                "timezone_id": "Asia/Shanghai"
90
            }
91
            if Path(AUTH_STATE_PATH).exists():
92
                context_kwargs["storage_state"] = AUTH_STATE_PATH
93

94
            context = await browser.new_context(**context_kwargs)
95
            # 注入 Stealth 反指纹探针
96
            await context.add_init_script(STEALTH_INJECT_CODE)
97
            page = await context.new_page()
98
            await self.setup_page_interception(page)
99

100
            # 挂载网络响应监听器：直接截获后端核心 API JSON
101
            async def on_response(response):
102
                if "/api/v1/goods/list" in response.url and response.status == 200:
103
                    try:
104
                        res_json = await response.json()
105
                        goods_list = res_json.get("data", {}).get("list", [])
106
                        for g in goods_list:
107
                            await self.buffer_data({
108
                                "sku_id": str(g.get("id")),
109
                                "name": str(g.get("title")),
110
                                "price": float(g.get("current_price", 0.0)),
111
                                "category": category_slug,
112
                                "scraped_at": datetime.datetime.now()
113
                            })
114
                    except Exception:
115
                        pass
116

117
            page.on("response", on_response)
118

119
            try:
120
                target_url = f"{TARGET_BASE_URL}?cat={category_slug}"
121
                await page.goto(target_url, wait_until="domcontentloaded", timeout=25000)
122

123
                # 动态模拟连续翻页/滚动加载
124
                for page_num in range(1, 4):
125
                    # 模拟人类随机微幅滚动行为
126
                    scroll_offset = random.randint(700, 1000)
127
                    await page.evaluate(f"window.scrollBy(0, {scroll_offset})")
128
                    await asyncio.sleep(random.uniform(1.2, 2.5))
129

130
                    # 尝试点击“下一页”或滚动触底
131
                    next_btn = page.get_by_role("button", name="下一页")
132
                    if await next_btn.is_visible():
133
                        await next_btn.click()
134
                        await page.wait_for_load_state("networkidle", timeout=10000)
135
                    else:
136
                        break
137

138
                print(f"[任务达成] 分类 {category_slug} 数据抓取完毕！")
139
            except Exception as task_err:
140
                print(f"[任务警告] 分类 {category_slug} 执行期间遇到扰动: {task_err}")
141
            finally:
142
                # 严密防范内存泄漏：必须确保上下文显式关闭
143
                await context.close()
144

145
    async def run_pipeline(self, category_list):
146
        start_time = time.time()
147
        print(">>> 启动自动化高并发无头抓取流水线 <<<")
148

149
        async with async_playwright() as p:
150
            # 启动全局单例 Chromium 进程，配置生产级容器安全参数
151
            browser = await p.chromium.launch(
152
                headless=True,
153
                args=[
154
                    "--disable-blink-features=AutomationControlled",
155
                    "--no-sandbox",
156
                    "--disable-dev-shm-usage"
157
                ]
158
            )
159

160
            # 并发派发任务集合
161
            tasks = [self.scrape_category_task(browser, cat) for cat in category_list]
162
            await asyncio.gather(*tasks, return_exceptions=True)
163

164
            # 优雅收尾
165
            await self.flush_buffer()
166
            await browser.close()
167

168
        elapsed = time.time() - start_time
169
        print("=" * 60)
170
        print(f"[流水线执行统计] 耗时: {elapsed:.2f} 秒 | 累计采集商品: {self.total_captured} 条")
171
        print(f"[结果归档位置] DuckDB 存储节点: {DB_OUTPUT_PATH}")
172
        print("=" * 60)