Python爬虫实战：Python动态网站爬虫实战 - 基于Playwright的DOM渲染与结构化字段抽取管理！

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

全文目录：

• • 🌟 开篇语
  • 📌 摘要（Abstract）
  • 🎯 背景与需求（Why）
  • • 为什么要爬动态网站？
    • 目标站点与字段清单
  • ⚖️ 合规与注意事项（必读）
  • • robots.txt协议
    • 频率控制
    • 数据使用边界
  • 🛠️ 技术选型与整体流程（What/How）
  • • 静态 vs 动态 vs API
    • 整体流程设计
  • 🔧 环境准备与依赖安装
  • • Python版本要求
    • 依赖安装
    • 项目目录结构
  • 🌐 核心实现：请求层（Fetcher）
  • 🔍 核心实现：解析层（Parser）
  • • 选择器配置文件 `config/selectors.yaml`
    • 解析器实现 `core/parser.py`
  • 💾 数据存储与导出（Storage）
  • • `core/storage.py`
  • ⚙️ 完整主程序 `main.py`
  • • 日志工具 `utils/logger.py`
    • 启动命令
    • 控制台输出示例
    • 输出文件位置
    • 示例数据（前5行）
  • 🔧 常见问题与排错
  • • 1️⃣ 403/429错误：访问被拒绝
    • 2️⃣ 抓到空壳HTML：动态内容未加载
    • 3️⃣ 选择器解析报错
    • 4️⃣ 编码/乱码问题
  • 🚀 进阶优化（生产级改造）
  • • 1️⃣ 并发爬取（asyncio协程池）
    • 2️⃣ 断点续爬（Redis/SQLite游标）
    • 3️⃣ 日志与监控（结构化日志）
    • 4️⃣ 定时任务（APScheduler）
  • 📚 总结与延伸阅读
  • • 我们完成了什么？
    • 下一步可以做什么？
    • 推荐阅读资源
  • 🎉 写在最后
  • • 🌟 文末
    • • 📌 专栏持续更新中｜建议收藏 + 订阅
      • ✅ 互动征集

🌟 开篇语

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

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

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

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

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

📌 摘要（Abstract）

本文将完整演示如何使用Playwright爬取动态渲染网站（以招聘平台职位列表为例），通过浏览器自动化获取JavaScript渲染后的DOM结构，并实现一套可维护的选择器管理体系，最终输出结构化CSV数据。

• 动态网站渲染原理的深刻理解与实战经验
• Playwright框架的核心使用技巧（页面等待、元素定位、反爬应对）
• 一套工程化的字段选择器映射管理方案（可复用、易维护）
• 完整可运行的爬虫代码（含异常处理、数据清洗、存储导出）

🎯 背景与需求（Why）

为什么要爬动态网站？

传统的requests + BeautifulSoup方案在面对现代Web应用时常常束手无策：

• 前后端分离架构：HTML只是空壳，数据通过Ajax异步加载
• JavaScript渲染：关键内容需要JS执行后才能呈现在DOM中
• **反、滑块验证、指纹识别

而招聘网站恰好是典型场景：职位列表通过接口懒加载，详情页依赖路由跳转，薪资范围、福利标签等核心字段都嵌套在多层组件中。

目标站点与字段清单

目标：某主流招聘平台的Python工程师职位数据（仅用于学习演示）

待采集字段：

⚖️ 合规与注意事项（必读）

robots.txt协议

在任何爬虫行为前，务必检查目标网站的/robots.txt文件。虽然该协议不具备法律约束力，但遵守它是技术从业者的基本素养。

# 示例代码：检查robots.txt
import urllib.robotparser

rp = urllib.robotparser.RobotFileParser()
rp.set_url("https://example.com/robots.txt")/jobs"))

频率控制

• 人类模拟：每页请求间隔2-5秒随机延迟
• 避免并发攻击：不使用多进程暴力采集
• User-Agent轮换：避免被识别为爬虫

数据使用边界

