Python爬虫实战：从零构建完整项目（数据采集+存储+异常处理）

爬虫不是简单的请求+解析，而是一个系统工程。本文将带你体验企业级爬虫开发的核心流程。

一、前言：为什么需要完整的爬虫项目？

作为初学者，你可能写过一些单文件的爬虫脚本，但当面对真实复杂的网络环境、海量数据存储需求及各种突发异常时，这些脚本往往脆弱不堪。真正的爬虫项目，是网络请求、数据解析、持久化存储和健壮性设计的交响乐。本次我们将综合运用Python核心技术，构建一个可投入生产环境的爬虫系统。

二、项目目标与技术栈

项目目标：爬取图书网站（以豆瓣读书为例）的书籍信息，包含：

书名
作者
评分
价格
ISBN
简介

技术栈：

网络请求：requests (处理HTTP请求/响应)
数据解析：BeautifulSoup4 (HTML解析)
数据存储：SQLite3 (轻量级数据库)
异常处理：Python内置异常机制 + 自定义重试
辅助工具：logging (日志记录), time (速率控制)

三、环境搭建（新手友好）

# 创建虚拟环境（可选但推荐）
python -m venv spider-env
source spider-env/bin/activate  # Linux/Mac
spider-env\Scripts\activate    # Windows# 安装核心库
pip install requests beautifulsoup4

📌 重要提示：真实项目中务必添加 requirements.txt 管理依赖！

四、分步构建爬虫系统

4.1 数据采集模块：与网络对话

核心任务：稳定获取目标网页内容

import requests
from fake_useragent import UserAgent  # pip install fake-useragentdef fetch_page(url, retries=3):"""获取网页内容，含简单重试机制:param url: 目标URL:param retries: 最大重试次数:return: HTML内容 (str) 或 None"""headers = {'User-Agent': UserAgent().random}  # 动态生成UAfor attempt in range(retries):try:response = requests.get(url, headers=headers, timeout=10)response.raise_for_status()  # 自动抛出HTTP错误return response.textexcept (requests.exceptions.RequestException, requests.exceptions.Timeout) as e:print(f"请求失败 (尝试 {attempt+1}/{retries}): {e}")time.sleep(2 ** attempt)  # 指数退避策略return None

关键技术解析：

User-Agent轮换：绕过基础反爬
timeout设置：防止请求阻塞
raise_for_status()：自动处理HTTP状态码（404, 500等）
指数退避重试：网络波动时的自愈能力

4.2 数据解析模块：从混沌中提取信息

核心任务：精准抽取目标数据

from bs4 import BeautifulSoupdef parse_book_page(html):"""解析图书详情页:param html: 网页HTML:return: 字典形式的结构化数据"""if not html:return Nonesoup = BeautifulSoup(html, 'html.parser')book_data = {}try:# 书名 (使用CSS选择器更稳定)book_data['title'] = soup.select_one('h1 span').text.strip()# 作者信息 (处理多作者情况)authors = [a.text.strip() for a in soup.select('.author a')]book_data['author'] = '; '.join(authors)# 评分 (处理可能缺失的情况)rating_tag = soup.select_one('.rating_num')book_data['rating'] = float(rating_tag.text) if rating_tag else 0.0# 使用更健壮的属性提取book_data['isbn'] = soup.find('meta', {'property': 'books:isbn'})['content']# 简介 (处理多段文本)summary_tag = soup.select_one('.intro')book_data['summary'] = '\n'.join([p.text for p in summary_tag.find_all('p')]) except (AttributeError, TypeError) as e:print(f"解析错误: {e}")return Nonereturn book_data

避坑指南：

优先使用CSS选择器而非XPath（更简洁）
所有解析操作都需考虑标签不存在的情况
多层数据使用try-except局部捕获
处理多值数据时用分隔符连接

4.3 数据存储模块：持久化艺术

为什么用数据库而非CSV？

避免重复爬取
支持复杂查询
数据一致性保障

