从零开始：Python爬虫实战——爬取豆瓣读书评分9.0以上高分图书（完整教程）

一、前言：为什么要写这个爬虫？

在数据驱动的时代，图书推荐系统、阅读社区、知识管理平台往往需要高质量的图书数据。豆瓣读书作为国内最具影响力的图书评价平台，其评分体系极具参考价值。而评分9.0以上的图书，通常被公认为“神作”或“经典”。手动逐页筛选这些高分图书效率极低，因此编写一个自动化爬虫，不仅可以高效获取数据，还能锻炼数据采集、解析、清洗和存储的全流程能力。

本文将带领你使用最新的Python技术栈（requests、BeautifulSoup、pandas、fake_useragent、retrying等），爬取豆瓣读书中评分≥9.0的图书信息，包括书名、作者、出版社、出版年份、评分、评分人数和一句话简介等。我们会重点讲解筛选与分页两大核心难点，并给出完整的工程化代码，最终将数据保存为CSV文件。

声明：本教程仅供学习交流使用，请遵守robots.txt协议，控制请求频率，切勿对目标网站造成压力。

目录

一、前言：为什么要写这个爬虫？

二、技术选型与环境搭建

2.1 核心技术栈

2.2 环境准备

2.3 豆瓣读书页面分析

三、爬虫设计思路

3.1 工作流程

3.2 关键难点与解决方案

四、代码实现（逐块详解）

4.1 导入库与配置

4.2 随机User-Agent

4.3 带重试机制的请求函数

4.4 解析单页图书数据

4.5 分页控制与主爬虫逻辑

4.6 数据保存与主函数

五、完整代码（整合版）

六、运行测试与结果分析

6.1 预期输出

6.2 生成的CSV示例

七、反爬虫策略与优化建议

7.1 进阶防封措施

7.2 数据清洗增强

7.3 增量爬取

八、总结与扩展

8.1 核心收获

8.2 可扩展方向

二、技术选型与环境搭建

2.1 核心技术栈

2.2 环境准备

bash

# 创建虚拟环境（推荐） python -m venv douban_spider source douban_spider/bin/activate # Windows: douban_spider\Scripts\activate # 安装依赖库 pip install requests beautifulsoup4 pandas fake_useragent retrying lxml

2.3 豆瓣读书页面分析

目标URL模式：

text

https://book.douban.com/tag/小说?start=0&type=T https://book.douban.com/tag/小说?start=20&type=T https://book.douban.com/tag/小说?start=40&type=T ...

• tag：图书分类（我们以“小说”为例，可自定义）

• start：起始索引，每页20条记录（0,20,40...）

• 我们只提取评分 ≥ 9.0 的条目，并持续跨页抓取直到某页完全没有高分图书。

三、爬虫设计思路

3.1 工作流程

1. 构造URL，设置请求头。

2. 发送GET请求，获取响应文本。

3. 使用BeautifulSoup解析每一页的图书列表。

4. 提取每本书的标题、链接、评分、评价人数、作者、出版社等信息。

5. 筛选评分 ≥ 9.0 的图书。

6. 自动翻页，直到最后一页（当前页没有任何评分≥9.0的图书）。

7. 将所有数据存入DataFrame并导出CSV。

3.2 关键难点与解决方案

四、代码实现（逐块详解）

4.1 导入库与配置

python

import requests import time import random import re import pandas as pd from bs4 import BeautifulSoup from fake_useragent import UserAgent from retrying import retry from typing import List, Dict, Optional # 全局配置 HEADERS = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3', 'Connection': 'keep-alive', } TIMEOUT = 10 MAX_RETRIES = 3 REQUEST_INTERVAL = (3, 6) # 随机间隔3~6秒

4.2 随机User-Agent

python

ua = UserAgent() def get_random_headers(): """随机生成User-Agent，降低封禁风险""" headers = HEADERS.copy() headers['User-Agent'] = ua.random return headers

4.3 带重试机制的请求函数

