Playwright 动态列表采集：网络响应、滚动加载与游标分页
在 Playwright 中观察列表数据响应，驱动滚动加载，按游标判断结束并按业务 ID 合并，避免虚拟列表、重复项和失败响应造成漏采。
无限滚动页面不一定把所有记录同时留在 DOM 中。滚动很久以后再一次性读取节点，可能只拿到最后一个窗口；看到条数不变，也可能是网络失败，而不是到了末尾。
本文提供一个完整的浏览器教学页面和三段分页响应。我们在动作前注册响应等待，逐段收集数据，用 next_cursor 判断结束，按商品 ID 合并。材料包含一条跨页重复，便于检查去重是否掩盖了覆盖问题。
安装与范围
使用 Python 3.12：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install playwright
python -m playwright install chromium示例用浏览器路由返回正文材料，不访问 example.16yun.cn 的真实服务。动态列表只有四个业务商品，详情地址没有被访问，不能把列表结果称为详情采集完成。
先定义浏览器会收到的数据
创建 cursor_list.py：
import asyncio
import json
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from playwright.async_api import async_playwright
PAGES = {
"": {"items": [{"id": "p1", "name": "商品一"}, {"id": "p2", "name": "商品二"}], "next_cursor": "second"},
"second": {"items": [{"id": "p2", "name": "商品二"}, {"id": "p3", "name": "商品三"}], "next_cursor": "last"},
"last": {"items": [{"id": "p4", "name": "商品四"}], "next_cursor": None},
}
HTML = """
<style>.card {height:600px;border:1px solid #ddd}</style>
<h1>教学列表</h1><div id="list"></div><p id="state"></p>
<script>
let cursor = '', loading = false, ended = false;
async function load() {
if (loading || ended) return;
loading = true;
try {
const response = await fetch('/api/items?cursor=' + encodeURIComponent(cursor));
if (!response.ok) throw new Error('API failed');
const data = await response.json();
const list = document.querySelector('#list');
for (const item of data.items) {
const card = document.createElement('div');
card.className = 'card'; card.textContent = item.name;
list.appendChild(card);
}
cursor = data.next_cursor; ended = cursor === null;
list.dataset.next = ended ? 'END' : cursor;
} catch (error) {
document.querySelector('#state').textContent = '加载失败';
} finally { loading = false; }
}
window.addEventListener('scroll', () => {
if (innerHeight + scrollY >= document.body.scrollHeight - 10) load();
});
load();
</script>
"""游标是服务给出的不透明标识，不解析 second 或 last 推算页码。真实接口可能使用长字符串或签名，原样携带，不能自行加一。
捕获响应并驱动后续滚动
继续在同一文件追加：
async def main():
records, cursors = {}, set()
async with async_playwright() as pw:
browser = await pw.chromium.launch()
try:
context = await browser.new_context(viewport={"width": 1000, "height": 400})
page = await context.new_page()
async def respond(route):
parsed = urlsplit(route.request.url)
if parsed.path == "/api/items":
cursor = parse_qs(parsed.query, keep_blank_values=True).get("cursor", [""])[0]
data = PAGES.get(cursor)
await route.fulfill(
status=200 if data is not None else 400,
content_type="application/json", body=json.dumps(data or {"error": "cursor"}),
)
else:
await route.fulfill(content_type="text/html; charset=utf-8", body=HTML)
await page.route("https://example.16yun.cn/**", respond)
def is_items(response):
return urlsplit(response.url).path == "/api/items" and response.request.method == "GET"
async with page.expect_response(is_items) as first:
await page.goto("https://example.16yun.cn/")
response = await first.value
for _ in range(10):
if response.status != 200:
raise ValueError("分页请求失败")
data = await response.json()
if not isinstance(data.get("items"), list):
raise ValueError("分页响应不是预期结构")
for item in data["items"]:
if not isinstance(item.get("id"), str) or not item.get("name"):
raise ValueError("商品 ID 或名称缺失")
if item["id"] in records and records[item["id"]] != item:
raise ValueError("同一 ID 跨页数据冲突")
records[item["id"]] = item
next_cursor = data.get("next_cursor")
marker = "END" if next_cursor is None else next_cursor
await page.wait_for_function(
"marker => document.querySelector('#list').dataset.next === marker",
arg=marker,
)
if next_cursor is None:
break
if not isinstance(next_cursor, str) or next_cursor in cursors:
raise ValueError("分页游标无效或循环")
cursors.add(next_cursor)
async with page.expect_response(is_items, timeout=15000) as pending:
await page.mouse.wheel(0, 5000)
response = await pending.value
else:
raise ValueError("超出本轮分页预算")
finally:
await browser.close()
Path("items.json").write_text(
json.dumps({"source_kind": "teaching_material", "items": list(records.values())},
ensure_ascii=False, indent=2), encoding="utf-8"
)
print({"unique_items": len(records)})
asyncio.run(main())执行：
python cursor_list.py预期接收三段分页材料，合并成 p1 至 p4 四条商品。第二段重复 p2 不增加唯一数，但不妨碍继续处理 next_cursor。这个数量来自正文材料，不是对线上站点的完整性测量。
为什么必须在动作之前等待响应
先滚动、再注册 expect_response，可能错过很快返回的请求。示例把滚动动作放在等待上下文中，保证先建立观察再触发加载。用法见 Playwright 网络文档。
接口响应结束后，页面渲染仍可能没完成。教学页将 next 标记写入 DOM 后再继续滚动，避免在列表尚未增高时重复滚到同一位置。真实站点应找到等价的加载完成或内容更新证据，而不是照搬该标记。
实际页面还可能有预取、广告和多个列表接口。路径匹配只是本例范围；迁移时应结合请求方法、目标 API、筛选参数与预期游标，避免把无关响应接到当前任务。
结束条件不应只看行数
明确的 null 游标是本例结束依据。重复页、空批次、网络失败或可见 DOM 不增长，都不能直接当成结束。
如果接口允许“空 items 但还有下一游标”，应按接口合同继续，并保留预算和循环检测。若它不允许这种状态，则应作为数据异常，不静默跳过。
虚拟化列表的 DOM 条数受窗口控制。保存已经观察到的响应，比等最后一次从 DOM 抓全更有解释性，但仍要确认接口里的条目确实属于当前筛选和用户可见范围。
常见问题与修复
滚动后等不到响应。 检查是否真的到达触发区、页面是否仍在 loading、观察路径是否正确，以及是否已到末尾。先看页面状态，不把超时改成“采集结束”。
同一游标反复出现。 保存当前请求与响应，停止循环，检查筛选参数、接口游标和重试逻辑。无限滚动不能靠无限重试解决。
记录越来越多但都是重复。 检查响应对应的游标和业务 ID。去重能控制输出，但不能证明发现了新页。
网络响应有字段，页面没有。 可能是隐藏数据、预取或另一列表。先确认任务是否应使用它，不自动把响应里所有字段都纳入业务数据。
页面不是 GET 分页。 POST body 也是分页合同的一部分，等待谓词和请求复用都要同步调整。不能只替换 URL。
中断后旧文件还在。 当前例子完成后才导出，没有持久恢复。为批次记录状态，不能用旧 items.json 判断失败运行的结果。
动态列表的可靠结束来自接口与页面状态，而不是滚动次数。游标、响应和业务记录之间的对应关系，才是排查漏采的基础。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。