1. 框架功能概述
(1) HttpSession 类：请求管理
- 功能：封装
requests库，实现带重试机制的 HTTP 请求（GET/POST）。 - 关键特性：
- 自动处理 429（请求过多）、5xx（服务器错误）等错误，最多重试 3 次。
- 自动设置请求头（
User-Agent、Accept等），降低被网站封禁的风险。 - 自动检测响应编码（
response.encoding = response.apparent_encoding）。
(2) DataParser 类：数据解析
- 功能：统一解析接口，支持三种解析方式：
- XPath：使用
lxml库，适合复杂层级结构解析。 - CSS 选择器：使用
BeautifulSoup，适合快速定位元素。 - 正则表达式：处理非结构化数据（如 JavaScript 生成的内容）。
- XPath：使用
- 使用示例：
parser = DataParser(response) titles = parser.css(".title") # CSS选择器 links = parser.xpath('//a/@href') # XPath numbers = parser.regex(r'\d+') # 正则表达式
(3) Spider 基类：爬虫流程模板
- 功能：定义爬虫的通用流程，强制子类实现核心逻辑。
- 关键方法：
start_requests()：生成初始请求 URL 列表（子类必须实现）。parse(response)：解析页面数据（子类必须实现）。run()：主流程控制，包括请求发送、解析、数据保存和防封禁延迟。
- 模板方法模式：子类只需聚焦业务逻辑（如 URL 生成和数据提取），框架自动处理请求重试、解析器初始化等通用逻辑。
2. 如何使用该框架？
步骤 1：创建类爬虫
新建一个 Python 文件（如 spider_framework.py）
'''
@Author : 小宇
@File : spider_framework.py
'''
import re
import time
import requests
from bs4 import BeautifulSoup
from lxml import etree
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
# ---------------------------
# 全局配置（请求头）
# ---------------------------
DEFAULT_HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) Chrome/114.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8"
}
# ---------------------------
# 请求类（带重试机制）
# ---------------------------
class HttpSession:
"""负责发送HTTP请求，支持失败重试"""
def __init__(self, retries=3, timeout=10):
self.session = requests.Session()
self.timeout = timeout # 请求超时时间（秒）
# 配置重试策略（遇到429/5xx错误时重试）
retry_strategy = Retry(
total=retries, # 最大重试次数
backoff_factor=1, # 重试间隔（秒）
status_forcelist=[429, 500, 502, 503, 504]
)
self.session.mount("http://", HTTPAdapter(max_retries=retry_strategy))
self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))
def get(self, url, headers=None):
"""发送GET请求"""
headers = headers or DEFAULT_HEADERS # 使用默认请求头
try:
response = self.session.get(url, headers=headers, timeout=self.timeout)
response.raise_for_status() # 失败时抛出异常
response.encoding = response.apparent_encoding # 自动识别编码
return response
except Exception as e:
print(f"GET请求失败: {url}, 错误: {str(e)}")
return None
def post(self, url, headers=None, data=None):
"""发送POST请求（简化版，实际项目可扩展）"""
headers = headers or DEFAULT_HEADERS
try:
response = self.session.post(url, headers=headers, data=data, timeout=self.timeout)
response.raise_for_status()
response.encoding = response.apparent_encoding
return response
except Exception as e:
print(f"POST请求失败: {url}, 错误: {str(e)}")
return None
# ---------------------------
# 解析类（支持3种解析方式）
# ---------------------------
class DataParser:
"""统一解析接口，接收响应后生成解析对象"""
def __init__(self, response):
self.text = response.text # 页面文本
self.soup = BeautifulSoup(self.text, "lxml") # 使用BeautifulSoup解析
self.html = etree.HTML(self.text) # 使用lxml解析XPath
def xpath(self, pattern):
"""用XPath提取数据，返回列表"""
return self.html.xpath(pattern)
def css(self, selector):
"""用CSS选择器提取数据，返回列表"""
return self.soup.select(selector)
def regex(self, pattern, flags=re.S):
"""用正则表达式提取数据，返回匹配列表"""
return re.findall(pattern, self.text, flags=flags)
# ---------------------------
# 爬虫基类（模板方法）
# ---------------------------
class Spider:
"""爬虫基类，定义通用流程，子类需实现关键方法"""
def __init__(self, name):
self.name = name # 爬虫名称
self.http = HttpSession() # 初始化请求对象
self.parser = None # 解析器对象
def start_requests(self):
"""生成初始请求URL列表（需子类实现）"""
raise NotImplementedError("请在子类中定义起始URL")
def parse(self, response):
"""解析页面数据（需子类实现）"""
raise NotImplementedError("请在子类中定义解析逻辑")
def run(self):
"""爬虫主流程"""
print(f"启动爬虫: {self.name}")
for url in self.start_requests(): # 遍历所有请求URL
response = self.http.get(url) # 发送请求
if not response: # 请求失败时跳过
continue
self.parser = DataParser(response) # 创建解析器
print(f"成功获取页面: {url}")
data = self.parse(response) # 解析数据
if data:
self.save_data(data) # 保存数据
time.sleep(1) # 暂停1秒防封禁
def save_data(self, data):
"""保存数据（默认打印到控制台，可自定义）"""
print(f"解析到数据: {data}")
步骤 2：创建子类爬虫
1：BeautifulSoup
新建一个 Python 文件（如 douban_BeautifulSoup.py），继承 Spider 基类并实现抽象方法：
'''
@Time : 2025/6/3 13:49
@Author : 小宇
@File : douban_BeautifulSoup .py
'''
from spider_framework import Spider
class DoubanSpider(Spider):
def __init__(self):
super().__init__(name="豆瓣电影TOP250爬虫")
self.base_url = "https://movie.douban.com/top250"
def start_requests(self):
"""生成多页 URL（每页 25 条数据）"""
return [f"{self.base_url}?start={i*25}" for i in range(10)]
def parse(self, response):
"""解析电影列表页，提取电影信息"""
movies = []
for item in self.parser.css(".item"): # 使用 CSS 选择器定位每个电影项
title = item.select_one(".title").get_text(strip=True) # 标题
rating = item.select_one(".rating_num").text # 评分
quote = item.select_one(".inq")?.text.strip() or "无引言" # 引言（处理可能不存在的情况）
cover_url = item.select_one(".pic a img")["src"] # 封面链接
movies.append({
"标题": title,
"评分": rating,
"引言": quote,
"封面": cover_url
})
return movies
if __name__ == "__main__":
spider = DoubanSpider()
spider.run() # 启动爬虫
2：XPath
'''
@Time : 2025/6/3 14:01
@Author : 小宇
@File : douban_XPath.py
'''
from spider_framework import Spider
class DoubanSpider(Spider):
def __init__(self):
super().__init__(name="豆瓣电影TOP250爬虫")
self.base_url = "https://movie.douban.com/top250"
def start_requests(self):
"""生成多页URL（每页25条）"""
return [f"{self.base_url}?start={i * 25}" for i in range(10)]
def parse(self, response):
movies = []
# 使用 XPath 选择所有电影项
for item in self.parser.xpath('//div[@class="item"]'): # lxml 元素
title = item.xpath('.//span[@class="title"]/text()')[0]
rating = item.xpath('.//span[@class="rating_num"]/text()')[0]
quote = item.xpath('.//span[@class="inq"]/text()')
quote = quote[0] if quote else "无引言"
# 使用 XPath 获取封面链接
cover_url = item.xpath('.//div[@class="pic"]/a/img/@src')[0]
movies.append({
"标题": title,
"评分": rating,
"引言": quote,
"封面": cover_url
})
return movies
if __name__ == "__main__":
spider = DoubanSpider()
spider.run()
步骤 2：运行爬虫
执行脚本，输出类似以下内容：
3. 框架扩展建议
(1) 数据存储扩展
- 需求：默认
save_data方法仅打印数据，实际应用中需保存到文件或数据库。 - 示例：保存到 CSV
import csv class DoubanSpider(Spider): def __init__(self): super().__init__(name="豆瓣电影TOP250爬虫") self.filename = "douban_movies.csv" self.csv_header = ["标题", "评分", "引言", "封面"] with open(self.filename, "w", newline="", encoding="utf-8") as f: self.writer = csv.DictWriter(f, fieldnames=self.csv_header) self.writer.writeheader() def save_data(self, data): """重写保存方法，写入 CSV 文件""" with open(self.filename, "a", newline="", encoding="utf-8") as f: self.writer.writerows(data) print(f"已保存 {len(data)} 条数据到 {self.filename}")
(2) 代理池集成
- 需求：避免 IP 被封禁，添加代理轮换功能。
- 修改
HttpSession类：class HttpSession: def __init__(self, retries=3, timeout=10, proxies=None): self.proxies = proxies # 代理列表 # ... 其他初始化代码 ... def get(self, url, headers=None): headers = headers or DEFAULT_HEADERS try: # 随机选择代理（示例：proxies = ["http://proxy1.com", "http://proxy2.com"]） proxy = random.choice(self.proxies) if self.proxies else None response = self.session.get( url, headers=headers, timeout=self.timeout, proxies={"http": proxy, "https": proxy} # 设置代理 ) # ... 其他请求逻辑 ... except Exception as e: # ... 错误处理 ...
(3) 异步请求优化
- 需求：提升爬取效率，使用异步框架（如
aiohttp+asyncio）。 - 说明：需重写
HttpSession为异步版本，并修改Spider.run()为异步流程，适合大规模数据爬取。
4. 常见问题与解决方案
(1) 解析器混用导致的错误
- 问题：误用
BeautifulSoup对象调用xpath方法，或反之。item = parser.css(".item")[0] # BeautifulSoup 对象 cover_url = item.xpath('.//img/@src') # 错误：BeautifulSoup 无 xpath 方法 - 解决方案：
- 统一解析方式：要么全用 CSS 选择器（
soup.select()），要么全用 XPath（html.xpath()）。 - 使用
DataParser的对应方法：# 使用 CSS 选择器 cover_url = item.select_one("img")["src"] # 或使用 XPath（需通过 parser.html 获取 lxml 根节点） cover_url = self.parser.xpath('//div[@class="item"][1]/img/@src')[0]
- 统一解析方式：要么全用 CSS 选择器（
(2) 反爬机制应对
- 现象：网站返回 403（禁止访问）或空白页面。
- 解决方案：
- 增加请求头（如
Referer、Cookie）。 - 调整
time.sleep(1)间隔（如改为随机延迟time.sleep(random.uniform(2, 5))）。 - 使用代理池或轮换 User-Agent。
- 增加请求头（如
(3) 动态内容爬取
- 现象：数据通过 JavaScript 生成，静态请求无法获取。
- 解决方案：
- 使用
Selenium或Playwright等浏览器自动化工具。 - 分析接口，直接请求后端 API（如通过浏览器开发者工具抓包）。
- 使用
5. 框架优势总结
- 模块化设计：请求、解析、流程控制分离，易于维护和扩展。
- 防封禁机制：自动重试、请求头设置、延迟等待，提升爬取稳定性。
- 多解析支持：灵活选择 CSS/XPath/正则，适应不同页面结构。
- 代码复用性：基类定义通用逻辑，子类只需实现业务相关方法，减少重复代码。
通过该框架，可快速开发各类静态网页爬虫，后续可根据具体需求逐步添加反爬策略、数据存储、异步请求等高级功能。