• ❌ 禁止采集：用户隐私数据（电话、邮箱）、付费内容
• ❌ 禁止绕过：登录墙、验证码（除非你有账号且合规使用）
• ✅ 允许场景：公开职位信息的个人学习、数据分析研究

法律提示：《数据安全法》和《个人信息保护法》要求数据采集必须合法、正当、必要。本文代码仅供技术学习，请勿用于商业用途。

🛠️ 技术选型与整体流程（What/How）

静态 vs 动态 vs API

本文选择Playwright的原因：

• 现代化异步API（比Selenium更快）
• 支持无头模式（headless）
• 内置等待机制（自动处理动态加载）
• 跨浏览器支持（Chromium/Firefox/WebKit）

整体流程设计

┌─────────────┐ ┌──────────────┐ ┌─────────────┐ ┌──────────────┐
│ 1. 启动浏览器 │ —> │ 2. 页面渲染等待 │ —> │ 3. DOM元素抽取 │ —> │ 4. 数据清洗转换 │
└─────────────┘ └──────────────┘ └─────────────┘ └──────────────┘
│
↓
┌─────────────────┐
│ 选择器管理配置 │
│ (YAML/JSON) │
└─────────────────┘
│
↓
┌─────────────┐ ┌──────────────┐ ┌─────────────┐
│ 5. 去重判断 │ <— │ 6. 数据持久化 │ <— │ 7. 异常记录 │
└─────────────┘ └──────────────┘ └─────────────┘

为什么不直接用Scrapy？

Scrapy的核心优势在于异步下载和中间件管道，但对于需要完整浏览器环境渲染的场景，Playwright的开发体验和调试效率更高。当然，生产环境中可以考虑Scrapy + Playwright插件的组合方案。

🔧 环境准备与依赖安装

Python版本要求

• 推荐：Python 3.9+（支持类型提示和新语法特性）
• 最低：Python 3.8

依赖安装

# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate # Windows: venv\\Scripts\\activate

# 安装核心依赖
pip install playwright==1.40.0
pip install pyyaml==6.0.1
pip install pandas==2.1.4

# 安装浏览器驱动（首次必须执行）
playwright install chromium

项目目录结构

dynamic_spider/
├── config/
│ └── selectors.yaml # 选择器配置文件
├── core/
│ ├── __init__.py
│ ├── fetcher.py # 页面请求层
│ ├── parser.py # DOM解析层
│ └── storage.py # 数据存储层
├── utils/
│ ├── __init__.py
│ └── logger.py # 日志工具
├── data/
│ ├── output/ # CSV输出目录
│ └── logs/ # 运行日志
├── main.py # 程序入口
└── requirements.txt # 依赖清单

创建基础目录：

mkdir -p dynamic_spider/{config,core,utils,data/{output,logs}}
cd dynamic_spider
touch core/__init__.py utils/__init__.py

🌐 核心实现：请求层（Fetcher）

这一层负责启动浏览器、加载页面等

from typing import Optional
from playwright.async_api import async_playwright, Page, Browser
from utils.logger import get_logger

logger = get_logger(**name**)

class DynamicFetcher:
"""动态页面抓取器"""

def __init__(self, headless: bool = True, slow_mo: int = 0):
"""
Args:
headless: 是否无头模式
slow_mo: 操作延迟（毫秒），调试时可设置为100
"""
self.headless = headless
self.slow_mo = slow_mo
self.browser: Optional[Browser] = None
self.context = None

async def __aenter__(self):
"""异步上下文管理器入口"""
playwright = await async_playwright().start()
self.browser = await playwright.chromium.launch(
headless=self.headless,
slow_mo=self.slow_mo
)

# 创建浏览器上下文（可设置User–Agent、视口等）
self.context = await self.browser.new_context(
user_agent=self._get_random_ua(),
viewport={'width': 1920, 'height': 1080},
locale='zh-CN',
timezone_id='Asia/Shanghai'
)

logger.info("浏览器启动成功")
return self

async def __aexit__(self, exc_type, exc_val, exc_tb):
"""异步上下文管理器出口"""
if self.context:
await self.context.close()
if self.browser:
await self.browser.close()
logger.info("浏览器已关闭")

