Crawlee Python 实战：动态职位列表、详情提取与 SQLite 保存
用 Crawlee Python 和 Playwright 处理延迟渲染的职位列表，跟随分页与详情，按职位编号写入 SQLite，并导出 CSV 和失败清单。
职位列表已经加载完成，抓取器却只拿到一个空容器，是动态页面里常见的情况。解决方法不是对空结果无限重试，而是先确认数据来自响应 HTML 还是 JavaScript 渲染，再选择相应的加载与等待方式。
本文使用 Crawlee Python 的 PlaywrightCrawler，完成两页动态职位列表、三条详情、SQLite 更新与 CSV 导出。正文内提供完整的教学页面材料，通过浏览器请求拦截返回，不需要另建网站或下载实验工程。换成真实招聘网站时，再替换来源和选择器。
安装与抓取器选择
使用 Python 3.12，终端采用 Bash：
python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'crawlee[playwright]==1.11.1'
python -m playwright install chromium如果职位、地点与要求已经在初始 HTML 中，BeautifulSoupCrawler 等 HTTP 解析路径通常更直接。本文列表会在 JavaScript 执行后出现，所以选 PlaywrightCrawler。两者不是“招聘站一定用浏览器”的行业规则，判断依据是目标字段的位置。用法见 Crawlee Python 官方项目。
先定义两页动态材料
创建 jobs.py，先放入下面的导入与材料函数：
import asyncio
import csv
import json
import sqlite3
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from crawlee.crawlers import PlaywrightCrawler
JOBS = {
"j1": {"title": "Python Developer", "location": "Remote", "requirements": "Python, SQL"},
"j2": {"title": "Data Engineer", "location": "Shanghai", "requirements": "ETL, SQL"},
"j3": {"title": "QA Engineer", "location": "Remote", "requirements": "Browser automation"},
}
START = "https://example.16yun.cn/jobs.html?page=1"
def teaching_html(url):
parsed = urlsplit(url)
if parsed.path == "/jobs.html":
number = int(parse_qs(parsed.query).get("page", ["1"])[0])
ids = ["j1", "j2"] if number == 1 else ["j3"]
cards = "".join(
f'<article class="job-card"><a href="/jobs/{key}">{JOBS[key]["title"]}</a></article>'
for key in ids
)
if number == 1:
cards += '<a class="next" href="/jobs.html?page=2">Next</a>'
return (
'<meta charset="utf-8"><h1>Jobs</h1><main id="results"></main>'
'<script>setTimeout(() => {'
f'document.querySelector("#results").innerHTML = {json.dumps(cards)};'
'document.querySelector("#results").dataset.ready = "true";'
'}, 50);</script>'
)
key = parsed.path.rsplit("/", 1)[-1]
job = JOBS.get(key)
if job is None:
return None
return (
'<meta charset="utf-8">'
f'<h1 class="job-title">{job["title"]}</h1>'
f'<p class="location">{job["location"]}</p>'
f'<div class="requirements">{job["requirements"]}</div>'
)example.16yun.cn 在这里是被拦截的教学地址，不要求该域名存在招聘服务。列表的原始 HTML 只有容器，职位卡片由脚本延迟添加。固定五十毫秒只是材料中模拟延迟的方法，抓取逻辑不靠固定 sleep 判断完成。
详情字段是简化结构。真实网站可能把要求放在富文本、折叠面板或 API 响应里，需要重新确认其读取方式。
连起加载、分页与存储
继续在同一个文件追加：
async def main():
db = sqlite3.connect("jobs.sqlite")
db.execute("""CREATE TABLE IF NOT EXISTS jobs(
job_id TEXT PRIMARY KEY, title TEXT NOT NULL,
location TEXT NOT NULL, requirements TEXT NOT NULL, source_url TEXT NOT NULL
)""")
failures = []
crawler = PlaywrightCrawler(max_requests_per_crawl=10, max_request_retries=1)
@crawler.pre_navigation_hook
async def material(context):
async def route_request(route):
html = teaching_html(route.request.url)
await route.fulfill(
status=200 if html is not None else 404,
content_type="text/html; charset=utf-8",
body=html or "Not found",
)
await context.page.route("https://example.16yun.cn/**", route_request)
@crawler.router.default_handler
async def listing(context):
await context.page.locator('#results[data-ready="true"]').wait_for()
await context.enqueue_links(selector=".job-card a", label="DETAIL")
await context.enqueue_links(selector="a.next")
@crawler.router.handler("DETAIL")
async def detail(context):
page = context.page
title = (await page.locator(".job-title").inner_text()).strip()
location = (await page.locator(".location").inner_text()).strip()
requirements = (await page.locator(".requirements").inner_text()).strip()
if not title or not location or not requirements:
raise ValueError("职位必填字段缺失")
job_id = urlsplit(context.request.url).path.rsplit("/", 1)[-1]
with db:
db.execute("""INSERT INTO jobs VALUES (?, ?, ?, ?, ?)
ON CONFLICT(job_id) DO UPDATE SET
title=excluded.title, location=excluded.location,
requirements=excluded.requirements, source_url=excluded.source_url""",
(job_id, title, location, requirements, page.url))
@crawler.failed_request_handler
async def failed(context, error):
failures.append({"url": context.request.url, "error_type": type(error).__name__})
try:
await crawler.run([START])
rows = db.execute(
"SELECT job_id,title,location,requirements,source_url FROM jobs ORDER BY job_id"
).fetchall()
with open("jobs.csv", "w", encoding="utf-8-sig", newline="") as stream:
writer = csv.writer(stream)
writer.writerow(["job_id", "title", "location", "requirements", "source_url"])
writer.writerows(rows)
Path("failed.json").write_text(json.dumps(failures, indent=2), encoding="utf-8")
print({"stored_jobs": len(rows), "failed_requests": len(failures)})
finally:
db.close()
asyncio.run(main())运行：
python jobs.py浏览器由 crawler 管理，不需要再猜一个 crawler.close 方法。SQLite 连接则由本程序创建，在 finally 中关闭。浏览器请求、业务存储和导出分别有明确的资源所有者。
为什么等待数据标记而不是网络空闲
DOMContentLoaded 只表示文档已解析，本文职位仍可能没出现。等待 #results[data-ready="true"] 使条件与“列表材料已经插入”直接关联。
真实站点的标记可能是加载指示消失、列表计数更新或 API 返回。长连接与后台请求会影响网络空闲判断，因此不能对所有 SPA 都套用同一条 networkidle 或固定等待。
分页链接仍保留 page 参数。把 page=1 与 page=2 强行规范化成同一个 URL，会让后者在调度层消失。请求上限十也不是职位数量；本例有两个列表加三个详情。
检查 SQLite、CSV 与失败记录
从教学材料推导，第一次运行应有 j1、j2、j3 三个唯一职位，CSV 含标题行与三条数据。示例没有实际运行记录，不将这些预期包装成测速结果。
检查 j3 是否存在，可以发现只处理第一页的问题。检查其要求是否是 Browser automation，可以发现字段读取错位。检查 Remote 与 Shanghai 是否正确，可以发现同一位置文本被复制的问题。
SQLite 使用 job_id 作为主键，重复抓到同一职位时更新字段，不追加重复行。它不自动判断职位已经下架，也不删除本轮没出现的历史记录。换成真实来源时，职位 ID 应使用来源业务编号，跨站存储再加入站点维度，不能只从任意 URL 最后一段猜通用 ID。
CSV 从数据库生成，是当前库的内容，不一定只属于刚才一轮。对长期任务，增加观测批次或更新时间，再明确导出历史还是本轮快照。事务和参数绑定见 Python SQLite 文档。
迁移到真实招聘页
删除教学请求拦截 hook，替换 START，更新列表、下一页和详情规则。不要保留拦截后又以为访问了真实站点。先选一个明确分类与两页数据，人工记下第二页的职位样本。
若真实列表有正常空状态，为它增加独立判断；不能把等待超时直接改成“零职位成功”。若没有下一页链接而是加载更多或游标 API，需要重新表达分页和终止条件，而不是只换按钮文本。
常见问题与修复
浏览器安装完成却找不到可执行文件。 检查运行账户、虚拟环境和浏览器缓存位置；容器与宿主机安装不是同一环境。
列表等待超时。 教学例子先检查代码三部分是否同在一个文件、hook 是否保留。真实页面则看数据就绪标记与页面身份，不先无限增加超时。
数据库三条但本轮失败。 旧数据可能仍在库里。对照 failed 与观测批次，不能以数据库总量证明本轮全部成功。
重新运行出现重复职位。 核对业务键是否稳定，是否把临时查询参数或列表序号当成 ID。修复键后按来源核对已有数据。
页面里有空要求。 确定是公开信息缺失还是选择器失效，再决定允许 null 还是隔离。当前合同要求非空，不用空字符串掩盖错误。
动态站能抓到，但批量不齐。 检查发现范围、分页参数和请求上限，再查详情失败。等待解决的是材料就绪，不会自动补齐没有进入队列的 URL。
从动态列表进入详情，再以稳定键写入和导出，才构成可维护的职位采集主线。浏览器只是其中负责材料获取的一环。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。