python

@retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=2000) def fetch_page(url: str) -> Optional[str]: """ 发送请求并返回HTML文本 重试最多3次，每次间隔2秒 """ try: headers = get_random_headers() response = requests.get(url, headers=headers, timeout=TIMEOUT) response.encoding = 'utf-8' # 豆瓣使用utf-8编码 if response.status_code == 200: return response.text else: print(f"请求失败，状态码：{response.status_code}，URL：{url}") return None except Exception as e: print(f"请求异常：{e}，URL：{url}") raise # 触发重试

4.4 解析单页图书数据

豆瓣图书列表页的HTML结构（2025年最新版依然稳定）：

• 每个图书条目位于<li class="subject-item">中

• 评分位于<span class="rating_nums">内

• 评价人数：<span class="pl">内括号中的数字

python

def parse_book_item(item) -> Optional[Dict]: """解析单个图书条目，返回字典或None（如果评分<9.0）""" # 1. 标题和链接 title_tag = item.find('div', class_='info').find('h2').find('a') if not title_tag: return None title = title_tag.get_text(strip=True) book_url = title_tag.get('href') # 2. 评分 rating_tag = item.find('span', class_='rating_nums') if not rating_tag: return None # 无评分则跳过 try: rating = float(rating_tag.get_text(strip=True)) except: rating = 0.0 # 核心筛选：只保留9.0分及以上 if rating < 9.0: return None # 3. 评价人数 people_tag = item.find('span', class_='pl') rating_people = 0 if people_tag: people_text = people_tag.get_text(strip=True) match = re.search(r'\((\d+)人评价\)', people_text) if match: rating_people = int(match.group(1)) # 4. 图书信息（作者/译者/出版社/出版年份） pub_tag = item.find('div', class_='pub') pub_info = pub_tag.get_text(strip=True) if pub_tag else '' # 示例格式："[法] 阿尔贝·加缪 / 金祎 / 江苏凤凰文艺出版社 / 2022-12 / 45.00元" # 使用正则或简单split拆分，这里为了通用性做粗略解析 parts = pub_info.split('/') author = parts[0].strip() if len(parts) > 0 else '未知' translator = '' # 如果存在译者（一般含有"译"或位于第二个位置且非数字） if len(parts) > 1 and ('译' in parts[1] or len(parts[1].strip()) < 10): translator = parts[1].strip() publisher = parts[2].strip() if len(parts) > 2 else '未知' year = parts[3].strip() if len(parts) > 3 else '未知' else: translator = '无' publisher = parts[1].strip() if len(parts) > 1 else '未知' year = parts[2].strip() if len(parts) > 2 else '未知' # 年份提取数字 year_match = re.search(r'\d{4}', year) year_num = year_match.group(0) if year_match else '未知' # 5. 简介（可选） abstract_tag = item.find('p', class_='detail') abstract = abstract_tag.get_text(strip=True) if abstract_tag else '' return { 'title': title, 'url': book_url, 'rating': rating, 'rating_people': rating_people, 'author': author, 'translator': translator, 'publisher': publisher, 'year': year_num, 'abstract': abstract }

4.5 分页控制与主爬虫逻辑

python

def crawl_douban_books(tag='小说', start=0, max_pages=50): """ 爬取豆瓣指定标签下的高分图书（评分≥9.0） :param tag: 图书分类标签 :param start: 起始偏移量 :param max_pages: 最大翻页数（防止无限循环） :return: 图书列表 """ base_url = f"https://book.douban.com/tag/{tag}" all_books = [] current_start = start page_num = 1 while page_num <= max_pages: url = f"{base_url}?start={current_start}&type=T" print(f"正在抓取第{page_num}页: {url}") html = fetch_page(url) if not html: print("获取页面失败，停止爬取") break soup = BeautifulSoup(html, 'lxml') book_items = soup.find_all('li', class_='subject-item') if not book_items: print("当前页没有找到任何图书条目，停止翻页") break high_rating_count = 0 for item in book_items: book_data = parse_book_item(item) if book_data: all_books.append(book_data) high_rating_count += 1 print(f"第{page_num}页共{len(book_items)}本，其中≥9.0分的有{high_rating_count}本") # 核心终止条件：如果当前页没有任何一本评分≥9.0，则停止爬取 if high_rating_count == 0: print("当前页已无高分图书，爬取结束") break # 翻页：增加20 current_start += 20 page_num += 1 # 随机休眠，模拟人类浏览行为 sleep_time = random.uniform(*REQUEST_INTERVAL) print(f"等待 {sleep_time:.2f} 秒后继续...") time.sleep(sleep_time) return all_books