async def fetch_page(self, url: str, wait_selector: str = None,
timeout: int = 30000) –> Page:
"""
加载页面并等待关键元素

Args:
url: 目标URL
wait_selector: 等待的CSS选择器（等到该元素出现才返回）
timeout: 超时时间（毫秒）

Returns:
Page对象
"""
page = await self.context.new_page()

try:
# 设置超时
page.set_default_timeout(timeout)

# 访问页面
logger.info(f"正在访问: {url}")
response = await page.goto(url, wait_until='domcontentloaded')

if response.status != 200:
logger.warning(f"响应状态码: {response.status}")

# 等待关键元素加载
if wait_selector:
await page.wait_for_selector(wait_selector, timeout=timeout)
logger.info(f"关键元素已加载: {wait_selector}")

# 模拟人类行为：随机滚动
await self._random_scroll(page)

# 随机延迟
await asyncio.sleep(random.uniform(2, 4))

return page

except Exception as e:
logger.error(f"页面加载失败: {url}, 错误: {str(e)}")
await page.close()
raise

async def _random_scroll(self, page: Page):
"""随机滚动页面（反爬措施）"""
scroll_distance = random.randint(300, 800)
await page.evaluate(f"window.scrollBy(0, {scroll_distance})")
await asyncio.sleep(random.uniform(0.5, 1.5))

@staticmethod
def _get_random_ua() –> str:
"""随机User-Agent"""
ua_list = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
]
return random.choice(ua_list)

关键点说明：

🔍 核心实现：解析层（Parser）

这是全文的核心创新点——结构化选择器管理。

选择器配置文件 config/selectors.yaml

# 职位列表页选择器配置
job_list:
container: "div.job-list-box" # 列表容器
item: "div.job-card" # 单个职位卡片

fields:
job_title:
selector: "div.job-title"
attribute: "text"
required: true

company_name:
selector: "div.company-name a"
attribute: "text"
required: true

salary_range:
selector: "span.salary"
attribute: "text"
required: true
transform: "clean_salary" # 调用清洗函数

work_location:
selector: "span.work-location"
attribute: "text"
required: false
default: "未知"

work_experience:
selector: "span.work-exp"
attribute: "text"
required: false

education:
selector: "span.education"
attribute: "text"
required: false

job_tags:
selector: "div.job-tags span"
attribute: "text"
multiple: true # 多个元素

company_size:
selector: "span.company-size"
attribute: "text"
required: false

job_url:
selector: "a.job-link"
attribute: "href"
required: true
transform: "complete_url" # 补全相对路径

解析器实现 core/parser.py

import re
from typing import Dict, List, Any, Optional
from playwright.async_api import Page, ElementHandle
import yaml
from utils.logger import get_logger

logger = get_logger(__name__)

class StructuredParser:
"""结构化字段解析器"""

def __init__(self, config_path: str):
"""加载选择器配置"""
with open(config_path, 'r', encoding='utf-8') as f:
self.config = yaml.safe_load(f)
logger.info(f"选择器配置加载成功: {config_path}")

async def parse_list_page(self, page: Page, section: str = 'job_list') –> List[Dict]:
"""
解析列表页

Args:
page: Playwright Page对象
section: 配置文件中的区域名

Returns:
字典列表，每个字典代表一条数据
"""
config = self.config.get(section, {})
container_sel = config.get('container')
item_sel = config.get('item')

if not container_sel or not item_sel:
raise ValueError(f"配置缺失: {section}")

# 等待容器加载
await page.wait_for_selector(container_sel)

# 获取所有列表项
items = await page.query_selector_all(f"{container_sel} {item_sel}")
logger.info(f"找到 {len(items)} 个列表项")

results = []
for idx, item in enumerate(items, 1):
try:
data = await self._parse_item(item, config['fields'])
data['_index'] = idx # 添加索引
results.append(data)
except Exception as e:
logger.error(f"解析第 {idx} 项失败: {str(e)}")
continue

return results

async def _parse_item(self, element: ElementHandle,
field_config: Dict) –> Dict:
"""解析单个元素"""
data = {}