import sqlite3DB_NAME = 'books.db'def init_database():"""初始化数据库表结构"""conn = sqlite3.connect(DB_NAME)c = conn.cursor()c.execute('''CREATE TABLE IF NOT EXISTS books(id INTEGER PRIMARY KEY AUTOINCREMENT,title TEXT NOT NULL,author TEXT,rating REAL,isbn TEXT UNIQUE,  -- ISBN唯一标识书籍summary TEXT,created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)''')conn.commit()conn.close()def save_book_data(book):"""保存单条图书数据"""if not book: return Falsetry:conn = sqlite3.connect(DB_NAME)c = conn.cursor()# 使用UPSERT操作避免重复c.execute('''INSERT OR IGNORE INTO books (title, author, rating, isbn, summary)VALUES (?, ?, ?, ?, ?)''',(book['title'], book['author'], book['rating'], book['isbn'], book['summary']))conn.commit()return c.rowcount > 0except sqlite3.Error as e:print(f"数据库错误: {e}")return Falsefinally:conn.close()  # 确保连接关闭

高级技巧：

使用UNIQUE约束防止重复数据
INSERT OR IGNORE实现去重插入
finally保证资源释放
添加时间戳追踪数据采集时间

4.4 异常处理模块：构建抗脆弱系统

爬虫常见崩溃原因：

网络波动（超时/断连）
网站改版（解析失败）
反爬机制（IP被封）
数据异常（类型转换错误）

class SpiderException(Exception):"""自定义爬虫异常基类"""passclass RetryExhaustedError(SpiderException):"""重试耗尽异常"""def __init__(self, url):self.url = urlsuper().__init__(f"重试耗尽: {url}")def safe_crawl(url):"""带完整异常处理的爬取流程:return: 成功保存返回True"""try:# 1. 网络请求html = fetch_page(url)if not html:raise SpiderException(f"获取页面失败: {url}")# 2. 数据解析book_data = parse_book_page(html)if not book_data:raise SpiderException(f"解析页面失败: {url}")# 3. 数据存储if not save_book_data(book_data):raise SpiderException(f"数据保存失败: {url}")print(f"成功处理: {url}")return Trueexcept SpiderException as e:print(f"业务流程异常: {e}")return Falseexcept Exception as e:  # 捕获未预料异常print(f"系统未知异常: {type(e).__name__} - {e}")# 此处可添加邮件/钉钉报警return False

异常处理金字塔：

基础异常：网络/解析/存储错误
业务异常：自定义SpiderException
全局兜底：捕获所有未处理异常
重试机制：网络请求层已实现

五、项目整合：让模块协同工作

import time
from urllib.parse import urljoinBASE_URL = "https://book.douban.com/tag/小说"
MAX_PAGES = 5  # 演示用控制页数def main():init_database()session = requests.Session()  # 复用连接提升性能for page in range(MAX_PAGES):list_url = f"{BASE_URL}?start={page*20}"print(f"正在爬取列表页: {list_url}")try:list_html = fetch_page(list_url)if not list_html: continuesoup = BeautifulSoup(list_html, 'html.parser')book_links = soup.select('.subject-item .info h2 a')for link in book_links:detail_url = link['href']safe_crawl(detail_url)  # 核心调度time.sleep(1.5)  # 礼貌爬取间隔except Exception as e:print(f"列表页处理异常: {e}")time.sleep(3)  # 页间延迟if __name__ == "__main__":main()

六、生产级优化建议

代理IP池：解决IP封锁问题

proxies = {"http": "http://10.10.1.10:3128"}
requests.get(url, proxies=proxies)

分布式任务队列：使用Celery+Redis
浏览器渲染：对JS渲染页面使用Selenium
增量爬取：基于数据库时间戳过滤
配置文件：分离敏感参数（数据库密码等）

七、常见问题排查手册

现象	可能原因	解决方案
返回403错误	IP被封/请求头异常	更换代理/更新UserAgent
解析到空数据	网站改版	重新分析DOM结构
数据库插入失败	字段超长/类型不匹配	增加字段长度检查
内存持续增长	未关闭数据库连接	使用`with`语句管理资源
被重定向到登录页	触发反爬	添加Cookie模拟登录