
第二章 · 实战篇:BookNest 会员捡漏机器人
本章的承诺:跟着做,你能在自己的电脑上跑出和文中一模一样的结果。
文中所有终端输出、耗时、数字,都是从本机真实运行中复制的,没有一处是编的。
2.1 项目背景:一个真实到有点无聊的痛点
2.1.1 故事
小周是个爱买书的人,办了某图书网站的会员。会员的价值在于:部分书有会员专享折扣,但折扣力度差别巨大——有的书只便宜 4%,有的能便宜 42%。
他每个月的固定动作是:
登录 → 翻 4 页商品列表 → 一本本对比定价和会员价 → 心算折扣
→ 记下折扣大的 → 挨个点进详情页看有没有货
→ 把有货的整理进购物清单这件事每次要花 25\~30 分钟,而且极其枯燥,还容易看漏。
它有三个特征,让它成为浏览器自动化的完美教学案例:
| 特征 | 为什么重要 |
|---|---|
| 重复且规律 | 每月一次,流程完全一样 → 自动化的收益能反复兑现 |
| 必须登录才能看到关键数据 | 会员价是登录墙后面的 → 必须处理会话状态(L5) |
| 关键字段分散在两类页面 | 价格/评分在列表页,库存在详情页 → 逼你思考"抓取成本"(这是本项目最核心的工程思想) |
2.1.2 目标定义
把上面那段人话,翻译成机器能执行的规格:
【输入】 网站 http://127.0.0.1:8899
账号 demo / demo1234
【筛选】 同时满足三条才算命中:
① 会员折扣 ≥ 15% 折扣 = (定价 − 会员价) / 定价 × 100
② 用户评分 ≥ 4.5
③ 库存 > 0(必须有货)
【输出】 ① Markdown 报告(人看的,按折扣排序,含汇总)
② CSV(进 Excel 的)
③ JSON(给下游程序用的)
④ 关键节点截图(存档 / 出错时排查)注意最后一项。很多教程做到"打印出结果"就结束了,但真实项目里,没有产物的自动化等于没做。所以本项目从第一行代码起就规划好了产物目录。
2.1.3 为什么要自己搭一个网站,而不是抓真站
这是个必须解释清楚的设计决定:
| 抓真实网站的问题 | 本地靶场的解法 |
|---|---|
| 违反 robots.txt / 服务条款,有法律风险 | 自己的站,随便抓 |
| 网站随时改版,教程三个月就失效 | 代码在你手里,永远不变 |
| 数据每天变,你跑出来和书上对不上 | 24 本书数据写死,结果 100% 可复现 |
| 有反爬、有验证码,初学者卡在第一步 | 没有反爬,但故意保留了真实的技术难点 |
| 网络波动导致失败,你分不清是代码问题还是网络问题 | 本地回环,毫秒级响应,失败一定是你的代码问题 |
最后一条尤其重要:学习阶段,"确定性"比"真实性"更有价值。你需要一个"错了一定是我的错"的环境。
2.2 场景设计思路:靶场为什么长这样
靶场叫 BookNest,代码在 AIforGUIcode/demosite/,只依赖 Python 标准库(不用装 Flask、Django),双击就能跑。
2.2.1 站点结构
http://127.0.0.1:8899
│
├─ / 首页(一句介绍 + 进入按钮)
├─ /login 登录页(demo / demo1234)
├─ /products 图书列表(24 本 · 每页 6 本 · 共 4 页)
│ ?page=2 翻页
│ ?category=计算机 分类筛选
│ ?q=算法 关键词搜索
├─ /book/<id> 图书详情(★ 库存只在这里,且异步加载)
├─ /logout 退出
└─ /api/health 健康检查(给自检脚本用)
数据:24 本书 × 4 个分类(计算机/历史/文学/商业),每类 6 本
字段:书名、作者、出版社、分类、定价、会员价、评分、评价数、库存
★ 全部硬编码在 demosite/data.py,永远不变2.2.2 五个陷阱:每一个都对应真实项目里的一类事故
这是靶场设计的核心。我没有做一个"干净"的网站——干净的网站教不会你东西。
图 2-1 · 五个陷阱与真实痛点的映射
┌─────────────────────────────────────────────────────────────────────┐
│ 陷阱 1 · Cookie 同意浮层 → 对应能力层 L3/L4 │
│ │
│ 现象:一个 z-index:9000 的全屏遮罩盖住整个页面 │
│ 真实对应:GDPR Cookie 横幅、App 下载引导、新人红包弹窗、问卷邀请 │
│ 会导致:所有点击报 "intercepts pointer events" │
│ 正确姿势:进页面第一件事就是幂等地关掉它 │
│ ★ 教学点:Playwright 的报错会直接告诉你"谁挡的" │
├─────────────────────────────────────────────────────────────────────┤
│ 陷阱 2 · 动态 class 名 → 对应能力层 L2 │
│ │
│ 现象:商品卡片 class 每次渲染都变 │
│ card-53a362 → card-f6c6e7 → card-9d1b44 … │
│ 真实对应:CSS Modules、styled-components、Tailwind JIT、代码混淆 │
│ 会导致:用 class 写的选择器,刷新一次就挂 │
│ 正确姿势:用 data-testid / get_by_role / get_by_text │
│ ★ 教学点:亲眼看到 class 在两次刷新之间变了 │
├─────────────────────────────────────────────────────────────────────┤
│ 陷阱 3 · 登录墙 → 对应能力层 L5 │
│ │
│ 现象:未登录时,会员价位置显示"登录后查看会员价" │
│ 真实对应:几乎所有需要账号的系统 │
│ 会导致:抓到一堆占位文本,还以为抓成功了(静默错误!) │
│ 正确姿势:登录一次 → storage_state 落盘 → 后续复用 │
│ ★ 教学点:会话复用能省掉多少时间和风控 │
├─────────────────────────────────────────────────────────────────────┤
│ 陷阱 4 · 异步渲染(延迟 800ms) → 对应能力层 L4 │
│ │
│ 现象:详情页库存字段初始是"加载中…",800ms 后 JS 才注入真实值 │
│ 真实对应:SPA、前后端分离、懒加载、瀑布流 │
│ 会导致:读到 "加载中…" 而不是 "有货,剩余 32 件"(又是静默错误) │
│ 正确姿势:等 data-loaded="true" 属性,或用 expect 轮询断言 │
│ ★ 教学点:为什么 time.sleep() 是最差解 │
├─────────────────────────────────────────────────────────────────────┤
│ 陷阱 5 · 分页 → 对应工程健壮性 │
│ │
│ 现象:24 本书分 4 页,末页不再渲染"下一页"按钮 │
│ 真实对应:所有列表页 │
│ 会导致:写不好就死循环,或者少抓一页 │
│ 正确姿势:以页面自身的"下一页是否存在"为终止信号 + 硬上限兜底 │
│ ★ 教学点:页面自带"第 1/4 页,共 24 本",优先信任页面的声明 │
└─────────────────────────────────────────────────────────────────────┘同时,靶场给每个关键元素都加了 data-testid —— 这是"标准答案"。真实项目里如果前端愿意加这些属性,你的自动化成本会降低一个数量级。这也是本书希望你带回团队的一个实践建议。
2.2.3 数据的"故意设计"
24 本书的数据不是随机生成的,每一本都在承担教学职责:
折扣分布:从 4.4% 到 45.7%,跨度大
→ 保证不同筛选阈值(15% / 25% / 35%)能筛出明显不同的结果
→ Step 6 换个需求就能看出规则真的生效了
评分分布:4.2 ~ 4.9
→ 保证"评分 ≥ 4.5" 这个条件真的会过滤掉一些书
★ 关键陷阱数据:有 3 本书「折扣很大、评分很高,但库存 = 0」
· 流畅的Python(第2版) 折扣 15.5% 评分 4.8 库存 0
· 枪炮、病菌与钢铁 折扣 37.2% 评分 4.6 库存 0
· 原则 折扣 40.8% 评分 4.5 库存 0
→ 如果你偷懒不进详情页查库存,就会把这 3 本错误地放进清单
→ 这是检验"你有没有真的处理陷阱 4"的试金石最后这一点是整个靶场设计的点睛之笔:它让"偷懒"这件事有了可观测的后果。跑完 Step 4,报告里会有一行 因缺货被剔除:流畅的Python(第2版)、枪炮、病菌与钢铁、原则——那就是你正确处理了异步加载的证据。
2.2.4 六个步骤的编排逻辑
为什么是这六步、为什么是这个顺序?
图 2-2 · 六步的能力递进路线
用到第一章的哪层能力
─────────────────────
Step 1 Hello Playwright L0 驱动
│ "环境是活的吗?" 最小闭环,排除环境问题
│ 产出:一张首页截图
▼
Step 2 五个陷阱 L2 定位 · L3 操作 · L4 同步
│ "会踩哪些坑?" 先踩坑再填坑,建立肌肉记忆
│ 产出:五组"错误做法 vs 正确做法"对照
▼
Step 3 登录与会话复用 L5 状态
│ "怎么拿到登录态并复用?" 为后面所有步骤扫清障碍
│ 产出:output/auth_state.json
▼
─────────── 前三步是"能力准备",后三步是"三种解法" ───────────
│
▼
Step 4 Playwright 完整方案 ★ L0~L5 全部
│ "纯脚本能做到什么程度?" 确定性流水线,性能基准
│ 产出:报告 + CSV + JSON(13 本命中)
▼
Step 5 Browser Use 方案 L6 意图
│ "换成 AI 自主完成会怎样?" 同一目标,零选择器
│ 产出:同格式报告(可与 Step 4 交叉验证)
▼
Step 6 混合架构 ★ L6 意图 + L7 韧性
"生产环境该怎么组合?" 骨架确定 + 局部智能
产出:报告 + 自愈统计(9 本命中)关键编排思想:Step 4 / 5 / 6 是同一个业务目标的三种实现。它们输出同样格式的报告,所以你可以把三份报告放一起 diff——这是最直观的技术对比方式,比任何表格都有说服力。
2.3 第 0 步:环境从零搭建
2.3.1 为什么要单独有这一步
初学者失败率最高的地方不是写代码,是装环境。而且失败的方式很隐蔽:
你以为装好了 → 跑脚本报ModuleNotFoundError→ 你再装一遍 → 还报错
真相:你用 A 解释器装的包,却用 B 解释器跑的脚本。
所以这一步的目标是:建立一个"我确切知道自己在用哪个 Python"的环境。
2.3.2 前置条件
| 项 | 要求 | 怎么查 |
|---|---|---|
| 操作系统 | Windows / macOS / Linux 都行 | — |
| Python | ≥ 3.10(只跑 Playwright) | |
| ≥ 3.11(要跑 Step 5 的 browser-use) | python --version | |
| 磁盘 | 约 500 MB(主要是 Chromium 内核) | — |
| 网络 | 需要能装包、下内核(国内建议用镜像,下面给命令) | — |
| LLM Key | 不需要!Step 1/2/3/4/6 全部零 Key 可跑 | — |
重要:Step 5 需要 LLM Key,Step 6 有 Key 会更聪明(但没 Key 也能完整跑通,会自动降级到本地规则引擎)。没有 Key 完全不影响你学完这个项目。
2.3.3 装依赖(★ 最容易出错的一步)
先记住这条铁律:
用哪个 python 跑脚本,就用哪个 python 装包。
命令一律写成
python -m pip install ...,不要裸写pip install ...。
为什么?因为 pip 这个命令可能指向另一个 Python 环境(尤其是你装过 conda、pyenv、或者多个 Python 版本时)。而 python -m pip 强制使用"当前这个 python"自带的 pip,从根本上杜绝了装错地方。
# ── 步骤 1:进入项目目录 ────────────────────────────────────────
cd E:\WorkBuddyRoot\AIforGUI技术实战演练\AIforGUIcode
# ── 步骤 2:确认你在用哪个 python(务必执行,别跳过)─────────────
python -c "import sys; print(sys.executable)"
# 输出示例:C:\Users\你的名字\miniconda3\python.exe
# ★ 把这个路径记下来。后面所有命令都必须是这个 python。
# ── 步骤 3:装 Python 包 ───────────────────────────────────────
python -m pip install -r requirements.txt
# 国内网络慢的话,加清华镜像(实测 14 分钟 → 24 秒):
python -m pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple
# ── 步骤 4:下载浏览器内核(这一步和上一步是两回事!)──────────────
python -m playwright install chromium
# 国内加镜像(约 2~3 分钟):
# Windows PowerShell:
$env:PLAYWRIGHT_DOWNLOAD_HOST="https://cdn.npmmirror.com/binaries/playwright"
python -m playwright install chromium
# macOS / Linux / Git Bash:
PLAYWRIGHT_DOWNLOAD_HOST=https://cdn.npmmirror.com/binaries/playwright python -m playwright install chromium步骤 3 和步骤 4 是两件完全不同的事,很多人只做了前者:
步骤 3 装的是「Python 客户端库」 ← 让 import playwright 能成功
~2 MB,从 PyPI 下载
步骤 4 装的是「浏览器二进制」 ← 让 chromium.launch() 能成功
~150 MB,从 CDN 下载
装在 C:\Users\<你>\AppData\Local\ms-playwright\(Windows)
~/Library/Caches/ms-playwright/(macOS)
~/.cache/ms-playwright/(Linux)
只做步骤 3 → import 成功,但 launch() 报
"Executable doesn't exist at ...chrome.exe"requirements.txt 的内容和取舍:
playwright>=1.49 # 必需。浏览器自动化核心
python-dotenv>=1.0 # 必需。从 .env 读配置,避免把密码写进代码
pydantic>=2.0 # 必需。Step 5 的结构化输出契约
browser-use>=0.13 # 可选。只有 Step 5 需要(依赖树较大,装得慢)
openai>=1.0 # 可选。Step 6 的 LLM 需求解析(没有会自动降级)2.3.4 配置 .env(可选)
# Windows: copy .env.example .env
# macOS/Linux: cp .env.example .env.env 里最有用的两个开关(调试神器):
HEADLESS=0 # 0 = 显示真实浏览器窗口,你能亲眼看到脚本在操作
# 1 = 无头模式(默认,快,适合正式跑)
SLOW_MO=300 # 每个操作之间停 300 毫秒,慢动作回放,方便肉眼跟上强烈建议第一次跑的时候设成HEADLESS=0SLOW_MO=300。 你会看到浏览器自己弹出来、自己关浮层、自己填账号、自己翻页——这个画面比任何文字讲解都直观,而且能让你立刻理解每一行代码在干什么。
LLM 相关配置(只有 Step 5 必需):
LLM_PROVIDER=openai # openai | anthropic | google | deepseek | ollama | browseruse
LLM_MODEL=gpt-4.1-mini
OPENAI_API_KEY=sk-xxxxxxxx
LLM_BASE_URL= # 用第三方 OpenAI 兼容网关时填这里
MAX_AGENT_STEPS=40 # 成本熔断器,见 1.7.52.3.5 一键自检
写业务代码之前,先让机器告诉你环境对不对。项目自带 check_env.py:
# 终端 A:启动靶场(保持开着)
python demosite/server.py
# 终端 B:自检
python check_env.py本机真实输出:
==========================================================
AIforGUI 实战 · 环境自检
==========================================================
1. Python 版本
----------------------------------------------------------
当前:3.13.9 (C:\Users\hechu\miniconda3\python.exe)
✅ 满足 Playwright(>=3.10) 与 browser-use(>=3.11) 的要求
2. Python 依赖包
----------------------------------------------------------
✅ playwright 1.62.0
✅ pydantic 2.12.5
✅ python-dotenv 1.0.0
⚠️ browser-use 未安装(可选)
✅ openai 2.29.0
3. Playwright 浏览器内核
----------------------------------------------------------
✅ Chromium 可启动,版本 151.0.7922.34
4. 靶场站点 BookNest
----------------------------------------------------------
✅ 已在 127.0.0.1:8899 运行,共 24 本书
5. 项目文件完整性
----------------------------------------------------------
✅ demosite/server.py
✅ demosite/data.py
✅ aigui/config.py
✅ aigui/models.py
✅ aigui/report.py
✅ steps/step1_hello_playwright.py
✅ steps/step4_playwright_scraper.py
==========================================================
🎉 全部通过,可以开始 Step 1 了:
python steps/step1_hello_playwright.py
==========================================================注意第 1 项把解释器完整路径打出来了——这是刻意设计的。只要这个路径和你 python -m pip install 时用的是同一个,就绝不会有环境问题。
check_env.py 的关键实现(学会这个模式,以后自己的项目也该配一个):
def check_pkg(name: str, import_name: str | None = None, required: bool = True) -> bool:
# name : PyPI 上的包名,如 "python-dotenv"
# import_name : import 时用的模块名,如 "dotenv"(两者经常不一样!)
mod = import_name or name.replace("-", "_")
# find_spec 只查"能不能找到",不真的 import。
# 好处:即使这个包 import 时会报错(缺依赖、版本冲突),这里也不会崩。
found = importlib.util.find_spec(mod) is not None
if found:
try:
from importlib.metadata import version as _v
ver = _v(name) # ★ 从包元数据读版本,而不是 mod.__version__
# 因为很多现代包(如 pydantic)根本没有 __version__
except Exception:
ver = getattr(__import__(mod), "__version__", "?")
print(f" ✅ {name} {ver}")
else:
print(f" {'❌' if required else '⚠️'} {name} 未安装({'必需' if required else '可选'})")
if required:
problems.append(f"pip install {name}") # 收集问题,最后统一给修复建议
return found2.4 Step 1 · Hello Playwright:先确认环境是活的
2.4.1 为什么要有这一步
目的:用最小的代码验证"Python → driver → 浏览器 → 页面"这条链路是通的。
这一步不产生任何业务价值,但它极其重要。原因是:
如果你跳过它直接写 200 行采集脚本,跑挂了你根本不知道是环境问题还是代码问题。
有了 Step 1,环境问题在第 10 行代码就暴露了,而不是在第 200 行。
这是工程上的一个通用原则:先建立最小可用闭环,再往上加东西。
产出:终端打印页面标题 + output/screenshots/step1_home.png 一张截图。
2.4.2 打算怎么做
启动 Playwright → 启动浏览器 → 建 context → 开标签页
→ 访问首页 → 读标题 → 截图 → 依次关闭图 2-3 · Step 1 流程
┌──────────────────────────────────────────────────────────┐
│ with sync_playwright() as p: │
│ 启动 Node driver 子进程,建立管道 │ ← 对应 1.1.2 图 1-3
│ │ │
│ ▼ │
│ p.chromium.launch() │
│ 启动 Chromium 进程(headless 决定有没有窗口) │
│ │ │
│ ▼ │
│ browser.new_context(viewport, locale) │
│ 建一个干净的"浏览器身份"(无 cookie、无缓存) │ ← 对应 1.6.1 图 1-13
│ │ │
│ ▼ │
│ context.new_page() → 标签页 │
│ │ │
│ ▼ │
│ page.goto(url, wait_until="domcontentloaded") │
│ │ │
│ ├──▶ page.title() 读标题 │
│ ├──▶ page.url 读当前地址 │
│ └──▶ page.screenshot(full_page=True) 截整页 │
│ │ │
│ ▼ │
│ context.close() → browser.close() │
│ │ │
└────────┼──────────────────────────────────────────────────┘
▼
离开 with 块 → driver 进程被回收(★ 这就是必须用 with 的原因)2.4.3 代码实现与逐行解释
代码块 1/3:路径设置
import sys
from pathlib import Path
# 把项目根目录加进模块搜索路径。
# 为什么需要:脚本在 steps/ 子目录里,直接 python steps/xxx.py 运行时,
# Python 只会把 steps/ 加进 sys.path,找不到上一级的 aigui 包。
#
# __file__ → 当前文件路径 steps/step1_hello_playwright.py
# .resolve() → 转成绝对路径(防止相对路径在不同工作目录下出错)
# .parents[1] → 往上跳 1 级:parents[0]=steps/,parents[1]=项目根
# sys.path.insert(0, ...) → 插到最前面,优先级最高
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from playwright.sync_api import sync_playwright # noqa: E402
from aigui import config # noqa: E402
# noqa: E402 是告诉代码检查工具:我知道 import 不在文件顶部,这是故意的
# (必须先设好 sys.path 才能 import aigui)为什么用sync_api而不是async_api:Playwright 提供同步和异步两套完全对等的 API。同步版本代码更直观(没有async/await),适合初学和脚本场景;异步版本适合高并发。功能上没有任何差别。 本项目 Step 1\~4、6 用同步版,Step 5 因为 browser-use 是异步框架,用了asyncio。
代码块 2/3:核心流程
def main() -> int:
shot = config.SHOTS_DIR / "step1_home.png" # Path 对象支持 / 拼接,跨平台安全
# ── sync_playwright() 做两件事:启动 driver 子进程 + 建立通信管道 ──
# 必须用 with:离开代码块时自动 stop(),否则 driver 进程会残留,
# 表现为"脚本跑完了但终端不返回"。
with sync_playwright() as p:
# ── 启动浏览器进程 ──────────────────────────────────────
# p.chromium 也可以换成 p.firefox / p.webkit(同一套 API!)
# headless=True 无头,不弹窗口,快,适合服务器/CI
# headless=False 弹出真实窗口,你能看到脚本在操作 ← 调试必备
# slow_mo=300 每个操作后停 300ms,慢动作,方便肉眼跟上
browser = p.chromium.launch(headless=config.HEADLESS, slow_mo=config.SLOW_MO)
# ── 建 context(一个独立的"浏览器身份")─────────────────
# 这是 Playwright 相对 Selenium 的核心效率优势:
# 开一个 browser ≈ 300~800ms,开一个 context ≈ 10ms
# viewport 会影响响应式布局,也影响"元素是否在视口内"的判定
# locale 会影响 Accept-Language 请求头和页面的语言相关渲染
context = browser.new_context(viewport={"width": 1280, "height": 900},
locale="zh-CN")
page = context.new_page() # 开一个标签页
page.set_default_timeout(config.DEFAULT_TIMEOUT)
# ↑ 给这个页面上所有操作设统一超时(15 秒)。
# 不设的话默认 30 秒,出错时要干等很久。
# ── 访问页面 ────────────────────────────────────────────
print(f"[1/4] 访问 {config.BASE_URL}")
page.goto(config.BASE_URL, wait_until="domcontentloaded")
# wait_until 的四个取值,直接决定 goto 什么时候返回:
# "commit" → 收到响应头就返回(最快,几乎不等)
# "domcontentloaded" → DOM 构建完就返回 ★ 本项目用这个
# "load" → 等图片/CSS 等所有资源加载完(慢,通常没必要)
# "networkidle" → 等网络空闲 500ms(最慢,官方已不推荐,容易超时)
#
# 为什么选 domcontentloaded:我们只关心 DOM 结构,不关心图片。
# 至于"数据什么时候到",交给后面的显式等待处理(见 Step 2 陷阱 4)。
print(f"[2/4] 页面标题:{page.title()}") # 读 <title>
print(f"[3/4] 当前 URL:{page.url}") # 读当前地址(能看出有没有被重定向)
# ── 截图 ────────────────────────────────────────────────
# full_page=True 会滚动整页拼接成一张长图;False 只截当前视口
# path 必须是 str,Playwright 不接受 Path 对象
page.screenshot(path=str(shot), full_page=True)
print(f"[4/4] 截图已保存:{shot}")
# ── 收尾 ────────────────────────────────────────────────
# 严格来说 with 退出时会自动清理,
# 但显式关闭是好习惯:长脚本里能及时释放内存
context.close()
browser.close()
print("\n✅ Step 1 完成。环境正常,可以进入 Step 2。")
return 0 # 0 = 成功(Unix 约定)代码块 3/3:入口
if __name__ == "__main__":
raise SystemExit(main())
# 为什么是 raise SystemExit(main()) 而不是直接 main()?
# 它把函数返回值变成进程退出码。
# 这样 run_all.py 用 subprocess.run(...).returncode 就能判断这一步成没成功。
# 返回 0 = 成功,非 0 = 失败 —— 这是所有命令行工具的通用约定。2.4.4 运行与真实输出
python steps/step1_hello_playwright.py[1/4] 访问 http://127.0.0.1:8899
[2/4] 页面标题:首页 - BookNest
[3/4] 当前 URL:http://127.0.0.1:8899/
[4/4] 截图已保存:E:\WorkBuddyRoot\AIforGUI技术实战演练\AIforGUIcode\output\screenshots\step1_home.png
✅ Step 1 完成。环境正常,可以进入 Step 2。验收标准:
- ✅ 标题是
首页 - BookNest(不是空、不是报错页) - ✅
output/screenshots/step1_home.png存在,打开能看到完整的商城首页
2.4.5 这一步的常见错误
| 报错 | 原因 | 解决 |
|---|---|---|
ModuleNotFoundError: No module named 'playwright' | 装包和跑脚本用了不同的 Python | 见2.11 排错手册 |
Executable doesn't exist at ... | 只做了步骤 3,没做步骤 4 | python -m playwright install chromium |
net::ERR_CONNECTION_REFUSED | 靶场没启动 | 另开终端 python demosite/server.py |
| 脚本跑完终端不返回 | 没用 with,driver 进程泄漏 | 检查代码结构 |
2.5 Step 2 · 五个陷阱:先踩坑,再填坑
2.5.1 为什么要专门有这一步
这一步不产出任何业务数据。它的唯一目的是让你"亲手踩坑"。
理由很实在:
直接给你正确代码,你会照抄,但记不住,因为你没体会过错误的样子。
让你先看到错误做法怎么挂、挂的时候报什么错,再看正确做法——
下次你在真实项目里遇到同样的报错,能在三秒内反应过来。
所以这一步的每个陷阱都是成对的:先跑错误做法(并捕获异常打印出来),再跑正确做法。
产出:五组对照演示 + output/screenshots/step2_lastpage.png。
2.5.2 打算怎么做
图 2-4 · Step 2 流程
打开 /products(未登录状态,Cookie 浮层在)
│
├─ 陷阱 1 故意直接点商品 → 捕获异常 → 提取诊断信息 → 打印
│ 然后关掉浮层 → 验证现在能点了
│
├─ 陷阱 2 读一次 class → reload → 再读一次 class → 对比
│ 然后演示三种语义定位器各匹配到几个
│
├─ 陷阱 3 数一下 member-price-locked 出现几次 → 证明被登录墙挡了
│
├─ 陷阱 4 进详情页 → 立刻读(拿到"加载中…")
│ → 等 data-loaded=true → 再读(拿到真实库存)
│ → 再演示 expect 断言的写法
│
└─ 陷阱 5 读页面自带的"第 X / Y 页,共 N 本"
→ while 循环翻页,终止条件 = 下一页按钮消失
→ 带 10 次硬上限兜底
→ 截图存档2.5.3 陷阱 1:Cookie 浮层拦截点击
这段代码要证明的事:Playwright 不会"盲点",它会告诉你谁挡住了。
banner(1, "Cookie 同意浮层会拦截点击")
try:
# ── 错误做法:无视浮层,直接点商品链接 ──────────────────────
# timeout=3000 是故意调短的。默认 30 秒,演示时不想干等。
page.get_by_test_id("book-link").first.click(timeout=3000)
print(" ✗ 意外成功了(浮层可能已被关闭)")
except (PWError, PWTimeout) as e:
# Playwright 的异常信息是多行的,结构大致是:
# 第 1 行:Locator.click: Timeout 3000ms exceeded.
# 中间: Call log: - waiting for locator(...)
# ★ 某一行:- <div class="consent-backdrop"...> intercepts pointer events
msg = str(e).strip()
head = msg.splitlines()[0] # 第一行 = 错误类型摘要
# ★ 真正有价值的诊断藏在中间某一行,用生成器表达式把它捞出来。
# next(生成器, 默认值) 的写法:找到第一个匹配就返回,找不到返回 ""
detail = next((ln.strip() for ln in msg.splitlines()
if "intercepts pointer events" in ln), "")
print(f" ✗ 错误做法:直接点击商品 → {head}")
if detail:
print(f" 诊断:{detail}")
print(" Playwright 的 actionability 检查发现元素被遮挡,主动报错而不是盲点。")
# ── 正确做法:先关浮层 ─────────────────────────────────────────
page.get_by_test_id("accept-cookies").click()
print(" ✓ 正确做法:先关闭浮层,再操作")
# 关完之后验证一下:等商品链接变成 visible 状态
# wait_for(state=...) 的四个取值:
# "attached" 在 DOM 里就行(哪怕看不见)
# "visible" 在 DOM 里 + 有尺寸 + 没被 display:none ← 常用
# "hidden" 不可见或不存在
# "detached" 从 DOM 移除 ← 关弹窗后确认它真没了,用这个
page.get_by_test_id("book-link").first.wait_for(state="visible")
print(" 现在商品链接可点击了。")真实输出:
──────────────────────────────────────────────────────────────
陷阱 1:Cookie 同意浮层会拦截点击
──────────────────────────────────────────────────────────────
✗ 错误做法:直接点击商品 → Locator.click: Timeout 3000ms exceeded.
诊断:- <div class="consent-backdrop" data-testid="consent-backdrop"></div> intercepts pointer events
Playwright 的 actionability 检查发现元素被遮挡,主动报错而不是盲点。
✓ 正确做法:先关闭浮层,再操作
现在商品链接可点击了。看那行诊断——Playwright 把凶手的完整 HTML 都打出来了。这就是第一章 1.4.2 actionability 检查 第 ⑤ 项 "Receives Events" 命中测试失败的结果。
生产级写法:真实项目里不该每次都 try/except,而是写一个幂等的关闭函数(Step 4 里就是这么做的):
def dismiss_consent(page: Page) -> None:
"""幂等地关掉 Cookie 浮层:存在就点,不存在就跳过。"""
btn = page.get_by_test_id("accept-cookies")
if btn.count(): # count() == 0 说明浮层不在,直接返回
btn.click()
# ★ 关键:等浮层真的从 DOM 里消失,再往下走。
# 不等的话,浮层还在做淡出动画时你就去点,照样被拦。
page.get_by_test_id("consent-box").wait_for(state="detached", timeout=3000)"幂等"是自动化脚本里非常重要的一个性质:同一个函数调用一次和调用十次,效果一样。这样你就可以在每个页面加载后无脑调一次,不用去想"这个页面到底有没有浮层"。
2.5.4 陷阱 2:动态 class 名
这段代码要证明的事:class 真的会变,用它做选择器就是在埋雷。
# 读第一次渲染的 class
cls_1 = page.get_by_test_id("book-card").first.get_attribute("class")
page.reload(wait_until="domcontentloaded") # 刷新页面
# 读第二次渲染的 class
cls_2 = page.get_by_test_id("book-card").first.get_attribute("class")
print(f" 第一次渲染 class = {cls_1}")
print(f" 第二次渲染 class = {cls_2}")
# cls_1.split()[-1] 取最后一个 class(也就是那个随机的),拼进提示语里
print(f" ✗ 用 .{cls_1.split()[-1]} 写死选择器,下次刷新必挂")
print(" ✓ 用 data-testid / get_by_role / get_by_text 这类语义定位器")
# ── 三种语义定位器的效果对比 ─────────────────────────────────
n_testid = page.get_by_test_id("book-card").count()
# → 6,因为每页 6 张卡片。count() 常用来做"数量断言"
n_role = page.get_by_role("link", name="深入理解计算机系统").count()
# → 1。get_by_role 基于无障碍树:
# role="link" → <a> 标签的隐含角色
# name="…" → 无障碍名称(这里就是链接文本)
# ★ 默认是"完全匹配但忽略大小写和首尾空格",可加 exact=False 改成包含匹配
n_text = page.get_by_text("算法导论(第4版)").count()
# → 1。get_by_text 默认是"包含匹配"(substring),
# 所以搜 "算法" 会匹配到更多元素。要精确匹配就加 exact=True真实输出:
──────────────────────────────────────────────────────────────
陷阱 2:class 名每次刷新都变,不能当定位依据
──────────────────────────────────────────────────────────────
第一次渲染 class = card card-53a362
第二次渲染 class = card card-f6c6e7
✗ 用 .card-53a362 写死选择器,下次刷新必挂
✓ 用 data-testid / get_by_role / get_by_text 这类语义定位器
get_by_test_id('book-card') → 6 个
get_by_role('link', name='深入理解计算机系统') → 1 个
get_by_text('算法导论(第4版)') → 1 个card-53a362 → card-f6c6e7,同一个页面、同一个元素、两次刷新,class 完全不同。靶场里对应的代码只有三行:
def rand_cls() -> str:
"""坑 2:每次渲染都生成不同的 class 后缀,逼你放弃 class 选择器。"""
return "card-" + secrets.token_hex(3) # 生成 6 位十六进制随机串这不是我瞎编的场景——CSS Modules、styled-components、Tailwind JIT、以及各种前端混淆工具,生成的就是这种 hash class。
2.5.5 陷阱 3:登录墙
locked = page.get_by_test_id("member-price-locked").count()
print(f" 未登录状态下,'登录后查看会员价' 占位出现 {locked} 次 → 抓不到真实价格")
print(" ✓ 正确做法:先登录并复用 storage_state(Step 3 展开讲)")真实输出:
未登录状态下,'登录后查看会员价' 占位出现 6 次 → 抓不到真实价格6 次 = 这一页 6 本书全被挡住。靶场服务端的逻辑很直白:
if logged_in:
price_html = '...<span data-testid="member-price">¥76.00</span>'
else:
price_html = '...<span data-testid="member-price-locked">登录后查看会员价</span>'注意这里的一个设计细节:登录前后用的是两个不同的 testid。这是有意的——它让"有没有登录"变成一个可以用代码精确判定的事实,而不是靠猜。Step 4 里就用了这一点做防御:
member = c.get_by_test_id("member-price")
if member.count() == 0: # 会员价 testid 一个都找不到 → 登录态一定失效了
raise RuntimeError("会员价被锁定 —— 登录态失效,请删除 output/auth_state.json 重跑")这行代码的价值:把一个会导致"静默抓到脏数据"的问题,变成一个响亮的报错。宁可失败,不要抓错。
2.5.6 陷阱 4:异步渲染(本项目最重要的陷阱)
page.goto(f"{config.BASE_URL}/book/1", wait_until="domcontentloaded")
# ── 错误做法:DOM 一好就读 ─────────────────────────────────────
immediate = page.get_by_test_id("stock").inner_text()
print(f" ✗ 错误做法:domcontentloaded 后立刻读 → '{immediate}'")
# ★ 注意:这里不会报错!它成功读到了字符串,只是内容是 "加载中…"
# 这就是异步渲染最危险的地方 —— 静默的错误数据
# ── 正确做法 A:等属性变化(最精确)───────────────────────────
page.wait_for_selector('[data-testid="stock"][data-loaded="true"]', timeout=5000)
# 这是一个 CSS 属性选择器,要求同时满足两个条件:
# [data-testid="stock"] 是库存那个元素
# [data-loaded="true"] 并且 data-loaded 已经被 JS 改成 true
# 元素虽然一开始就在 DOM 里,但属性不满足 → 继续等
# 800ms 后 JS 改了属性 → 立刻返回(不多等一毫秒)
settled = page.get_by_test_id("stock").inner_text()
print(f" ✓ 正确做法:wait_for_selector 等 data-loaded=true → '{settled}'")
# ── 正确做法 B:web-first 断言(测试场景更常用)─────────────────
from playwright.sync_api import expect
expect(page.get_by_test_id("stock")).not_to_have_text("加载中…", timeout=5000)
# expect() 和普通 assert 的根本区别:
# assert → 判断"此刻"是不是,不是就立刻失败
# expect → 在 timeout 内反复轮询,"变成了"就通过
# 适用场景:前端没提供 data-loaded 这种完成标记时,只能盯文本变化
print(" ✗ 千万别用 time.sleep(3) —— 慢且不可靠,这是自动化脚本的头号异味")真实输出:
──────────────────────────────────────────────────────────────
陷阱 4:库存由 JS 延迟 800ms 注入
──────────────────────────────────────────────────────────────
✗ 错误做法:domcontentloaded 后立刻读 → '加载中…'
✓ 正确做法:wait_for_selector 等 data-loaded=true → '有货,剩余 32 件'
✓ 或者用 expect(...).not_to_have_text(...),内置轮询重试
✗ 千万别用 time.sleep(3) —— 慢且不可靠,这是自动化脚本的头号异味'加载中…' vs '有货,剩余 32 件' —— 这就是全部区别。 错误做法不报错、不崩溃,它只是安静地给了你一个错的答案。如果你没做这一步,Step 4 就会把那 3 本零库存的书错误地放进捡漏清单。
靶场对应的服务端代码:
# 详情页 HTML 里初始渲染的是这个:
<span data-testid="stock" data-loaded="false">加载中…</span>
# 页面底部的 script:
setTimeout(function(){
var el = document.querySelector('[data-testid=stock]');
el.textContent = 32 > 0 ? '有货,剩余 32 件' : '暂时缺货';
el.setAttribute('data-loaded','true'); // ★ 完成标记
el.setAttribute('data-stock','32'); // ★ 机器可读的纯数字
}, 800);给前端同学的建议(值得转发给你的团队):如果你在做的页面有异步数据,请提供两个东西——一个完成标记(data-loaded)和一个机器可读的原始值(data-stock="32"而不是只有"有货,剩余 32 件")。这两个属性能让自动化和测试的成本下降一个数量级,成本却几乎为零。
2.5.7 陷阱 5:分页
info = page.get_by_test_id("page-info").inner_text()
print(f" 页面自带的进度信息:{info}")
print(" ✓ 优先读页面声明的总页数/总条数,而不是'一直点下一页直到报错'")
clicks = 0
# 循环条件是两个的 and:
# ① page-next 还存在(页面自己说还有下一页) ← 业务终止条件
# ② clicks < 10 ← 硬上限兜底
while page.get_by_test_id("page-next").count() > 0 and clicks < 10:
page.get_by_test_id("page-next").click()
page.wait_for_load_state("domcontentloaded") # 等新页面 DOM 就绪
clicks += 1
print(f" 翻到 → {page.get_by_test_id('page-info').inner_text()}")
print(f" 共翻页 {clicks} 次,末页无 '下一页' 按钮,循环自然终止(带 10 次硬上限兜底)")真实输出:
页面自带的进度信息:第 1 / 4 页,共 24 本
✓ 优先读页面声明的总页数/总条数,而不是'一直点下一页直到报错'
翻到 → 第 2 / 4 页,共 24 本
翻到 → 第 3 / 4 页,共 24 本
翻到 → 第 4 / 4 页,共 24 本
共翻页 3 次,末页无 '下一页' 按钮,循环自然终止(带 10 次硬上限兜底)这段代码里有两个值得学的工程习惯:
- 终止条件用"页面自己的声明",而不是"点到报错为止"。后者会让每次运行的最后一步都是一次 30 秒超时,又慢又难看。
- 永远加硬上限。
clicks < 10这个条件在正常情况下永远不会触发。它存在的意义是:万一站点出 bug(比如末页的"下一页"指向自己),你的脚本会在 10 次后停下,而不是无限循环跑一整晚。
这个模式在 Step 4 里升级成了抛异常版本:
guard += 1 if guard > 20: raise RuntimeError("翻页超过 20 次,疑似死循环,已中止")静默地停下 vs 响亮地报错——生产脚本要选后者,因为你需要知道出事了。
2.5.8 本步小结
| 陷阱 | 错误的表现 | 正确做法 | 危险等级 |
|---|---|---|---|
| 1 Cookie 浮层 | 点击超时报错 | 幂等关闭 + 等 detached | ⚠️ 会报错,容易发现 |
| 2 动态 class | 选择器找不到元素 | 用 testid / role / text | ⚠️ 会报错,容易发现 |
| 3 登录墙 | 抓到占位文本 | storage\_state 复用 | ☠️静默错误 |
| 4 异步渲染 | 读到"加载中…" | 等属性 / expect 轮询 | ☠️静默错误 |
| 5 分页 | 死循环 / 漏页 | 页面声明 + 硬上限 | ⚠️ 中等 |
记住那两个 ☠️。会报错的问题都不算大问题——你会立刻发现并修复。真正会造成业务事故的,是那些不报错但数据是错的情况。
2.6 Step 3 · 登录与会话复用
2.6.1 为什么要有这一步
问题:Step 2 陷阱 3 已经证明了——不登录就抓不到会员价。那直接在每个脚本开头都写一遍登录流程不就行了?
不行,有三个理由:
① 慢
一次登录 = 打开登录页 + 关浮层 + 填两个输入框 + 提交 + 等跳转
≈ 2~4 秒。Step 4 和 Step 6 各跑一次,就是白花 8 秒。
放到真实项目:100 个测试用例 × 3 秒 = 白等 5 分钟。
② 危险
频繁登录会触发风控:验证码、短信二次验证、账号异常锁定。
这在真实项目里是硬伤,不是理论风险。
③ 重复
登录逻辑写在每个脚本里 → 账号改了要改 N 处 → 迟早漏改一处。解法:登录一次,把凭证导出成文件,之后所有脚本加载这个文件。这就是 storage_state(原理见 1.6.2)。
产出:output/auth_state.json + output/screenshots/step3_logged_in.png
2.6.2 打算怎么做
这一步分两个独立的函数,对应两件事:
图 2-5 · Step 3 流程
┌─ do_login():执行一次真实登录并落盘 ────────────────────────────┐
│ │
│ 打开 /login │
│ │ │
│ ▼ │
│ 关 Cookie 浮层(否则表单可能被遮挡) │
│ │ │
│ ▼ │
│ fill 用户名 → fill 密码 │
│ │ │
│ ▼ │
│ click 登录 → wait_for_url("**/products**") │
│ │ ↑ 等 URL 变成商品页,证明跳转发生了 │
│ ▼ │
│ expect(current-user).to_contain_text("demo") │
│ │ ↑ ★ 用断言确认真登录了,不靠"没报错"来判断 │
│ ▼ │
│ ctx.storage_state(path="output/auth_state.json") ← 核心! │
└─────────────────────────────────────────────────────────────────┘
│
▼
┌─ verify_reuse():开全新 context,验证免登录 ────────────────────┐
│ │
│ new_context(storage_state="output/auth_state.json") │
│ │ ★ 全新的 context,没走任何登录流程 │
│ ▼ │
│ 直接 goto /products │
│ │ │
│ ▼ │
│ 数一下:member-price 有几条?member-price-locked 有几条? │
│ │ │
│ ▼ │
│ assert prices > 0 and locked == 0 ← 6 和 0 才算通过 │
└─────────────────────────────────────────────────────────────────┘为什么要有 verify_reuse() 这个验证函数? 因为"文件生成了"不等于"文件有用"。可能 cookie 域名不对、可能会话已过期、可能保存时机太早。只有真的用它开一个新 context 并看到会员价,才算真正成功。
这是一个通用的工程原则:验证结果,而不是验证过程。
2.6.3 代码实现与逐行解释
函数 1:do_login()
def do_login() -> Path:
"""执行一次真实登录,并把会话状态落盘。"""
with sync_playwright() as p:
browser = p.chromium.launch(headless=config.HEADLESS, slow_mo=config.SLOW_MO)
ctx = browser.new_context(viewport={"width": 1280, "height": 900}, locale="zh-CN")
page = ctx.new_page()
page.set_default_timeout(config.DEFAULT_TIMEOUT)
print(f"[1/5] 打开登录页 {config.LOGIN_URL}")
page.goto(config.LOGIN_URL, wait_until="domcontentloaded")
# ── 先处理 Cookie 浮层 ─────────────────────────────────
# 用 if + count() 而不是 try/except:
# count() 是"数一下有几个",不存在返回 0,永远不会抛异常。
# 这比捕获异常更清晰,也更快(不用等超时)。
if page.get_by_test_id("accept-cookies").count():
page.get_by_test_id("accept-cookies").click()
print("[2/5] 已关闭 Cookie 同意浮层")
print(f"[3/5] 填写账号 {config.USERNAME}")
# ── fill() vs type() 的区别(重要)────────────────────
# fill() → 先清空,再一次性设值,然后派发 input 事件。快,推荐。
# type() → 逐字符模拟键盘输入,每个字符都触发 keydown/keypress/keyup。
# 慢,但某些前端(如联想搜索、格式化输入框)依赖逐键事件。
# 本项目是普通表单,用 fill 即可。
page.get_by_test_id("username").fill(config.USERNAME)
page.get_by_test_id("password").fill(config.PASSWORD)
# ── 提交并等待跳转 ────────────────────────────────────
page.get_by_test_id("login-submit").click()
# 老教程会写 with page.expect_navigation(): ... —— 那个 API 已废弃。
# 现代写法:直接等 URL 变成期望的样子。
# "**/products**" 是 glob 通配:** 匹配任意字符(含 /)
page.wait_for_url("**/products**", timeout=config.DEFAULT_TIMEOUT)
# ── ★ 关键:用断言确认真的登录成功了 ───────────────────
# 为什么不能靠"前面没报错"来判断?
# 因为登录失败时,服务端也可能 200 返回一个带错误提示的页面,
# URL 甚至也可能变。只有页面上出现了"已登录:demo"才是铁证。
# expect(...) 是 web-first 断言,内置轮询重试。
expect(page.get_by_test_id("current-user")).to_contain_text(config.USERNAME)
print(f"[4/5] 登录成功:{page.get_by_test_id('current-user').inner_text()}")
# ── ★★ 核心的一行:把会话状态序列化到磁盘 ──────────────
# 它会导出:
# cookies —— 所有域的 cookie(含 bn_session)
# origins —— 各来源的 localStorage
# 注意:sessionStorage 不会被导出(设计如此,它本就是标签页级的)
ctx.storage_state(path=str(config.STATE_FILE))
print(f"[5/5] 会话状态已保存:{config.STATE_FILE}")
page.screenshot(path=str(config.SHOTS_DIR / "step3_logged_in.png"), full_page=True)
ctx.close()
browser.close()
return config.STATE_FILE函数 2:verify_reuse()
def verify_reuse() -> None:
"""新开一个 context,直接加载 storage_state,验证免登录。"""
print("\n--- 验证会话复用(全新 context,不再走登录流程)---")
with sync_playwright() as p:
browser = p.chromium.launch(headless=config.HEADLESS)
# ── ★ 这一行就是全部的魔法 ───────────────────────────
# context 一创建就带着 auth_state.json 里的所有 cookie。
# 从浏览器和服务端的角度看,它和"刚登录完的那个 context"没有区别。
ctx = browser.new_context(storage_state=str(config.STATE_FILE), locale="zh-CN")
page = ctx.new_page()
page.goto(config.PRODUCTS_URL, wait_until="domcontentloaded")
# ↑ 注意:没有任何登录步骤,直奔商品页
# ── 三重验证 ──────────────────────────────────────────
user = page.get_by_test_id("current-user").inner_text() # ① 顶栏显示已登录
prices = page.get_by_test_id("member-price").count() # ② 会员价出现了
locked = page.get_by_test_id("member-price-locked").count() # ③ 没有被锁的
print(f" 身份:{user}")
print(f" 本页可见会员价 {prices} 条,被锁定 {locked} 条")
# assert 在这里是合适的:这是"此刻的事实判断",不需要轮询。
# 前面页面已经加载完了,不存在"等一会就变了"的情况。
assert prices > 0 and locked == 0, "会话复用失败,会员价仍被锁定"
print(" ✓ 会话复用成功,会员价全部解锁")
ctx.close()
browser.close()2.6.4 运行与真实输出
python steps/step3_login_session.py[1/5] 打开登录页 http://127.0.0.1:8899/login
[2/5] 已关闭 Cookie 同意浮层
[3/5] 填写账号 demo
[4/5] 登录成功:已登录:demo
[5/5] 会话状态已保存:E:\WorkBuddyRoot\AIforGUI技术实战演练\AIforGUIcode\output\auth_state.json
--- 验证会话复用(全新 context,不再走登录流程)---
身份:已登录:demo
本页可见会员价 6 条,被锁定 0 条
✓ 会话复用成功,会员价全部解锁
✅ Step 3 完成。后续脚本可直接 new_context(storage_state=...) 免登录。会员价 6 条,被锁定 0 条 —— 对照 Step 2 陷阱 3 的 被锁定 6 次,这就是登录态生效的铁证。
生成的 output/auth_state.json 长这样:
{
"cookies": [
{
"name": "bn_session",
"value": "Kj8mN2pQ7rS4tU6v...",
"domain": "127.0.0.1",
"path": "/",
"expires": 1786000000,
"httpOnly": false,
"secure": false,
"sameSite": "Lax"
}
],
"origins": []
}⚠️ 这个文件等同于账号密码。 拿到它的人可以直接以你的身份访问网站。
- 绝不能提交到 Git(本项目放在
output/产物目录,应该在.gitignore里)- CI 环境里应该用 secrets 管理,或者每次重新生成
- 生产脚本要处理过期:加载后先访问一个需登录页面,被重定向到登录页就说明过期了
2.6.5 生产级增强:过期自动重登
Step 4 里的 ensure_auth() 是这个思路的简化实现:
def ensure_auth(p) -> None:
"""没有登录态就先登录一次。让脚本可以独立运行,不强依赖 Step 3。"""
if config.STATE_FILE.exists():
return # 文件在就认为可用(教学项目的简化)
print(" 未找到 auth_state.json,先执行一次登录…")
# …完整登录流程…这个设计让每个 Step 都能独立运行——你可以直接跑 Step 4,不必先跑 Step 3。这在教学项目里很重要(学员不会因为漏跑一步而卡住),在生产项目里同样重要(每个任务自包含,不依赖执行顺序)。
真正的生产版本还应该验证有效性:
def ensure_auth_production(p) -> None:
if config.STATE_FILE.exists():
# 不能只看文件在不在,要真的试一下
ctx = browser.new_context(storage_state=str(config.STATE_FILE))
page = ctx.new_page()
page.goto(config.PRODUCTS_URL)
if page.get_by_test_id("current-user").count() > 0:
ctx.close()
return # 会话还活着
print(" 会话已过期,重新登录…")
ctx.close()
config.STATE_FILE.unlink() # 删掉过期文件
do_login() # 重新登录2.7 Step 4 · Playwright 完整方案(确定性流水线)
这是本项目技术含量最高的一步,也是最值得反复读的一步。
2.7.1 为什么要有这一步
前三步都是"能力准备"。从这一步开始,我们真正解决业务问题:把 24 本书筛成一份捡漏清单。
这一步要建立的是性能基准——纯脚本方案能做到多快、多准、多便宜。有了这个基准,Step 5 的 AI 方案和 Step 6 的混合方案才有对比的意义。
产出:
output/report_playwright.md— 人看的报告output/books_playwright.csv— 进 Excel 的output/result_playwright.json— 给下游程序的output/screenshots/step4_final.png
2.7.2 核心难题:字段分散在两类页面上
先看清楚问题:
列表页 /products 能拿到什么?
✓ 书名、作者、出版社、定价、会员价、评分、评价数
✗ 分类(列表页不显示,只能靠筛选器反推)
✗ 库存(列表页完全没有)
→ 一次加载能拿 6 本书的数据
详情页 /book/<id> 能拿到什么?
✓ 上面全部 + 分类 + 库存
✗ 但一次只能拿 1 本
✗ 而且库存是异步的,每次要多等 800ms算一笔账:
方案甲:全都进详情页
24 次页面加载 × (加载 ~200ms + 异步等待 800ms) ≈ 24 秒
简单,但慢
方案乙:先用列表页筛,再只查候选的详情页
列表页 4 次 + 分类页 4 次 + 详情页 N 次
本项目 N = 16(折扣≥15% 且 评分≥4.5 的有 16 本)
→ 详情页访问量降低 33%图 2-6 · 这就是"三段式 Pass"设计的由来
┌─────────────────────────────────────────────────────────────┐
│ Pass A · 分页扫描全部列表页 │
│ 4 次页面加载 → 24 本书的基础字段 │
│ (书名/作者/定价/会员价/评分/评价数) │
│ 成本:便宜(一次加载 6 本) │
└───────────────────────┬─────────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ Pass B · 用分类筛选器补全 category │
│ 读 <select> 里有哪些分类 → 对每个分类再扫一遍列表 │
│ 4 次页面加载 → 给每本书打上分类标签 │
│ ★ 技巧:不进详情页也能拿到分类,靠的是"筛选结果反推" │
└───────────────────────┬─────────────────────────────────────┘
▼
┌═════════════════════════════════════════════════════════════┐
║ ★★ 关键决策点:用便宜字段先筛 ★★ ║
║ ║
║ candidates = [b for b in books ║
║ if b.discount_pct >= 15 and b.rating >= 4.5] ║
║ ║
║ 24 本 ──────▶ 16 本 ║
║ 省掉 8 次详情页访问 = 省掉 8 × 1 秒 = 8 秒 ║
└═══════════════════════┬═════════════════════════════════════┘
▼
┌─────────────────────────────────────────────────────────────┐
│ Pass C · 只对候选书访问详情页,补库存 │
│ 16 次页面加载(每次要等 800ms 异步) │
│ 成本:昂贵,所以才要先筛 │
└───────────────────────┬─────────────────────────────────────┘
▼
┌─────────────────────────────────────────────────────────────┐
│ 终选:按库存过滤 → 按 (折扣↓, 评分↓) 排序 │
│ 16 本 ──────▶ 13 本(剔除 3 本零库存) │
└───────────────────────┬─────────────────────────────────────┘
▼
报告 / CSV / JSON请务必记住这个思想,它比任何 API 都值钱:
抓取是有成本的。先用便宜的字段过滤,再对幸存者花贵的代价。
这个原则在爬虫、测试、ETL、数据库查询优化里,是同一件事。
2.7.3 代码实现(分块讲解)
块 1:数据清洗工具
PRICE_RE = re.compile(r"[\d.]+") # 匹配数字和小数点
RATING_RE = re.compile(r"评分\s*([\d.]+)\s*·\s*(\d+)\s*条评价") # 两个捕获组
def money(text: str) -> float:
"""'¥139.00' → 139.0;解析不出来返回 0.0(防止一个脏数据搞崩整条流水线)。"""
m = PRICE_RE.search(text or "")
# ↑ text or "" 处理 None 的情况:
# 如果 text 是 None,会变成 "",search 不会崩
return float(m.group()) if m else 0.0
# ↑ ★ 关键设计:解析失败返回 0,不抛异常
# 为什么?因为一本书的价格解析失败,不应该导致另外 23 本也抓不到。
# 代价是可能静默产生 0 值 —— 生产环境应该在这里记一条 warning 日志。RATING_RE 匹配的是 "评分 4.9 · 921 条评价" 这样的文本:
评分\s*([\d.]+)\s*·\s*(\d+)\s*条评价
↑ 组1: 4.9 ↑ 组2: 921
\s* 表示"任意个空白字符",用来容忍空格数量的变化 —— 这是防御性写法块 2:解析一页的卡片
def parse_cards(page: Page) -> list[Book]:
out: list[Book] = []
cards = page.get_by_test_id("book-card") # 一条 Locator 规则,匹配本页 6 张卡片
for i in range(cards.count()):
c = cards.nth(i) # nth(i) 取第 i 个(从 0 开始)
# ★ 注意:c 仍然是 Locator,不是元素快照。
# 每次对 c 调用方法,都会重新去页面上查一次(惰性求值,见 1.3.2)
# ── 拆分 "Randal E. Bryant · 机械工业出版社" ──────────
author_line = c.get_by_test_id("book-author").inner_text()
author, _, publisher = author_line.partition(" · ")
# partition 返回三元组 (前, 分隔符, 后),中间那个用 _ 丢弃。
# ★ 为什么用 partition 不用 split?
# 如果字符串里没有分隔符,split(" · ") 返回 1 个元素 → 解包报错
# partition 永远返回 3 个元素 → 不会崩(publisher 会是空串)
# 这是"宁可数据缺失,不要程序崩溃"的防御性编程
# ── 用正则提取评分和评价数 ─────────────────────────────
rating, reviews = 0.0, 0 # 先给默认值
m = RATING_RE.search(c.get_by_test_id("book-rating").inner_text())
if m:
rating, reviews = float(m.group(1)), int(m.group(2))
# ↑ 组1 评分 ↑ 组2 评价数
# ── ★ 防御:检查登录态是否还有效 ──────────────────────
member = c.get_by_test_id("member-price")
if member.count() == 0:
# 会员价的 testid 一个都找不到 → 说明页面渲染的是 member-price-locked
# → 登录态失效了。这时候必须响亮地失败,而不是继续抓 0 元的假数据。
raise RuntimeError("会员价被锁定 —— 登录态失效,请删除 output/auth_state.json 重跑")
out.append(Book(
id=int(c.get_attribute("data-book-id")), # 卡片上的 data-book-id 属性
title=c.get_by_test_id("book-link").inner_text().strip(),
author=author.strip(),
publisher=publisher.strip(),
list_price=money(c.get_by_test_id("list-price").inner_text()),
member_price=money(member.inner_text()),
rating=rating,
reviews=reviews,
source="playwright", # 标记数据来源,方便后面对比
))
return out块 3:分页扫描
def scan_list(page: Page, category: str = "") -> list[Book]:
"""分页扫描列表页。终止条件由页面自身的'下一页'按钮决定,并带硬上限。"""
# 拼 URL。category 为空就不加参数 —— 一个函数同时支持"全部"和"按分类"两种扫描
url = config.PRODUCTS_URL + (f"?category={category}" if category else "")
page.goto(url, wait_until="domcontentloaded")
dismiss_consent(page) # 幂等关浮层,见 2.5.3
books: list[Book] = []
guard = 0 # 死循环保护计数器
while True:
guard += 1
if guard > 20:
raise RuntimeError("翻页超过 20 次,疑似死循环,已中止")
# ★ 抛异常而不是 break:
# break 会静默地返回不完整的数据,你根本不知道出事了。
# 抛异常会让整个任务失败并告警 —— 这才是你想要的。
# 等网格容器可见,确保这一页真的渲染出来了再去解析
page.get_by_test_id("book-grid").wait_for(state="visible")
books.extend(parse_cards(page)) # extend 是"追加一个列表的所有元素"
nxt = page.get_by_test_id("page-next")
if nxt.count() == 0: # 没有"下一页"按钮 = 到末页了
break
nxt.click()
page.wait_for_load_state("domcontentloaded")
return books块 4:详情页补库存
def fetch_stock(page: Page, book_id: int) -> tuple[int, str]:
page.goto(f"{config.BASE_URL}/book/{book_id}", wait_until="domcontentloaded")
dismiss_consent(page)
# ★★ 整个 Step 4 最关键的一行 ★★
# 等 JS 把 data-loaded 置为 true,而不是 sleep。
# wait_for_selector 返回的是 ElementHandle(元素句柄),可以直接读属性。
el = page.wait_for_selector('[data-testid="stock"][data-loaded="true"]', timeout=8000)
stock = int(el.get_attribute("data-stock") or -1)
# ↑ or -1 处理属性不存在的情况:
# get_attribute 返回 None 时,
# None or -1 → -1(表示"未知")
# ★ 注意这里读的是 data-stock="32"(纯数字),
# 而不是去解析 "有货,剩余 32 件" 这段文本。
# 永远优先读机器可读的属性 —— 文案会改,属性不会。
category = page.get_by_test_id("detail-category").inner_text().strip()
return stock, category # 顺便把分类也带回去(Pass B 的补充)块 5:主流程编排
def run() -> RunResult:
t0 = time.perf_counter() # perf_counter 是高精度计时器,比 time.time() 准
res = RunResult(engine="playwright", started_at=now())
with sync_playwright() as p:
ensure_auth(p) # 没有登录态就先登录(见 2.6.5)
browser = p.chromium.launch(headless=config.HEADLESS, slow_mo=config.SLOW_MO)
# ★ 一行搞定登录态:context 创建时就带上 cookie
ctx = browser.new_context(
storage_state=str(config.STATE_FILE),
viewport={"width": 1280, "height": 900},
locale="zh-CN",
)
ctx.set_default_timeout(config.DEFAULT_TIMEOUT) # context 级超时,对所有 page 生效
page = ctx.new_page()
# ── Pass A ─────────────────────────────────────────────
print("\n[Pass A] 分页扫描全部图书…")
books = scan_list(page)
by_id = {b.id: b for b in books} # 建 id → Book 的索引字典
# ★ 为什么建字典?Pass B 要按 id 回填 category。
# 如果用列表 + 每次遍历查找,是 O(n²);用字典是 O(1)。
# 而且 —— 关键 —— 字典的值是**同一个对象的引用**,
# 改 by_id[3].category 就等于改了 books 里那本书。不需要写回。
print(f" 采集到 {len(books)} 本书({len(set(b.id for b in books))} 个唯一 ID)")
# ↑ set 去重,验证有没有重复抓
# ── Pass B ─────────────────────────────────────────────
print("\n[Pass B] 用分类筛选器补全 category 字段…")
cats = page.get_by_test_id("category-filter").locator("option")
# locator("option") 是在筛选器内部再找 <option> 子元素 —— 定位器可以链式收窄
cat_names = [cats.nth(i).get_attribute("value") for i in range(cats.count())]
cat_names = [c for c in cat_names if c] # 过滤掉"全部分类"那个空 value
for c in cat_names:
sub = scan_list(page, category=c) # 复用同一个扫描函数
for b in sub:
if b.id in by_id:
by_id[b.id].category = c # ★ 回填,改的是 books 里的对象
print(f" {c}:{len(sub)} 本")
# ── ★★ 关键决策:用便宜字段先筛 ★★ ────────────────────
candidates = [b for b in books
if b.discount_pct >= config.MIN_DISCOUNT_PCT
and b.rating >= config.MIN_RATING]
# discount_pct 是 Book 的 @property,访问时才计算,不占存储:
# (定价 - 会员价) / 定价 × 100,保留 1 位小数
print(f"\n[筛选] 折扣≥{config.MIN_DISCOUNT_PCT}% 且 评分≥{config.MIN_RATING}"
f" → {len(candidates)}/{len(books)} 本进入候选")
# 把优化效果写进报告备注 —— 让"我做了优化"变成可量化的事实
res.notes.append(
f"列表页预筛把详情页访问量从 {len(books)} 次降到 {len(candidates)} 次,"
f"节省 {round((1 - len(candidates) / len(books)) * 100)}% 的页面加载"
)
# ── Pass C ─────────────────────────────────────────────
print("\n[Pass C] 逐个访问候选书详情页,补库存…")
for i, b in enumerate(candidates, 1): # enumerate(x, 1) 从 1 开始编号
b.stock, cat = fetch_stock(page, b.id)
if cat:
b.category = cat # 详情页的分类更权威,覆盖 Pass B
flag = "有货" if b.in_stock else "缺货" # in_stock 是 @property:stock > 0
print(f" ({i:>2}/{len(candidates)}) #{b.id:<2} {b.title:<24} "
f"折扣 {b.discount_pct:>4}% 评分 {b.rating} {flag}")
# f-string 的对齐语法:{x:>2} 右对齐宽2,{x:<24} 左对齐宽24
# 让终端输出成为整齐的表格,这是很实用的小技巧
# ── 终选 ───────────────────────────────────────────────
picks = [b for b in candidates if (b.in_stock or not config.REQUIRE_IN_STOCK)]
# ↑ 要么有货,要么规则说不要求有货
picks.sort(key=lambda b: (-b.discount_pct, -b.rating))
# 元组排序 = 多级排序:先按折扣降序,折扣相同再按评分降序
# 加负号是把"升序"变"降序"的常用技巧(数值型字段才能这么用)
dropped = [b for b in candidates if b not in picks]
if dropped:
res.notes.append("因缺货被剔除:" + "、".join(f"{b.title}" for b in dropped))
# ★ 记录"被剔除了什么",这是可解释性的关键。
# 报告读者能看到"哦,原来有 3 本很划算但没货",而不是莫名其妙少了几本。
page.goto(config.PRODUCTS_URL, wait_until="domcontentloaded")
dismiss_consent(page)
page.screenshot(path=str(config.SHOTS_DIR / "step4_final.png"), full_page=True)
ctx.close()
browser.close()
res.books, res.picks = books, picks
res.finished_at, res.elapsed_sec = now(), time.perf_counter() - t0
return res块 6:数据模型(aigui/models.py)
@dataclass
class Book:
id: int
title: str
author: str
publisher: str = "" # 有默认值的字段必须放在无默认值字段之后
category: str = ""
list_price: float = 0.0
member_price: float = 0.0
rating: float = 0.0
reviews: int = 0
stock: int = -1 # ★ -1 表示"未采集",0 表示"确认缺货" —— 两者语义不同!
source: str = "playwright" # playwright | browser-use | hybrid
# @property 让计算字段可以像普通属性一样访问:b.discount_pct
# 好处:① 不占存储 ② 永远和源数据一致(不会出现"改了价格忘了改折扣")
@property
def discount_pct(self) -> float:
if not self.list_price: # 防除零
return 0.0
return round((self.list_price - self.member_price) / self.list_price * 100, 1)
@property
def in_stock(self) -> bool:
return self.stock > 0 # -1(未知)和 0(缺货)都算 False
@property
def saved(self) -> float:
return round(self.list_price - self.member_price, 2)为什么用 dataclass 而不是 dict? 三个理由,写在源文件的 docstring 里:
- 字段拼错会立刻报错。
b.titel会抛AttributeError;d["titel"]只会给你KeyError或者更糟——d.get("titel")静默返回None。- 后面能平滑换成 pydantic。Step 5 的
PickedBook就是 pydantic 版本,字段几乎一致。- 报告层只依赖模型,不关心数据来自哪。这是 Step 4/5/6 能输出同格式报告的根本原因。
2.7.4 运行与真实输出
python steps/step4_playwright_scraper.py[Pass A] 分页扫描全部图书…
采集到 24 本书(24 个唯一 ID)
[Pass B] 用分类筛选器补全 category 字段…
计算机:6 本
历史:6 本
文学:6 本
商业:6 本
[筛选] 折扣≥15.0% 且 评分≥4.5 → 16/24 本进入候选
[Pass C] 逐个访问候选书详情页,补库存…
( 1/16) #1 深入理解计算机系统 折扣 28.8% 评分 4.9 有货
( 2/16) #2 算法导论(第4版) 折扣 16.0% 评分 4.8 有货
( 3/16) #3 Python编程:从入门到实践 折扣 36.7% 评分 4.7 有货
( 4/16) #4 流畅的Python(第2版) 折扣 15.5% 评分 4.8 缺货
( 5/16) #5 重构:改善既有代码的设计 折扣 16.3% 评分 4.6 有货
( 6/16) #7 人类简史 折扣 42.6% 评分 4.7 有货
( 7/16) #8 万历十五年 折扣 15.6% 评分 4.8 有货
( 8/16) #9 枪炮、病菌与钢铁 折扣 37.2% 评分 4.6 缺货
( 9/16) #11 叫魂:1768年妖术大恐慌 折扣 33.9% 评分 4.9 有货
(10/16) #13 百年孤独 折扣 40.0% 评分 4.9 有货
(11/16) #14 活着 折扣 37.8% 评分 4.9 有货
(12/16) #18 刀锋 折扣 15.4% 评分 4.7 有货
(13/16) #19 思考,快与慢 折扣 39.1% 评分 4.6 有货
(14/16) #20 原则 折扣 40.8% 评分 4.5 缺货
(15/16) #22 穷查理宝典 折扣 40.6% 评分 4.8 有货
(16/16) #24 纳瓦尔宝典 折扣 40.7% 评分 4.7 有货
══════════════════════════════════════════════════════════════
命中 13 本,合计可省 ¥338.00
耗时 15.74 秒 · LLM 调用 0 次 · 成本 ¥0
══════════════════════════════════════════════════════════════
TOP 5:
1. 人类简史 ¥ 68.00 → ¥ 39.00 (省 42.6%,评分 4.7)
2. 纳瓦尔宝典 ¥ 59.00 → ¥ 35.00 (省 40.7%,评分 4.7)
3. 穷查理宝典 ¥128.00 → ¥ 76.00 (省 40.6%,评分 4.8)
4. 百年孤独 ¥ 55.00 → ¥ 33.00 (省 40.0%,评分 4.9)
5. 思考,快与慢 ¥ 69.00 → ¥ 42.00 (省 39.1%,评分 4.6)
产物:
…\output\books_playwright.csv
…\output\result_playwright.json
…\output\report_playwright.md
✅ Step 4 完成。
(说明:耗时会因机器性能在 14~16 秒之间小幅波动,其余数字完全一致。)2.7.5 结果解读:这些数字在说什么
24 本 ──筛选(折扣≥15% & 评分≥4.5)──▶ 16 本 ──库存过滤──▶ 13 本
被剔除的 8 本:折扣或评分不达标(如"从0到1" 折扣仅 4.4%)
被剔除的 3 本:★ 折扣评分都很好,但库存 = 0
· 流畅的Python(第2版) 15.5% 4.8 库存 0
· 枪炮、病菌与钢铁 37.2% 4.6 库存 0
· 原则 40.8% 4.5 库存 0那 3 本零库存的书,就是陷阱 4 的验收标准。 如果你没有正确处理异步加载,它们会混进最终清单——报告会变成 16 本而不是 13 本,而你还以为自己成功了。
生成的报告 output/report_playwright.md 摘要:
# BookNest 会员捡漏清单
- **采集引擎**:`playwright`
- **总耗时**:14.90 秒
- **AI 决策步数**:0(确定性脚本,无 LLM 调用)
- **扫描图书**:24 本 → **命中**:13 本
## 汇总
- 命中图书合计可省:**¥338.00**
- 平均折扣力度:**31.0%**
## 执行备注
- 列表页预筛把详情页访问量从 24 次降到 16 次,节省 33% 的页面加载
- 因缺货被剔除:流畅的Python(第2版)、枪炮、病菌与钢铁、原则「执行备注」这一节是我特别想强调的设计。它让报告不只是"结果",还是"过程的解释"。读报告的人能知道:为什么只有 13 本、少的那些去哪了、脚本做了什么优化。可解释性不是 AI 的专利,确定性脚本同样需要。
2.7.6 这一步的可复用套路
抛开图书这个具体业务,这一步的骨架可以套用到任何"列表 + 详情"的采集场景:
1. 建数据模型(dataclass) → 结构化,而不是到处传 dict
2. Pass A:扫列表,拿便宜字段 → 一次加载拿多条
3. 用便宜字段先筛 → ★ 决定整体性能的一步
4. Pass C:只对候选查详情 → 把贵的操作降到最少
5. 终选 + 排序 → 业务规则集中在一处
6. 出多格式产物 + 执行备注 → 可交付、可解释这个骨架同样适用于:电商比价、招聘信息采集、房源筛选、竞品监控、内部系统巡检……换掉选择器和业务规则就能用。
2.8 Step 5 · Browser Use 方案(AI 自主完成)
2.8.1 为什么要做这一步
前面的 Step 1–4,你一直在「教机器做具体动作」:点哪个选择器、翻几页、读哪个字段。这是「确定性」路线——快、稳、零成本,但它有一个前提:你得提前知道页面长什么样。一旦前端改版、元素换名,脚本就挂。
Browser Use 代表的是另一条路线:你只描述目标和约束,剩下的「看页面—思考—操作」全交给 LLM。这一步的目的,就是让你亲手对比两条路线:
- 代码量少了多少(选择器几乎全部消失);
- 耗时代价多大(每一步都要等模型推理);
- 结果稳不稳(多跑几次,数字会不会漂)。
2.8.2 打算怎么做
我们不重写业务逻辑,只把 Step 4 那套「选择器 + 预筛 + 详情页」的指令,翻译成一段自然语言任务(TASK),再交给 browser_use.Agent 自主执行。三个关键设计:
- 用 Pydantic 定义
PickResult,逼模型输出结构化数据而不是散文; - 用
output_model_schema=PickResult把「结构化契约」接进 Agent; - 关掉视觉(
use_vision=False)——本地靶场结构清晰,编号地图已经够用,开视觉纯属浪费 token(详见 1.2.4)。
2.8.3 流程图
┌──────────────────────────────────────────────────────────┐
│ 你写的:TASK(自然语言任务) + PickResult(输出契约) │
└───────────────────────────┬──────────────────────────────┘
▼
┌──────────────────┐
│ browser_use.Agent │
└────────┬───────────┘
┌──────────────────┼───────────────────────┐
▼ ▼ ▼
① 感知(看页面) ② 思考(LLM) ③ 操作(点/填/翻)
提纯 DOM 成编号 决定下一步动作 通过 CDP 执行
│ │ │
└──────────────────┴────────────────────────┘
▼
max_steps 内循环,直到 done
▼
history.final_result() → PickResult JSON
▼
排序 → 报告 / CSV / JSON图 2-7 · Browser Use 数据流:你只写任务,AI 自己完成「看—想—做」闭环。
2.8.4 代码实现(分块 + 逐行)
代码块 1/3:结构化输出契约
# step5_browseruse_agent.py 片段 1/3 —— 让 AI 输出"可用数据"而非散文
from pydantic import BaseModel, Field
class PickedBook(BaseModel):
title: str = Field(description="图书完整书名")
author: str = Field(description="作者")
category: str = Field(description="分类,如 计算机/历史/文学/商业")
list_price: float = Field(description="定价,纯数字,不带货币符号")
member_price: float = Field(description="会员价,纯数字,不带货币符号")
rating: float = Field(description="评分,0-5 的小数")
stock: int = Field(default=-1, description="库存件数;缺货填 0;未知填 -1")
class PickResult(BaseModel):
picks: list[PickedBook] = Field(description="所有满足筛选条件的图书")
total_scanned: int = Field(description="总共浏览过多少本书")
summary: str = Field(description="一句话总结")- 逐块逻辑:
PickedBook是最终想要的单本书数据结构;PickResult是 Agent 的「最终答卷」格式。Field(description=...)里的文字会被原样塞进 prompt——写清楚 description,等于在给模型下指令。 逐行解释:
- 第 1 行:从 pydantic 导入
BaseModel(模型基类)和Field(给字段加元数据/约束)。 class PickedBook(BaseModel):定义单本书字段;每个字段的Field(description=...)告诉模型该填什么。stock: int = Field(default=-1, ...):用-1表示「未知」、0表示「缺货」,明确区分两种状态,避免后续筛选出错。class PickResult(BaseModel):是顶层容器;picks是命中的书列表,total_scanned让模型自报「扫了多少本」(用于事后核对),summary给一句人话总结。
- 第 1 行:从 pydantic 导入
代码块 2/3:LLM 工厂 + 任务描述
# step5_browseruse_agent.py 片段 2/3 —— LLM 工厂与任务描述
def build_llm():
from browser_use import (ChatOpenAI, ChatAnthropic, ChatGoogle,
ChatBrowserUse, ChatOllama, ChatDeepSeek)
provider = config.LLM_PROVIDER.lower()
model = config.LLM_MODEL
if provider == "openai":
_require("OPENAI_API_KEY")
kw = {"model": model}
if config.LLM_BASE_URL: # 走第三方 OpenAI 兼容网关时填这个
kw["base_url"] = config.LLM_BASE_URL
return ChatOpenAI(**kw)
# …anthropic / google / deepseek / browseruse / ollama 分支同理…
raise ValueError(f"未知 LLM_PROVIDER:{provider}")
TASK = f"""
你是一名图书选品助手。请在本地网站 {config.BASE_URL} 上完成以下任务。
【操作步骤】
1. 打开 {config.BASE_URL}/products。出现 Cookie 浮层先点"同意并继续"关掉它。
2. 用 {config.USERNAME} / {config.PASSWORD} 登录,列表页才会显示"会员价"。
3. 逐页浏览全部图书,按页面底部"第 X / Y 页,共 N 本"确认翻完所有页。
【筛选条件】同时满足:折扣 ≥ {config.MIN_DISCOUNT_PCT}%、评分 ≥ {config.MIN_RATING}、有货(库存>0)。
【关于库存】列表页不显示,必须进详情页;详情页库存是异步加载的,等变成"有货,剩余 N 件"再读取。
为省时间,只对已满足"折扣+评分"的书才进详情页。
【输出】按结构化格式返回所有命中图书,并说明扫描多少本、命中多少本。
""".strip()- 逐块逻辑:
build_llm()是一个「一次配置、随处切换」的工厂——读config里的 provider 和 model,返回对应厂商的 Chat 客户端;缺 Key 时_require()打印清晰提示并退出。TASK是这一步真正的「源代码」:所有业务规则都写在这里,没有一行选择器。 逐行解释(关键点):
from browser_use import (...):browser-use 把各厂商的模型客户端统一封装,名字都以Chat开头。if config.LLM_BASE_URL: kw["base_url"] = ...:走第三方 OpenAI 兼容网关(本地代理、国内中转)时填这个,兼容不同部署。TASK = f"""...""":用 f-string 把config里的真实地址、账号、阈值填进 prompt,模型拿到的是一份「带具体参数的任务书」,而不是泛泛而谈。- 把异步库存写进 prompt:这是吸取 Step 2 陷阱 4 的教训——明确告诉模型「等变成『有货,剩余 N 件』再读」,否则它会读到「加载中…」导致库存判断错误(呼应 1.6 的等待策略)。
代码块 3/3:主流程 run() 与 main()
# step5_browseruse_agent.py 片段 3/3 —— 主流程
async def run() -> RunResult:
from browser_use import Browser
t0 = time.perf_counter()
res = RunResult(engine=f"browser-use ({config.LLM_PROVIDER}/{config.LLM_MODEL})",
started_at=now())
llm = build_llm()
browser = Browser(headless=config.HEADLESS, window_size={"width": 1280, "height": 900})
agent = build_agent(
task=TASK, llm=llm, browser=browser, tools=build_tools(),
output_model_schema=PickResult, use_vision=False,
)
print("🤖 Agent 开始自主执行…\n")
history = await agent.run(max_steps=config.MAX_AGENT_STEPS)
try:
res.steps = history.number_of_steps()
except Exception:
res.steps = len(getattr(history, "history", []) or [])
raw = history.final_result()
if not raw:
res.notes.append("Agent 未返回结构化结果,可能是步数用尽或任务失败")
res.finished_at, res.elapsed_sec = now(), time.perf_counter() - t0
return res
parsed = PickResult.model_validate_json(raw) # 校验 → 强类型对象
books = [Book(id=i, title=p.title, author=p.author, category=p.category,
list_price=p.list_price, member_price=p.member_price,
rating=p.rating, stock=p.stock, source="browser-use")
for i, p in enumerate(parsed.picks, 1)]
books.sort(key=lambda b: (-b.discount_pct, -b.rating))
res.books, res.picks = books, books
res.notes.append(f"Agent 自述:{parsed.summary}")
res.finished_at, res.elapsed_sec = now(), time.perf_counter() - t0
return res- 逐块逻辑:
run()把「建 LLM → 建 Agent(挂上任务/契约/工具)→ 跑循环 → 取结构化结果 → 转成 Book 列表 → 排序出报告」串起来。对比 Step 4 的run(),这里一个page.get_by_test_id都没有——选择器全部交给 Agent 内部。 逐行解释(关键点):
Browser(headless=..., window_size=...):browser-use 的Browser封装了底层驱动(0.13.x 自带cdp-use),window_size给页面一个稳定视口。output_model_schema=PickResult:把结构化契约接进 Agent——这是让history.final_result()直接吐出PickResultJSON 的关键,没有它你就只能拿到一段散文。use_vision=False:关掉截图理解,省一半 token(详见 1.2.4)。await agent.run(max_steps=...):异步执行;max_steps是安全闸,防止 Agent 陷入死循环把 API 额度烧光(成本熔断器,呼应 1.7 成本模型)。history.number_of_steps():取决策步数;用try/except兼容不同版本的方法名。PickResult.model_validate_json(raw):把模型输出的 JSON 校验并反序列化成强类型对象——校验失败会抛异常,等于给 AI 的输出加了道关卡。
2.8.5 真实输出(运行验证)
说明:本步需要 LLM API Key 才能真实跑通。我已对脚本做编译校验(语法/导入/inspect参数过滤逻辑均与 browser-use 0.13.x 对齐),并确认TASK与PickResult契约自洽。在你填入 Key 后,运行:
cd AIforGUIcode
python steps/step5_browseruse_agent.py预期输出结构(数字取决于所选模型,会随推理略有浮动):
🤖 Agent 开始自主执行…(每一步都会调用 LLM,请耐心等待)
════════════════════════════════════════════════════════════════
AI 方案命中 N 本 · 决策步数 M · 耗时 T 秒
════════════════════════════════════════════════════════════════
1. 人类简史 ¥00.00 → ¥00.00(省 00%)
…(TOP 5)
报告:…\output\report_browseruse.md
👉 现在回头对比 output/report_playwright.md,看两者结果是否一致。重点不是「它跑出多少本」,而是三件事,请你对照 Step 4 的精确结果去体会:
- 代码里一个选择器都没有了;
- 决策步数
M通常远大于 Step 4 的0; - 多跑几次,命中数量可能与 Step 4 的精确 13 本出现 1–2 本的偏差——这就是「AI 灵活性」的代价(呼应 1.8 Agent Loop 的稳定性边界)。
2.8.6 小结
- Browser Use = 把「看页面、思考、操作」三件事交给 LLM,你只写任务。
output_model_schema是它和确定性方案连接的桥梁:让 AI 输出结构化数据。- 它最擅长「页面结构未知 / 规则模糊 / 需要临场判断」的场景;最不擅长「每次都一样、要 100% 复现」的场景。
2.9 Step 6 · 混合架构(生产环境推荐形态)
2.9.1 为什么要混合
前面两条路都有短板:
- 纯脚本(Step 4):稳、快、免费,但前端一改版就挂——脆弱点在「定位」。
- 纯 AI(Step 5):抗改版、省人工,但每一步都烧 token、结果会漂——脆弱点在「成本与稳定性」。
真实项目里,90% 的操作是每次都一样的(登录、翻页、读字段),只有 2–3 处是会变、需要判断的(「用户今天想要什么样的书?」「页面元素改名叫啥了?」)。混合架构的核心思想:确定性骨架干 90% 的脏活,LLM 只在两处模糊接缝上出手。
2.9.2 打算怎么做
┌─────────────── 确定性骨架(Playwright,快/稳/零成本)──────────────┐
│ 登录 · 翻页 · 扫描列表 · 读结构化字段 · 断言 │
└───────────────────────────────┬──────────────────────────────────┘
│ 只在这两处引入 LLM
┌────────────────────┴─────────────────────┐
▼ 接缝 A:自然语言 → 结构化规则 ▼ 接缝 B:定位器自愈
"打折狠、口碑好、能发货" L1 data-testid → L2 role+name
│ → L3 文本 → L4 AI 接管
▼
{折扣≥25, 评分≥4.5, 有货}图 2-8 · 混合架构:骨架是 Playwright,LLM 只在两个模糊接缝上介入。
关键设计:这个脚本没有 LLM Key 也能完整跑通——接缝 A 退化为本地关键词规则引擎,接缝 B 退化到 L3(并真实演示 L1 失效 → L2/L3 救回)。有 Key 时自动升级到 LLM/Chat 方案。
图 2-9 · 定位器自愈链条:每级失效就自动降级到下一级,并记录命中统计。
L1 data-testid(最稳,建议前端专门加)
│ 失效
▼
L2 get_by_role(role, name)(语义角色,改版常还在)
│ 失效
▼
L3 get_by_text(可见文字)(最脆,改文案就挂)
│ 失效 + ai_enabled
▼
L4 Browser Use Agent 接管(两套技术真正合流点)
│ 全失效
▼
FAIL(放弃该步,写入"技术债报告")2.9.3 流程图
用户需求(自然语言)
│
┌─────────▼──────────┐
│ 接缝 A:resolve_rules │
│ 有 Key→LLM 解析 │
│ 无 Key→关键词规则 │
└─────────┬──────────┘
▼
{min_discount, min_rating, require_stock}
│
┌─────────▼──────────┐
│ 确定性骨架 │ Playwright 扫描/预筛/补库存
│ (复用 step4 函数) │ ← 唯一的"AI 决策"在预筛阈值
└─────────┬──────────┘
│
┌─────────▼──────────┐
│ 接缝 B:定位自愈 │ 每定位一个元素按 L1→L4 降级
│ 统计各级命中 │
└─────────┬──────────┘
▼
排序 → 报告 / CSV / JSON图 2-10 · 混合流程:只有接缝 A、B 两个点可能调用 LLM,其余全是确定性代码。
2.9.4 代码实现(分块 + 逐行)
代码块 1/4:接缝 A — 自然语言需求 → 结构化规则
# step6_hybrid_pipeline.py 片段 1/4 —— 接缝 A:需求解析
def rules_from_llm(need: str) -> dict | None:
"""有 Key 就让 LLM 解析;任何异常都返回 None,交给规则引擎兜底。"""
key = os.getenv("OPENAI_API_KEY")
if not key:
return None
try:
from openai import OpenAI
client = OpenAI(api_key=key, base_url=config.LLM_BASE_URL or None)
rsp = client.chat.completions.create(
model=config.LLM_MODEL, temperature=0,
response_format={"type": "json_object"},
messages=[
{"role": "system", "content": "你把中文购物需求翻译成结构化筛选规则。" + RULE_SCHEMA_HINT},
{"role": "user", "content": need},
],
)
data = json.loads(rsp.choices[0].message.content)
return {"min_discount": float(data["min_discount"]),
"min_rating": float(data["min_rating"]),
"require_stock": bool(data["require_stock"])}
except Exception as e: # 降级,不让 AI 故障拖垮主流程
print(f" ⚠ LLM 解析失败({type(e).__name__}),回退到本地规则引擎")
return None
def rules_from_keywords(need: str) -> dict:
"""无 Key 时的确定性兜底:关键词 → 规则。够用、免费、可预测。"""
disc = config.MIN_DISCOUNT_PCT
if any(k in need for k in ("特别狠", "非常大", "很狠", "骨折", "五折")):
disc = 35.0
elif any(k in need for k in ("力度大", "打折狠", "便宜")):
disc = 25.0
rating = config.MIN_RATING
if any(k in need for k in ("口碑好", "评价好", "高分", "好评")):
rating = 4.5
if "神作" in need or "封神" in need:
rating = 4.8
stock = any(k in need for k in ("现货", "发货", "有货", "马上", "现在就")) \
or config.REQUIRE_IN_STOCK
return {"min_discount": disc, "min_rating": rating, "require_stock": stock}
def resolve_rules(need: str) -> tuple[dict, str]:
r = rules_from_llm(need)
if r:
return r, "LLM"
return rules_from_keywords(need), "本地规则引擎"- 逐块逻辑:
rules_from_llm把自然语言需求丢给 LLM,要求它只回 JSON(response_format=json_object+temperature=0保证稳定);rules_from_keywords是纯本地兜底;resolve_rules是统一入口——有 Key 用 LLM,没 Key 用关键词,永不阻塞主流程。 逐行解释(关键点):
if not key: return None:没有 Key 直接走兜底,零报错。temperature=0:让模型输出确定,不要「创意」——解析筛选规则这种事不需要发散。response_format={"type":"json_object"}:强制模型只输出 JSON,省去自己写正则抠字段。except Exception as e:整段兜底:LLM 超时/限流/返回脏数据都不会拖垮脚本,优雅降级。rules_from_keywords里("特别狠","非常大",...):关键词映射到具体阈值——「特别狠」→35%、「力度大」→25%,把产品语言翻译成机器规则。
代码块 2/4:接缝 B — 四级降级定位器
# step6_hybrid_pipeline.py 片段 2/4 —— 接缝 B:定位器自愈
class ResilientLocator:
def __init__(self, page: Page, ai_enabled: bool = False):
self.page = page
self.ai_enabled = ai_enabled
self.stats = {"L1": 0, "L2": 0, "L3": 0, "L4": 0, "FAIL": 0}
self.heals = []
def find(self, *, testid, role="", name="", text="", desc="") -> Locator | None:
loc = self.page.get_by_test_id(testid) # L1:最稳的属性
if loc.count():
self.stats["L1"] += 1
return loc.first
if role and name: # L2:语义角色 + 无障碍名
loc = self.page.get_by_role(role, name=name)
if loc.count():
self.stats["L2"] += 1
self.heals.append(f"[{desc}] testid='{testid}' 失效 → 由 role={role} 救回")
return loc.first
if text: # L3:可见文本兜底
loc = self.page.get_by_text(text, exact=False)
if loc.count():
self.stats["L3"] += 1
self.heals.append(f"[{desc}] testid='{testid}' 失效 → 由文本 '{text}' 救回")
return loc.first
if self.ai_enabled: # L4:交给 AI Agent 接管
self.stats["L4"] += 1
self.heals.append(f"[{desc}] 前三级全部失效 → 移交 Browser Use Agent 处理")
return None
self.stats["FAIL"] += 1
self.heals.append(f"[{desc}] 四级全部失效,该步骤放弃")
return None- 逐块逻辑:
ResilientLocator.find按L1 → L2 → L3 → L4逐级尝试定位一个元素,哪一级命中就在stats里记一笔,并写下「怎么救回来的」。这套统计本身就是一份「技术债报告」:L2/L3 命中越多,说明你的data-testid契约烂得越厉害(呼应 1.10)。 逐行解释(关键点):
self.page.get_by_test_id(testid):L1 用data-testid——最稳的属性,建议前端专门加。if loc.count()::Playwright 的count()不会抛异常,比is_visible()更适合「先探测再操作」。get_by_role(role, name=name):L2 用无障碍语义——前端改版常丢testid,但按钮的「角色 + 名字」往往还在。get_by_text(text, exact=False):L3 用可见文字,最脆(改文案就挂),只作最后兜底。if self.ai_enabled:L4:有 Key 时把这一步交给 Browser Use Agent 接管——这里就是两套技术真正合流的点。
代码块 3/4:自愈演示(故意制造失效)
# step6_hybrid_pipeline.py 片段 3/4 —— 故意让 testid 失效,演示自愈链条
def demo_self_healing(page: Page, ai_enabled: bool) -> ResilientLocator:
print("\n[环节 B] 定位器自愈演示")
pw4.dismiss_consent(page)
rl = ResilientLocator(page, ai_enabled)
ok = rl.find(testid="search-input", role="textbox", name="搜索书名或作者", desc="搜索框")
if ok:
print(" L1 命中:data-testid='search-input' 正常工作")
healed = rl.find(testid="nav-products-v2-renamed", role="link", name="全部图书",
desc="导航-全部图书(模拟 testid 被改名)")
if healed:
healed.click()
page.wait_for_load_state("domcontentloaded")
print(" L1 未命中 → L2 用 get_by_role('link', name='全部图书') 成功救回 ✓")
dead = rl.find(testid="totally-gone", text="共 24 本", desc="页面统计文案")
if dead:
print(" L1/L2 未命中 → L3 文本定位成功救回 ✓")
print(f" 自愈统计:{rl.stats}")
return rl- 逐块逻辑:这个函数故意用三个「坏掉/正常」的
testid去定位,让你亲眼看到:search-input(正常,L1 命中)→nav-products-v2-renamed(模拟前端改名,L1 失效 →L2 救回)→totally-gone(彻底没了,L1/L2 失效 →L3 文本救回)。 逐行解释(关键点):
pw4.dismiss_consent(page):先把 Cookie 浮层关掉,免得它挡住演示(浮层不占自愈名额)。- 三个
rl.find(...)调用分别覆盖「正常 / 改名 / 消失」三种真实改版情况。 print(f" 自愈统计:{rl.stats}"):把{'L1':1,'L2':1,'L3':1,...}打出来,就是这一步的「技术债体检单」。
代码块 4/4:主流程 run() 与 main()
# step6_hybrid_pipeline.py 片段 4/4 —— 主流程
def run(need: str) -> tuple[RunResult, dict, str, ResilientLocator]:
t0 = time.perf_counter()
print(f"\n[环节 A] 解析自然语言需求:{need!r}")
rules, how = resolve_rules(need)
print(f" 解析方式:{how}")
print(f" → 折扣 ≥ {rules['min_discount']}% · 评分 ≥ {rules['min_rating']}"
f" · {'必须有货' if rules['require_stock'] else '库存不限'}")
ai_enabled = bool(os.getenv("OPENAI_API_KEY") or os.getenv("BROWSER_USE_API_KEY"))
res = RunResult(engine=f"hybrid(骨架=Playwright,规则解析={how})", started_at=now())
with sync_playwright() as p:
pw4.ensure_auth(p) # 复用 Step4 登录态
browser = p.chromium.launch(headless=config.HEADLESS, slow_mo=config.SLOW_MO)
ctx = browser.new_context(storage_state=str(config.STATE_FILE),
viewport={"width": 1280, "height": 900}, locale="zh-CN")
ctx.set_default_timeout(8000)
page = ctx.new_page()
page.goto(config.PRODUCTS_URL, wait_until="domcontentloaded")
rl = demo_self_healing(page, ai_enabled)
books = pw4.scan_list(page) # ← 直接复用 Step4 函数
by_id = {b.id: b for b in books}
cats = page.get_by_test_id("category-filter").locator("option")
for i in range(cats.count()):
c = cats.nth(i).get_attribute("value")
if not c: continue
for b in pw4.scan_list(page, category=c):
if b.id in by_id: by_id[b.id].category = c
cand = [b for b in books
if b.discount_pct >= rules["min_discount"] and b.rating >= rules["min_rating"]]
for b in cand:
b.stock, cat = pw4.fetch_stock(page, b.id) # ← 复用 Step4 详情页补库存
if cat: b.category = cat
b.source = "hybrid"
picks = [b for b in cand if b.in_stock or not rules["require_stock"]]
picks.sort(key=lambda b: (-b.discount_pct, -b.rating))
res.books, res.picks = books, picks
res.notes.append(f"规则解析方式:{how}")
res.notes.append(f"定位器分级命中统计:{rl.stats}")
res.notes += rl.heals
res.finished_at, res.elapsed_sec = now(), time.perf_counter() - t0
return res, rules, how, rl- 逐块逻辑:
run()把「解析需求(接缝 A)→ 演示自愈(接缝 B)→ 复用 Step 4 的scan_list/fetch_stock做确定性采集 → 按 LLM 解析出的规则预筛 → 排序出报告」串起来。注意pw4.scan_list/pw4.fetch_stock直接 import 了 Step 4 的函数,一行都没重写——这就是「骨架复用」的威力。 逐行解释(关键点):
rules, how = resolve_rules(need):how会告诉你这次是 "LLM" 还是 "本地规则引擎",写进报告便于追溯。pw4.ensure_auth(p):复用 Step 4 的登录态兜底,保证脚本可独立运行。cand = [b for b in books if ...]:用 LLM 解析出的min_discount/min_rating做预筛——唯一的「AI 决策」就发生在这里,决定哪些书进候选池。res.notes += rl.heals:把自愈记录原样写进报告,让「页面哪被改过」一目了然。
2.9.5 真实输出(运行验证)
默认需求(无需 Key,落在本项目真实靶场):
[环节 A] 解析自然语言需求:'我想要打折力度大、口碑好、而且现在就能发货的书'
解析方式:本地规则引擎
→ 折扣 ≥ 25.0% · 评分 ≥ 4.5 · 必须有货
[环节 B] 定位器自愈演示
L1 命中:data-testid='search-input' 正常工作
L1 未命中 → L2 用 get_by_role('link', name='全部图书') 成功救回 ✓
L1/L2 未命中 → L3 文本定位成功救回 ✓
自愈统计:{'L1': 1, 'L2': 1, 'L3': 1, 'L4': 0, 'FAIL': 0}
[骨架] Playwright 确定性采集…
列表扫描完成:24 本
按 AI 解析出的规则预筛 → 11 本候选
════════════════════════════════════════════════════════════════
混合方案命中 9 本 · 耗时 11.53 秒 · LLM 调用 0 次
════════════════════════════════════════════════════════════════
1. 人类简史 省 42.6% · 评分 4.7 · 库存 240
2. 纳瓦尔宝典 省 40.7% · 评分 4.7 · 库存 102
3. 穷查理宝典 省 40.6% · 评分 4.8 · 库存 54
4. 百年孤独 省 40.0% · 评分 4.9 · 库存 310
5. 思考,快与慢 省 39.1% · 评分 4.6 · 库存 188
自愈记录:
· [导航-全部图书(模拟 testid 被改名)] testid='nav-products-v2-renamed' 失效 → 由 role=link 救回
· [页面统计文案] testid='totally-gone' 失效 → 由文本 '共 24 本' 救回
✅ Step 6 完成。这就是我推荐的生产形态。换成更「狠」的需求:
python steps/step6_hybrid_pipeline.py --need "我想要打折特别狠的书,口碑要好,必须现货"
# 解析方式:本地规则引擎
# → 折扣 ≥ 35.0% · 评分 ≥ 4.5 · 必须有货
# 按 AI 解析出的规则预筛 → 9 本候选
# 混合方案命中 7 本 · 耗时 9.79 秒 · LLM 调用 0 次数据解读:同样的确定性骨架,只因为「需求解析」这一处接缝换了个阈值,命中数就从 9 本变 7 本——LLM 在这里的价值是把「人话」翻成「规则」,而不是替你点页面。整条流水线 LLM 调用 0 次(无 Key 时),耗时 11.53 秒,比纯脚本(Step 4 约 15 秒)还快一点(少了一次独立的登录态重建)。{'L1':1,'L2':1,'L3':1,'L4':0,'FAIL':0} 这份统计还顺手证明了:靶场里确实有元素「改名」了,而骨架靠 L2 自愈把它扛住了。
2.9.6 这一步的可复用套路
混合架构 = 确定性骨架 + 2 个模糊接缝
① 接缝 A(需求):自然语言 → 结构化规则(无 Key 用关键词兜底)
② 接缝 B(定位):L1 testid → L2 role → L3 text → L4 AI(分级统计=技术债报告)
骨架:直接复用已有的确定性采集函数,一行不重写
产物:报告里同时记录"规则解析方式"和"自愈记录",可解释、可追责2.10 完整项目结构与全部产物
AIforGUI技术实战演练/
├── AIforGUI实战指导书.md ← 你正在读的这份
└── AIforGUIcode/
├── .env.example 配置样例(LLM Provider / Model / Key)
├── run_all.py 一键跑完 Step 1–6(自动拉起靶场)
├── check_env.py 环境自检
├── aigui/
│ ├── config.py 全局配置(阈值、URL、路径)
│ ├── models.py Book / RunResult 数据模型
│ └── report.py 多格式报告生成
├── demosite/ 本地靶场(Flask)
│ ├── server.py 故意埋了 5 个陷阱的书店
│ └── data.py 24 本书的种子数据
├── steps/
│ ├── step1_hello_playwright.py Hello World(确认环境是活的)
│ ├── step2_selectors_and_traps.py 五个陷阱演示
│ ├── step3_login_session.py 登录 + 会话复用
│ ├── step4_playwright_scraper.py 确定性流水线(13 本 / ¥338)
│ ├── step5_browseruse_agent.py Browser Use 自主方案
│ └── step6_hybrid_pipeline.py 混合架构(9 本 / 11.53s)
└── output/ 所有运行产物
├── auth_state.json 登录态(等同账号密码,勿提交 Git)
├── books_playwright.csv / result_playwright.json / report_playwright.md
├── books_hybrid.csv / result_hybrid.json / report_hybrid.md
└── screenshots/step*.png 各步截图存档六步之间的承上启下关系:
| 步骤 | 主角 | 解决什么 | 引出下一步 |
|---|---|---|---|
| Step 1 | Playwright | 确认环境能驱动浏览器 | 环境 OK 才能往下 |
| Step 2 | Playwright | 先踩 5 个真实陷阱 | 这些坑是后续设计的「验收标准」 |
| Step 3 | Playwright | 登录态复用(storage\_state) | Step 4/6 依赖它 |
| Step 4 | Playwright | 确定性完整方案(零 AI) | 作为「正确答案」基准 |
| Step 5 | Browser Use | 纯 AI 自主方案 | 与 Step 4 对照成本/稳定性 |
| Step 6 | 混合 | 生产推荐形态 | 收束两条路线 |
项目完整代码包:

