简单爬虫框架实现

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 生成的内容）。
• 使用示例：

  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 responseexcept Exception as e:print(f"GET请求失败: {url}, 错误: {str(e)}")return Nonedef post(self, url, headers=None, data=None):"""发送POST请求（简化版，实际项目可扩展）"""headers = headers or DEFAULT_HEADERStry:response = self.session.post(url, headers=headers, data=data, timeout=self.timeout)response.raise_for_status()response.encoding = response.apparent_encodingreturn responseexcept 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解析XPathdef 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():  # 遍历所有请求URLresponse = self.http.get(url)  # 发送请求if not response:  # 请求失败时跳过continueself.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 Spiderclass 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 moviesif __name__ == "__main__":spider = DoubanSpider()spider.run()  # 启动爬虫

2：XPath

'''
@Time    : 2025/6/3 14:01
@Author  : 小宇
@File    : douban_XPath.py
'''from spider_framework import Spiderclass 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 moviesif __name__ == "__main__":spider = DoubanSpider()spider.run()

步骤 2：运行爬虫

执行脚本，输出类似以下内容：

3. 框架扩展建议

(1) 数据存储扩展

• 需求：默认 save_data 方法仅打印数据，实际应用中需保存到文件或数据库。
• 示例：保存到 CSV

  import csvclass 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_HEADERStry:# 随机选择代理（示例：proxies = ["http://proxy1.com", "http://proxy2.com"]）proxy = random.choice(self.proxies) if self.proxies else Noneresponse = 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]  
    

(2) 反爬机制应对

• 现象：网站返回 403（禁止访问）或空白页面。
• 解决方案：
  • 增加请求头（如 Referer、Cookie）。
  • 调整 time.sleep(1) 间隔（如改为随机延迟 time.sleep(random.uniform(2, 5))）。
  • 使用代理池或轮换 User-Agent。

(3) 动态内容爬取

• 现象：数据通过 JavaScript 生成，静态请求无法获取。
• 解决方案：
  • 使用 Selenium 或 Playwright 等浏览器自动化工具。
  • 分析接口，直接请求后端 API（如通过浏览器开发者工具抓包）。

5. 框架优势总结

• 模块化设计：请求、解析、流程控制分离，易于维护和扩展。
• 防封禁机制：自动重试、请求头设置、延迟等待，提升爬取稳定性。
• 多解析支持：灵活选择 CSS/XPath/正则，适应不同页面结构。
• 代码复用性：基类定义通用逻辑，子类只需实现业务相关方法，减少重复代码。

通过该框架，可快速开发各类静态网页爬虫，后续可根据具体需求逐步添加反爬策略、数据存储、异步请求等高级功能。