for field_name, field_spec in field_config.items():
try:
value = await self._extract_field(element, field_spec)
data[field_name] = value
except Exception as e:
# 处理必填字段缺失
if field_spec.get('required', False):
logger.warning(f"必填字段缺失: {field_name}")
raise
else:
data[field_name] = field_spec.get('default', None)

return data

async def _extract_field(self, element: ElementHandle,
spec: Dict) –> Any:
"""提取单个字段"""
selector = spec['selector']
attribute = spec.get('attribute', 'text')
multiple = spec.get('multiple', False)
transform = spec.get('transform')

if multiple:
# 提取多个元素
elements = await element.query_selector_all(selector)
if attribute == 'text':
values = [await el.text_content() for el in elements]
else:
values = [await el.get_attribute(attribute) for el in elements]
result = [v.strip() for v in values if v]
else:
# 提取单个元素
el = await element.query_selector(selector)
if not el:
return None

if attribute == 'text':
result = await el.text_content()
else:
result = await el.get_attribute(attribute)

result = result.strip() if result else None

# 应用转换函数
if transform and result:
transform_func = getattr(self, transform, None)
if transform_func:
result = transform_func(result)

return result

# === 数据清洗转换函数 ===

@staticmethod
def clean_salary(salary: str) –> str:
"""清洗薪资字段"""
# 示例: "25K-45K·14薪" -> "25K-45K"
match = re.search(r'(\\d+K?-\\d+K?)', salary)
return match.group(1) if match else salary

@staticmethod
def complete_url(url: str, base_url: str = "https://www.example.com") –> str:
"""补全相对URL"""
if url.startswith('http'):
return url
return base_url.rstrip('/') + '/' + url.lstrip('/')

设计亮点：

💾 数据存储与导出（Storage）

core/storage.py

import csv
import hashlib
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Set
import pandas as pd
from utils.logger import get_logger

logger = get_logger(__name__)

class DataStorage:
"""数据存储管理器"""

def __init__(self, output_dir: str = 'data/output'):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(parents=True, exist_ok=True)

self.seen_urls: Set[str] = set() # URL去重集合
self.data_buffer: List[Dict] = [] # 数据缓存

def add_record(self, record: Dict) –> bool:
"""
添加一条记录（带去重）

Returns:
是否成功添加（True=新记录，False=重复）
"""
url = record.get('job_url', '')

# URL去重
url_hash = self._hash_url(url)
if url_hash in self.seen_urls:
logger.debug(f"重复记录: {url}")
return False