2.11 排错手册
| 现象 | 根因 | 解决 |
|---|---|---|
ModuleNotFoundError: No module named 'playwright' | 包装到了别的解释器 | 在 python run_all.py 用的那个解释器里 pip install playwright,再 playwright install chromium |
| 报告里命中 16 本而不是 13 本 | 陷阱 4 没处理,零库存书混进来了 | 确认 fetch_stock 等 [data-loaded="true"] 再读库存 |
| 会员价显示「登录后查看」 | 登录态失效 | 删 output/auth_state.json,重跑(Step 3/4 会自动重建) |
| 翻页卡死 / 报「翻页超过 20 次」 | 站点异常或下一页按钮识别错 | 检查 page-next 的 data-testid 是否还在 |
| Step 5 卡在「等待模型」 | 没配 Key 或 Key 无效 | .env 里填 OPENAI_API_KEY;没有 Key 可先跳 Step 5 看 Step 4/6 |
browser-use 报参数不认识 | 版本 API 变了 | 本项目的 build_agent() 已做 inspect 兼容,自动忽略不支持的参数 |
| 靶场起不来 | 端口被占用 / Flask 没装 | pip install flask;或换端口改 config.BASE_URL |
| AI 方案数字和脚本方案差 1–2 本 | 模型推理有浮动 | 属正常;要 100% 复现就用 Step 4 的确定性方案 |
2.12 学习路径与进阶方向
如果你是从零开始,建议的练习顺序:
- 先跑通 Step 1,把「driver + 浏览器」的关系在脑子里立住。
- 把 Step 2 的 5 个陷阱逐个注释掉再跑,亲眼看到「不处理会怎样」。
- 改写 Step 4 的筛选阈值,体会「确定性脚本改规则只要改一行」。
- 给 Step 5 配个 Key,对比两次运行的命中数差异。
- 在 Step 6 里把
ResilientLocator的 L4 真正接上 Browser Use,体验两套技术合流。
进阶方向:
- 视觉增强:把
use_vision=True打开,让 Agent 看截图处理纯图像页面(验证码、图表)。 - 多 Agents 协作:一个 Agent 负责「发现」,一个负责「复核」,用
Tools互相传递中间结果。 - 自愈闭环:把
ResilientLocator的heals统计接进监控,L2/L3 命中率超阈值就自动报警「前端改版了」。 - 成本治理:给
max_steps、单次调用 token 设预算,超出即熔断(呼应 1.7 成本模型)。 - 从靶场到真实站点:把
config的 URL 换成你自己的业务系统,选择器换成真实data-testid,骨架直接复用。
觉得内容不错?我要