4.6 数据保存与主函数

python

def save_to_csv(books: List[Dict], filename: str = 'douban_high_score_books.csv'): """保存数据到CSV文件""" if not books: print("无数据可保存") return df = pd.DataFrame(books) # 去重（根据URL） df.drop_duplicates(subset=['url'], keep='first', inplace=True) # 按评分降序排列 df.sort_values(by=['rating', 'rating_people'], ascending=[False, False], inplace=True) df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"成功保存 {len(df)} 条记录到 {filename}") def main(): """主入口函数""" print("豆瓣读书高分爬虫启动...") books = crawl_douban_books(tag='小说', max_pages=30) print(f"总共抓取到 {len(books)} 本评分≥9.0的图书") if books: # 展示前5本 for i, book in enumerate(books[:5], 1): print(f"{i}.《{book['title']}》 评分：{book['rating']} 评价人数：{book['rating_people']}") save_to_csv(books) else: print("未获取到任何数据，请检查网络或目标标签") if __name__ == '__main__': main()

五、完整代码（整合版）

python

# douban_spider.py import requests import time import random import re import pandas as pd from bs4 import BeautifulSoup from fake_useragent import UserAgent from retrying import retry from typing import List, Dict, Optional # ---------- 配置 ---------- HEADERS = { 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3', } TIMEOUT = 10 MAX_RETRIES = 3 REQUEST_INTERVAL = (3, 6) ua = UserAgent() def get_random_headers(): headers = HEADERS.copy() headers['User-Agent'] = ua.random return headers @retry(stop_max_attempt_number=MAX_RETRIES, wait_fixed=2000) def fetch_page(url: str) -> Optional[str]: try: response = requests.get(url, headers=get_random_headers(), timeout=TIMEOUT) response.encoding = 'utf-8' if response.status_code == 200: return response.text else: print(f"请求失败，状态码：{response.status_code}") return None except Exception as e: print(f"请求异常：{e}") raise def parse_book_item(item) -> Optional[Dict]: try: title_tag = item.find('div', class_='info').find('h2').find('a') if not title_tag: return None title = title_tag.get_text(strip=True) book_url = title_tag.get('href') rating_tag = item.find('span', class_='rating_nums') if not rating_tag: return None rating = float(rating_tag.get_text(strip=True)) if rating < 9.0: return None people_tag = item.find('span', class_='pl') rating_people = 0 if people_tag: match = re.search(r'\((\d+)人评价\)', people_tag.get_text(strip=True)) if match: rating_people = int(match.group(1)) pub_tag = item.find('div', class_='pub') pub_info = pub_tag.get_text(strip=True) if pub_tag else '' parts = [p.strip() for p in pub_info.split('/')] author = parts[0] if len(parts) > 0 else '未知' translator = '无' publisher = '未知' year = '未知' if len(parts) > 1: if '译' in parts[1] or len(parts[1]) < 10: translator = parts[1] publisher = parts[2] if len(parts) > 2 else '未知' year = parts[3] if len(parts) > 3 else '未知' else: publisher = parts[1] year = parts[2] if len(parts) > 2 else '未知' year_num = re.search(r'\d{4}', year) year = year_num.group(0) if year_num else '未知' abstract_tag = item.find('p', class_='detail') abstract = abstract_tag.get_text(strip=True) if abstract_tag else '' return { 'title': title, 'url': book_url, 'rating': rating, 'rating_people': rating_people, 'author': author, 'translator': translator, 'publisher': publisher, 'year': year, 'abstract': abstract } except Exception as e: print(f"解析条目出错：{e}") return None def crawl_douban_books(tag='小说', start=0, max_pages=50): base_url = f"https://book.douban.com/tag/{tag}" all_books = [] current_start = start page_num = 1 while page_num <= max_pages: url = f"{base_url}?start={current_start}&type=T" print(f"[第{page_num}页] 抓取 {url}") html = fetch_page(url) if not html: break soup = BeautifulSoup(html, 'lxml') items = soup.find_all('li', class_='subject-item') if not items: print("无图书条目，停止翻页") break high_count = 0 for item in items: data = parse_book_item(item) if data: all_books.append(data) high_count += 1 print(f"本页高分图书数量：{high_count}") if high_count == 0: print("已无更多高分图书，结束爬取") break current_start += 20 page_num += 1 sleep_time = random.uniform(*REQUEST_INTERVAL) print(f"休眠 {sleep_time:.2f} 秒\n") time.sleep(sleep_time) return all_books def save_to_csv(books, filename='douban_top_books.csv'): if not books: return df = pd.DataFrame(books) df.drop_duplicates(subset=['url'], inplace=True) df.sort_values(['rating', 'rating_people'], ascending=False, inplace=True) df.to_csv(filename, index=False, encoding='utf-8-sig') print(f"已保存 {len(df)} 条数据至 {filename}") def main(): books = crawl_douban_books(tag='小说', max_pages=30) print(f"总计获取：{len(books)} 本") if books: save_to_csv(books) print("\n示例数据：") for b in books[:3]: print(f"《{b['title']}》 {b['rating']}分 {b['rating_people']}人评价") if __name__ == '__main__': main()