# 添加时间戳
record['crawl_time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

self.seen_urls.add(url_hash)
self.data_buffer.append(record)
return True

def save_to_csv(self, filename: str = None) –> str:
"""保存为CSV文件"""
if not self.data_buffer:
logger.warning("没有数据可保存")
return None

if filename is None:
timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
filename = f"jobs_{timestamp}.csv"

filepath = self.output_dir / filename

# 使用pandas处理（自动处理列表字段）
df = pd.DataFrame(self.data_buffer)

# 列表字段转字符串
for col in df.columns:
if df[col].apply(lambda x: isinstance(x, list)).any():
df[col] = df[col].apply(lambda x: '|'.join(x) if isinstance(x, list) else x)

# 指定字段顺序
columns_order = [
'job_title', 'company_name', 'salary_range', 'work_location',
'work_experience', 'education', 'job_tags', 'company_size',
'job_url', 'crawl_time'
]
df = df[[c for c in columns_order if c in df.columns]]

df.to_csv(filepath, index=False, encoding='utf-8-sig') # Excel兼容编码
logger.info(f"数据已保存: {filepath} ({len(df)} 条记录)")

return str(filepath)

def get_statistics(self) –> Dict:
"""获取统计信息"""
return {
'total_records': len(self.data_buffer),
'unique_urls': len(self.seen_urls),
'duplicate_count': len(self.seen_urls) – len(self.data_buffer)
}

@staticmethod
def _hash_url(url: str) –> str:
"""URL哈希（用于去重）"""
return hashlib.md5(url.encode()).hexdigest()

字段映射表：

去重策略：基于job_url的MD5哈希去重，避免重复采集同一职位。

⚙️ 完整主程序 main.py

import asyncio
from core.fetcher import DynamicFetcher
from core.parser import StructuredParser
from core.storage import DataStorage
from utils.logger import get_logger

logger = get_logger('main')

async def crawl_jobs(keyword: str = 'Python', city: str = '北京', max_pages: int = 3):
"""
爬取职位信息

Args:
keyword: 搜索关键词
city: 城市
max_pages: 最大页数
"""
# 初始化组件
parser = StructuredParser('config/selectors.yaml')
storage = DataStorage('data/output')

async with DynamicFetcher(headless=True) as fetcher:
for page_num in range(1, max_pages + 1):
# 构造URL（此处为示例，实际需根据目标站点调整）
url = f"https://www.example.com/jobs?keyword={keyword}&city={city}&page={page_num}"

try:
logger.info(f"开始抓取第 {page_num} 页")

# 1. 加载页面
page = await fetcher.fetch_page(
url,
wait_selector='div.job-list-box' # 等待列表容器
)

# 2. 解析数 {page_num} 页解析到 {len(jobs)} 条职位")

# 3. 存储数据
new_count = 0
for job in jobs:
if storage.add_record(job):
new_count += 1

logger.info(f"第 {page_num} 页新增 {new_count} 条有效数据")

# 关闭页面释放资源
await page.close()

except Exception as e:
logger.error(f"第 {page_num} 页爬取失败: {str(e)}")
continue

# 保存数据
output_file = storage.save_to_csv()

# 打印统计
stats = storage.get_statistics()
logger.info(f"爬取完成! 统计: {stats}")
logger.info(f"输出文件: {output_file}")

if __name__ == '__main__':
asyncio.run(crawl_jobs(
keyword='Python',
city='北京',
max_pages=3
))

日志工具 utils/logger.py

import logging
from pathlib import Path

def get_logger(name: str) –> logging.Logger:
"""获取配置好的logger"""
logger = logging.getLogger(name)

if not logger.handlers:
logger.setLevel(logging.INFO)

# 控制台输出
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 文件输出
log_dir = Path('data/logs')
log_dir.mkdir(parents=True, exist_ok=True)
file_handler = logging.FileHandler(
log_dir / 'spider.log',
encoding='utf-8'
)
file_handler.setLevel(logging.DEBUG)

# 格式化
formatter = logging.Formatter(
'%(asctime)s [%(name)s] %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
console_handler.setFormatter(formatter)
file_handler.setFormatter(formatter)

logger.addHandler(console_handler)
logger.addHandler(file_handler)

return logger

一行方式与结果展示

启动命令

# 确保在项目根目录
cd dynamic_spider

# 运行爬虫
python main.py

控制台输出示例

2026–02–04 10:30:12 [main] INFO: 开始抓取第 1 页
2026–02–04 10:30:15 [core.fetcher] INFO: 正在访问: https://www.example.com/jobs?keyword=Python&city=北京&page=1
2026–02–04 10:30:18 [core.fetcher] INFO: 关键元素已加载: div.job–list–box
2026–02–04 10:30:20 [core.parser] INFO: 找到 30 个列表项
2026–02–04 10:30:25 [main] INFO: 第 1 页解析到 30 条职位
2026–02–04 10:30:25 [main] INFO: 第 1 页新增 30 条有效数据
2026–02–04 10:30:30 [main] INFO: 开始抓取第 2 页
…
2026–02–04 10:32:45 [core.storage] INFO: 数据已保存: data/output/jobs_20260204_103245.csv (85 条记录)
2026–02–04 10:32:45 [main] INFO: 爬取完成! 统计: {'total_records': 85, 'unique_urls': 85, 'duplicate_count': 0}

输出文件位置

data/output/jobs_20260204_103245.csv

示例数据（前5行）

🔧 常见问题与排错

1️⃣ 403/429错误：访问被拒绝

症状：

playwright._impl._api_types.Error: Response status code 403

原因与解决方案：

代理示例：

self.context = await self.browser.new_context(
proxy={
"server": "http://proxy.example.com:8080",
"username": "user",
"password": "pass"
}
)

2️⃣ 抓到空壳HTML：动态内容未加载

症状：

logger.error: 找到 0 个列表项

排查步骤：

如果确认是接口渲染：

# 监听网络请求
async def handle_response(response):
if 'api/jobs' in response.url:
data = await response.json()
print(data) # 分析接口结构

page.on('response', handle_response)

3️⃣ 选择器解析报错

症状：

AttributeError: 'NoneType' object has no attribute 'text_content'

原因：

• 页面结构变化（选择器失效）
• 元素未加载完成
• A/B测试导致不同用户看到不同DOM

解决方案：

job_title:
selectors: # 数组形式，依次尝试
– "div.job-title"
– "h3.title"
– "span.position-name"
attribute: "text"

# 在parse_item中添加
logger.debug(await element.inner_html()) # 查看实际HTML结构

4️⃣ 编码/乱码问题

症状： CSV文件中出现中文等乱码

解决方案：

# storage.py中已使用utf-8-sig编码
df.to_csv(filepath, index=False, encoding='utf-8-sig')

Excel打开乱码：

• 使用utf-8-sig（已实现）
• 或者改用gbk编码：encoding='gbk'

🚀 进阶优化（生产级改造）

1️⃣ 并发爬取（asyncio协程池）

import asyncio
from asyncio import Semaphore

async def crawl_with_concurrency(urls: List[str], max_concurrent: int = 3):
"""限制并发数的爬取"""
semaphore = Semaphore(max_concurrent)

async def fetch_one(url):
async with semaphore:
# 获取信号量后才执行
return await fetcher.fetch_page(url)

tasks = [fetch_one(url) for url in urls]
results = await asyncio.gather(*tasks, return_exceptions=True)

# 过滤异常
return [r for r in results if not isinstance(r, Exception)]

2️⃣ 断点续爬（Redis/SQLite游标）

class CrawlState:
"""爬取状态管理"""

def __init__(self, state_file='data/state.json'):
self.state_file = state_file
self.state = self._load_state()

def _load_state(self):
if Path(self.state_file).exists():
with open(self.state_file, 'r') as f:
return json.load(f)
return {'last_page': 0, 'crawled_urls': []}

def save_progress(self, page_num, url):
self.state['last_page'] = page_num
self.state['crawled_urls'].append(url)
with open(self.state_file, 'w') as f:
json.dump(self.state, f)

def should_skip(self, url):
return url in self.state['crawled_urls']

3️⃣ 日志与监控（结构化日志）

import json
from datetime import datetime

class MetricsLogger:
"""指标记录器"""

def __init__(self):
self.metrics = {
'start_time': datetime.now().isoformat(),
'pages_crawled': 0,
'records_extracted': 0,
'errors': []
}

def log_page(self, page_num, record_count):
self.metrics['pages_crawled'] += 1
self.metrics['records_extracted'] += record_count

def log_error(self, error_type, message):
self.metrics['errors'].append({
'type': error_type,
'message': message,
'timestamp': datetime.now().isoformat()
})

def save_report(self):
self.metrics['end_time'] = datetime.now().isoformat()
self.metrics['success_rate'] = (
1 – len(self.metrics['errors']) / max(self.metrics['pages_crawled'], 1)
) * 100

with open('data/logs/metrics.json', 'w') as f:
json.dump(self.metrics, f, indent=2, ensure_ascii=False)

4️⃣ 定时任务（APScheduler）

from apscheduler.schedulers.asyncio import AsyncIOScheduler

scheduler = AsyncIOScheduler()

@scheduler.scheduled_job('cron', hour=2) # 每天凌晨2点
async def daily_crawl():
await crawl_jobs(keyword='Python', city='北京', max_pages=10)

scheduler.start()
asyncio.get_event_loop().run_forever()

📚 总结与延伸阅读

我们完成了什么？

通过这篇文章，你已经掌握了：

✅ 动态网站爬取的核心技术：Playwright浏览器自动化、异步等待、反爬应对 ✅ 工程化的选择器管理：YAML配置驱动、字段映射、转换管道 ✅ 完整的数据处理流程：抓取 → 解析 → 清洗 → 去重 → 存储 ✅ 生产级代码规范：异步上下文管理、结构化日志、异常容错

这套方案不仅适用于招聘网站，同样可以迁移到：

• 电商平台（商品列表 + 评论爬取）
• 新闻网站（文章聚合 + 评论分析）
• 社交媒体（动态内容 + 用户信息）

下一步可以做什么？

1. 深度学习Scrapy框架 Playwright适合小规模精细化爬取，但大规模场景下Scrapy的异步架构和中间件生态更强大。可以尝试：

scrapy startproject advanced_spider
scrapy-playwright # Scrapy集成Playwright的插件

2. 探索更智能的反爬策略

• 机器学习识别验证码（Tesseract OCR、深度学习模型）
• 浏览器指纹伪装（Canvas、WebGL、Audio指纹）
• 分布式IP池（自建代理隧道）

3. 分布式爬虫架构

• Scrapy-Redis：基于Redis的分布式队列
• Celery任务调度：异步任务分发
• Kubernetes部署：容器化爬虫集群

4. 数据分析与可视化

# 薪资分析示例
import matplotlib.pyplot as plt

df = pd.read_csv('data/output/jobs_20260204.csv')
df['salary_avg'] = df['salary_range'].apply(lambda x:
sum(map(int, re.findall(r'\\d+', x))) / 2 if x else 0
)

df.groupby('company_name')['salary_avg'].mean().plot(kind='bar')
plt.title('各公司平均薪资对比')
plt.show()

推荐阅读资源

📖 官方文档：

• Playwright Python文档
• Scrapy官方教程

📖 进阶书籍：

• 《精通Scrapy网络爬虫》- 刘硕
• 《Python网络爬虫权威指南》- Ryan Mitchell

📖 法律合规：

• 《数据安全法》全文
• GDPR数据保护条例

🎉 写在最后

动态网站爬虫不是简单的"复制粘贴"代码，而是一个需要持续迭代的工程问题。网站结构会变、反爬策略会升级、数据格式会调整——这正是这个领域的魅力所在。

记住三个原则：

希望这篇文章能成为你爬虫进阶路上的一块坚实基石。现在，打开你的编辑器，去征服那些动态渲染的网站吧！

🌟 文末

好啦～以上就是本期 《Python爬虫实战》的全部内容啦！如果你在实践过程中遇到任何疑问，欢迎在评论区留言交流，我看到都会尽量回复～咱们下期见！

小伙伴们在批阅的过程中，如果觉得文章不错，欢迎点赞、收藏、关注哦～ 三连就是对我写作道路上最好的鼓励与支持！ ❤️🔥

📌 专栏持续更新中｜建议收藏 + 订阅

专栏 👉 《Python爬虫实战》，我会按照“入门 → 进阶 → 工程化 → 项目落地”的路线持续更新，争取让每一篇都做到：

✅ 讲得清楚（原理）｜✅ 跑得起来（代码）｜✅ 用得上（场景）｜✅ 扛得住（工程化）

📣 想系统提升的小伙伴：强烈建议先订阅专栏，再按目录顺序学习，效率会高很多～

✅ 互动征集

想让我把【某站点/某反爬/某验证码/某分布式方案】写成专栏实战？

评论区留言告诉我你的需求，我会优先安排更新 ✅

⭐️ 若喜欢我，就请关注我叭～（更新不迷路） ⭐️ 若对你有用，就请点赞支持一下叭～（给我一点点动力） ⭐️ 若有疑问，就请评论留言告诉我叭～（我会补坑 & 更新迭代）

免责声明：本文仅用于学习与技术研究，请在合法合规、遵守站点规则与 Robots 协议的前提下使用相关技术。严禁将技术用于任何非法用途或侵害他人权益的行为。技术无罪，责任在人！！！