六、运行测试与结果分析

6.1 预期输出

text

[第1页] 抓取 https://book.douban.com/tag/小说?start=0&type=T 本页高分图书数量：8 休眠 4.23 秒 [第2页] 抓取 https://book.douban.com/tag/小说?start=20&type=T 本页高分图书数量：5 休眠 3.67 秒 ... [第6页] 抓取 https://book.douban.com/tag/小说?start=100&type=T 本页高分图书数量：0 已无更多高分图书，结束爬取 总计获取：42 本 已保存 42 条数据至 douban_top_books.csv

6.2 生成的CSV示例

七、反爬虫策略与优化建议

7.1 进阶防封措施

• IP代理池：使用免费或付费代理（如requests+proxies参数）。

• 请求头轮转：除UA外，也随机更换Accept-Language等。

• Cookies维持：从浏览器复制一个已登录的Cookie字符串，提高信任度。

• 异步爬取：使用aiohttp提升效率，但需更谨慎控制并发。

7.2 数据清洗增强

• 使用jieba分词对简介做关键词提取。

• 通过isbnlib库根据书名自动补全ISBN号。

• 出版社名称标准化（“中信出版社”与“中信出版集团”合并）。

7.3 增量爬取

记录上次爬取的最后一页URL，下次启动时从该页开始，避免重复。

八、总结与扩展

8.1 核心收获

• 分页控制：通过start参数递增，并设计合理的终止条件（当前页无高分图书）。

• 评分筛选：在解析阶段进行浮点数比较，提前丢弃低分数据，节省内存。

• 健壮性设计：重试机制、异常捕获、随机延时，保障长时间稳定运行。

8.2 可扩展方向

1. 多标签并行爬取：如“历史”“科幻”“哲学”等，对比不同类别的高分密度。

2. 存入数据库：使用SQLite或PostgreSQL，方便后续查询分析。

3. 可视化分析：用matplotlib/seaborn绘制评分分布、年份趋势图。

4. 构建推荐系统：基于爬取的数据和协同过滤算法，做简单的图书推荐。